80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
"""Bearer token authentication and authorization."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from enum import StrEnum
|
|
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
|
|
from app.core.config import Settings, get_settings
|
|
|
|
|
|
class Permission(StrEnum):
|
|
READ = "read"
|
|
TRADE_WRITE = "trade-write"
|
|
ADMIN = "admin"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AuthContext:
|
|
token: str
|
|
permission: Permission
|
|
|
|
|
|
security = HTTPBearer(auto_error=False)
|
|
|
|
|
|
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 _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,
|
|
credentials: HTTPAuthorizationCredentials | None = Depends(security),
|
|
settings: Settings = Depends(get_settings),
|
|
) -> AuthContext:
|
|
if credentials is None or credentials.scheme.lower() != "bearer":
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token")
|
|
|
|
permission = _resolve_permission(credentials.credentials, settings)
|
|
if permission is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
|
|
|
|
if not _has_permission(permission, required):
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permission")
|
|
|
|
return AuthContext(token=credentials.credentials, permission=permission)
|
|
|
|
|
|
def permission_dependency(required: Permission):
|
|
def dependency(
|
|
credentials: HTTPAuthorizationCredentials | None = Depends(security),
|
|
settings: Settings = Depends(get_settings),
|
|
) -> AuthContext:
|
|
return require_permission(required, credentials=credentials, settings=settings)
|
|
|
|
return dependency
|
|
|
|
|
|
require_read = permission_dependency(Permission.READ)
|
|
require_trade_write = permission_dependency(Permission.TRADE_WRITE)
|
|
require_admin = permission_dependency(Permission.ADMIN)
|