feat: tenant-isolated BYO routing contracts for SaaS

.env.example

@@ -106,6 +106,13 @@ ANTSEED_MAX_OUTPUT=1000
 # NOTE this endpoint is UNTRUSTED — the funding keeper below treats its readings
 # as vetoes and alerts only, never as permission to spend.
 ANTSEED_WALLET_RPC_URL=
+
+# Optional external control plane ("bring your own control plane"): a separate
+# service owning consumer keys / tenants (e.g. unhardcoded-cloud). Both must be
+# set to enable; otherwise the router runs operator-only. The secret gates the
+# ingress /internal/usage surface AND authenticates outbound resolve calls.
+CONTROL_PLANE_URL=
+CONTROL_PLANE_INTERNAL_SECRET=
 # --- 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

antseed/byo-gateway.js

@@ -0,0 +1,85 @@
+// Optional customer-owned BYO gateway. Separate from wallet control: this token
+// can spend through inference, but cannot deposit, withdraw or export keys.
+// Terminate HTTPS at the customer's reverse proxy; do not expose the raw proxy.
+'use strict';
+const http = require('http');
+const { timingSafeEqual } = require('crypto');
+
+function createGateway({ token, snapshot, proxyPort = 8377 }) {
+  if (!token || token.length < 32) throw new Error('ANTSEED_BYO_TOKEN must have at least 32 characters');
+  const expected = Buffer.from('Bearer ' + token);
+  return http.createServer(async (req, res) => {
+    const auth = Buffer.from(req.headers.authorization || '');
+    const send = (code, data) => {
+      res.writeHead(code, { 'content-type': 'application/json', 'cache-control': 'no-store' });
+      res.end(JSON.stringify(data));
+    };
+    if (auth.length !== expected.length || !timingSafeEqual(auth, expected)) {
+      return send(401, { error: 'unauthorized' });
+    }
+    if (req.method === 'GET' && req.url === '/snapshot') {
+      try { return send(200, await snapshot()); }
+      catch (_) { return send(503, { error: 'buyer_state_unavailable' }); }
+    }
+    if (req.method !== 'POST' || req.url !== '/v1/chat/completions') {
+      return send(404, { error: 'unsupported_endpoint' });
+    }
+    // Fixed local upstream, no caller-chosen URL or forwarded credentials.
+    const chunks = [];
+    let bytes = 0;
+    try {
+      for await (const chunk of req) {
+        bytes += chunk.length;
+        if (bytes > 1024 * 1024) return send(413, { error: 'request_too_large' });
+        chunks.push(chunk);
+      }
+    } catch (_) { return res.destroy(); }
+    const headers = { 'content-type': 'application/json' };
+    const peer = req.headers['x-antseed-pin-peer'];
+    if (typeof peer !== 'string' || !peer || peer.length > 256) {
+      return send(400, { error: 'peer_pin_required' });
+    }
+    headers['x-antseed-pin-peer'] = peer;
+    const upstream = http.request({ hostname: '127.0.0.1', port: proxyPort,
+      path: '/v1/chat/completions', method: 'POST', headers, timeout: 120000 }, response => {
+      res.writeHead(response.statusCode, {
+        'content-type': response.headers['content-type'] || 'application/json',
+        'cache-control': 'no-store',
+      });
+      response.on('error', () => res.destroy());
+      response.pipe(res);
+    });
+    upstream.on('timeout', () => upstream.destroy());
+    upstream.on('error', () => {
+      if (!res.headersSent) send(502, { error: 'buyer_unavailable' });
+      else res.destroy();
+    });
+    res.on('close', () => upstream.destroy());
+    upstream.end(Buffer.concat(chunks));
+  });
+}
+
+if (require.main === module && process.env.ANTSEED_BYO_TOKEN) {
+  const { Pool } = require('pg');
+  const { pgConfig } = require('./db.js');
+  const pool = new Pool({ ...pgConfig(process.env.DATABASE_URL), statement_timeout: 5000 });
+  const pid = process.env.ANTSEED_BUYER_PID || 'antseed';
+  const snapshot = async () => {
+    const [offers, status] = await Promise.all([
+      pool.query(`SELECT peer_id, service, price_in, price_out, price_cached_in,
+                         max_concurrency, reputation, last_seen, last_reached_at, observed_at
+                  FROM peer_offers WHERE observed_at >= $1 LIMIT 10000`, [Date.now() - 900000]),
+      pool.query(`SELECT pinned_peer_id, deposits_available, fetched_at
+                  FROM buyer_status WHERE pid = $1`, [pid]),
+    ]);
+    return { peer_offers: offers.rows.map(row => ({ ...row, observed_at: Number(row.observed_at),
+        last_reached_at: row.last_reached_at == null ? null : Number(row.last_reached_at),
+        last_seen: row.last_seen == null ? null : Number(row.last_seen) })),
+      buyer_status: status.rows[0] ? { ...status.rows[0], fetched_at: Number(status.rows[0].fetched_at) } : null };
+  };
+  createGateway({ token: process.env.ANTSEED_BYO_TOKEN, snapshot,
+    proxyPort: Number(process.env.ANTSEED_PROXY_PORT || 8377) })
+    .listen(Number(process.env.ANTSEED_BYO_PORT || 8380), '0.0.0.0');
+}
+
+module.exports = { createGateway };

