165 lines
7.2 KiB
Python
165 lines
7.2 KiB
Python
"""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:
|
|
self._repository.get_user(payload.owner_user_id)
|
|
return self._repository.create_bot_principal(
|
|
owner_user_id=payload.owner_user_id,
|
|
name=payload.name,
|
|
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("비활성화된 봇에는 토큰을 발급할 수 없습니다")
|
|
|
|
owner = self._repository.get_user(bot.owner_user_id)
|
|
owner_permission = Permission(owner.permission)
|
|
requested = Permission(payload.permission or owner.permission)
|
|
if not _has_permission(owner_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(
|
|
owner.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) |