"""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, BotPrincipal, User from app.schemas.auth import ( AuthTokenResponse, BotPrincipalResponse, CreateBotPrincipalRequest, CreateBotTokenRequest, CreateClientTokenRequest, IssuedBotTokenResponse, IssuedClientTokenResponse, LoginRequest, TokenResponse, ) from app.schemas.user import UserResponse from app.services.auth_service import AuthService, IssuedBotToken, IssuedClientToken, 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, bot_principal_id=token.bot_principal_id, expires_at=token.expires_at, last_used_at=token.last_used_at, revoked_at=token.revoked_at, created_at=token.created_at, ) def _to_bot_principal_response(bot: BotPrincipal) -> BotPrincipalResponse: return BotPrincipalResponse( id=bot.id, name=bot.name, permission=bot.permission, description=bot.description, is_active=bot.is_active, created_at=bot.created_at, updated_at=bot.updated_at, deactivated_at=bot.deactivated_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 _to_issued_token_response(result: IssuedClientToken) -> IssuedClientTokenResponse: return IssuedClientTokenResponse( token=result.token, auth_token=_to_auth_token_response(result.auth_token), ) def _to_issued_bot_token_response(result: IssuedBotToken) -> IssuedBotTokenResponse: return IssuedBotTokenResponse( token=result.token, auth_token=_to_auth_token_response(result.auth_token), ) 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: 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.post("/tokens", response_model=IssuedClientTokenResponse, status_code=status.HTTP_201_CREATED, dependencies=[Depends(require_read)]) def create_token( payload: CreateClientTokenRequest, auth_context: AuthContext = Depends(require_read), service: AuthService = Depends(get_auth_service), ) -> IssuedClientTokenResponse: try: return _to_issued_token_response(service.issue_client_token(auth_context, payload)) 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("/bots", response_model=BotPrincipalResponse, status_code=status.HTTP_201_CREATED, dependencies=[Depends(require_admin)]) def create_bot_principal( payload: CreateBotPrincipalRequest, service: AuthService = Depends(get_auth_service), ) -> BotPrincipalResponse: try: return _to_bot_principal_response(service.create_bot_principal(payload)) except ValueError as exc: _raise_auth_error(exc) @router.get("/bots", response_model=list[BotPrincipalResponse], dependencies=[Depends(require_admin)]) def list_bot_principals( is_active: bool | None = None, service: AuthService = Depends(get_auth_service), ) -> list[BotPrincipalResponse]: try: return [_to_bot_principal_response(bot) for bot in service.list_bot_principals(is_active=is_active)] except ValueError as exc: _raise_auth_error(exc) @router.get("/bots/{bot_id}", response_model=BotPrincipalResponse, dependencies=[Depends(require_admin)]) def get_bot_principal( bot_id: str, service: AuthService = Depends(get_auth_service), ) -> BotPrincipalResponse: try: return _to_bot_principal_response(service.get_bot_principal(bot_id)) except ValueError as exc: _raise_auth_error(exc) @router.delete("/bots/{bot_id}", response_model=BotPrincipalResponse, dependencies=[Depends(require_admin)]) def deactivate_bot_principal( bot_id: str, service: AuthService = Depends(get_auth_service), ) -> BotPrincipalResponse: try: return _to_bot_principal_response(service.deactivate_bot_principal(bot_id)) except ValueError as exc: _raise_auth_error(exc) @router.post( "/bots/{bot_id}/tokens", response_model=IssuedBotTokenResponse, status_code=status.HTTP_201_CREATED, dependencies=[Depends(require_admin)], ) def issue_bot_token( bot_id: str, payload: CreateBotTokenRequest, service: AuthService = Depends(get_auth_service), ) -> IssuedBotTokenResponse: try: return _to_issued_bot_token_response(service.issue_bot_token(bot_id, payload)) except ValueError as exc: _raise_auth_error(exc) @router.get("/bots/{bot_id}/tokens", response_model=list[AuthTokenResponse], dependencies=[Depends(require_admin)]) def list_bot_tokens( bot_id: str, service: AuthService = Depends(get_auth_service), ) -> list[AuthTokenResponse]: try: return [_to_auth_token_response(token) for token in service.list_bot_tokens(bot_id)] except ValueError as exc: _raise_auth_error(exc) @router.delete( "/bots/{bot_id}/tokens/{token_id}", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(require_admin)], ) def revoke_bot_token( bot_id: str, token_id: str, service: AuthService = Depends(get_auth_service), ) -> Response: try: service.revoke_bot_token(bot_id, token_id) except ValueError as exc: _raise_auth_error(exc) return Response(status_code=status.HTTP_204_NO_CONTENT)