"""API Gateway — control plane HTTP/gRPC gateway for external access. Provides: - REST API for agent orchestration, task submission, status queries - Authentication & authorization middleware - Request validation and rate limiting - OpenAPI schema generation """ from __future__ import annotations import hashlib import hmac import logging import time from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from enum import Enum from typing import Any, Protocol from uuid import uuid4 logger = logging.getLogger(__name__) # Bucket key used when a request carries neither an authenticated principal nor # a client address. It is a single shared bucket on purpose: an unidentified # caller is throttled rather than trusted to be rare, and the cost of the # choice is bounded (see ``APIGateway._derive_rate_limit_key``). Deliberately # namespaced so it can never collide with a real principal or address. UNIDENTIFIED_RATE_LIMIT_KEY = "unidentified" class HTTPMethod(str, Enum): GET = "GET" POST = "POST" PUT = "PUT" PATCH = "PATCH" DELETE = "DELETE" @dataclass class Route: """API route definition.""" path: str method: HTTPMethod handler: Callable[..., Awaitable[Any]] name: str summary: str = "" tags: list[str] = field(default_factory=list) requires_auth: bool = True rate_limit: int | None = None # requests per minute @dataclass class APIRequest: """Incoming API request. ``client_addr`` is the caller-supplied remote address (when available) used as the rate-limit key for anonymous traffic. ``request_id`` is an opaque correlation id for the request and is deliberately never used as a rate-limit key (a per-request UUID would create a fresh empty bucket each call and make limiting ineffective). """ method: HTTPMethod path: str headers: dict[str, str] query_params: dict[str, str] body: Any path_params: dict[str, str] = field(default_factory=dict) client_addr: str | None = None request_id: str = field(default_factory=lambda: uuid4().hex) @dataclass class APIResponse: """API response.""" status_code: int body: Any headers: dict[str, str] = field(default_factory=dict) request_id: str = "" class AuthProvider(Protocol): """Authentication provider protocol.""" async def authenticate(self, request: APIRequest) -> tuple[str | None, dict[str, Any]]: """ Authenticate a request. Returns (user_id, claims) or (None, {}) if unauthenticated. """ ... class APIGateway: """HTTP API Gateway for CyberBrain26 control plane. Routes requests to registered handlers, enforces auth, validates input, and applies rate limiting. Security model ------------- The gateway is fail-closed by default: * A route with ``requires_auth=True`` is denied (HTTP 401) when no ``auth_provider`` is configured, rather than silently bypassing authentication. Anonymous access is never the implicit fallback when ``requires_auth`` is set; this closes the fail-open bypass where a gateway constructed without a provider would serve every authenticated route to anonymous callers. * Anonymous routes (``requires_auth=False``) are served only when the caller explicitly opts in via ``allow_unauthenticated=True``. This makes public access intentional and documented instead of accidental, so a forgotten provider does not silently expose routes. * Handler exception text is never returned to clients. A generic error plus an opaque ``correlation_id`` (the request id) is returned instead; exception detail is disclosed only when ``debug=True`` is explicitly set. * Rate limiting is keyed on the authenticated principal (``user_id``) or the caller-supplied ``client_addr``; it is never keyed on the per-request ``request_id``. * Rate limiting is never silently skipped. A request that carries neither a principal nor a ``client_addr`` is throttled through one documented shared ``unidentified`` bucket rather than escaping the limiter; see ``_derive_rate_limit_key``. """ def __init__( self, auth_provider: AuthProvider | None = None, default_rate_limit: int = 60, *, allow_unauthenticated: bool = False, debug: bool = False, ) -> None: self._routes: dict[str, Route] = {} self._auth_provider = auth_provider self._default_rate_limit = default_rate_limit # Anonymous routes are opt-in only (default deny) to prevent accidental # public exposure when no auth provider is wired up. self._allow_unauthenticated = allow_unauthenticated # Debug mode is opt-in and off by default; it MAY disclose handler # exception text to callers. Never enable in production. self._debug = debug self._rate_limit_buckets: dict[str, list[float]] = {} def add_route( self, path: str, method: HTTPMethod, handler: Callable[..., Awaitable[Any]], *, name: str | None = None, summary: str = "", tags: list[str] | None = None, requires_auth: bool = True, rate_limit: int | None = None, ) -> Route: """Register a route.""" route_key = f"{method.value}:{path}" route = Route( path=path, method=method, handler=handler, name=name or f"{method.value.lower()}_{path.strip('/').replace('/', '_')}", summary=summary, tags=tags or [], requires_auth=requires_auth, rate_limit=rate_limit or self._default_rate_limit, ) self._routes[route_key] = route logger.info("Registered route %s %s", method.value, path) return route def route(self, path: str, method: HTTPMethod) -> Route | None: """Find route for path/method (supports path params).""" route_key = f"{method.value}:{path}" if route_key in self._routes: return self._routes[route_key] # Try pattern matching for path params for registered_path, route in self._routes.items(): if route.method != method: continue if self._match_path(registered_path.split(":", 1)[1], path): return route return None def _match_path(self, pattern: str, path: str) -> bool: """Simple path parameter matching. Splits both pattern and path on ``/`` and compares segment by segment. Because the segment counts must be equal and every non-parameter segment must match exactly, this avoids the prefix-matching foot-gun where ``/admin`` would erroneously match ``/administrator``. """ pattern_parts = pattern.split("/") path_parts = path.split("/") if len(pattern_parts) != len(path_parts): return False for p_pat, p_path in zip(pattern_parts, path_parts): if p_pat.startswith("{"): continue if p_pat != p_path: return False return True def _extract_path_params(self, pattern: str, path: str) -> dict[str, str]: """Extract path parameters from matched route.""" params = {} pattern_parts = pattern.split("/") path_parts = path.split("/") for p_pat, p_path in zip(pattern_parts, path_parts): if p_pat.startswith("{") and p_pat.endswith("}"): param_name = p_pat[1:-1] params[param_name] = p_path return params async def handle(self, request: APIRequest) -> APIResponse: """Process an incoming request.""" route = self.route(request.path, request.method) if not route: return APIResponse(status_code=404, body={"error": "Not found"}, request_id=request.request_id) # --- Authentication (fail-closed; see class docstring, D1) --- user_id: str | None = None claims: dict[str, Any] = {} if route.requires_auth: # A route that requires auth but has no usable provider must be # denied rather than silently allowing unauthenticated access. if self._auth_provider is None: return APIResponse( status_code=401, body={"error": "Unauthorized"}, request_id=request.request_id, ) user_id, claims = await self._auth_provider.authenticate(request) if user_id is None: return APIResponse( status_code=401, body={"error": "Unauthorized"}, request_id=request.request_id, ) elif not self._allow_unauthenticated: # Anonymous routes are opt-in only; deny unless explicitly allowed. return APIResponse( status_code=401, body={"error": "Unauthorized"}, request_id=request.request_id, ) # --- Rate limiting (keyed on principal, never per-request UUID; D3) --- # The key is always derivable: an absent principal and an absent # ``client_addr`` resolve to the shared ``unidentified`` bucket rather # than to "no key", because a falsy key used to short-circuit this # condition and disable rate limiting entirely for exactly the # unauthenticated traffic that needs it most. rate_key = self._derive_rate_limit_key(user_id, request.client_addr) if route.rate_limit and not self._check_rate_limit(rate_key, route.rate_limit): return APIResponse( status_code=429, body={"error": "Rate limit exceeded"}, headers={"Retry-After": "60"}, request_id=request.request_id, ) # --- Handler execution --- try: # Inject request context. ``claims`` never contains raw credentials # (see SimpleAPIKeyAuth) so handler logging cannot leak tokens. context = { "request": request, "user_id": user_id, "claims": claims, "path_params": self._extract_path_params(route.path, request.path), } result = await route.handler(**context) if isinstance(result, APIResponse): result.request_id = request.request_id return result return APIResponse(status_code=200, body=result, request_id=request.request_id) except Exception as exc: # Log the full traceback server-side, but never echo it to the # client. Return a generic error plus an opaque correlation id. logger.exception("Handler error for %s %s", request.method.value, request.path) body: dict[str, Any] = { "error": "Internal server error", "correlation_id": request.request_id, } if self._debug: # Opt-in only; never disclose detail by default. body["detail"] = str(exc) return APIResponse(status_code=500, body=body, request_id=request.request_id) def _derive_rate_limit_key(self, user_id: str | None, client_addr: str | None) -> str: """Derive the rate-limit bucket key for a request. The rule lives in one place so it is auditable: the authenticated principal wins, the caller-supplied address is next, and a request that offers neither is attributed to a single shared ``unidentified`` bucket. Every key is namespaced by its kind so a principal named like an address (or vice versa) can never share quota with it. Security rationale: this helper is deliberately total — it never returns ``None`` and never reports "no key". The previous inline ``user_id or request.client_addr`` guard short-circuited the rate-limit check when both were absent, so an unauthenticated request (the ``allow_unauthenticated=True`` path) was served with no bucket at all and no limit could ever fire: fail-open, and fail-open precisely on the floodable path. Trade-off, chosen deliberately: the shared bucket means one unidentified caller can consume the unidentified quota and lock out other unidentified callers for the rest of the window (a denial-of-service on that bucket only). The alternative — having no bucket for such requests — lets one caller exhaust the process itself, and that limit is unbounded and permanent for the lifetime of the gateway. Bounding the damage to a shared quota of the anonymous path is the smaller and reversible failure; authenticated principals keep their own independent buckets and are unaffected either way. Args: user_id: Authenticated principal id, or ``None`` when the request is anonymous or unauthenticated. client_addr: Caller-supplied remote address, or ``None`` when the transport did not supply one. Returns: A non-empty, namespaced bucket key, always safe to use. """ if user_id: return f"principal:{user_id}" if client_addr: return f"addr:{client_addr}" return UNIDENTIFIED_RATE_LIMIT_KEY def _check_rate_limit(self, key: str, limit: int) -> bool: """Simple in-memory sliding window rate limiter. ``key`` MUST be a stable bucket identifier produced by ``_derive_rate_limit_key`` (a principal, a client address, or the shared ``unidentified`` bucket). Keying on a fresh per-request UUID would create a new empty bucket every call and the limit could never trigger. """ now = time.time() window_start = now - 60 # 1 minute window if key not in self._rate_limit_buckets: self._rate_limit_buckets[key] = [] bucket = self._rate_limit_buckets[key] # Remove old entries bucket[:] = [ts for ts in bucket if ts > window_start] if len(bucket) >= limit: return False bucket.append(now) return True def list_routes(self) -> list[Route]: """List all registered routes.""" return list(self._routes.values()) def generate_openapi(self, title: str = "CyberBrain26 API", version: str = "1.0.0") -> dict[str, Any]: """Generate OpenAPI 3.0 schema. The schema reflects only route metadata (summary, tags, security requirements). It never embeds credentials, API keys, or handler claims, so it is safe to expose publicly for documentation. """ paths: dict[str, Any] = {} for route in self._routes.values(): if route.path not in paths: paths[route.path] = {} paths[route.path][route.method.value.lower()] = { "summary": route.summary, "operationId": route.name, "tags": route.tags, "security": [{"bearerAuth": []}] if route.requires_auth else [], "responses": { "200": {"description": "Success"}, "401": {"description": "Unauthorized"}, "429": {"description": "Rate limited"}, "500": {"description": "Internal error"}, }, } return { "openapi": "3.0.0", "info": {"title": title, "version": version}, "paths": paths, "components": { "securitySchemes": { "bearerAuth": {"type": "http", "scheme": "bearer"}, } }, } # Simple API key auth provider for development class SimpleAPIKeyAuth: """Simple API key (bearer token) authentication for development. Security properties: * Plaintext API keys are never retained. The constructor hashes each key with SHA-256 and stores only the digest, so a memory/core dump or an accidental log of the provider cannot disclose live credentials. * Lookup is performed with ``hmac.compare_digest`` instead of a plaintext dict membership test, avoiding the timing side-channel of a plain ``dict.get`` which could leak information about which keys are valid. * Returned ``claims`` never contain the raw bearer token, preventing credential leakage into handler logs or response bodies. Known gaps (documented, intentionally not implemented): keys never expire and there is no rotation support. Production deployments should use a dedicated auth service with short-lived tokens and key rollover. """ def __init__(self, valid_keys: dict[str, str] | None = None) -> None: """Build a provider from a plaintext ``api_key -> user_id`` mapping. Keys are hashed on construction and the plaintext input is discarded; the provider holds only SHA-256 digests internally. """ self._keys: dict[bytes, str] = { hashlib.sha256(api_key.encode("utf-8")).digest(): user_id for api_key, user_id in (valid_keys or {}).items() } async def authenticate(self, request: APIRequest) -> tuple[str | None, dict[str, Any]]: auth_header = request.headers.get("authorization", "") if not auth_header.startswith("Bearer "): return None, {} api_key = auth_header[7:] digest = hashlib.sha256(api_key.encode("utf-8")).digest() user_id: str | None = None for stored_digest, uid in self._keys.items(): if hmac.compare_digest(digest, stored_digest): user_id = uid break if user_id is not None: # Intentionally omit the raw credential from claims so it cannot # surface via handler logging or be echoed in responses. return user_id, {"sub": user_id} return None, {} __all__ = [ "UNIDENTIFIED_RATE_LIMIT_KEY", "APIGateway", "APIRequest", "APIResponse", "AuthProvider", "HTTPMethod", "Route", "SimpleAPIKeyAuth", ]