from __future__ import annotations import hmac from collections.abc import Generator from typing import Annotated from fastapi import Depends, Header, Request from sqlalchemy.orm import Session from fan_passport.auth import Actor, DevHeaderAuthBackend from fan_passport.config import Settings, get_settings from fan_passport.database import get_session from fan_passport.errors import ForbiddenError, UnauthorizedError def get_app_settings(request: Request) -> Settings: return getattr(request.app.state, "settings", get_settings()) def get_db_session() -> Generator[Session, None, None]: yield from get_session() def get_current_actor( request: Request, settings: Annotated[Settings, Depends(get_app_settings)], ) -> Actor: backend = DevHeaderAuthBackend() actor = backend.authenticate(request.headers) if actor: return actor if settings.enable_dev_auth and settings.is_localish: return Actor(subject="demo-user", display_name="Demo Fan", roles=("fan",)) raise UnauthorizedError() def require_admin( settings: Annotated[Settings, Depends(get_app_settings)], x_fan_admin_key: Annotated[str | None, Header(alias="X-Fan-Admin-Key")] = None, ) -> Actor: if settings.admin_api_key: if x_fan_admin_key and hmac.compare_digest(x_fan_admin_key, settings.admin_api_key): return Actor(subject="admin-api-key", roles=("admin",)) raise ForbiddenError("Invalid or missing admin API key.") if settings.is_localish: return Actor(subject="local-admin", roles=("admin",)) raise ForbiddenError("Admin API key is not configured.")