portfolio-hub/app/core/auth.py

"""Bearer token authentication and authorization."""

from __future__ import annotations

from dataclasses import dataclass
from enum import StrEnum
from typing import Generator

from fastapi import Depends, HTTPException, Request, status

from app.core.config import Settings, get_settings
from app.core.security import hash_token
from app.repositories.factory import get_portfolio_repository_from_settings
from app.repositories.portfolio_repository import PortfolioRepository


class Permission(StrEnum):
    READ = "read"
    TRADE_WRITE = "trade-write"
    ADMIN = "admin"


@dataclass(frozen=True)
class AuthContext:
    token: str
    permission: Permission
    user_id: str | None = None
    token_id: str | None = None
    token_type: str | None = None
    token_source: str = "env"
    runtime_id: str | None = None
    daemon_id: str | None = None

def get_auth_repository(settings: Settings = Depends(get_settings)) -> Generator[PortfolioRepository, None, None]:
    repository = get_portfolio_repository_from_settings(settings)
    try:
        yield repository
    finally:
        if callable(getattr(type(repository), "close", None)):
            repository.close()


def _resolve_permission(token: str, settings: Settings) -> Permission | None:
    if token in settings.admin_tokens:
        return Permission.ADMIN
    if token in settings.trade_write_tokens:
        return Permission.TRADE_WRITE
    if token in settings.read_tokens:
        return Permission.READ
    return None


def _resolve_auth_context(token: str, settings: Settings, repository: PortfolioRepository) -> AuthContext | None:
    permission = _resolve_permission(token, settings)
    if permission is not None:
        return AuthContext(token=token, permission=permission, token_source="env")

    if not settings.db_enabled:
        return None

    principal = repository.get_token_principal(hash_token(token))
    if principal is None:
        return None
    return AuthContext(
        token=token,
        permission=Permission(principal.permission),
        user_id=principal.user_id,
        token_id=principal.token_id,
        token_type=principal.token_type,
        token_source="db",
        runtime_id=principal.runtime_id,
        daemon_id=principal.daemon_id,
    )


def _has_permission(current: Permission, required: Permission) -> bool:
    if current == Permission.ADMIN:
        return True
    if current == required:
        return True
    if current == Permission.TRADE_WRITE and required == Permission.READ:
        return True
    return False


def require_permission(
    required: Permission,
    request: Request,
    settings: Settings = Depends(get_settings),
    repository: PortfolioRepository = Depends(get_auth_repository),
) -> AuthContext:
    return authorize_request(required, request, settings, repository=repository)


def authorize_request(
    required: Permission,
    request: Request,
    settings: Settings,
    *,
    repository: PortfolioRepository | None = None,
) -> AuthContext:
    authorization = request.headers.get("Authorization", "").strip()
    if not authorization:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token")
    scheme, _, token = authorization.partition(" ")
    if scheme.lower() != "bearer" or not token:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token")

    managed_repository = repository is None
    if repository is None:
        repository = get_portfolio_repository_from_settings(settings)

    try:
        auth_context = _resolve_auth_context(token, settings, repository)
    finally:
        if managed_repository and callable(getattr(type(repository), "close", None)):
            repository.close()

    if auth_context is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")

    if not _has_permission(auth_context.permission, required):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permission")

    return auth_context


def permission_dependency(required: Permission):
    def dependency(
        request: Request,
        settings: Settings = Depends(get_settings),
        repository: PortfolioRepository = Depends(get_auth_repository),
    ) -> AuthContext:
        return require_permission(required, request=request, settings=settings, repository=repository)

    return dependency


require_read = permission_dependency(Permission.READ)
require_trade_write = permission_dependency(Permission.TRADE_WRITE)
require_admin = permission_dependency(Permission.ADMIN)