antseed/byo-gateway.test.js

@@ -0,0 +1,35 @@
+const { test } = require('node:test');
+const assert = require('node:assert/strict');
+const http = require('node:http');
+const { createGateway } = require('./byo-gateway.js');
+
+test('BYO token permits snapshot/inference, never wallet ops or unpinned traffic', async () => {
+  const received = [];
+  const proxy = http.createServer((req, res) => {
+    received.push(req.headers);
+    res.setHeader('content-type', 'text/event-stream');
+    res.end('data: {"ok":true}\n\n');
+  });
+  await new Promise(resolve => proxy.listen(0, '127.0.0.1', resolve));
+  const token = 't'.repeat(40);
+  const gateway = createGateway({token, proxyPort: proxy.address().port,
+    snapshot: async () => ({ peer_offers: [], buyer_status: { deposits_available: 3 } })});
+  await new Promise(resolve => gateway.listen(0, '127.0.0.1', resolve));
+  const base = 'http://127.0.0.1:' + gateway.address().port;
+  const headers = { authorization: 'Bearer ' + token };
+  try {
+    assert.equal((await fetch(base + '/snapshot')).status, 401);
+    assert.equal((await (await fetch(base + '/snapshot', {headers})).json()).buyer_status.deposits_available, 3);
+    for (const path of ['/deposit','/withdraw','/reclaim','/status','/v1/models']) {
+      assert.equal((await fetch(base + path, {method:'POST', headers})).status, 404);
+    }
+    assert.equal((await fetch(base + '/v1/chat/completions', {method:'POST',headers,body:'{}'})).status, 400);
+    const result = await fetch(base + '/v1/chat/completions', {method:'POST',
+      headers:{...headers,'x-antseed-pin-peer':'peer-a'},body:'{}'});
+    assert.equal(await result.text(), 'data: {"ok":true}\n\n');
+    assert.equal(received[0]['x-antseed-pin-peer'], 'peer-a');
+    assert.equal(received[0].authorization, undefined);
+  } finally {
+    await Promise.all([new Promise(resolve => gateway.close(resolve)), new Promise(resolve => proxy.close(resolve))]);
+  }
+});

antseed/entrypoint.sh

@@ -62,6 +62,9 @@ socat "TCP-LISTEN:${PORT_PUBLIC},fork,reuseaddr" "TCP:127.0.0.1:${PORT_PROXY}" &
 # Self-disables when ANTSEED_CONTROL_TOKEN is unset. See antseed/control.js.
 node "$LIB/control.js" &
 
