"""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), )