"""FastAPI auth dependency for the outbound pull surface. ``require_consumer(*scopes)`` is the single gate every Directory / SCIM endpoint sits behind. It mirrors the defence stack the MCP PAT middleware proved out (``app.mcp.middleware``), adapted to a router dependency: * uniform 401 for every credential failure (no oracle about which check failed), * per-TOKEN failed-auth window peeked BEFORE the DB lookup, so one stuck client retrying a dead token stops buying DB round-trips, * per-IP failed-auth window so a token scanner can't flood the audit table, * per-CONSUMER fixed-window rate limit on the success path (consumer-configurable cap, platform default), * auth failures audited via an independent committed session, * last-used + daily-usage bookkeeping committed in its own short session so read-only requests still leave an investigation trail. All Redis windows fail OPEN (with a small delay) — a Redis outage must degrade rate limiting, not take the integration surface down. """ from __future__ import annotations import asyncio import hashlib import logging from dataclasses import dataclass, field from datetime import UTC, datetime from typing import Any from uuid import UUID, uuid4 from fastapi import Depends, Request from sqlalchemy.dialects.mysql import insert as mysql_insert from sqlalchemy.ext.asyncio import AsyncSession from app.core import database as _db from app.core.config import settings from app.core.database import get_db from app.core.exceptions import AppException, ForbiddenError, UnauthorizedError from app.identity.consumers.models import ( IdentityConsumerToken, IdentityConsumerUsageDaily, ) from app.identity.consumers.token_service import ( SVC_TOKEN_PREFIX, ConsumerTokenService, ) logger = logging.getLogger(__name__) _AUTH_FAILURE_ACTION = "directory_api.auth_failure" @dataclass(slots=True) class ConsumerContext: """Request-scoped snapshot of the authenticated consumer. Plain values (not the ORM rows) so handlers can't accidentally trip lazy-loads or mutate authorization state mid-request. """ consumer_id: UUID slug: str name: str scopes: frozenset[str] #: Least-privilege person-field grant (field_catalog keys). The #: ``/persons`` serializer projects to exactly this set. granted_fields: frozenset[str] = field(default_factory=frozenset) filter_config: dict[str, Any] = field(default_factory=dict) token_id: UUID | None = None token_prefix: str = "" def has_scope(self, scope: str) -> bool: return scope in self.scopes def reads_pii(self) -> bool: """True iff any granted field is in the PII tier — drives the ``directory_api.pii_read`` audit on the data endpoints.""" from app.identity.consumers.field_catalog import PII_FIELD_KEYS return bool(self.granted_fields & PII_FIELD_KEYS) # ── Redis windows ───────────────────────────────────────────────────── #: Process-local fallback windows for when Redis is down. A plain #: dict {key: (window_started_monotonic, count)} — coarse (per uvicorn #: worker, not cluster-wide) but it means a Redis outage degrades the #: caps by a small worker-count factor instead of removing them #: entirely. On a DATA-EGRESS surface "no limit at all during the #: outage window" is the wrong failure mode (security review #: 2026-06-11). Bounded: keys = consumers + failure buckets, pruned #: opportunistically on each touch. _LOCAL_WINDOWS: dict[str, tuple[float, int]] = {} _LOCAL_WINDOWS_MAX = 4096 def _local_window_incr(key: str, window: int) -> int: import time now = time.monotonic() started, count = _LOCAL_WINDOWS.get(key, (now, 0)) if now - started >= window: started, count = now, 0 count += 1 if len(_LOCAL_WINDOWS) >= _LOCAL_WINDOWS_MAX and key not in _LOCAL_WINDOWS: # Drop expired entries; if everything is live, drop arbitrary # oldest-inserted (dict order) — a degraded limiter beats an # unbounded dict. for k, (st, _c) in list(_LOCAL_WINDOWS.items()): if now - st >= window: _LOCAL_WINDOWS.pop(k, None) while len(_LOCAL_WINDOWS) >= _LOCAL_WINDOWS_MAX: _LOCAL_WINDOWS.pop(next(iter(_LOCAL_WINDOWS)), None) _LOCAL_WINDOWS[key] = (started, count) return count async def _fixed_window_incr(key: str, window: int) -> int | None: """Atomic fixed-window counter (SET NX + INCR). On Redis failure, falls back to the process-local window above and returns its count — degraded (per-worker) but never uncapped. """ from app.core.redis_client import get_redis try: redis = get_redis() async with redis.pipeline(transaction=True) as pipe: await pipe.set(key, 0, ex=window, nx=True) await pipe.incr(key) results = await pipe.execute() return int(results[1]) except Exception as exc: # noqa: BLE001 — degrade to local window logger.warning( "directory-api rate window Redis error (local fallback): %s", type(exc).__name__, ) await asyncio.sleep(0.05) return _local_window_incr(key, window) async def _window_ttl(key: str, fallback: int) -> int: from app.core.redis_client import get_redis try: ttl = await get_redis().ttl(key) return int(ttl) if ttl and ttl > 0 else fallback except Exception: # noqa: BLE001 return fallback def _token_failure_key(token: str) -> str: digest = hashlib.sha256(token.encode("utf-8")).hexdigest()[:32] return f"dirapi_authfail:tok:{digest}" async def _peek_token_failures(token: str) -> tuple[bool, int]: """Read-only check of the per-token failed-auth window (pre-DB).""" from app.core.redis_client import get_redis key = _token_failure_key(token) cap = settings.DIRECTORY_API_AUTH_FAILURE_LIMIT_PER_IP_MAX window = settings.DIRECTORY_API_AUTH_FAILURE_LIMIT_PER_IP_WINDOW_SECONDS try: raw = await get_redis().get(key) if raw is not None and int(raw) > cap: return False, await _window_ttl(key, window) return True, 0 except Exception as exc: # noqa: BLE001 logger.warning( "directory-api token-failure peek Redis error (local check): %s", type(exc).__name__, ) await asyncio.sleep(0.05) # Redis down: consult the process-local failure window so the # brute-force gate still bites. _record_auth_failure writes the # same key into _LOCAL_WINDOWS on the failure side; reading it # here closes the "count written but never checked" gap (logic # review round 2, 2026-06-11). Read-only — never mutate here. local = _LOCAL_WINDOWS.get(key) if local is not None and local[1] > cap: return False, window return True, 0 async def _record_auth_failure( *, reason: str, request: Request, token: str | None, ) -> None: """Bump failure windows + audit. Never raises.""" window = settings.DIRECTORY_API_AUTH_FAILURE_LIMIT_PER_IP_WINDOW_SECONDS if token: await _fixed_window_incr(_token_failure_key(token), window) client_ip = _client_ip(request) ip_count = await _fixed_window_incr(f"dirapi_authfail:{client_ip}", window) if ( ip_count is not None and ip_count > settings.DIRECTORY_API_AUTH_FAILURE_LIMIT_PER_IP_MAX ): # Over the per-IP cap: skip the audit write so a scanner can't # flood the table; the 401 below still goes out. return try: from app.audit.service import log_denied_action await log_denied_action( trace_id=getattr(request.state, "trace_id", None) or str(uuid4()), actor_type="anonymous", action=_AUTH_FAILURE_ACTION, resource_type="directory_api", resource_id="auth", detail={"reason": reason[:200], "path": request.url.path[:200]}, ip_address=client_ip, user_agent=(request.headers.get("user-agent") or "")[:512] or None, ) except Exception: # noqa: BLE001 — audit must never block the 401 logger.exception("directory-api auth-failure audit write failed") def _client_ip(request: Request) -> str: """Proxy-aware client IP, same trust rules as the REST surface.""" try: from app.core.dependencies import get_real_ip return get_real_ip(request) except Exception: # noqa: BLE001 return request.client.host if request.client else "unknown" # ── Bookkeeping (independent session, best-effort) ─────────────────── async def _record_usage( *, token_id: UUID, consumer_id: UUID, client_ip: str, ) -> None: """last_used_* on the token + daily counter upsert, own session. Best-effort: a DB blip here must not 500 an otherwise-valid request. Committed independently so the trail survives read-only requests that never commit their request transaction. """ now = datetime.now(UTC) try: async with _db.async_session_factory() as session: token = await session.get(IdentityConsumerToken, token_id) if token is not None: token.last_used_at = now token.last_used_ip = client_ip[:64] stmt = mysql_insert(IdentityConsumerUsageDaily).values( id=str(uuid4()), consumer_id=str(consumer_id), usage_date=now.date(), request_count=1, last_request_at=now, ) stmt = stmt.on_duplicate_key_update( request_count=IdentityConsumerUsageDaily.request_count + 1, last_request_at=now, ) await session.execute(stmt) await session.commit() except Exception as exc: # noqa: BLE001 logger.warning("directory-api usage bookkeeping failed: %s", exc) async def audit_pii_read( *, request: Request, ctx, persons_served: int, ) -> None: """Audit row for a successful plaintext-PII read (independent committed session — the data endpoints are read-only and never commit their request transaction). Auth failures and scope probes were already audited; without this the one question forensics actually asks — "which consumers read plaintext PII, when, how much" — had no audit_logs answer (security review 2026-06-11). Volume is bounded by the per-consumer rate limit and PII consumers are expected to be few. Best-effort. """ if persons_served <= 0: return try: from app.audit.service import AuditService async with _db.async_session_factory() as audit_db: await AuditService(audit_db).log( trace_id=getattr(request.state, "trace_id", None) or str(uuid4()), actor_type="service", action="directory_api.pii_read", status="success", resource_type="identity_consumer", resource_id=str(ctx.consumer_id), detail={ "consumer_slug": ctx.slug, "path": request.url.path[:200], "persons_served": persons_served, }, ip_address=_client_ip(request), ) await audit_db.commit() except Exception: # noqa: BLE001 — audit must not break the read logger.exception("directory-api pii-read audit write failed") # ── The dependency ──────────────────────────────────────────────────── def require_consumer(*required_scopes: str): """Dependency factory: authenticate a ``helix_svc_`` bearer and enforce that the consumer holds every scope in ``required_scopes``. """ async def _dependency( request: Request, db: AsyncSession = Depends(get_db), ) -> ConsumerContext: auth_header = request.headers.get("authorization") or "" if not auth_header.lower().startswith("bearer "): await _record_auth_failure( reason="missing_or_malformed_authorization", request=request, token=None, ) raise UnauthorizedError("Missing or malformed Authorization header.") token = auth_header.split(" ", 1)[1].strip() if not token.startswith(SVC_TOKEN_PREFIX): await _record_auth_failure( reason="non_service_bearer_token", request=request, token=None, ) raise UnauthorizedError( "Only service tokens (helix_svc_…) are accepted on this endpoint.", ) allowed, retry_after = await _peek_token_failures(token) if not allowed: raise AppException( code="RATE_LIMITED", message=f"Too many failed attempts. Try again in {retry_after}s.", status_code=429, headers={"Retry-After": str(retry_after)}, ) resolved = await ConsumerTokenService(db).authenticate(token) if resolved is None: await _record_auth_failure( reason="invalid_or_expired_service_token", request=request, token=token, ) raise UnauthorizedError("Invalid or expired token.") consumer, token_row = resolved scopes = frozenset( s for s in (consumer.scopes or []) if isinstance(s, str) ) # Field grant: an explicit list narrows the persons bundle; # an empty/missing list (pre-field-level row not yet backfilled) # falls back to the full bundle default so the row keeps working. from app.identity.consumers.field_catalog import default_fields_for_scopes granted_fields = frozenset( f for f in (consumer.granted_fields or []) if isinstance(f, str) ) or frozenset(default_fields_for_scopes(scopes)) # Per-consumer fixed-window rate limit — enforced BEFORE the scope # check so a valid-but-under-privileged token cannot hammer a scoped # endpoint and amplify unbounded ``scope_denied`` audit INSERTs (each # 403 below writes an audit row; the per-IP failed-auth cap does NOT # fire on a 403 because the credential itself is valid). Mirrors the # api_clients rate-limit-before-scope ordering fix. cap = consumer.rate_limit_per_minute or settings.DIRECTORY_API_RATE_LIMIT_PER_MINUTE rl_key = f"dirapi_ratelimit:{consumer.id}" count = await _fixed_window_incr(rl_key, 60) if count is not None and count > cap: retry_after = await _window_ttl(rl_key, 60) raise AppException( code="RATE_LIMITED", message=f"Rate limit exceeded. Try again in {retry_after}s.", status_code=429, headers={"Retry-After": str(retry_after)}, ) missing = [s for s in required_scopes if s not in scopes] if missing: # Scope failures are 403 (the credential is real, the grant # isn't) and are audited with the consumer named — this is # the trail that catches a consumer probing beyond its grant. try: from app.audit.service import log_denied_action await log_denied_action( trace_id=getattr(request.state, "trace_id", None) or str(uuid4()), actor_type="service", action="directory_api.scope_denied", resource_type="identity_consumer", resource_id=str(consumer.id), detail={ "consumer_slug": consumer.slug, "missing_scopes": missing, "path": request.url.path[:200], }, ip_address=_client_ip(request), ) except Exception: # noqa: BLE001 logger.exception("directory-api scope-denied audit failed") raise ForbiddenError( f"Consumer lacks required scope(s): {', '.join(missing)}", ) client_ip = _client_ip(request) await _record_usage( token_id=token_row.id, consumer_id=consumer.id, client_ip=client_ip, ) logger.info( "directory-api request consumer=%s path=%s ip=%s", consumer.slug, request.url.path, client_ip, ) return ConsumerContext( consumer_id=consumer.id, slug=consumer.slug, name=consumer.name, scopes=scopes, granted_fields=granted_fields, filter_config=dict(consumer.filter_config or {}), token_id=token_row.id, token_prefix=token_row.token_prefix, ) return _dependency __all__ = ["ConsumerContext", "audit_pii_read", "require_consumer"]