+# Separate inference + read-only discovery token; no wallet control authority.
+node "$LIB/byo-gateway.js" &
+
 write_market() {
     raw="$MARKET_DIR/.market.raw.$$"
     timeout -k 5 "$CLI_TIMEOUT" antseed network browse --services --top "$TOP" --json > "$raw" 2>/dev/null || true

auth_proxy.py

@@ -24,7 +24,9 @@
 from fastapi.responses import HTMLResponse, JSONResponse, Response, StreamingResponse
 from starlette.background import BackgroundTask
 
+import control_plane_client
 import host_store
+import internal_api
 from env_secrets import load_env_secrets
 from shim import _CACHE_READ_FACTOR   # the billing cache-read discount — one source
 
@@ -85,6 +87,10 @@
 SYNTHETIC_PROBE_INITIAL_DELAY_S = float(os.getenv("DASHBOARD_SYNTHETIC_PROBE_INITIAL_DELAY_S", "45"))
 SYNTHETIC_PROBE_TIMEOUT_S = float(os.getenv("DASHBOARD_SYNTHETIC_PROBE_TIMEOUT_S", "45"))
 SYNTHETIC_PROBE_CALLER = os.getenv("DASHBOARD_SYNTHETIC_PROBE_CALLER", "dashboard-probe")
+# Optional namespace for control-plane-resolved callers (e.g. "cp:") so tenant
+# slugs can't merge attribution with operator-minted consumer names. Default off:
+# calls.caller == tenant slug, which is what the control plane's usage reads key on.
+CONTROL_PLANE_CALLER_PREFIX = os.getenv("CONTROL_PLANE_CALLER_PREFIX", "")
 
 
 
@@ -119,6 +125,10 @@ def _bootstrap_caller_key_hashes(raw: str) -> Dict[str, str]:
 logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"), format="%(message)s")
 log = logging.getLogger("llm-router-auth-proxy")
 app = FastAPI(title="llm-router auth proxy", docs_url=None, redoc_url=None)
+# Control-plane metering surface (/internal/usage[/recent]). Registered before
+# any route definition below so Starlette matches it ahead of the catch-all
+# proxy; hidden (404) unless CONTROL_PLANE_INTERNAL_SECRET is configured.
+app.include_router(internal_api.router)
 _client: httpx.AsyncClient | None = None
 _probe_task: asyncio.Task[None] | None = None
 _started_wall = time.time()
@@ -734,6 +744,42 @@ def _caller_auth(token: str | None) -> dict[str, Any]:
     return {"ok": True, "caller": caller, "digest": digest, "storage": storage, "meta": meta}
 
 
+async def _caller_auth_async(token: str | None) -> dict[str, Any]:
+    """_caller_auth plus the external-control-plane fallback.
+
+    Local key stores stay authoritative: a locally known key — including a
+    revoked/inactive one — never falls through to the control plane; only a
+    pure local miss consults it (and only when the feature is configured, see
+    control_plane_client.enabled()). On a resolve hit the caller becomes the
+    control-plane consumer name (tenant slug) and the plan's rate limits ride
+    the returned meta; an explicit local record for that same name keeps
+    precedence (operator kill-switch: mark the slug inactive locally)."""
+    auth = await asyncio.to_thread(_caller_auth, token)
+    if auth.get("ok") or auth.get("caller") or not token or not control_plane_client.enabled():
+        return auth
+    digest = hashlib.sha256(token.encode()).hexdigest()
+    resolved = await control_plane_client.resolve_key(digest)
+    if resolved is None or not resolved.active or not resolved.consumer:
+        return auth
+    caller = f"{CONTROL_PLANE_CALLER_PREFIX}{resolved.consumer}"
+    meta = await asyncio.to_thread(_consumer_meta, caller)
+    if meta.get("keys"):
+        # A local consumer already answers to this name — attribution merges.
+        control_plane_client.log_collision_once(caller)
+    if meta.get("status") != "active":
+        return {"ok": False, "caller": caller, "digest": digest,
+                "storage": "control_plane", "error_code": "caller_inactive"}
+    if meta.get("rate_per_min") is None:
+        meta["rate_per_min"] = resolved.rate_per_min
+    if meta.get("burst") is None:
+        meta["burst"] = resolved.burst
+    out = {"ok": True, "caller": caller, "digest": digest,
+           "storage": "control_plane", "meta": meta}
+    if resolved.tenant_id is not None:
+        out["tenant_id"] = resolved.tenant_id
+    return out
+
+
 def _rate_ok(caller: str, meta: dict[str, Any] | None = None) \
         -> tuple[bool, bool, float]:
     meta = meta or _consumer_meta(caller)
@@ -773,7 +819,8 @@ def _route_matches(pattern: str, route: str) -> bool:
 
 def _route_allowed(caller: str, route: str | None,
                    meta: dict[str, Any] | None = None) -> bool:
-    allowed_routes = (meta or _consumer_meta(caller)).get("allowed_routes") or []
+    meta = meta if meta is not None else _consumer_meta(caller)
+    allowed_routes = meta.get("allowed_routes") or []
     if not allowed_routes:
         return True
     if not route:
@@ -1251,6 +1298,7 @@ async def shutdown() -> None:
         _probe_task = None
     if _client:
         await _client.aclose()
+    await control_plane_client.close()
 
 
 @app.get("/healthz")
@@ -2029,7 +2077,7 @@ async def consumer_skill(request: Request) -> Response:
     and the same key auth (revoked/expired/inactive keys are rejected)."""
     auth = request.headers.get("authorization") or ""
     token = auth[7:].strip() if auth.lower().startswith("bearer ") else auth.strip()
-    if not _caller_auth(token).get("ok"):
+    if not (await _caller_auth_async(token)).get("ok"):
         return JSONResponse(status_code=401, content={"error": {
             "message": "invalid or inactive API key",
             "type": "auth_error", "code": "consumer_auth"}})
@@ -2048,7 +2096,7 @@ async def consumer_skill(request: Request) -> Response:
 # or mutate state: they admit/normalize a term, preview the ranking, or list the
 # field vocabulary — all derived from the live catalog the skill already shows.
 async def _consumer_x_proxy(request: Request, method: str, path: str) -> Response:
-    if not _caller_auth(_extract_token(request)).get("ok"):
+    if not (await _caller_auth_async(_extract_token(request))).get("ok"):
         return JSONResponse(status_code=401, content={"error": {
             "message": "invalid or inactive API key",
             "type": "auth_error", "code": "consumer_auth"}})
@@ -2689,7 +2737,7 @@ async def dashboard_key_usage(request: Request) -> Response:
 @app.get("/api/usage")
 async def key_usage(request: Request) -> Response:
     token = _extract_token(request)
-    auth = _caller_auth(token)
+    auth = await _caller_auth_async(token)
     if not auth.get("ok"):
         code = auth.get("error_code") or "caller_auth"
         status_code = 403 if code in {"caller_inactive", "caller_key_revoked", "caller_key_expired"} else 401
@@ -2717,7 +2765,7 @@ async def session_view(sid: str, request: Request) -> Response:
     itself sends as X-Unhardcoded-Session. Lets a harness (e.g. the opencode
     plugin) show live router economics without operator access to /x/*."""
     token = _extract_token(request)
-    auth = _caller_auth(token)
+    auth = await _caller_auth_async(token)
     if not auth.get("ok"):
         code = auth.get("error_code") or "caller_auth"
         status_code = 403 if code in {"caller_inactive", "caller_key_revoked", "caller_key_expired"} else 401
@@ -3857,7 +3905,7 @@ async def proxy(path: str, request: Request) -> Response:
         return JSONResponse(status_code=404, content={"error": {"message": "not found", "type": "invalid_request_error", "code": None}})
     started = time.perf_counter()
     token = _extract_token(request)
-    auth = await asyncio.to_thread(_caller_auth, token)
+    auth = await _caller_auth_async(token)
     caller = auth.get("caller")
     if not auth.get("ok"):
         code = auth.get("error_code") or "caller_auth"
@@ -3882,6 +3930,28 @@ async def proxy(path: str, request: Request) -> Response:
     consumer_meta = auth.get("meta") or {}
     body = await request.body()
     requested_route = _requested_route_from(path, body)
+    published_route = None
+    if auth.get("tenant_id") is not None:
+        # SaaS keys address published routes only. Raw policies, profiles and
+        # pins cannot replace the workspace's published restrictions.
+        import re
+        if (request.method != "POST" or path not in {"v1/chat/completions", "v1/responses"}
+                or not re.fullmatch(r"route:[a-z0-9][a-z0-9-]{0,79}", requested_route or "")):
+            return JSONResponse(status_code=400, content={"error": {
+                "message": "Use a published route, for example model='route:production'.",
+                "type": "invalid_request_error", "code": "route_required"}})
+        from route_contract import apply_contract, PreferenceNotAllowed
+        try:
+            published_route = await control_plane_client.resolve_route(
+                auth["tenant_id"], requested_route[6:])
+            payload, published_route = apply_contract(json.loads(body), published_route)
+            body = json.dumps(payload, separators=(",", ":")).encode()
+        except PreferenceNotAllowed as exc:
+            return JSONResponse(status_code=400, content={'error': {
+                'message': str(exc), 'type': 'invalid_request_error', 'code': 'preference_not_allowed'}})
+        except control_plane_client.RouteUnavailable as exc:
+            return JSONResponse(status_code=503, content={"error": {
+                "message": str(exc), "type": "server_error", "code": "route_unavailable"}})
     if not _route_allowed(caller, requested_route, consumer_meta):
         _record_reject(reason="route_not_allowed", path="/" + path, caller=caller, status=403, route=requested_route)
         _log({"event": "reject", "reason": "route_not_allowed", "caller": caller, "path": "/" + path, "route": requested_route})
@@ -3953,9 +4023,21 @@ async def proxy(path: str, request: Request) -> Response:
 
     headers = {
         k: v for k, v in request.headers.items()
-        if k.lower() not in {"authorization", "host", "connection", "content-length"}
+        if k.lower() not in {"authorization", "host", "connection", "content-length",
+                             "x-llm-router-tenant", "x-internal-secret",
+                             "x-unhardcoded-route", "x-unhardcoded-revision", "x-unhardcoded-policy-id", "x-unhardcoded-preference"}
     }
     headers["x-llm-router-caller"] = caller
+    # Tenant identity for per-tenant provider credentials (BYO keys): set ONLY
+    # from the authenticated resolve — the client-sent header is stripped above,
+    # so it can never be smuggled past auth.
+    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
+        headers["x-unhardcoded-route"] = requested_route
+        headers["x-unhardcoded-revision"] = str(published_route["revision"])
+        headers["x-unhardcoded-policy-id"] = published_route["policy_id"]
+        headers['x-unhardcoded-preference'] = published_route.get('routing_preference', 'default')
 
     status = 502
     provider = None
@@ -3979,11 +4061,16 @@ async def proxy(path: str, request: Request) -> Response:
     session_id = request.headers.get("x-unhardcoded-session")
 
     def _finish():
-        nonlocal capacity_released
+        nonlocal capacity_released, decision_trace
         if capacity_released:
             return
         capacity_released = True
         latency_ms = round((time.perf_counter() - started) * 1000, 1)
+        if published_route:
+            decision_trace = {**(decision_trace or {}), 'route': requested_route,
+                              'route_revision': published_route['revision'],
+                              'policy_id': published_route['policy_id'],
+                              'routing_preference': published_route.get('routing_preference', 'default')}
         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)