LegacyHUB/app/api/security.py

"""Optional API-key auth.

Behaviour:

- If ``API_KEY`` is empty (default) every request is allowed - matches the
  original dev configuration.
- If ``API_KEY`` is set, every request to a route under ``app_api_prefix``
  must carry either ``X-API-Key: <value>`` or ``Authorization: Bearer <value>``.
- ``/health`` is intentionally exempt so external probes (compose healthcheck,
  reverse proxy, monitoring) keep working without leaking the key.
- The root ``/`` page stays open so the OpenAPI banner and docs links remain
  reachable.

This is a defence-in-depth layer behind whatever reverse proxy / OAuth gateway
runs in production - not a replacement.
"""

from __future__ import annotations

import hmac
from collections.abc import Awaitable, Callable

from fastapi import FastAPI, HTTPException, Request, Response, status
from fastapi.responses import JSONResponse
from starlette.types import ASGIApp

EXEMPT_PATHS: tuple[str, ...] = ("/", "/docs", "/redoc", "/openapi.json")
EXEMPT_SUFFIXES: tuple[str, ...] = ("/health",)
# Machine (service-to-service) endpoints authenticate through the mandatory
# ``require_service_api_key`` route dependency, which accepts INGEST_API_KEY or
# API_KEY. The global middlewares must not swallow them: the API_KEY middleware
# only knows the global key (the ingest key would be rejected), and the
# trusted-identity middleware expects user headers machine callers never have.
# Exempting them here does NOT open the endpoints — their own dependency stays
# mandatory (see tests/test_ingest_auth.py).
MACHINE_PATH_SUFFIXES: tuple[str, ...] = ("/knowledge-ingest", "/dispatch/inbox")


def _extract_token(request: Request) -> str | None:
    header = request.headers.get("x-api-key")
    if header:
        return header.strip()
    auth = request.headers.get("authorization") or ""
    if auth.lower().startswith("bearer "):
        return auth[7:].strip()
    return None


def _valid_service_keys() -> list[str]:
    from app.config import settings

    keys: list[str] = []
    if settings.ingest_api_key and settings.ingest_api_key.strip():
        keys.append(settings.ingest_api_key.strip())
    if settings.api_key and settings.api_key.strip():
        keys.append(settings.api_key.strip())
    return keys


def require_service_api_key(request: Request) -> None:
    """FastAPI dependency: mandatory service auth for participant endpoints.

    Independent of the global ``install_api_key_auth`` middleware. The caller
    must present ``X-API-Key`` / ``Authorization: Bearer`` matching the
    configured service key (``INGEST_API_KEY`` or ``API_KEY``). When no key is
    configured at all (pure local dev) the check is skipped; production forces
    ``API_KEY`` so the endpoint is always authenticated there. This is the
    machine layer and is distinct from the user ``X-TeamHub-*`` identity layer.
    """
    valid = _valid_service_keys()
    if not valid:
        return
    token = _extract_token(request)
    if not token or not any(hmac.compare_digest(token, k) for k in valid):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="invalid or missing service api key",
            headers={"WWW-Authenticate": "Bearer"},
        )


def install_api_key_auth(app: FastAPI) -> None:
    """Attach the middleware. Always safe to call; becomes a no-op when no key
    is configured.

    Reads ``app.config.settings`` lazily so test fixtures can reload the config
    module and have the new ``API_KEY`` value take effect on the next install.
    """
    from app.config import settings as fresh_settings  # re-resolve after reloads

    settings = fresh_settings
    expected = settings.api_key.strip() if settings.api_key else ""
    if not expected:
        return

    @app.middleware("http")
    async def _api_key_middleware(  # type: ignore[no-redef]
        request: Request,
        call_next: Callable[[Request], Awaitable[Response]],
    ) -> Response:
        path = request.url.path
        if request.method == "OPTIONS":
            return await call_next(request)
        if path in EXEMPT_PATHS:
            return await call_next(request)
        if any(path.endswith(s) for s in EXEMPT_SUFFIXES):
            return await call_next(request)
        if any(path.endswith(s) for s in MACHINE_PATH_SUFFIXES):
            return await call_next(request)
        if not path.startswith(settings.app_api_prefix):
            return await call_next(request)

        token = _extract_token(request)
        if not token or not hmac.compare_digest(token, expected):
            return JSONResponse(
                status_code=401,
                content={"detail": "invalid or missing api key"},
                headers={"WWW-Authenticate": "Bearer"},
            )
        return await call_next(request)


def install_trusted_headers_auth(app: FastAPI) -> None:
    """Enforce presence of gateway trusted-identity headers when enabled.

    No-op unless ``AUTH_REQUIRE_IDENTITY`` is true. When enabled, every request
    under ``app_api_prefix`` (except ``/health``, ``/``, docs) must carry a
    valid ``X-TeamHub-Actor`` header injected by the platform gateway. This is
    the *user-context* layer and is independent of the ``X-API-Key`` *machine*
    layer above. Trust requires the module to be reachable only behind the
    gateway (network isolation) - see ``docker-compose.teamhub.yml``.
    """
    from app.config import settings as fresh_settings
    from app.integrations.identity import parse_identity

    if not fresh_settings.auth_require_identity:
        return

    @app.middleware("http")
    async def _trusted_headers_middleware(  # type: ignore[no-redef]
        request: Request,
        call_next: Callable[[Request], Awaitable[Response]],
    ) -> Response:
        path = request.url.path
        if request.method == "OPTIONS":
            return await call_next(request)
        if path in EXEMPT_PATHS or any(path.endswith(s) for s in EXEMPT_SUFFIXES):
            return await call_next(request)
        if any(path.endswith(s) for s in MACHINE_PATH_SUFFIXES):
            return await call_next(request)
        if not path.startswith(fresh_settings.app_api_prefix):
            return await call_next(request)

        identity = parse_identity(request.headers)
        if identity is None:
            return JSONResponse(
                status_code=401,
                content={"detail": "missing trusted identity headers"},
            )
        request.state.identity = identity
        return await call_next(request)


__all__ = [
    "install_api_key_auth",
    "install_trusted_headers_auth",
    "require_service_api_key",
]
_ = ASGIApp  # re-export hint to keep mypy happy on older Starlette versions