loyaly-catalogue/backend/app/api/deps.py

"""
Request-scoped authentication dependencies.

Guards are attached per route, not as middleware matching on paths. Two
reasons that matters here:

  * A path-matching middleware silently stops guarding a route the moment
    somebody renames it. A ``Depends`` on the route function cannot drift out
    of sync with the route it protects.
  * FastAPI reflects these into the OpenAPI schema, so ``/docs`` shows which
    operations need a credential instead of implying everything is open.

The guard therefore holds regardless of which host the request arrives on -
through the frontend's nginx on ``{$DOMAIN}``, or directly on ``api.{$DOMAIN}``.

Usage::

    @router.post("/thing", dependencies=[Depends(require_admin)])
    def create_thing(): ...

    @router.post("/other", dependencies=[Depends(require_permission("add_product"))])
    def other_thing(): ...

    @router.post("/who", ...)
    def who(principal: Principal = Depends(get_principal)): ...
"""
from __future__ import annotations

from typing import Callable, Optional

from fastapi import Depends, HTTPException, status
from fastapi.security import APIKeyHeader, HTTPAuthorizationCredentials, HTTPBearer

from app.infrastructure.security import (
    AuthError,
    Principal,
    anonymous_principal,
    decode_access_token,
    principal_for_api_key,
)
from app.infrastructure.settings import AUTH_ENABLED

# auto_error=False on both: with two accepted credential types, letting either
# scheme raise on its own would reject a request that carried the *other* one.
# get_principal decides, once it has seen both.
_bearer_scheme = HTTPBearer(auto_error=False, description="Access token from POST /api/auth/login")
_api_key_scheme = APIKeyHeader(
    name="X-API-Key",
    auto_error=False,
    description="Static key for machine consumers (see API_KEYS)",
)

_UNAUTHENTICATED = HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED,
    detail="Not authenticated. Send a bearer token from POST /api/auth/login, or an X-API-Key header.",
    headers={"WWW-Authenticate": "Bearer"},
)


def get_principal(
    credentials: Optional[HTTPAuthorizationCredentials] = Depends(_bearer_scheme),
    api_key: Optional[str] = Depends(_api_key_scheme),
) -> Principal:
    """Resolve the caller, or raise 401. Use this to require *any* valid credential."""
    if not AUTH_ENABLED:
        return anonymous_principal()

    if credentials is not None and credentials.credentials:
        try:
            return decode_access_token(credentials.credentials)
        except AuthError as exc:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail=str(exc),
                headers={"WWW-Authenticate": "Bearer"},
            ) from exc

    if api_key:
        try:
            return principal_for_api_key(api_key)
        except AuthError as exc:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED, detail=str(exc)
            ) from exc

    raise _UNAUTHENTICATED


def get_optional_principal(
    credentials: Optional[HTTPAuthorizationCredentials] = Depends(_bearer_scheme),
    api_key: Optional[str] = Depends(_api_key_scheme),
) -> Optional[Principal]:
    """
    Resolve the caller if they presented a valid credential, else None.

    For endpoints that are public but behave differently when signed in. A
    credential that is present but *invalid* still raises - failing open there
    would mean a typo'd token silently downgrades to anonymous access.
    """
    if not AUTH_ENABLED:
        return anonymous_principal()
    if credentials is None and not api_key:
        return None
    return get_principal(credentials, api_key)


def require_role(*roles: str) -> Callable[[Principal], Principal]:
    """Require the caller to hold one of ``roles``."""
    allowed = frozenset(roles)

    def _dependency(principal: Principal = Depends(get_principal)) -> Principal:
        if principal.role not in allowed:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=(
                    f"This operation requires the {' or '.join(sorted(allowed))} role; "
                    f"you are signed in as '{principal.role}'."
                ),
            )
        return principal

    return _dependency


def require_permission(permission: str) -> Callable[[Principal], Principal]:
    """
    Require a specific permission. ``admin`` passes every check - see
    ``Principal.has_permission``.
    """

    def _dependency(principal: Principal = Depends(get_principal)) -> Principal:
        if not principal.has_permission(permission):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"This operation requires the '{permission}' permission.",
            )
        return principal

    return _dependency


# The two guards used most often, named so route decorators stay readable.
require_admin = require_role("admin")
require_authenticated = get_principal