"""Authentication middleware and utilities.""" import secrets from contextvars import ContextVar from typing import Optional from fastapi import Depends, HTTPException, Query, Request, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from .config import settings security = HTTPBearer(auto_error=False) # Context variable to store current request's token label token_label_var: ContextVar[str] = ContextVar("token_label", default="unknown") # Per-request correlation ID — generated in middleware if upstream didn't send one. request_id_var: ContextVar[str] = ContextVar("request_id", default="-") def auth_enabled() -> bool: """Check if authentication is enabled (i.e. at least one token is configured).""" return bool(settings.api_tokens) def get_token_label(token: str) -> Optional[str]: """Get the label for a token. Returns None if token is invalid. Args: token: The token to look up Returns: The label for the token, or None if invalid """ for label, spec in settings.api_tokens.items(): if secrets.compare_digest(spec.token, token): return label return None def token_has_scope(label: str, required: str) -> bool: """Whether the token identified by `label` grants `required` scope.""" spec = settings.api_tokens.get(label) if spec is None: # Unknown label = no auth or anonymous; treat as full access only # when auth is disabled entirely (matches existing behaviour). return not auth_enabled() return spec.grants(required) def require_scope(scope: str): """Build a FastAPI dependency that enforces the given scope. Use as ``Depends(require_scope("admin"))`` on management endpoints. When auth is disabled the dependency is a no-op (anonymous access). """ async def _checker(label: str = Depends(verify_token)) -> str: if not auth_enabled(): return label if not token_has_scope(label, scope): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"Token '{label}' lacks required scope: {scope}", ) return label return _checker async def verify_token( request: Request, credentials: HTTPAuthorizationCredentials = Depends(security), ) -> str: """Verify the API token from the Authorization header. When no tokens are configured, authentication is skipped entirely. Reuses the label from middleware context when already validated. Returns: The token label (or "anonymous" when auth is disabled) Raises: HTTPException: If the token is missing or invalid (only when auth enabled) """ if not auth_enabled(): token_label_var.set("anonymous") return "anonymous" # Reuse label already set by middleware to avoid redundant O(n) scan existing = token_label_var.get("unknown") if existing != "unknown": return existing if credentials is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing authentication token", headers={"WWW-Authenticate": "Bearer"}, ) label = get_token_label(credentials.credentials) if label is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authentication token", headers={"WWW-Authenticate": "Bearer"}, ) token_label_var.set(label) return label class TokenAuth: """Dependency class for token authentication.""" def __init__(self, auto_error: bool = True): self.auto_error = auto_error async def __call__( self, request: Request, credentials: HTTPAuthorizationCredentials = Depends(security), ) -> str | None: """Verify the token and return the label or raise an exception.""" if not auth_enabled(): token_label_var.set("anonymous") return "anonymous" if credentials is None: if self.auto_error: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing authentication token", headers={"WWW-Authenticate": "Bearer"}, ) return None label = get_token_label(credentials.credentials) if label is None: if self.auto_error: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authentication token", headers={"WWW-Authenticate": "Bearer"}, ) return None # Set label in context for logging token_label_var.set(label) return label async def verify_token_or_query( credentials: HTTPAuthorizationCredentials = Depends(security), token: Optional[str] = Query(None, description="API token as query parameter"), ) -> str: """Verify the API token from header or query parameter. Useful for endpoints that need to be accessed via URL (like images). Args: credentials: The bearer token credentials from header token: Token from query parameter Returns: The token label Raises: HTTPException: If the token is missing or invalid """ if not auth_enabled(): token_label_var.set("anonymous") return "anonymous" # Reuse label already set by middleware existing = token_label_var.get("unknown") if existing != "unknown": return existing label = None # Try header first if credentials is not None: label = get_token_label(credentials.credentials) # Try query parameter if label is None and token is not None: label = get_token_label(token) if label is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing or invalid authentication token", headers={"WWW-Authenticate": "Bearer"}, ) token_label_var.set(label) return label