"""Authentication and token issuance service.""" from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timedelta, timezone from fastapi import Depends from app.core.auth import AuthContext, Permission, _has_permission from app.core.config import Settings, get_settings from app.core.security import generate_token, hash_token, token_prefix, verify_password from app.domain.models import AuthToken, BotPrincipal, User from app.repositories.factory import get_portfolio_repository_from_settings from app.repositories.portfolio_repository import PortfolioRepository from app.schemas.auth import CreateBotPrincipalRequest, CreateBotTokenRequest, CreateClientTokenRequest, LoginRequest @dataclass class LoginResult: token: str auth_token: AuthToken user: User @dataclass class IssuedClientToken: token: str auth_token: AuthToken @dataclass class IssuedBotToken: token: str auth_token: AuthToken def get_auth_service_repository(settings: Settings = Depends(get_settings)) -> PortfolioRepository: return get_portfolio_repository_from_settings(settings) class AuthService: def __init__(self, repository: PortfolioRepository, settings: Settings) -> None: self._repository = repository self._settings = settings def login(self, payload: LoginRequest) -> LoginResult: user = self._repository.get_user_by_username(payload.username) if user is None or not user.is_active or not verify_password(payload.password, user.password_hash): raise ValueError("잘못된 사용자명 또는 비밀번호입니다") raw_token = generate_token() expires_at = datetime.now(timezone.utc) + timedelta(hours=self._settings.session_token_ttl_hours) auth_token = self._repository.create_auth_token( user.id, name="session", token_hash=hash_token(raw_token), token_prefix=token_prefix(raw_token), permission=user.permission, token_type="session", expires_at=expires_at, ) self._repository.update_user_last_login(user.id) refreshed_user = self._repository.get_user(user.id) return LoginResult(token=raw_token, auth_token=auth_token, user=refreshed_user) def issue_client_token(self, auth_context: AuthContext, payload: CreateClientTokenRequest) -> IssuedClientToken: if auth_context.bot_principal_id is not None: raise ValueError("봇 토큰으로는 사용자 클라이언트 토큰을 발급할 수 없습니다") if auth_context.user_id is None: raise ValueError("DB 사용자 컨텍스트에서만 클라이언트 토큰을 발급할 수 있습니다") requested = Permission(payload.permission or auth_context.permission.value) if not _has_permission(auth_context.permission, requested): raise ValueError("요청한 토큰 권한이 현재 권한을 초과합니다") raw_token = generate_token() expires_at = self._build_expiration(payload.expires_in_days, self._settings.client_token_default_ttl_days) auth_token = self._repository.create_auth_token( auth_context.user_id, name=payload.name, token_hash=hash_token(raw_token), token_prefix=token_prefix(raw_token), permission=requested.value, token_type="client", expires_at=expires_at, ) return IssuedClientToken(token=raw_token, auth_token=auth_token) def list_tokens(self, auth_context: AuthContext) -> list[AuthToken]: if auth_context.bot_principal_id is not None: raise ValueError("봇 토큰으로는 사용자 토큰 목록을 조회할 수 없습니다") if auth_context.user_id is None: raise ValueError("DB 사용자 컨텍스트에서만 토큰 목록을 조회할 수 있습니다") return self._repository.list_auth_tokens(auth_context.user_id) def revoke_token(self, auth_context: AuthContext, token_id: str) -> None: if auth_context.bot_principal_id is not None: raise ValueError("봇 토큰으로는 사용자 토큰을 폐기할 수 없습니다") if auth_context.user_id is None: raise ValueError("DB 사용자 컨텍스트에서만 토큰을 폐기할 수 있습니다") self._repository.revoke_auth_token(auth_context.user_id, token_id) def create_bot_principal(self, payload: CreateBotPrincipalRequest) -> BotPrincipal: return self._repository.create_bot_principal( name=payload.name, permission=payload.permission, description=payload.description, ) def list_bot_principals(self, *, is_active: bool | None = None) -> list[BotPrincipal]: return self._repository.list_bot_principals(is_active=is_active) def get_bot_principal(self, bot_principal_id: str) -> BotPrincipal: return self._repository.get_bot_principal(bot_principal_id) def deactivate_bot_principal(self, bot_principal_id: str) -> BotPrincipal: return self._repository.deactivate_bot_principal(bot_principal_id) def issue_bot_token(self, bot_principal_id: str, payload: CreateBotTokenRequest) -> IssuedBotToken: bot = self._repository.get_bot_principal(bot_principal_id) if not bot.is_active: raise ValueError("비활성화된 봇에는 토큰을 발급할 수 없습니다") bot_permission = Permission(bot.permission) requested = Permission(payload.permission or bot.permission) if not _has_permission(bot_permission, requested): raise ValueError("요청한 토큰 권한이 봇 권한을 초과합니다") raw_token = generate_token() expires_at = self._build_expiration(payload.expires_in_days, self._settings.bot_token_default_ttl_days) auth_token = self._repository.create_auth_token( bot.id, name=payload.name, token_hash=hash_token(raw_token), token_prefix=token_prefix(raw_token), permission=requested.value, token_type="bot", bot_principal_id=bot.id, expires_at=expires_at, ) return IssuedBotToken(token=raw_token, auth_token=auth_token) def list_bot_tokens(self, bot_principal_id: str) -> list[AuthToken]: self._repository.get_bot_principal(bot_principal_id) return self._repository.list_bot_auth_tokens(bot_principal_id) def revoke_bot_token(self, bot_principal_id: str, token_id: str) -> None: self._repository.get_bot_principal(bot_principal_id) self._repository.revoke_bot_auth_token(bot_principal_id, token_id) @staticmethod def _build_expiration(expires_in_days: int | None, default_days: int) -> datetime: days = expires_in_days if expires_in_days is not None else default_days return datetime.now(timezone.utc) + timedelta(days=days) def get_auth_service( settings: Settings = Depends(get_settings), repository: PortfolioRepository = Depends(get_auth_service_repository), ) -> AuthService: return AuthService(repository, settings)