feat: enforce SaaS project scopes and isolate environment usage

.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

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)

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}

control_plane_client.py

@@ -29,13 +29,15 @@
 import time
 from dataclasses import dataclass
 from typing import Any, Mapping
+from urllib.parse import urlsplit
 
 import httpx
 
 log = logging.getLogger("llm-router-control-plane")
 
 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,13 +144,25 @@ 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,
         tenant_id=_opt_int(data.get("tenant_id")) if active else 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

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.