genichinandClaude Sonnet 4.6
c815ce39b6
BAT-006 (REQ-016/017/018): - client token 발급 경로 제거 (issue_client_token, POST /auth/tokens, 관련 스키마/테스트) - service-token rotate unique index revoked_at IS NULL 조건 추가 (Alembic CONCURRENTLY 마이그레이션) - POST /auth/service-tokens body daemon_id forward 누락 보정 (1-line fix) BAT-007 (REQ-022/023): - daemon deactivate 시 바인딩 service token 자동 revoke (CLI --force 옵션 + revoke_tokens_by_daemon_id) - NULL daemon_id legacy token 진단 CLI (migrate-null-daemon --dry-run) 거버넌스 문서: dcy-003/004 Discovery, REQ-016~023/025~032 신규, BAT-006/007 전체 산출물 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
124 lines
4.2 KiB
Python
124 lines
4.2 KiB
Python
"""Authentication API routes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import NoReturn
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Response, status
|
|
|
|
from app.core.auth import AuthContext, require_admin, require_read
|
|
from app.domain.models import AuthToken, User
|
|
from app.schemas.auth import (
|
|
AuthTokenResponse,
|
|
IssueServiceTokenRequest,
|
|
IssueServiceTokenApiResponse,
|
|
LoginRequest,
|
|
TokenResponse,
|
|
)
|
|
from app.schemas.user import UserResponse
|
|
from app.services.auth_service import AuthService, LoginResult, get_auth_service
|
|
|
|
router = APIRouter(prefix="/auth", tags=["auth"])
|
|
|
|
|
|
def _to_user_response(user: User) -> UserResponse:
|
|
return UserResponse(
|
|
id=user.id,
|
|
username=user.username,
|
|
email=user.email,
|
|
permission=user.permission,
|
|
is_active=user.is_active,
|
|
created_at=user.created_at,
|
|
updated_at=user.updated_at,
|
|
last_login_at=user.last_login_at,
|
|
)
|
|
|
|
|
|
def _to_auth_token_response(token: AuthToken) -> AuthTokenResponse:
|
|
return AuthTokenResponse(
|
|
id=token.id,
|
|
user_id=token.user_id,
|
|
name=token.name,
|
|
token_prefix=token.token_prefix,
|
|
permission=token.permission,
|
|
token_type=token.token_type,
|
|
expires_at=token.expires_at,
|
|
last_used_at=token.last_used_at,
|
|
revoked_at=token.revoked_at,
|
|
created_at=token.created_at,
|
|
)
|
|
|
|
|
|
def _to_login_response(result: LoginResult) -> TokenResponse:
|
|
return TokenResponse(
|
|
access_token=result.token,
|
|
expires_at=result.auth_token.expires_at,
|
|
permission=result.auth_token.permission,
|
|
user=_to_user_response(result.user),
|
|
)
|
|
|
|
|
|
def _raise_auth_error(exc: ValueError) -> NoReturn:
|
|
detail = str(exc)
|
|
if "잘못된 사용자명 또는 비밀번호" in detail:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=detail) from exc
|
|
if "초과" in detail or "권한이 없습니다" in detail:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=detail) from exc
|
|
if "찾을 수 없습니다" in detail:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=detail) from exc
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=detail) from exc
|
|
|
|
|
|
@router.post("/login", response_model=TokenResponse)
|
|
def login(payload: LoginRequest, service: AuthService = Depends(get_auth_service)) -> TokenResponse:
|
|
try:
|
|
return _to_login_response(service.login(payload))
|
|
except ValueError as exc:
|
|
_raise_auth_error(exc)
|
|
|
|
|
|
@router.get("/tokens", response_model=list[AuthTokenResponse], dependencies=[Depends(require_read)])
|
|
def list_tokens(
|
|
auth_context: AuthContext = Depends(require_read),
|
|
service: AuthService = Depends(get_auth_service),
|
|
) -> list[AuthTokenResponse]:
|
|
try:
|
|
return [_to_auth_token_response(token) for token in service.list_tokens(auth_context)]
|
|
except ValueError as exc:
|
|
_raise_auth_error(exc)
|
|
|
|
|
|
@router.delete("/tokens/{token_id}", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(require_read)])
|
|
def revoke_token(
|
|
token_id: str,
|
|
auth_context: AuthContext = Depends(require_read),
|
|
service: AuthService = Depends(get_auth_service),
|
|
) -> Response:
|
|
try:
|
|
service.revoke_token(auth_context, token_id)
|
|
except ValueError as exc:
|
|
_raise_auth_error(exc)
|
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
|
|
|
|
|
@router.post("/service-tokens", response_model=IssueServiceTokenApiResponse, status_code=status.HTTP_201_CREATED, dependencies=[Depends(require_admin)])
|
|
def issue_service_token(
|
|
payload: IssueServiceTokenRequest,
|
|
service: AuthService = Depends(get_auth_service),
|
|
) -> IssueServiceTokenApiResponse:
|
|
try:
|
|
result = service.issue_service_token(
|
|
payload.name,
|
|
payload.username,
|
|
permission=payload.permission,
|
|
expires_in_days=payload.expires_in_days,
|
|
no_expiry=payload.no_expiry,
|
|
daemon_id=payload.daemon_id,
|
|
)
|
|
except ValueError as exc:
|
|
_raise_auth_error(exc)
|
|
return IssueServiceTokenApiResponse(
|
|
token=result.token,
|
|
runtime_id=result.auth_token.runtime_id,
|
|
auth_token=_to_auth_token_response(result.auth_token),
|
|
) |