svcforge/services/api/deps.py

"""Dependency injection: how a handler gets a pool, a repo, a catalog, and a team.

Everything expensive — the pool, the JWKS client, the parsed catalog — is built once in
`lifespan` and parked on `app.state`; these functions only hand it out. A `Depends` that
does I/O per request does that I/O on every request forever.
"""

from __future__ import annotations

import asyncio
from typing import Annotated, Any

import jwt
from fastapi import Depends, HTTPException, Request, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jwt import PyJWKClient

from svcforge_core.adapters.redis import RateLimiterProto
from svcforge_core.domain.models import CatalogEntry
from svcforge_core.repo.db import DictPool
from svcforge_core.repo.instances import InstanceRepo
from svcforge_core.repo.tasks import TaskRepo
from svcforge_core.settings import Settings

# The algorithm allow-list is not configuration. Without it, `jwt.decode` accepts whatever
# the *token* claims in its own header — including `none`, and including HS256 verified
# with the RSA public key as an HMAC secret. Both are forgery.
ALLOWED_ALGORITHMS = ["RS256"]

# What `auth_disabled` returns. Settings.check_production() refuses that flag in prod.
DEV_TEAM = "platform"

TEAM_CLAIM = "team"

# auto_error=False is load-bearing. HTTPBearer(auto_error=True) answers a *missing*
# Authorization header with 403, not 401 — an old FastAPI wart. The spec (and every
# client that knows what to do about it) wants 401, so the error is raised here.
_bearer = HTTPBearer(auto_error=False)


def _unauthorized() -> HTTPException:
    """One shape for every auth failure.

    Expired, wrong issuer, wrong audience, bad signature, malformed, no header: the same 401
    with the same body. Naming which one turns the endpoint into an oracle a forger can tune
    against.
    """
    return HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail={"code": "unauthorized", "message": "invalid or missing credentials"},
        headers={"WWW-Authenticate": "Bearer"},
    )


def get_settings(request: Request) -> Settings:
    """The Settings that create_app() was handed."""
    settings: Settings = request.app.state.settings
    return settings


async def get_pool(request: Request) -> DictPool:
    """Return the pool that lifespan put on app.state."""
    pool: DictPool = request.app.state.pool
    return pool


def get_catalog(request: Request) -> dict[str, CatalogEntry]:
    """The catalog, parsed once at startup.

    Read per request, a mid-flight edit to catalog.yaml would change the answer between two
    requests of the same deploy. A catalog change is a restart.
    """
    catalog: dict[str, CatalogEntry] = request.app.state.catalog
    return catalog


def get_instance_repo(pool: Annotated[DictPool, Depends(get_pool)]) -> InstanceRepo:
    """An InstanceRepo bound to the app's pool. Cheap: it is a handle, not a connection."""
    return InstanceRepo(pool)


def get_task_repo(pool: Annotated[DictPool, Depends(get_pool)]) -> TaskRepo:
    """A TaskRepo bound to the app's pool."""
    return TaskRepo(pool)


async def get_current_team(
    request: Request,
    creds: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)],
    settings: Annotated[Settings, Depends(get_settings)],
) -> str:
    """Verify the JWT against the cached JWKS. Check aud/iss/exp and the alg allow-list.

    Returns the team claim. Raises HTTPException(401) on any failure — never leaks why.
    """
    if settings.auth_disabled:
        return DEV_TEAM

    if creds is None or not creds.credentials:
        raise _unauthorized()

    jwks_client: PyJWKClient | None = getattr(request.app.state, "jwks_client", None)
    if jwks_client is None:
        # Auth is on but there is no key source. Fail closed. A 500 would be honest about
        # the cause and would also let a forger tell a misconfigured deploy from a bad token.
        raise _unauthorized()

    try:
        # PyJWKClient keeps a TTL cache, so this is a dict lookup on the hot path and only
        # blocks on a miss (key rotation) — hence to_thread, a thread hop a few times a day
        # rather than an event loop stalled on someone else's HTTP call.
        signing_key = await _signing_key(jwks_client, creds.credentials)
        claims: dict[str, Any] = jwt.decode(
            creds.credentials,
            signing_key.key,
            algorithms=ALLOWED_ALGORITHMS,
            audience=settings.jwt_audience,
            issuer=settings.jwt_issuer,
            options={
                "require": ["exp", "aud", "iss"],
                "verify_exp": True,
                "verify_aud": True,
                "verify_iss": settings.jwt_issuer is not None,
                "verify_signature": True,
            },
        )
    except Exception as exc:  # deliberate catch-all: every failure becomes one opaque 401
        raise _unauthorized() from exc

    team = claims.get(TEAM_CLAIM)
    if not isinstance(team, str) or not team:
        raise _unauthorized()
    return team


async def _signing_key(client: PyJWKClient, token: str) -> jwt.PyJWK:
    """Fetch the signing key off the event loop.

    `get_signing_key_from_jwt()` does a synchronous urlopen on a cache miss. Called directly
    from `async def` it blocks the loop: every other in-flight request stops until the IdP
    answers, and if the IdP hangs so does the pod, with /readyz still saying it is fine.
    """
    return await asyncio.to_thread(client.get_signing_key_from_jwt, token)


def get_rate_limiter(request: Request) -> RateLimiterProto | None:
    """The limiter lifespan built, or None when no Redis is configured."""
    limiter: RateLimiterProto | None = getattr(request.app.state, "rate_limiter", None)
    return limiter


async def rate_limit(
    request: Request,
    team: Annotated[str, Depends(get_current_team)],
) -> None:
    """Per-team rate limiting. One Redis command per check, and it fails OPEN.

    Redis holds derived state, so losing it must degrade the platform rather than stop it: a
    limiter that fails closed turns a cache outage into a total outage, a worse incident
    than the burst it was guarding against. `RateLimiter.check` swallows its own errors and
    returns `allowed=True`, so the 429 below only comes from a real, counted overage.
    """
    limiter = get_rate_limiter(request)
    if limiter is None:
        return

    result = await limiter.check(team)
    if not result.allowed:
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail={"code": "rate_limited", "message": "too many requests"},
            headers={"Retry-After": str(result.retry_after_s)},
        )


async def idempotency_key(request: Request) -> str | None:
    """`Idempotency-Key` handling. Seam only — Module 10 fills this in (Redis store).

    Until then the real idempotency anchor is `instances.release_name`, which is unique in
    the schema and deterministic from (team, service_type, id).
    """
    return request.headers.get("Idempotency-Key")


PoolDep = Annotated[DictPool, Depends(get_pool)]
TeamDep = Annotated[str, Depends(get_current_team)]
InstanceRepoDep = Annotated[InstanceRepo, Depends(get_instance_repo)]
TaskRepoDep = Annotated[TaskRepo, Depends(get_task_repo)]
CatalogDep = Annotated[dict[str, CatalogEntry], Depends(get_catalog)]