diff --git a/.env.example b/.env.example index 0ec92fe..30e4fe4 100644 --- a/.env.example +++ b/.env.example @@ -113,6 +113,8 @@ ANTSEED_WALLET_RPC_URL= # ingress /internal/usage surface AND authenticates outbound resolve calls. CONTROL_PLANE_URL= CONTROL_PLANE_INTERNAL_SECRET= +# HTTPS is required for control-plane and trusted router hops. Local dev only: +CP_ALLOW_INSECURE_HTTP=0 # --- funding autonomy (wallet_keeper.py) ------------------------------------- # The keeper reclaims stuck payment channels and tops the escrow up on its own. # It SHIPS DARK: 0 = off (default). Arm it deliberately, from the dashboard diff --git a/auth_proxy.py b/auth_proxy.py index 1a8ffc1..4c6a19b 100644 --- a/auth_proxy.py +++ b/auth_proxy.py @@ -777,6 +777,8 @@ async def _caller_auth_async(token: str | None) -> dict[str, Any]: "storage": "control_plane", "meta": meta} if resolved.tenant_id is not None: out["tenant_id"] = resolved.tenant_id + if resolved.scope_version == 2: + out.update(scope_version=2, project_id=resolved.project_id, environment_id=resolved.environment_id) return out @@ -3942,8 +3944,10 @@ async def proxy(path: str, request: Request) -> Response: "type": "invalid_request_error", "code": "route_required"}}) from route_contract import apply_contract, PreferenceNotAllowed try: + scope = ({"project_id": auth["project_id"], "environment_id": auth["environment_id"], + "key_digest": auth["digest"]} if auth.get("scope_version") == 2 else {}) published_route = await control_plane_client.resolve_route( - auth["tenant_id"], requested_route[6:]) + auth["tenant_id"], requested_route[6:], **scope) payload, published_route = apply_contract(json.loads(body), published_route) body = json.dumps(payload, separators=(",", ":")).encode() except PreferenceNotAllowed as exc: @@ -4006,6 +4010,10 @@ async def proxy(path: str, request: Request) -> Response: "code": "consumer_budget_exhausted", "budget_usd": budget_usd, "spent_usd": spent_usd}}) + if auth.get("tenant_id") is not None and not control_plane_client.trusted_transport_ok(UPSTREAM): + return JSONResponse(status_code=503, content={"error": { + "message": "Trusted router bridge requires HTTPS", "type": "server_error", + "code": "bridge_transport_unavailable"}}) assert _client is not None if not await _capacity_acquire(): _record_reject(reason="router_overloaded", path="/" + path, @@ -4025,6 +4033,7 @@ async def proxy(path: str, request: Request) -> Response: k: v for k, v in request.headers.items() if k.lower() not in {"authorization", "host", "connection", "content-length", "x-llm-router-tenant", "x-internal-secret", + "x-unhardcoded-scope-version", "x-unhardcoded-project", "x-unhardcoded-environment", "x-unhardcoded-route", "x-unhardcoded-revision", "x-unhardcoded-policy-id", "x-unhardcoded-preference"} } headers["x-llm-router-caller"] = caller @@ -4034,6 +4043,10 @@ async def proxy(path: str, request: Request) -> Response: if auth.get("tenant_id") is not None: headers["x-llm-router-tenant"] = str(auth["tenant_id"]) headers["x-internal-secret"] = control_plane_client.CONTROL_PLANE_INTERNAL_SECRET + if auth.get("scope_version") == 2: + headers["x-unhardcoded-scope-version"] = "2" + headers["x-unhardcoded-project"] = str(auth["project_id"]) + headers["x-unhardcoded-environment"] = str(auth["environment_id"]) headers["x-unhardcoded-route"] = requested_route headers["x-unhardcoded-revision"] = str(published_route["revision"]) headers["x-unhardcoded-policy-id"] = published_route["policy_id"] @@ -4071,6 +4084,8 @@ def _finish(): 'route_revision': published_route['revision'], 'policy_id': published_route['policy_id'], 'routing_preference': published_route.get('routing_preference', 'default')} + if auth.get('scope_version') == 2: + decision_trace.update(project_id=auth['project_id'], environment_id=auth['environment_id']) try: _record_request(caller=caller, method=request.method, path="/" + path, status=status, latency_ms=latency_ms, provider=provider, model_family=model_family, served_model_id=served_model_id, served_by=served_by, requested_model=requested_model, session=session_id, tokens_in=tokens_in, tokens_out=tokens_out, tokens_total=tokens_total, tokens_cached=tokens_cached, cost_usd=cost_usd, cost_basis=cost_basis, decision_trace=decision_trace, error_type=error_type, error_code=error_code, error_message=error_message, key_sha256=auth.get("digest")) _metric_request(status, latency_ms) diff --git a/compose.yml b/compose.yml index 8df3ed0..72ed74f 100644 --- a/compose.yml +++ b/compose.yml @@ -30,6 +30,7 @@ services: # Optional external control plane: per-tenant BYO provider credentials # (fetched by tenant id from the trusted x-llm-router-tenant header the # ingress stamps). Both unset -> feature off, platform keys only. + CP_ALLOW_INSECURE_HTTP: ${CP_ALLOW_INSECURE_HTTP:-0} CONTROL_PLANE_URL: ${CONTROL_PLANE_URL:-} CONTROL_PLANE_INTERNAL_SECRET: ${CONTROL_PLANE_INTERNAL_SECRET:-} SAAS_SHARED_PROVIDERS: ${SAAS_SHARED_PROVIDERS:-} @@ -93,6 +94,7 @@ services: # Optional external control plane: consumer-key resolve fallback plus the # /internal/usage metering surface. Both unset -> feature off (/internal/* # answers 404, unknown keys just 401). + CP_ALLOW_INSECURE_HTTP: ${CP_ALLOW_INSECURE_HTTP:-0} CONTROL_PLANE_URL: ${CONTROL_PLANE_URL:-} CONTROL_PLANE_INTERNAL_SECRET: ${CONTROL_PLANE_INTERNAL_SECRET:-} LOG_LEVEL: ${LOG_LEVEL:-INFO} diff --git a/control_plane_client.py b/control_plane_client.py index 1e209e3..48f3204 100644 --- a/control_plane_client.py +++ b/control_plane_client.py @@ -29,6 +29,7 @@ import time from dataclasses import dataclass from typing import Any, Mapping +from urllib.parse import urlsplit import httpx @@ -36,6 +37,7 @@ CONTROL_PLANE_URL = os.getenv("CONTROL_PLANE_URL", "").rstrip("/") CONTROL_PLANE_INTERNAL_SECRET = os.getenv("CONTROL_PLANE_INTERNAL_SECRET", "") +ALLOW_INSECURE_HTTP = os.getenv("CP_ALLOW_INSECURE_HTTP", "0").lower() in {"1", "true", "yes"} RESOLVE_TTL_S = float(os.getenv("CP_RESOLVE_TTL_S", "60")) NEGATIVE_TTL_S = float(os.getenv("CP_NEGATIVE_TTL_S", "15")) RESOLVE_STALE_GRACE_S = float(os.getenv("CP_RESOLVE_STALE_GRACE_S", "300")) @@ -75,6 +77,9 @@ class ResolvedKey: rate_per_min: int | None burst: int | None fetched_at: float # time.monotonic() + scope_version: int = 1 + project_id: int | None = None + environment_id: int | None = None _resolve_cache: dict[str, ResolvedKey] = {} # sha256 hex -> entry (positive AND negative) @@ -87,8 +92,21 @@ class ResolvedKey: _collision_logged: set[str] = set() +def trusted_transport_ok(url: str) -> bool: + """Bridge secrets require TLS; local HTTP requires explicit operator opt-in.""" + try: + parsed = urlsplit(url) + return bool(parsed.hostname and not parsed.username and not parsed.password + and not parsed.query and not parsed.fragment + and (parsed.scheme == "https" or (parsed.scheme == "http" and ALLOW_INSECURE_HTTP))) + except ValueError: + return False + + def _get_client() -> httpx.AsyncClient: global _client + if not trusted_transport_ok(CONTROL_PLANE_URL): + raise httpx.UnsupportedProtocol("Control-plane bridge requires HTTPS") if _client is None: _client = httpx.AsyncClient(timeout=_TIMEOUT) return _client @@ -126,6 +144,15 @@ def _opt_int(value: Any) -> int | None: consumer = str(data.get("consumer") or "").strip() or None active = bool(data.get("active")) and consumer is not None and _opt_int(data.get("tenant_id")) is not None + version = data.get("scope_version", 1) + project_id, environment_id = data.get("project_id"), data.get("environment_id") + if type(version) is not int or version not in (1, 2): + active = False + elif version == 2: + if any(type(value) is not int or value <= 0 for value in (project_id, environment_id)): + active = False + elif project_id is not None or environment_id is not None: + active = False # Never downgrade a partial scoped identity to tenant-wide. return ResolvedKey( active=active, consumer=consumer if active else None, @@ -133,6 +160,9 @@ def _opt_int(value: Any) -> int | None: rate_per_min=_opt_int(data.get("rate_per_min")) if active else None, burst=_opt_int(data.get("burst")) if active else None, fetched_at=time.monotonic(), + scope_version=version if active else 1, + project_id=project_id if active and version == 2 else None, + environment_id=environment_id if active and version == 2 else None, ) @@ -142,7 +172,7 @@ async def _fetch_resolve(digest: str) -> ResolvedKey | None: try: resp = await _get_client().get( f"{CONTROL_PLANE_URL}/internal/keys/resolve", - params={"sha256": digest}, + params={"sha256": digest, "scope_version": "2"}, headers={"x-internal-secret": CONTROL_PLANE_INTERNAL_SECRET}, ) except httpx.HTTPError as exc: @@ -244,24 +274,35 @@ async def tenant_env(tenant_id: int) -> dict[str, str]: return env -async def tenant_connections(tenant_id: int, allowed_env: set[str]) -> tuple[dict, dict]: +async def tenant_connections(tenant_id: int, allowed_env: set[str], *, project_id=None, environment_id=None) -> tuple[dict, dict]: """Fresh closed credential scope and encrypted structured BYO connections. The loaded catalog, not a second hardcoded list, declares credential names. An unavailable control plane grants nothing; no stale authorization.""" if not enabled(): + if project_id is not None or environment_id is not None: + raise RouteUnavailable("Scoped credentials unavailable.") return {}, {} + scoped = project_id is not None or environment_id is not None + if scoped and any(type(value) is not int or value <= 0 for value in (project_id, environment_id)): + raise RouteUnavailable("Invalid credential scope.") + path = (f"/internal/tenants/{tenant_id}/projects/{project_id}/environments/{environment_id}/provider-env" + if scoped else f"/internal/tenants/{int(tenant_id)}/provider-env") try: response = await _get_client().get( - f"{CONTROL_PLANE_URL}/internal/tenants/{int(tenant_id)}/provider-env", + f"{CONTROL_PLANE_URL}{path}", headers={"x-internal-secret": CONTROL_PLANE_INTERNAL_SECRET}) response.raise_for_status() data = response.json() + if scoped: + _validate_scope(data, tenant_id, project_id, environment_id) raw, connections = data.get('env'), data.get('connections', {}) if not isinstance(raw, dict) or not isinstance(connections, dict): raise ValueError('invalid connection scope') return ({k: v for k, v in raw.items() if k in allowed_env and isinstance(v, str) and v}, {p: c for p, c in connections.items() if p in {'bedrock', 'antseed'} and isinstance(c, dict)}) - except (httpx.HTTPError, ValueError, AttributeError): + except (httpx.HTTPError, ValueError, AttributeError) as exc: + if scoped: + raise RouteUnavailable("Scoped credentials unavailable.") from exc return {}, {} @@ -285,15 +326,33 @@ class RouteUnavailable(RuntimeError): pass -async def resolve_route(tenant_id: int, name: str) -> dict: +def _validate_scope(data, tenant_id, project_id, environment_id): + expected = {"scope_version": 2, "tenant_id": tenant_id, + "project_id": project_id, "environment_id": environment_id} + if not isinstance(data, dict) or any(type(data.get(k)) is not int or data[k] != v for k, v in expected.items()): + raise ValueError("Mismatched project/environment scope") + + +async def resolve_route(tenant_id: int, name: str, *, project_id=None, environment_id=None, key_digest=None) -> dict: """Resolve the published revision on every call, so publish/pause is immediate.""" try: + scoped = project_id is not None or environment_id is not None + if scoped: + if (any(type(value) is not int or value <= 0 for value in (project_id, environment_id)) + or not isinstance(key_digest, str) or len(key_digest) != 64): + raise ValueError("Invalid route scope") + path = f"/internal/tenants/{tenant_id}/projects/{project_id}/environments/{environment_id}/routes/{name}" + else: + path = f"/internal/tenants/{tenant_id}/routes/{name}" response = await _get_client().get( - f"{CONTROL_PLANE_URL}/internal/tenants/{tenant_id}/routes/{name}", + f"{CONTROL_PLANE_URL}{path}", + params={"key_sha256": key_digest} if scoped else None, headers={"x-internal-secret": CONTROL_PLANE_INTERNAL_SECRET}, ) response.raise_for_status() data = response.json() + if scoped: + _validate_scope(data, tenant_id, project_id, environment_id) if (not isinstance(data.get("policy_ir"), list) or not isinstance(data.get("policy_id"), str) or len(data["policy_id"]) != 64 or not isinstance(data.get("revision"), int) or data["revision"] <= 0 diff --git a/docs/saas-project-scopes.md b/docs/saas-project-scopes.md new file mode 100644 index 0000000..6483ab9 --- /dev/null +++ b/docs/saas-project-scopes.md @@ -0,0 +1,86 @@ +# SaaS project/environment bridge (version 2) + +This extends the existing control-plane bridge; it does not introduce a router, +policy algebra or evaluator. Legacy operator authentication and non-tenant calls +are unchanged. The engine submodule is unchanged. + +## Authentication and scope + +The ingress advertises `scope_version=2` when resolving a consumer key. A scoped +response must contain positive integer `project_id` and `environment_id`, with +`scope_version: 2` and the existing `tenant_id`. Partial or unknown versions are +inactive, never downgraded to tenant-wide access. Old control planes that return +the original tenant-only response remain supported. + +The cloud must reject scoped keys when this capability is absent. Deploy this +dataplane before activating scoped keys in cloud; an old ingress cannot safely +serve a scoped key. + +On every inference request the ingress resolves: + +```text +GET /internal/tenants/{tenant}/projects/{project}/environments/{environment}/routes/{name} + ?key_sha256={authenticated_key_digest} +``` + +The cloud checks that the key still exists, is unrevoked/unexpired and belongs to +that exact scope, then resolves the active unpaused revision. This check is not +cached: stale ingress authentication cannot bypass key revocation. Requests that +already passed authorization may finish; revocation is not cancellation. + +The ingress strips client-supplied scope headers and stamps only resolved values: +`x-unhardcoded-scope-version`, `x-unhardcoded-project`, +`x-unhardcoded-environment`, plus the existing trusted tenant/internal headers. +Chat Completions and Responses follow the same contract resolver and Lua engine. + +The shim validates the scope and fetches fresh credentials from the corresponding +`.../provider-env` endpoint. Both route and credential responses must echo the +matching version, tenant, project and environment. Missing/mismatched scopes or +control-plane failures reject scoped requests; no legacy credential fallback is +allowed. Sessions and provider-discovery caches include the environment identity. +The bounded routing summary records project/environment IDs, not secrets. + +## Rollout boundary + +This protocol is only one delivery block. The cloud must also enforce project +permissions in the dashboard, assign all issued keys/routes to environments and +restrict provider assignments by development/production grants before claiming +complete environment isolation. Do not expose a UI-only project boundary. + +Cloud migration and deployment need a coordinated SaaS-only write pause so an old +control-plane instance cannot keep editing the original credential tables after +their encrypted blobs are migrated. Preserve originals for recovery; do not roll +back across subsequent project configuration changes without a validated restore +plan. No production deployment is authorized by these changes. + +## Scoped usage and activity + +The internal `/internal/usage` and `/internal/usage/recent` endpoints accept +`project_id` and `environment_id` together with the required caller slug. Partial +or nonpositive scopes are rejected. PostgreSQL applies all three filters before +aggregation or the recent-call limit; historical rows without a scope do not +appear in environment reports. Responses echo `scope_version: 2` and both IDs so +Cloud can reject an old ingress that ignores the new filters. Database errors +return 503; reads run off the ingress event loop and have a statement deadline. +The ledger is best-effort telemetry, not billing-grade accounting. + +The companion Cloud #3 now wires scope through the entire dashboard. It includes +an opt-in two-process test using the actual Django bridge and this dataplane, +with concurrent Chat Completions/Responses, streaming/fallback, scope forgery, +revoked cached keys and removed connection assignments. Provider calls are fake; +this verifies isolation/recovery rather than a production throughput target. + +## Bridge transport + +HTTPS is required by default for all control-plane HTTP requests and the ingress +hop that carries trusted tenant headers to the shim. TLS certificate verification +remains enabled. Configure trusted internal certificates (including a CA bundle +where needed); do not expose the shim or internal control-plane endpoints publicly. + +`CP_ALLOW_INSECURE_HTTP=1` is an explicit exception for isolated local development +or an operator-managed transport boundary. It does not add encryption. The local +Compose/test fixtures set it deliberately; SaaS production must leave it disabled. +Cloud's production settings also reject plaintext router/ingress URLs. Existing +operator-only deployments are unaffected because their control-plane integration +is disabled. Existing HTTP control-plane deployments must configure HTTPS or +explicitly opt into their existing insecure transport before adopting this release. diff --git a/host_store.py b/host_store.py index fc1d99c..6d8fa4e 100644 --- a/host_store.py +++ b/host_store.py @@ -386,7 +386,7 @@ def routing_summary(trace) -> dict | None: or the full candidate catalog in the tenant activity ledger.""" if not isinstance(trace, dict): return None - summary = {key: str(trace[key])[:100] for key in ('route', 'route_revision', 'policy_id', 'routing_preference') + summary = {key: str(trace[key])[:100] for key in ('route', 'route_revision', 'policy_id', 'routing_preference', 'project_id', 'environment_id') if trace.get(key) is not None} summary['attempts'] = [ {key: str(step[key])[:160] for key in ('provider_id', 'model_family', 'error_kind') @@ -529,18 +529,24 @@ def observe_route_call_async(row: dict[str, Any]) -> None: _enqueue(lambda: _insert_route_observation(snap)) -def recent_calls(limit: int = 100, caller: "str | None" = None) -> list[dict[str, Any]]: +def recent_calls(limit: int = 100, caller: "str | None" = None, *, project_id=None, environment_id=None, strict=False) -> list[dict[str, Any]]: """The most recent calls, newest first (operator view / verification). Optionally scoped to one caller (control-plane activity feed).""" try: where, params = ("", []) if caller is None else (" WHERE caller = %s", [caller]) + if project_id is not None or environment_id is not None: + where, params = _usage_where(caller=caller, project_id=project_id, environment_id=environment_id) with _get_pool().connection() as conn: + if strict: + _set_dashboard_statement_timeout(conn) with conn.cursor(row_factory=dict_row) as cur: cur.execute(f"SELECT * FROM calls{where} ORDER BY id DESC LIMIT %s", params + [int(limit)]) return cur.fetchall() except Exception as exc: # noqa: BLE001 _log.warning("host_store recent_calls failed: %s", exc) + if strict: + raise return [] @@ -1267,7 +1273,7 @@ def cost_by_route(window_s: int = 86_400) -> list[dict[str, Any]]: def _usage_where(since_ts: "int | None" = None, caller: "str | None" = None, caller_is_null: bool = False, consumer_sha: "str | None" = None, provider: "str | None" = None, - model_family: "str | None" = None) -> "tuple[str, list[Any]]": + model_family: "str | None" = None, project_id=None, environment_id=None) -> "tuple[str, list[Any]]": """The shared WHERE for every calls-derived usage read. "all" (since_ts=None) is still bounded to the retention horizon so the read is ALWAYS time-bounded (never a bare table scan): rows older than retention are pruned anyway, so the @@ -1290,6 +1296,11 @@ def _usage_where(since_ts: "int | None" = None, caller: "str | None" = None, clauses.append("provider_id = %s"); params.append(provider) if model_family is not None: clauses.append("model_family = %s"); params.append(model_family) + if project_id is not None or environment_id is not None: + if not caller or any(type(v) is not int or v <= 0 for v in (project_id, environment_id)): + raise ValueError("A caller and complete positive project/environment scope are required") + clauses.extend(["routing_summary->>'project_id' = %s", "routing_summary->>'environment_id' = %s"]) + params.extend([str(project_id), str(environment_id)]) return " WHERE " + " AND ".join(clauses), params @@ -1425,7 +1436,7 @@ def _empty_usage_aggregate() -> dict[str, Any]: def usage_aggregate(since_ts: "int | None" = None, caller: "str | None" = None, caller_is_null: bool = False, consumer_sha: "str | None" = None, provider: "str | None" = None, - model_family: "str | None" = None) -> dict[str, Any]: + model_family: "str | None" = None, *, project_id=None, environment_id=None, strict=False) -> dict[str, Any]: """Every dashboard stats aggregate in ONE window scan: overall totals plus the by_caller / by_provider / by_model_family / by_route / by_served_model / by_status / by_day breakdowns, as raw counters (see _agg_counter). Fail-soft @@ -1434,7 +1445,7 @@ def usage_aggregate(since_ts: "int | None" = None, caller: "str | None" = None, where, params = _usage_where(since_ts=since_ts, caller=caller, caller_is_null=caller_is_null, consumer_sha=consumer_sha, provider=provider, - model_family=model_family) + model_family=model_family, project_id=project_id, environment_id=environment_id) sql = ("SELECT grouping(caller_k, provider_k, family_k, route_k," " served_k, status_k, day_k) AS gset," " caller_k, provider_k, family_k, route_k, served_k, status_k," @@ -1463,18 +1474,20 @@ def usage_aggregate(since_ts: "int | None" = None, caller: "str | None" = None, return out except Exception as exc: # noqa: BLE001 _log.warning("host_store usage_aggregate failed: %s", exc) + if strict: + raise return _empty_usage_aggregate() def usage_totals(since_ts: "int | None" = None, - caller: "str | None" = None) -> dict[str, Any]: + caller: "str | None" = None, *, project_id=None, environment_id=None, strict=False) -> dict[str, Any]: """One-row window totals over `calls`, including cached tokens (which the dashboard aggregate doesn't sum) — the control-plane metering read. Fail-soft -> zeros (same keys).""" zeros = {"requests": 0, "errors": 0, "tokens_in": 0, "tokens_out": 0, "tokens_cached": 0, "tokens_total": 0, "cost_usd": 0.0, "priced": 0} try: - where, params = _usage_where(since_ts=since_ts, caller=caller) + where, params = _usage_where(since_ts=since_ts, caller=caller, project_id=project_id, environment_id=environment_id) sql = ( "SELECT count(*)," " count(*) FILTER (WHERE COALESCE(status,0) >= 400)," @@ -1487,6 +1500,7 @@ def usage_totals(since_ts: "int | None" = None, " count(cost_usd)" f" FROM calls{where}") with _get_pool().connection() as conn: + _set_dashboard_statement_timeout(conn) row = conn.execute(sql, params).fetchone() requests, errors, tin, tout, tcached, ttotal, cost, priced = row return {"requests": int(requests), "errors": int(errors), @@ -1495,6 +1509,8 @@ def usage_totals(since_ts: "int | None" = None, "cost_usd": float(cost), "priced": int(priced)} except Exception as exc: # noqa: BLE001 _log.warning("host_store usage_totals failed: %s", exc) + if strict: + raise return zeros diff --git a/internal_api.py b/internal_api.py index 9375a18..939c3d0 100644 --- a/internal_api.py +++ b/internal_api.py @@ -14,6 +14,7 @@ from __future__ import annotations import time +import asyncio from typing import Any from fastapi import APIRouter, Request @@ -38,16 +39,27 @@ def _gate(request: Request) -> JSONResponse | None: @router.get("/internal/usage") async def internal_usage(request: Request, caller: str = "", since_ts: int | None = None, - bucket: str | None = None) -> JSONResponse: + bucket: str | None = None, + project_id: int | None = None, environment_id: int | None = None) -> JSONResponse: denied = _gate(request) if denied is not None: return denied + scope = {} + if project_id is not None or environment_id is not None: + if project_id is None or environment_id is None or min(project_id, environment_id) <= 0: + return JSONResponse({"error": "invalid_scope"}, status_code=400) + scope = {"project_id": project_id, "environment_id": environment_id} caller = caller.strip() if not caller: return JSONResponse({"error": "caller_required"}, status_code=400) - totals = host_store.usage_totals(since_ts=since_ts, caller=caller) + try: + totals = await asyncio.to_thread(host_store.usage_totals, since_ts=since_ts, caller=caller, strict=True, **scope) + by_day = (await asyncio.to_thread(host_store.usage_aggregate, since_ts=since_ts, caller=caller, strict=True, **scope))["by_day"] if bucket == "day" else {} + except Exception: + return JSONResponse({"error": "ledger_unavailable"}, status_code=503) out: dict[str, Any] = { "caller": caller, + **({"scope_version": 2, **scope} if scope else {}), "window": {"since_ts": since_ts, "until_ts": int(time.time())}, "runs": totals["requests"], "errors": totals["errors"], @@ -58,7 +70,6 @@ async def internal_usage(request: Request, caller: str = "", "cost_usd": totals["cost_usd"], } if bucket == "day": - by_day = host_store.usage_aggregate(since_ts=since_ts, caller=caller)["by_day"] out["buckets"] = [ {"date": day, "runs": counter["requests"], "cost_usd": counter["cost_usd"]} for day, counter in sorted(by_day.items()) @@ -68,16 +79,26 @@ async def internal_usage(request: Request, caller: str = "", @router.get("/internal/usage/recent") async def internal_usage_recent(request: Request, caller: str = "", - limit: int = 50) -> JSONResponse: + limit: int = 50, project_id: int | None = None, + environment_id: int | None = None) -> JSONResponse: denied = _gate(request) if denied is not None: return denied + scope = {} + if project_id is not None or environment_id is not None: + if project_id is None or environment_id is None or min(project_id, environment_id) <= 0: + return JSONResponse({"error": "invalid_scope"}, status_code=400) + scope = {"project_id": project_id, "environment_id": environment_id} caller = caller.strip() if not caller: return JSONResponse({"error": "caller_required"}, status_code=400) limit = max(1, min(int(limit), _RECENT_LIMIT_MAX)) calls = [] - for row in host_store.recent_calls(limit=limit, caller=caller): + try: + rows = await asyncio.to_thread(host_store.recent_calls, limit=limit, caller=caller, strict=True, **scope) + except Exception: + return JSONResponse({"error": "ledger_unavailable"}, status_code=503) + for row in rows: calls.append({ "ts": row.get("ts"), "status": row.get("status"), @@ -95,4 +116,4 @@ async def internal_usage_recent(request: Request, caller: str = "", "routing_summary": row.get("routing_summary"), "key_sha256_prefix": (row.get("consumer_sha") or "")[:12] or None, }) - return JSONResponse({"caller": caller, "calls": calls}) + return JSONResponse({"caller": caller, "calls": calls, **({"scope_version": 2, **scope} if scope else {})}) diff --git a/llm_router_host.py b/llm_router_host.py index a2bc6f0..ea96c1c 100644 --- a/llm_router_host.py +++ b/llm_router_host.py @@ -256,7 +256,7 @@ def normalize_policy(self, policy_ir: list, *, admit: bool = False) -> dict: "policy_id": hashlib.sha256(encoded.encode()).hexdigest(), } - def for_tenant(self, tenant_id: int, env: dict, managed_providers=(), connections=None): + def for_tenant(self, tenant_id: int, env: dict, managed_providers=(), connections=None, *, project_id=None, environment_id=None): """A request-local engine: credential failures cannot disable other tenants. Reuses the same engine, catalog files, live discovery and HTTP adapters. @@ -268,6 +268,8 @@ def for_tenant(self, tenant_id: int, env: dict, managed_providers=(), connection managed = set(managed_providers).intersection(shared_provider_ids(catalog)) scoped_env = dict(env) scoped_env['SAAS_TENANT_SCOPE'] = str(tenant_id) + if environment_id is not None: + scoped_env['SAAS_TENANT_SCOPE'] = f"{tenant_id}:{project_id}:{environment_id}" allowed = set(managed) for pid, provider in (catalog.get('providers') or {}).items(): key = auth_env(provider) @@ -289,6 +291,8 @@ def for_tenant(self, tenant_id: int, env: dict, managed_providers=(), connection child.config.providers = _to_lua(child.lua, catalog.get('providers') or {}) child.config.models = _to_lua(child.lua, catalog.get('models') or {}) child._tenant_id = tenant_id + child._project_id = project_id + child._environment_id = environment_id child._tenant_allowed = allowed child._tenant_managed = managed child._tenant_connections = byo diff --git a/saas_routes.py b/saas_routes.py index 4bf7bbe..90889ba 100644 --- a/saas_routes.py +++ b/saas_routes.py @@ -27,7 +27,8 @@ def __getattr__(self, name): async def execute_async(self, contract, **kwargs): host = _active.get() or self.base if host._tenant_id is not None and contract.get("session"): - contract = {**contract, "session": f"tenant:{host._tenant_id}:{contract['session']}"} + scope_id = host._env.get('SAAS_TENANT_SCOPE', str(host._tenant_id)) + contract = {**contract, "session": f"tenant:{scope_id}:{contract['session']}"} result = await host.execute_async(contract, **kwargs) if revision := _revision.get(): result.setdefault("trace", {}).update(revision) @@ -168,7 +169,11 @@ def install(app, scoped, handle_chat, chat_request): @app.middleware("http") async def tenant_context(request: Request, call_next): raw = request.headers.get("x-llm-router-tenant") + scope_headers = [request.headers.get(name) for name in ( + "x-unhardcoded-scope-version", "x-unhardcoded-project", "x-unhardcoded-environment")] if raw is None: + if any(value is not None for value in scope_headers): + return JSONResponse({"error": {"message": "Scope requires trusted tenant context"}}, status_code=403) return await call_next(request) if not cp.enabled() or not cp.internal_secret_ok(request.headers): return JSONResponse({"error": {"message": "Untrusted tenant context"}}, status_code=403) @@ -178,13 +183,27 @@ async def tenant_context(request: Request, call_next): raise ValueError() except ValueError: return JSONResponse({"error": {"message": "Invalid tenant"}}, status_code=400) + scope = {} + if any(value is not None for value in scope_headers): + try: + version, project, environment = scope_headers + if version != "2" or any(not value or not value.isascii() or not value.isdigit() for value in (project, environment)): + raise ValueError() + scope = {"project_id": int(project), "environment_id": int(environment)} + if min(scope.values()) <= 0: + raise ValueError() + except (ValueError, TypeError): + return JSONResponse({"error": {"message": "Invalid project/environment scope"}}, status_code=400) allowed = {"/v1/chat/completions", "/v1/responses", "/x/saas/catalog", "/x/saas/preview", "/x/saas/test", "/x/saas/connections"} if request.url.path not in allowed: return JSONResponse({"error": {"message": "Unsupported tenant endpoint"}}, status_code=404) # Read current credentials for each request: revocation/rotation is immediate. # The bounded request-local VM cannot retain stale tenant secrets/health. - env, connections_config = await cp.tenant_connections(tenant_id, credential_names(scoped.base.catalog())) - child = await asyncio.to_thread(scoped.base.for_tenant, tenant_id, env, (), connections_config) + try: + env, connections_config = await cp.tenant_connections(tenant_id, credential_names(scoped.base.catalog()), **scope) + except cp.RouteUnavailable: + return JSONResponse({"error": {"message": "Scoped credentials unavailable"}}, status_code=503) + child = await asyncio.to_thread(scoped.base.for_tenant, tenant_id, env, (), connections_config, **scope) if request.url.path != '/x/saas/connections': from tenant_providers import prepare await prepare(child) @@ -196,6 +215,7 @@ async def tenant_context(request: Request, call_next): "route_revision": request.headers.get("x-unhardcoded-revision"), "policy_id": request.headers.get("x-unhardcoded-policy-id"), 'routing_preference': request.headers.get('x-unhardcoded-preference'), + **scope, }) try: return await call_next(request) diff --git a/tenant_providers.py b/tenant_providers.py index e334518..3113eab 100644 --- a/tenant_providers.py +++ b/tenant_providers.py @@ -109,7 +109,7 @@ async def one(name, load, ids): # operator source after timeout, malformed data or revoked credentials. host._tenant_offers.update({pid: [] for pid in ids}) digest = hashlib.sha256(json.dumps(configs[name], sort_keys=True).encode()).hexdigest() - key = (host._tenant_id, name, digest) + key = (host._tenant_id, getattr(host, '_environment_id', None), name, digest) cached = _cache.get(key) try: if cached and time.monotonic() - cached[0] < (15 if name == 'antseed' else 300): diff --git a/tests/conftest.py b/tests/conftest.py index 32e93f8..8d0ffc8 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -42,6 +42,9 @@ # waiting on psycopg_pool's 30s default connection checkout timeout. os.environ.setdefault("HOST_STORE_POOL_TIMEOUT", "1") +# Test HTTP endpoints are local/fake; production requires TLS by default. +os.environ["CP_ALLOW_INSECURE_HTTP"] = "1" + import host_store # noqa: E402 _HOST_STORE_UNAVAILABLE: str | None = None diff --git a/tests/test_auth_proxy_control_plane.py b/tests/test_auth_proxy_control_plane.py index 83abe53..8f87453 100644 --- a/tests/test_auth_proxy_control_plane.py +++ b/tests/test_auth_proxy_control_plane.py @@ -276,7 +276,8 @@ async def unavailable(*args): @pytest.mark.parametrize('path', ['/v1/chat/completions', '/v1/responses']) -def test_full_ingress_to_engine_alias_and_failover(monkeypatch, path): +@pytest.mark.parametrize('scoped', [False, True]) +def test_full_ingress_to_engine_alias_and_failover(monkeypatch, path, scoped): """Actual ingress + shim + Lua, with only CP HTTP and providers faked.""" require_host_store() import asyncio @@ -293,7 +294,12 @@ async def call(request): return {'ok':False, 'error_kind':'server_error'} return {'ok':True, 'latency_ms':10, 'response':{'text':'Fallback works', 'tokens_in':2, 'tokens_out':3}} host.set_async_call_hook(call) + scope = {'scope_version': 2, 'tenant_id': 7, 'project_id': 8, 'environment_id': 9} if scoped else {} def cp_http(request): + if scoped and ('/provider-env' in request.url.path or '/routes/' in request.url.path): + assert '/projects/8/environments/9/' in request.url.path + if scoped and '/routes/' in request.url.path: + assert request.url.params['key_sha256'] == hashlib.sha256(b'full-chain').hexdigest() if request.url.path.endswith('/provider-env'): data = {'env':{'OPENAI_API_KEY':'sk-tenant', 'ANTHROPIC_API_KEY':'sk-backup'}} elif '/routes/' in request.url.path: @@ -301,18 +307,24 @@ def cp_http(request): 'revision':3, 'execution':{'timeout_ms':8000}, 'route':'route:production'} else: data = {'active':True, 'consumer':'acme', 'tenant_id':7} - return httpx.Response(200, json=data) + return httpx.Response(200, json={**data, **scope}) cp_client = httpx.AsyncClient(transport=httpx.MockTransport(cp_http)) shim_client = httpx.AsyncClient(transport=httpx.ASGITransport(app=create_app(host)), base_url='http://router.test') monkeypatch.setattr(cpc, '_client', cp_client) monkeypatch.setattr(auth_proxy, '_client', shim_client) body = {'model':'route:production', 'messages':[{'role':'user','content':'hi'}]} if 'chat' in path else {'model':'route:production','input':'hi'} try: - r = TestClient(auth_proxy.app).post(path, headers={'Authorization':'Bearer full-chain'}, json=body) + r = TestClient(auth_proxy.app).post(path, headers={'Authorization':'Bearer full-chain', + 'x-unhardcoded-scope-version': '2', 'x-unhardcoded-project': '999', + 'x-unhardcoded-environment': '999'}, json=body) assert r.status_code == 200, r.text trace = r.json()['x_router']['decision_trace'] assert trace['route_revision'] == '3' assert trace['route'] == 'route:production' + if scoped: + assert trace['project_id'] == 8 and trace['environment_id'] == 9 + else: + assert 'project_id' not in trace and 'environment_id' not in trace assert [s['provider_id'] for s in trace['decision_path'] if s['event'] == 'attempted'] == ['openai', 'anthropic'] assert seen == [('openai','sk-tenant'), ('anthropic','sk-tenant')] host_store._write_q.join() @@ -399,3 +411,61 @@ def test_internal_usage_is_not_proxied_upstream(monkeypatch): headers={"x-internal-secret": "s3cret"}) assert r.status_code == 400 # caller_required, answered locally assert upstream.requests == [] + + +def test_scoped_usage_filters_before_aggregate_and_recent_limit(): + require_host_store() + now = int(time.time()) + for caller, project, env, tokens in [('acme', 1, 10, 7), ('acme', 1, 11, 90), + ('acme', 2, 10, 800), ('other', 1, 10, 900), + ('acme', None, None, 1000)]: + trace = {'project_id': project, 'environment_id': env} if project else None + host_store.insert_call({'ts': now, 'caller': caller, 'status': 200, 'tokens_in': tokens, + 'decision_trace': trace, 'cost_usd': tokens / 1000}) + client = TestClient(auth_proxy.app) + headers = {'x-internal-secret': 's3cret'} + scope = {'caller': 'acme', 'project_id': 1, 'environment_id': 10} + response = client.get('/internal/usage', params={**scope, 'bucket': 'day'}, headers=headers) + assert response.status_code == 200 + data = response.json() + assert data['runs'] == 1 and data['tokens_in'] == 7 + assert data['cost_usd'] == pytest.approx(.007) + assert sum(b['runs'] for b in data['buckets']) == 1 + assert {k: data[k] for k in scope} == scope + assert data['scope_version'] == 2 + response = client.get('/internal/usage/recent', params={**scope, 'limit': 1}, headers=headers) + assert response.status_code == 200 + assert response.json()['calls'][0]['tokens_in'] == 7 + for path in ('/internal/usage', '/internal/usage/recent'): + for invalid in ({'project_id': 1}, {'environment_id': 10}, {'project_id': 0, 'environment_id': 10}): + assert client.get(path, params={'caller': 'acme', **invalid}, headers=headers).status_code == 400 + + +def test_scoped_metering_outage_is_unavailable_not_successful_zero(monkeypatch): + def unavailable(*args, **kwargs): + raise RuntimeError('database unavailable') + monkeypatch.setattr(host_store, '_get_pool', unavailable) + client = TestClient(auth_proxy.app) + for path in ('/internal/usage', '/internal/usage/recent'): + result = client.get(path, params={'caller': 'acme', 'project_id': 1, 'environment_id': 2}, + headers={'x-internal-secret': 's3cret'}) + assert result.status_code == 503 + assert result.json() == {'error': 'ledger_unavailable'} + + +def test_plaintext_upstream_never_receives_trusted_scope_secret(monkeypatch): + _upstream(monkeypatch) + monkeypatch.setattr(cpc, 'ALLOW_INSECURE_HTTP', False) + monkeypatch.setattr(auth_proxy, 'UPSTREAM', 'http://router.test') + async def auth(token): + return {'ok': True, 'caller': 'transport-fixture', 'tenant_id': 1, 'digest': 'a' * 64, 'meta': {}} + async def route(*args, **kwargs): + return {'route': 'route:production', 'revision': 1, 'policy_id': 'a' * 64, + 'policy_ir': ['policy'], 'execution': {}} + monkeypatch.setattr(auth_proxy, '_caller_auth_async', auth) + monkeypatch.setattr(cpc, 'resolve_route', route) + monkeypatch.setattr(auth_proxy, '_rate_ok', lambda *args: True) + result = _post_chat(TestClient(auth_proxy.app), 'fixture') + assert result.status_code == 503 + assert result.json()['error']['code'] == 'bridge_transport_unavailable' + assert auth_proxy._client.requests == [] diff --git a/tests/test_control_plane_client.py b/tests/test_control_plane_client.py index 4cd1ab4..7677c81 100644 --- a/tests/test_control_plane_client.py +++ b/tests/test_control_plane_client.py @@ -94,10 +94,52 @@ def test_resolve_fetches_then_serves_from_cache(monkeypatch): assert len(fake.calls) == 1 call = fake.calls[0] assert call["url"].endswith("/internal/keys/resolve") - assert call["params"] == {"sha256": DIGEST} + assert call["params"] == {"sha256": DIGEST, "scope_version": "2"} assert call["headers"] == {"x-internal-secret": "s3cret"} +@pytest.mark.parametrize('extra', [ + {'scope_version': 2, 'project_id': 4}, + {'scope_version': 2, 'project_id': 4, 'environment_id': 0}, + {'scope_version': 2, 'project_id': True, 'environment_id': 9}, + {'scope_version': 2, 'project_id': '4', 'environment_id': 9}, + {'scope_version': 3, 'project_id': 4, 'environment_id': 9}, + {'project_id': 4, 'environment_id': 9}, +]) +def test_partial_unknown_or_malformed_scopes_never_downgrade(extra): + assert not cpc._parse_resolved({**_active(), **extra}).active + + +def test_scoped_identity_survives_cache(monkeypatch): + _install(monkeypatch, [_FakeResp(200, {**_active(), 'scope_version': 2, 'project_id': 4, 'environment_id': 9})]) + first = asyncio.run(cpc.resolve_key(DIGEST)) + assert first.active and first.scope_version == 2 + assert first.project_id == 4 and first.environment_id == 9 + assert asyncio.run(cpc.resolve_key(DIGEST)) is first + + +def test_scoped_credentials_require_echoed_matching_scope(monkeypatch): + identity = {'scope_version': 2, 'tenant_id': 7, 'project_id': 4, 'environment_id': 9} + fake = _install(monkeypatch, [_FakeResp(200, {**identity, 'env': {'OPENAI_API_KEY': 'private'}, 'connections': {}})]) + assert asyncio.run(cpc.tenant_connections(7, {'OPENAI_API_KEY'}, project_id=4, environment_id=9))[0] == {'OPENAI_API_KEY': 'private'} + assert '/projects/4/environments/9/provider-env' in fake.calls[0]['url'] + for extra in ({'environment_id': 10}, {'scope_version': 1}, {'project_id': '4'}): + _install(monkeypatch, [_FakeResp(200, {**identity, **extra, 'env': {}, 'connections': {}})]) + with pytest.raises(cpc.RouteUnavailable): + asyncio.run(cpc.tenant_connections(7, set(), project_id=4, environment_id=9)) + + +def test_scoped_route_rechecks_key_on_every_request(monkeypatch): + identity = {'scope_version': 2, 'tenant_id': 7, 'project_id': 4, 'environment_id': 9} + contract = {**identity, 'policy_ir': ['policy'], 'policy_id': 'a'*64, 'revision': 1, 'execution': {}} + fake = _install(monkeypatch, [_FakeResp(200, contract), _FakeResp(403, {})]) + assert asyncio.run(cpc.resolve_route(7, 'assistant', project_id=4, environment_id=9, key_digest=DIGEST))['revision'] == 1 + with pytest.raises(cpc.RouteUnavailable): + asyncio.run(cpc.resolve_route(7, 'assistant', project_id=4, environment_id=9, key_digest=DIGEST)) + assert len(fake.calls) == 2 + assert all(call['params'] == {'key_sha256': DIGEST} for call in fake.calls) + + def test_negative_answer_is_cached(monkeypatch): fake = _install(monkeypatch, [_FakeResp(200, {"active": False})]) first = asyncio.run(cpc.resolve_key(DIGEST)) @@ -248,3 +290,27 @@ def test_internal_secret_ok(monkeypatch): assert cpc.internal_secret_ok({}) is False monkeypatch.setattr(cpc, "CONTROL_PLANE_INTERNAL_SECRET", "") assert cpc.internal_secret_ok({"x-internal-secret": ""}) is False + + +def test_trusted_transport_requires_tls_without_explicit_local_opt_in(monkeypatch): + monkeypatch.setattr(cpc, 'ALLOW_INSECURE_HTTP', False) + assert cpc.trusted_transport_ok('https://router.internal:18080') + for url in ('http://router:18080', 'ftp://router', 'https://user:secret@router', 'https:///path', 'https://router?secret=x'): + assert not cpc.trusted_transport_ok(url) + monkeypatch.setattr(cpc, 'ALLOW_INSECURE_HTTP', True) + assert cpc.trusted_transport_ok('http://router:18080') + assert not cpc.trusted_transport_ok('ftp://router') + + +def test_plaintext_control_plane_sends_no_secret(monkeypatch): + monkeypatch.setattr(cpc, 'CONTROL_PLANE_URL', 'http://cp.test') + monkeypatch.setattr(cpc, 'ALLOW_INSECURE_HTTP', False) + class NoNetwork: + async def get(self, *args, **kwargs): + pytest.fail('plaintext bridge must fail before sending headers') + monkeypatch.setattr(cpc, '_client', NoNetwork()) + assert asyncio.run(cpc._fetch_resolve('a' * 64)) is None + with pytest.raises(cpc.RouteUnavailable): + asyncio.run(cpc.tenant_connections(1, set(), project_id=2, environment_id=3)) + with pytest.raises(cpc.RouteUnavailable): + asyncio.run(cpc.resolve_route(1, 'assistant', project_id=2, environment_id=3, key_digest='a' * 64))