138 lines
4.4 KiB
Python
138 lines
4.4 KiB
Python
"""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"
|
|
|
|
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",
|
|
)
|
|
|
|
|
|
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)
|