From 9800a25bf0f757d6693529cfe28225e3d0c9ac79 Mon Sep 17 00:00:00 2001 From: jmlago Date: Sat, 26 Sep 2026 17:16:27 +0100 Subject: [PATCH 1/2] Security audit 2026-09-26: harden router, ingress, sidecar and engine Engine (core -> 2f43c47): route keys restored after every Xform, sampling-only set_param names, caller breaker/disable limits, bounded retries and params. Router: route/credential pinned from the catalog before every call; loop yields each step with a hard attempt cap; tenant BYO outcomes kept out of shared route_observations; tenant keys never unlock operator-local providers; flows run under the request deadline and cancel on disconnect; session reads scoped to the caller and time/statement bounded; /x/* admin endpoints require x-internal-secret (loopback-only without one); BYO reads byte-capped; BYO peer gates tenant-scoped; NAT64/IPv4-embedded egress rejected; codex OAuth pinned to the configured upstream; broker URL requires TLS off-cluster; overlay/env-secrets cannot name or override infrastructure variables. Ingress: normalized path allowlist (no traversal to router internals), streams drained after client disconnect so budgets book usage, consumer views/writes fail closed on store errors, tenant/consumer name collisions rejected, /v1/usage bounded and rate-limited, sid validated, byte-safe secret comparisons, /internal/usage window validated. Sidecar/images: optional bearer-protected :8378 proxy, control server on loopback with constant-time token check, bounded peer ids/services, ingress as default image CMD, OAuth files excluded from the build context, compose requires POSTGRES_PASSWORD. Co-Authored-By: Claude Opus 5.5 (1M context) --- .dockerignore | 4 + Dockerfile | 5 +- SECURITY.md | 9 + antseed/auth.js | 22 ++ antseed/auth.test.js | 24 ++ antseed/control.js | 11 +- antseed/entrypoint.sh | 11 +- antseed/public-proxy.js | 43 ++ antseed/public-proxy.test.js | 45 +++ antseed/write-market.js | 9 +- antseed/write-market.test.js | 31 ++ auth_proxy.py | 310 ++++++++++++--- byo_http.py | 15 +- codex_backend.py | 4 +- codex_broker.py | 2 +- compose.yml | 25 +- control_plane_client.py | 9 +- core | 2 +- env_secrets.py | 20 +- host_store.py | 66 +++- internal_api.py | 4 + llm_router_host.py | 60 ++- provider_adapters/decisions.py | 2 +- provider_adapters/openai_compatible.py | 75 +++- provider_overlay.py | 43 +- remote_codex.py | 23 ++ shim.py | 95 +++-- sources/antseed.py | 4 +- streaming.py | 2 +- tests/conftest.py | 3 + tests/test_antseed_node.py | 12 + tests/test_provider_overlay.py | 9 +- tests/test_security_ingress.py | 459 +++++++++++++++++++++ tests/test_security_router.py | 528 +++++++++++++++++++++++++ tests/test_security_sidecar.py | 42 ++ 35 files changed, 1888 insertions(+), 140 deletions(-) create mode 100644 antseed/auth.js create mode 100644 antseed/auth.test.js create mode 100644 antseed/public-proxy.js create mode 100644 antseed/public-proxy.test.js create mode 100644 tests/test_security_ingress.py create mode 100644 tests/test_security_router.py create mode 100644 tests/test_security_sidecar.py diff --git a/.dockerignore b/.dockerignore index 1c69684..7f69a52 100644 --- a/.dockerignore +++ b/.dockerignore @@ -8,3 +8,7 @@ __pycache__ !.env.example runtime secrets +# Codex OAuth tokens (compose's default CODEX_AUTH_PATH) and dashboard-added +# accounts: `COPY . .` would bake them into the image. +codex-auth*.json +codex-accounts diff --git a/Dockerfile b/Dockerfile index e1f8b21..06d71e8 100644 --- a/Dockerfile +++ b/Dockerfile @@ -20,4 +20,7 @@ RUN test -f core/router.lua && test -f core/llm_policy.lua EXPOSE 8080 -CMD ["python", "serve.py", "--config", "config.internal.lua", "--default-profile", "agent", "--host", "0.0.0.0", "--port", "8080"] +# Default to the authenticated ingress. The router (serve.py) has no caller auth +# of its own and must only ever be reached through it, so every deployment that +# runs the router names that command explicitly (compose, k8s, CI). +CMD ["uvicorn", "auth_proxy:app", "--host", "0.0.0.0", "--port", "8080", "--proxy-headers"] diff --git a/SECURITY.md b/SECURITY.md index e3cbca7..7142d74 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -22,3 +22,12 @@ extra care: as critical and rotate immediately. This provider is opt-in (off by default). - **Codex provider** uses an unofficial ChatGPT-subscription auth path and is ToS-risky; see `docs/OPENAI-CODEX.md`. +- **Router admin endpoints** (`/x/wallet*`, `/x/providers`, `/x/provider-key`, + `/x/config/reload`, `/x/codex/reload`, `/x/calls`, `/x/sessions`, + `/x/session/*`) require `x-internal-secret` = `CONTROL_PLANE_INTERNAL_SECRET` + (the ingress sends it). With no secret configured only a loopback (same-pod) + caller is accepted; `ROUTER_ADMIN_ALLOW_UNAUTHENTICATED=1` restores the old + open behaviour for a trusted local stack only. +- **AntSeed buyer proxy** (`:8378`) spends the funded wallet: set + `ANTSEED_PROXY_TOKEN` (>= 32 chars) on the sidecar and the router so the port + requires a bearer. Unset keeps the legacy unauthenticated forwarder. diff --git a/antseed/auth.js b/antseed/auth.js new file mode 100644 index 0000000..4880451 --- /dev/null +++ b/antseed/auth.js @@ -0,0 +1,22 @@ +// Shared-secret checks and bind addresses for the sidecar's HTTP listeners. +// Dependency-free so auth.test.js runs outside the sidecar image (control.js +// pulls in `pg`, which only exists inside it). +'use strict'; +const { timingSafeEqual } = require('crypto'); + +// Constant-time: a `!==` compare leaks the token prefix through response timing +// to anyone who can reach the port. Length mismatch is not secret (it is fixed +// by the deployment), so it short-circuits. +function tokenMatches(presented, token) { + if (typeof presented !== 'string' || typeof token !== 'string' || !token) return false; + const a = Buffer.from(presented), b = Buffer.from(token); + return a.length === b.length && timingSafeEqual(a, b); +} + +// Pod-local by default: in k8s the router shares the pod and dials 127.0.0.1. +// Compose runs the sidecar as its own service and opts in to 0.0.0.0 explicitly. +function listenHost(env, name) { + return (env && env[name]) || '127.0.0.1'; +} + +module.exports = { tokenMatches, listenHost }; diff --git a/antseed/auth.test.js b/antseed/auth.test.js new file mode 100644 index 0000000..74ec64b --- /dev/null +++ b/antseed/auth.test.js @@ -0,0 +1,24 @@ +const { test } = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const { tokenMatches, listenHost } = require('./auth.js'); + +test('token compare is exact and total', () => { + const t = 'x'.repeat(32); + assert.equal(tokenMatches(t, t), true); + assert.equal(tokenMatches(t + 'y', t), false); + assert.equal(tokenMatches(t.slice(1), t), false); + assert.equal(tokenMatches(undefined, t), false); + assert.equal(tokenMatches(['a'], t), false); + assert.equal(tokenMatches('', ''), false); // unset token never matches + assert.equal(tokenMatches('é'.repeat(16), t), false); // multibyte, same char count +}); + +test('control server compares in constant time and binds pod-local by default', () => { + const src = fs.readFileSync(__dirname + '/control.js', 'utf8'); + assert.doesNotMatch(src, /!==\s*TOKEN/); + assert.match(src, /tokenMatches\(req\.headers\['x-antseed-control-token'\], TOKEN\)/); + assert.match(src, /server\.listen\(PORT, HOST,/); + assert.equal(listenHost({}, 'ANTSEED_CONTROL_HOST'), '127.0.0.1'); + assert.equal(listenHost({ ANTSEED_CONTROL_HOST: '0.0.0.0' }, 'ANTSEED_CONTROL_HOST'), '0.0.0.0'); +}); diff --git a/antseed/control.js b/antseed/control.js index 52b8a92..3963183 100644 --- a/antseed/control.js +++ b/antseed/control.js @@ -6,8 +6,9 @@ // -> THIS server (:8379) -> `antseed buyer ` // // Guarded by a shared token (ANTSEED_CONTROL_TOKEN); if unset the server does -// not start (feature disabled, router degrades to 503). Listens on the pod/ -// container network only — never published. Subcommands are whitelisted and run +// not start (feature disabled, router degrades to 503). Binds 127.0.0.1 (the +// router shares the pod) unless ANTSEED_CONTROL_HOST says otherwise — compose, +// where the sidecar is its own service, sets 0.0.0.0. Never published. Subcommands are whitelisted and run // via execFile with array args (no shell); amounts are strictly validated. 'use strict'; const http = require('http'); @@ -26,11 +27,13 @@ const { createQueue } = require('./queue.js'); // deserve a file. See antseed/broadcast.js. const { classifyCliFailure } = require('./broadcast.js'); const { walletCommandArgs } = require('./cli-args.js'); +const { tokenMatches, listenHost } = require('./auth.js'); const path = require('path'); const PORT = parseInt(process.env.ANTSEED_CONTROL_PORT || '8379', 10); const TOKEN = process.env.ANTSEED_CONTROL_TOKEN || ''; +const HOST = listenHost(process.env, 'ANTSEED_CONTROL_HOST'); const PID = process.env.ANTSEED_BUYER_PID || 'antseed'; const DEPOSIT_TIMEOUT_MS = 120000; // on-chain tx const STATUS_TIMEOUT_MS = 30000; @@ -179,7 +182,7 @@ function inconclusive(res, status, error) { // a broadcast cannot be ruled ou } const server = http.createServer(async (req, res) => { - if (req.headers['x-antseed-control-token'] !== TOKEN) { + if (!tokenMatches(req.headers['x-antseed-control-token'], TOKEN)) { return refuse(res, 401, 'unauthorized'); } const url = (req.url || '').split('?')[0]; @@ -283,4 +286,4 @@ const server = http.createServer(async (req, res) => { return refuse(res, 404, 'not found'); }); -server.listen(PORT, () => console.error('[control] listening on :' + PORT)); +server.listen(PORT, HOST, () => console.error('[control] listening on ' + HOST + ':' + PORT)); diff --git a/antseed/entrypoint.sh b/antseed/entrypoint.sh index 54f1805..09b0f6b 100644 --- a/antseed/entrypoint.sh +++ b/antseed/entrypoint.sh @@ -56,7 +56,16 @@ antseed buyer start -p "$PORT_PROXY" \ PROXY_PID=$! # Expose the localhost-only proxy on the container network for the router. -socat "TCP-LISTEN:${PORT_PUBLIC},fork,reuseaddr" "TCP:127.0.0.1:${PORT_PROXY}" & +# Every call through it spends the funded wallet, so with ANTSEED_PROXY_TOKEN +# set the network face demands that bearer (public-proxy.js). Unset keeps the +# legacy unauthenticated socat forwarder for routers that do not send it yet — +# then only network policy stands between the pod network and the wallet. +if [ -n "${ANTSEED_PROXY_TOKEN:-}" ]; then + node "$LIB/public-proxy.js" & +else + echo "[entrypoint] WARNING: ANTSEED_PROXY_TOKEN unset — :${PORT_PUBLIC} forwards to the funded buyer without auth" >&2 + socat "TCP-LISTEN:${PORT_PUBLIC},fork,reuseaddr" "TCP:127.0.0.1:${PORT_PROXY}" & +fi # Wallet control server (deposit/withdraw/status from the dashboard, no kubectl). # Self-disables when ANTSEED_CONTROL_TOKEN is unset. See antseed/control.js. diff --git a/antseed/public-proxy.js b/antseed/public-proxy.js new file mode 100644 index 0000000..63d04d9 --- /dev/null +++ b/antseed/public-proxy.js @@ -0,0 +1,43 @@ +// Authenticated network face of the buyer proxy. The buyer binds 127.0.0.1:8377 +// and spends the funded wallet on every call, so whatever exposes it beyond the +// pod must check a credential: with ANTSEED_PROXY_TOKEN set, entrypoint.sh runs +// THIS instead of the bare socat forwarder and every request needs +// `Authorization: Bearer ` (the router sends it via the antseed +// provider's auth). Transparent otherwise: any method/path, streamed both ways. +'use strict'; +const http = require('http'); +const { tokenMatches } = require('./auth.js'); + +function createPublicProxy({ token, proxyPort = 8377 }) { + if (!token || token.length < 32) throw new Error('ANTSEED_PROXY_TOKEN must have at least 32 characters'); + return http.createServer((req, res) => { + if (!tokenMatches(req.headers.authorization, 'Bearer ' + token)) { + res.writeHead(401, { 'content-type': 'application/json', 'cache-control': 'no-store' }); + return res.end(JSON.stringify({ error: 'unauthorized' })); + } + const headers = { ...req.headers }; + delete headers.authorization; // the buyer proxy never needs our credential + const upstream = http.request({ hostname: '127.0.0.1', port: proxyPort, + path: req.url, method: req.method, headers }, response => { + res.writeHead(response.statusCode, response.headers); + response.on('error', () => res.destroy()); + response.pipe(res); + }); + upstream.on('error', () => { + if (!res.headersSent) { + res.writeHead(502, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ error: 'buyer_unavailable' })); + } else res.destroy(); + }); + res.on('close', () => upstream.destroy()); + req.pipe(upstream); + }); +} + +if (require.main === module) { + createPublicProxy({ token: process.env.ANTSEED_PROXY_TOKEN, + proxyPort: Number(process.env.ANTSEED_PROXY_PORT || 8377) }) + .listen(Number(process.env.ANTSEED_PUBLIC_PORT || 8378), '0.0.0.0'); +} + +module.exports = { createPublicProxy }; diff --git a/antseed/public-proxy.test.js b/antseed/public-proxy.test.js new file mode 100644 index 0000000..daf7950 --- /dev/null +++ b/antseed/public-proxy.test.js @@ -0,0 +1,45 @@ +const { test } = require('node:test'); +const assert = require('node:assert/strict'); +const http = require('node:http'); +const fs = require('node:fs'); +const { createPublicProxy } = require('./public-proxy.js'); + +test('network face of the buyer proxy demands the bearer and forwards transparently', async () => { + const received = []; + const buyer = http.createServer((req, res) => { + let body = ''; + req.on('data', c => { body += c; }); + req.on('end', () => { + received.push({ method: req.method, url: req.url, headers: req.headers, body }); + res.setHeader('content-type', 'text/event-stream'); + res.end('data: {"ok":true}\n\n'); + }); + }); + await new Promise(resolve => buyer.listen(0, '127.0.0.1', resolve)); + const token = 'p'.repeat(40); + assert.throws(() => createPublicProxy({ token: 'short' })); + const proxy = createPublicProxy({ token, proxyPort: buyer.address().port }); + await new Promise(resolve => proxy.listen(0, '127.0.0.1', resolve)); + const base = 'http://127.0.0.1:' + proxy.address().port; + try { + assert.equal((await fetch(base + '/v1/chat/completions', { method: 'POST', body: '{}' })).status, 401); + assert.equal((await fetch(base + '/v1/chat/completions', { method: 'POST', body: '{}', + headers: { authorization: 'Bearer ' + 'q'.repeat(40) } })).status, 401); + assert.equal(received.length, 0); + const r = await fetch(base + '/v1/chat/completions', { method: 'POST', body: '{"a":1}', + headers: { authorization: 'Bearer ' + token, 'x-antseed-pin-peer': 'peer-a' } }); + assert.equal(r.status, 200); + assert.equal(await r.text(), 'data: {"ok":true}\n\n'); + assert.equal(received[0].url, '/v1/chat/completions'); + assert.equal(received[0].body, '{"a":1}'); + assert.equal(received[0].headers['x-antseed-pin-peer'], 'peer-a'); + assert.equal(received[0].headers.authorization, undefined); + } finally { + await Promise.all([new Promise(r => proxy.close(r)), new Promise(r => buyer.close(r))]); + } +}); + +test('entrypoint uses the authenticated forwarder whenever a proxy token is configured', () => { + const src = fs.readFileSync(__dirname + '/entrypoint.sh', 'utf8'); + assert.match(src, /if \[ -n "\$\{ANTSEED_PROXY_TOKEN:-\}" \]; then\n\s+node "\$LIB\/public-proxy\.js" &\nelse/); +}); diff --git a/antseed/write-market.js b/antseed/write-market.js index a6f3472..d639e17 100644 --- a/antseed/write-market.js +++ b/antseed/write-market.js @@ -28,6 +28,12 @@ function numOrNull(v) { const n = Number(v); return Number.isFinite(n) ? n : null; } +// peerId/service are seller-announced strings that end up as a DB key, an +// x-antseed-pin-peer header and dashboard text. Bound them: an oversized or +// control-char id is not one a real peer announces, so the row is dropped. +const PEER_ID_RE = /^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$/; +const SERVICE_RE = /^[A-Za-z0-9][A-Za-z0-9_.:@\/+-]{0,127}$/; +const validId = (re, v) => typeof v === "string" && re.test(v); function posIntOrNull(v) { // maxConcurrency: positive int, else ungated (null) const n = Number(v); return Number.isFinite(n) && n > 0 ? Math.trunc(n) : null; @@ -43,7 +49,7 @@ if (fresh === null || typeof fresh !== "object" || !Array.isArray(fresh.peers)) // row per (peerId, service). No services -> the peer contributes no rows. const rows = []; for (const peer of fresh.peers) { - if (!peer || !peer.peerId) continue; + if (!peer || !validId(PEER_ID_RE, peer.peerId)) continue; const maxc = posIntOrNull(peer.maxConcurrency); const rep = numOrNull(peer.onChainReputationScore); const lastSeen = numOrNull(peer.lastSeen); @@ -54,6 +60,7 @@ for (const peer of fresh.peers) { const lastReachedAt = numOrNull(peer.lastReachedAt); for (const [provider, pricing] of Object.entries(peer.providerPricing || {})) { for (const [service, sp] of Object.entries((pricing || {}).services || {})) { + if (!validId(SERVICE_RE, service)) continue; const rawProtocols = peer.providerServiceApiProtocols?.[provider]?.services?.[service]; const protocols = Array.isArray(rawProtocols) && rawProtocols.every(p => typeof p === 'string') ? rawProtocols : null; diff --git a/antseed/write-market.test.js b/antseed/write-market.test.js index ecc4a70..f7f490e 100644 --- a/antseed/write-market.test.js +++ b/antseed/write-market.test.js @@ -34,3 +34,34 @@ test('browse writer preserves decision protocols and supports older announcement assert.equal(inserts[2].params[12], null); assert.match(inserts[0].sql, /protocols=COALESCE\(EXCLUDED.protocols, peer_offers.protocols\)/); }); + +test('browse writer drops peers and services with oversized or non-id names', async () => { + const queries = []; + const svc = name => ({ [name]: { inputUsdPerMillion: 1, outputUsdPerMillion: 2 } }); + const peer = (id, services) => ({ peerId: id, providerPricing: { p: { services } } }); + const input = JSON.stringify({ peers: [ + peer('4668854ba3e8b094e6f48fbeb59cec1cfde162f2', { ...svc('anthropic/claude-opus-4.8'), ...svc('x'.repeat(129)), + ...svc('evil\r\nx-injected: 1'), ...svc('qwen3-235b-a22b@fast') }), + peer('a'.repeat(129), svc('ok-service')), + peer('bad peer\n', svc('ok-service')), + peer({ nested: true }, svc('ok-service')), + ] }); + class Client { + async connect() {} + async end() {} + async query(sql, params) { queries.push({ sql, params }); } + } + const processMock = { env: {}, stderr: { write() {} }, exit(code) { throw new Error('exit ' + code); } }; + vm.runInNewContext(fs.readFileSync(__dirname + '/write-market.js', 'utf8'), { + process: processMock, console, + require(name) { + if (name === 'pg') return { Client }; + if (name === './db.js') return { pgConfig: () => ({}) }; + if (name === 'fs') return { readFileSync: () => input }; + throw new Error('unexpected import'); + }, + }); + await new Promise(resolve => setImmediate(resolve)); + const inserts = queries.filter(q => q.sql.startsWith('INSERT')); + assert.deepEqual(inserts.map(q => q.params[1]), ['anthropic/claude-opus-4.8', 'qwen3-235b-a22b@fast']); +}); diff --git a/auth_proxy.py b/auth_proxy.py index 40f4330..d83aa61 100644 --- a/auth_proxy.py +++ b/auth_proxy.py @@ -19,6 +19,7 @@ from pathlib import Path from threading import Lock, RLock from typing import Any, Dict +from urllib.parse import quote import httpx from fastapi import FastAPI, Request @@ -38,6 +39,19 @@ load_env_secrets() UPSTREAM = os.getenv("ROUTER_UPSTREAM", "http://router:18080").rstrip("/") + + +def _router_admin_headers(extra: dict[str, str] | None = None) -> dict[str, str]: + """Headers for the router's operator /x/* endpoints (wallet, providers, + provider-key, reloads, sessions/calls), which require the shared internal + secret. Only ever sent on these server-originated calls, never on proxied + client traffic.""" + headers = dict(extra or {}) + if control_plane_client.CONTROL_PLANE_INTERNAL_SECRET: + headers["x-internal-secret"] = control_plane_client.CONTROL_PLANE_INTERNAL_SECRET + return headers + + CALLER_KEYS_JSON = os.getenv("CALLER_KEYS_JSON", "{}") CALLER_KEYS_SHA256_JSON = os.getenv("CALLER_KEYS_SHA256_JSON", "{}") # GitOps/bootstrap keys arrive from a reconciled workload Secret. Keep them in @@ -426,9 +440,24 @@ def _write_json_file(path: Path, data: Any) -> None: path.chmod(0o600) +def _provider_env_error(key: str, value: str) -> str | None: + """Why KEY=value may not be written to the env-secrets file (applied over + the process env at boot), or None. Only provider-credential names, never an + infrastructure/security variable, and no control characters: a newline in + the value would inject arbitrary extra KEY=value lines.""" + from provider_overlay import AUTH_ENV_RE, is_forbidden_auth_env + if not AUTH_ENV_RE.match(key or "") or is_forbidden_auth_env(key): + return f"auth_env {key!r} is not an allowed provider credential name" + if any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in f"{key}{value}"): + return "key must not contain control characters" + return None + + def _upsert_env_line(path: Path, key: str, value: str) -> None: """Set KEY=value in an env file (used for provider API keys). Same semantics as _upsert_env_json but for plain string values.""" + if (error := _provider_env_error(key, value)) is not None: + raise ValueError(error) path.parent.mkdir(parents=True, exist_ok=True) lines = path.read_text().splitlines() if path.exists() else [] rendered = f"{key}={value}" @@ -642,6 +671,29 @@ def _consumer_meta(consumer: str) -> dict[str, Any]: return records.get(consumer, _normalize_consumer_record(consumer, {})) +def _consumer_store_unavailable() -> JSONResponse: + return JSONResponse(status_code=503, content={"error": { + "message": "consumer key store is temporarily unavailable", + "type": "server_error", "code": "consumer_store_unavailable"}}) + + +def _writable_consumer_records() -> dict[str, dict[str, Any]] | None: + """Records for a read-modify-write, or None when the store read failed: the + fallback view holds only env-configured consumers, and writing it back would + wipe every issued record (keys, budgets, revocations).""" + records = _issued_consumer_records() + return None if _issued_keys_load_failed else records + + +async def _consumer_view_unavailable(role: str) -> JSONResponse | None: + # A consumer dashboard view is scoped to that consumer's records; if the + # store cannot confirm them, fail closed rather than render unscoped data. + if role == "admin": + return None + _, ok = await asyncio.to_thread(host_store.get_consumer_keys) + return None if ok else _consumer_store_unavailable() + + def _active_key_rows(keys: list[dict[str, Any]]) -> list[dict[str, Any]]: now = int(time.time()) rows = [] @@ -781,9 +833,15 @@ async def _caller_auth_async(token: str | None) -> dict[str, Any]: 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. + if meta.get("keys") or caller in CALLER_KEYS.values() or caller in CALLER_KEY_HASHES.values(): + # A local consumer (its own keys) already answers to this name: serving + # the tenant under it would merge attribution, limits and budgets across + # two principals. Fail closed until the operator renames one of them + # (or sets CONTROL_PLANE_CALLER_PREFIX — which also renames the tenant's + # usage caller). A key-less local record stays a kill-switch/override. control_plane_client.log_collision_once(caller) + return {"ok": False, "caller": caller, "digest": digest, + "storage": "control_plane", "error_code": "caller_name_collision"} if meta.get("status") != "active": return {"ok": False, "caller": caller, "digest": digest, "storage": "control_plane", "error_code": "caller_inactive"} @@ -913,7 +971,7 @@ def _dashboard_session_context(request: Request) -> dict[str, Any] | None: return None body, sig = raw.rsplit(".", 1) expected = hmac.new(DASHBOARD_SESSION_SECRET.encode(), body.encode(), hashlib.sha256).hexdigest() - if not hmac.compare_digest(sig, expected): + if not control_plane_client.secret_equal(sig, expected): return None try: payload = json.loads(_b64d(body)) @@ -943,7 +1001,7 @@ def _dashboard_password_ok(password: str) -> bool: if not DASHBOARD_PASSWORD_SHA256: return False got = hashlib.sha256(password.encode()).hexdigest() - return hmac.compare_digest(got, DASHBOARD_PASSWORD_SHA256) + return control_plane_client.secret_equal(got, DASHBOARD_PASSWORD_SHA256) def _require_dashboard_context(request: Request) -> dict[str, Any] | None: if DASHBOARD_NO_AUTH: @@ -951,7 +1009,7 @@ def _require_dashboard_context(request: Request) -> dict[str, Any] | None: if DASHBOARD_TRUSTED_USER_HEADER: trusted_user = (request.headers.get(DASHBOARD_TRUSTED_USER_HEADER) or "").strip() trusted_secret = (request.headers.get("x-dashboard-trusted-secret") or "").strip() - if trusted_user and DASHBOARD_TRUSTED_USER_SECRET and hmac.compare_digest(trusted_secret, DASHBOARD_TRUSTED_USER_SECRET): + if trusted_user and DASHBOARD_TRUSTED_USER_SECRET and control_plane_client.secret_equal(trusted_secret, DASHBOARD_TRUSTED_USER_SECRET): return {"role": "admin", "user": trusted_user, "viewer": f"dashboard:{trusted_user}", "consumer": None, "key_sha256": None} return _dashboard_session_context(request) @@ -1851,6 +1909,8 @@ async def dashboard_stats(request: Request) -> Response: consumer = str(ctx.get("consumer") or "").strip() or requested_consumer key_sha256 = str(ctx.get("key_sha256") or "").strip() or None role = str(ctx.get("role") or "admin") + if (unavailable := await _consumer_view_unavailable(role)) is not None: + return unavailable # _stats_snapshot performs synchronous PostgreSQL aggregation. Keep it off # Uvicorn's event loop so a slow query cannot stall health checks and all # proxied traffic handled by this worker. @@ -1890,6 +1950,8 @@ async def _dashboard_full_snapshot(request: Request) -> Response: consumer = str(ctx.get("consumer") or "").strip() or requested_consumer key_sha256 = str(ctx.get("key_sha256") or "").strip() or None role = str(ctx.get("role") or "admin") + if (unavailable := await _consumer_view_unavailable(role)) is not None: + return unavailable stats = await asyncio.to_thread( _stats_snapshot, viewer=caller, upstream_status=upstream_status, upstream_health=upstream_health, consumer=consumer, @@ -2833,7 +2895,34 @@ async def key_usage(request: Request) -> Response: options = _usage_query_options(request.query_params) except ValueError as exc: return JSONResponse(status_code=400, content={"error": {"message": str(exc), "type": "invalid_request_error", "code": "invalid_usage_window"}}) - return JSONResponse(content=_key_usage_snapshot(viewer=f"consumer:{caller}", key_sha256=str(auth.get("digest")), caller=caller, options=options)) + if (limited := await _consumer_rate_limited(caller, auth)) is not None: + return limited + # Postgres reads + per-row pricing: off the event loop. + snap = await asyncio.to_thread(_key_usage_snapshot, viewer=f"consumer:{caller}", + key_sha256=str(auth.get("digest")), caller=caller, options=options) + return JSONResponse(content=snap) + + +# Opaque client-chosen session ids (opencode `ses_...`, UUIDs, `tenant:x:y`); +# never a bare dot segment, which would re-target the upstream path. +_SESSION_ID_RE = re.compile(r"[A-Za-z0-9_:-][A-Za-z0-9_.:-]{0,127}") + + +async def _consumer_rate_limited(caller: str, auth: dict[str, Any]) -> JSONResponse | None: + """The caller's request rate limit, shared with proxied traffic, for the + consumer read endpoints (/v1/usage, /v1/session).""" + result = await asyncio.to_thread(_rate_ok, caller, auth.get("meta") or {}) + allowed, store_ok, retry_after_s = result if isinstance(result, tuple) else (bool(result), True, 0.0) + if not store_ok: + _metric_store_error("rate_limit") + return JSONResponse(status_code=503, content={"error": { + "message": "caller rate-limit state is temporarily unavailable", + "type": "server_error", "code": "caller_rate_limit_unavailable"}}) + if not allowed: + return JSONResponse(status_code=429, headers={"Retry-After": str(max(1, math.ceil(retry_after_s)))}, + content={"error": {"message": "caller rate limit exceeded", + "type": "rate_limit_error", "code": "caller_rate_limit"}}) + return None @app.get("/v1/session/{sid}") @@ -2857,9 +2946,13 @@ async def session_view(sid: str, request: Request) -> Response: # answers 404 (not 403) for a sid this caller does not own, so the endpoint # never confirms that someone else's sid exists. caller = str(auth.get("caller") or "") + if not _SESSION_ID_RE.fullmatch(sid): + return JSONResponse(status_code=400, content={"error": {"message": "invalid session id", "type": "invalid_request_error", "code": "invalid_session_id"}}) + if (limited := await _consumer_rate_limited(caller, auth)) is not None: + return limited try: - r = await _client.get(f"{UPSTREAM}/x/session/{sid}", timeout=5.0, - headers={"x-llm-router-caller": caller}) + r = await _client.get(f"{UPSTREAM}/x/session/{quote(sid, safe='')}", timeout=5.0, + headers=_router_admin_headers({"x-llm-router-caller": caller})) return JSONResponse(status_code=r.status_code, content=r.json()) except Exception as exc: return JSONResponse(status_code=502, content={"error": {"message": f"upstream: {exc}", "type": "api_error", "code": "upstream_error"}}) @@ -2880,7 +2973,9 @@ async def dashboard_update_consumer(consumer: str, request: Request) -> Response return JSONResponse(status_code=400, content={"error": {"message": str(exc), "type": "invalid_request_error", "code": "invalid_consumer"}}) except Exception: return JSONResponse(status_code=400, content={"error": {"message": "invalid JSON body", "type": "invalid_request_error", "code": "invalid_json"}}) - records = _issued_consumer_records() + records = _writable_consumer_records() + if records is None: + return _consumer_store_unavailable() meta = records.get(consumer, _normalize_consumer_record(consumer, {})) if "status" in data: status = str(data.get("status") or "").strip().lower() @@ -2919,7 +3014,9 @@ async def dashboard_revoke_key(request: Request) -> Response: return JSONResponse(status_code=400, content={"error": {"message": str(exc), "type": "invalid_request_error", "code": "invalid_key"}}) except Exception: return JSONResponse(status_code=400, content={"error": {"message": "invalid JSON body", "type": "invalid_request_error", "code": "invalid_json"}}) - records = _issued_consumer_records() + records = _writable_consumer_records() + if records is None: + return _consumer_store_unavailable() meta = records.get(consumer, _normalize_consumer_record(consumer, {})) now = int(time.time()) found = False @@ -2982,9 +3079,11 @@ async def dashboard_add_provider(request: Request) -> Response: try: from provider_overlay import load_overlay, save_overlay, validate_entry catalog = _load_policy_config() # includes existing overlay entries - errors = validate_entry(pid, entry, catalog) + errors = validate_entry(pid, entry, catalog, key_supplied=bool(key)) if not key: errors.append("key is required") + elif (env_error := _provider_env_error(auth_env, key)) is not None: + errors.append(env_error) if errors: return JSONResponse(status_code=400, content={"error": {"message": "; ".join(errors), "type": "invalid_request", "code": "provider_add"}}) _upsert_env_line(Path(DASHBOARD_KEY_ENV_PATH), auth_env, key) @@ -3002,7 +3101,7 @@ async def dashboard_add_provider(request: Request) -> Response: if _client is not None: r = await _client.post(f"{UPSTREAM}/x/providers", json={ "id": pid, **{k: v for k, v in entry.items() if k != "added_at"}, - "key": key}, timeout=10.0) + "key": key}, timeout=10.0, headers=_router_admin_headers()) applied_live = r.status_code == 200 if not applied_live: apply_error = f"router /x/providers returned {r.status_code}" @@ -3046,6 +3145,8 @@ async def dashboard_update_provider_key(request: Request) -> Response: auth_env = str(provider.get("auth_env") or "").strip() if not auth_env: return JSONResponse(status_code=400, content={"error": {"message": f"provider {pid!r} has no auth_env (e.g. oauth/codex); use the Codex account flow instead", "type": "invalid_request", "code": "provider_no_auth_env"}}) + if (env_error := _provider_env_error(auth_env, key)) is not None: + return JSONResponse(status_code=400, content={"error": {"message": env_error, "type": "invalid_request", "code": "provider_update"}}) try: _upsert_env_line(Path(DASHBOARD_KEY_ENV_PATH), auth_env, key) except Exception as exc: @@ -3057,7 +3158,8 @@ async def dashboard_update_provider_key(request: Request) -> Response: try: if _client is not None: r = await _client.post(f"{UPSTREAM}/x/provider-key", json={ - "provider": pid, "auth_env": auth_env, "key": key}, timeout=10.0) + "provider": pid, "auth_env": auth_env, "key": key}, timeout=10.0, + headers=_router_admin_headers()) applied_live = r.status_code == 200 if not applied_live: apply_error = f"router /x/provider-key returned {r.status_code}" @@ -3084,7 +3186,8 @@ async def _wallet_proxy(request: Request, op: str, *, body: dict | None = None, return JSONResponse(status_code=502, content={"error": { "message": "router client unavailable", "type": "wallet_error", "code": "wallet"}}) try: - r = await _client.post(f"{UPSTREAM}/x/wallet/{op}", json=(body or {}), timeout=timeout) + r = await _client.post(f"{UPSTREAM}/x/wallet/{op}", json=(body or {}), timeout=timeout, + headers=_router_admin_headers()) except Exception as exc: return JSONResponse(status_code=502, content={"error": { "message": f"router /x/wallet/{op} unreachable: {exc}", @@ -3156,7 +3259,7 @@ async def dashboard_wallet_get(request: Request) -> Response: return JSONResponse(status_code=502, content={"error": { "message": "router client unavailable", "type": "wallet_error", "code": "wallet"}}) try: - r = await _client.get(f"{UPSTREAM}/x/wallet", timeout=10.0) + r = await _client.get(f"{UPSTREAM}/x/wallet", timeout=10.0, headers=_router_admin_headers()) return JSONResponse(status_code=r.status_code, content=r.json()) except Exception as exc: return JSONResponse(status_code=502, content={"error": { @@ -3242,7 +3345,7 @@ async def _reload_codex_router() -> tuple[bool, str | None]: headers = {"authorization": f"Bearer {CODEX_BROKER_TOKEN}"} else: url = f"{UPSTREAM}/x/codex/reload" - headers = None + headers = _router_admin_headers() r = await _client.post(url, headers=headers, timeout=10.0) if r.status_code == 200: return True, None @@ -3671,7 +3774,8 @@ async def dashboard_set_config(request: Request) -> Response: note = None try: if _client is not None: - r = await _client.post(f"{UPSTREAM}/x/config/reload", timeout=10.0) + r = await _client.post(f"{UPSTREAM}/x/config/reload", timeout=10.0, + headers=_router_admin_headers()) applied_live = r.status_code == 200 if not applied_live: note = f"router /x/config/reload returned {r.status_code}" @@ -3769,7 +3873,9 @@ async def dashboard_create_key(request: Request) -> Response: now = int(time.time()) token = f"{DASHBOARD_KEY_PREFIX}_{secrets.token_urlsafe(32)}" token_hash = hashlib.sha256(token.encode()).hexdigest() - records = _issued_consumer_records() + records = _writable_consumer_records() + if records is None: + return _consumer_store_unavailable() meta = records.get(consumer, _normalize_consumer_record(consumer, {})) if rotate: # With sha256_prefix, rotation targets ONLY that active key; without it, @@ -3838,7 +3944,9 @@ async def dashboard_create_key_batch(request: Request) -> Response: "message": "invalid JSON body", "type": "invalid_request_error", "code": "invalid_json"}}) - records = _issued_consumer_records() + records = _writable_consumer_records() + if records is None: + return _consumer_store_unavailable() collisions = [consumer for _, consumer in members if consumer in records] if collisions: return JSONResponse(status_code=409, content={"error": { @@ -3978,10 +4086,33 @@ async def _synthetic_probe_loop() -> None: await _synthetic_probe_once(route) await asyncio.sleep(max(SYNTHETIC_PROBE_INTERVAL_S, 1.0)) +_PROXY_SEGMENT_RE = re.compile(r"[A-Za-z0-9_.:@+~-]+") +_PROXY_PROFILE_RE = re.compile(r"[A-Za-z0-9_-]{1,64}") + + +def _proxy_path(request: Request) -> str | None: + """The consumer API path to forward (`v1/...`, `/v1/...`, `api/...`), + or None. An allowlist over the DECODED path, with dot/empty segments and + encoded separators refused outright: the router would otherwise resolve + `/v1/..%2fx/calls` (or `%2e%2e`, `./x`, `//x`) to its internal /x/* API.""" + raw = (request.scope.get("raw_path") or b"").decode("latin-1").split("?", 1)[0].lower() + if "%2f" in raw or "%5c" in raw or "\\" in raw: + return None + segs = str(request.scope.get("path") or "").split("/")[1:] + if not segs or any(s in (".", "..") or not _PROXY_SEGMENT_RE.fullmatch(s) for s in segs): + return None + if (segs[0] in ("v1", "api") and len(segs) > 1) or ( + len(segs) > 2 and segs[1] == "v1" and segs[0] != "x" and _PROXY_PROFILE_RE.fullmatch(segs[0])): + return "/".join(segs) + return None + + @app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"]) async def proxy(path: str, request: Request) -> Response: - if path == "x" or path.startswith("x/"): - # Router-internal diagnostics (/x/runtime): never proxied to callers. + path = _proxy_path(request) + if path is None: + # Router-internal diagnostics (/x/*) and anything off the consumer API + # surface: never proxied to callers. return JSONResponse(status_code=404, content={"error": {"message": "not found", "type": "invalid_request_error", "code": None}}) started = time.perf_counter() token = _extract_token(request) @@ -4294,53 +4425,110 @@ async def _close_upstream_stream() -> None: upstream_closed = True await r.aclose() - async def _passthrough(): + # The upstream is read by a pump task, not by the client-facing + # generator: a client that disconnects before the final event would + # otherwise record the call at $0 and walk past budget_usd. After a + # disconnect the pump keeps draining (bounded in time and bytes). + queue: asyncio.Queue = asyncio.Queue(maxsize=64) + client_gone_at: float | None = None + drain_complete = pump_finished = False + + def _account(tail: bytes) -> None: nonlocal provider, model_family, served_model_id, served_by, \ tokens_in, tokens_out, tokens_total, tokens_cached, cost_usd, \ cost_basis, decision_trace, error_type, error_code, error_message + try: + meta = _parse_stream_tail(tail) + xr = meta.get("x_router") or {} + if xr: + provider = xr.get("provider") + model_family = xr.get("model_family") + served_model_id = xr.get("served_model_id") + served_by = xr.get("served_by") + decision_trace = xr.get("decision_trace") if isinstance(xr.get("decision_trace"), dict) else None + if isinstance(xr.get("cost_usd"), (int, float)): + cost_usd = float(xr["cost_usd"]) + cost_basis = xr.get("cost_basis") + if isinstance(xr.get("tokens_cached"), int): + tokens_cached = xr["tokens_cached"] + usage = meta.get("usage") + if isinstance(usage, dict): + tokens_in = int(usage.get("prompt_tokens") or 0) + tokens_out = int(usage.get("completion_tokens") or 0) + tokens_total = int(usage.get("total_tokens") or (tokens_in + tokens_out)) + err = meta.get("error") + if isinstance(err, dict): + error_type = err.get("type") + error_code = err.get("code") + error_message = err.get("message") + except Exception: + pass + if client_gone_at is not None and not drain_complete and cost_usd is None and status < 400: + # Cut off before the usage event: never book a served call + # as free. Same floor Cloud key budgets reserve per call. + cost_usd, cost_basis = CLOUD_BUDGET_RESERVATION_USD, "estimated_disconnect" + + def _client_left() -> None: + nonlocal client_gone_at + if client_gone_at is not None or pump_finished: + return + client_gone_at = time.monotonic() + while not queue.empty(): + queue.get_nowait() + asyncio.get_running_loop().call_later(STREAM_DRAIN_TIMEOUT_S, pump.cancel) + + async def _pump() -> None: + nonlocal drain_complete, pump_finished tail = bytearray() + drained = 0 + failure: BaseException | None = None try: async for chunk in r.aiter_raw(): tail.extend(chunk) _trim_sse_tail(tail) - yield chunk + if client_gone_at is None: + try: + # A reader that stalls this long is treated as gone. + await asyncio.wait_for(queue.put(chunk), STREAM_DRAIN_TIMEOUT_S) + continue + except asyncio.TimeoutError: + _client_left() + drained += len(chunk) + if drained > STREAM_DRAIN_MAX_BYTES: + break + else: + drain_complete = True + except asyncio.CancelledError: + pass + except Exception as exc: # noqa: BLE001 + failure = exc finally: await _close_upstream_stream() - try: - meta = _parse_stream_tail(bytes(tail)) - xr = meta.get("x_router") or {} - if xr: - provider = xr.get("provider") - model_family = xr.get("model_family") - served_model_id = xr.get("served_model_id") - served_by = xr.get("served_by") - decision_trace = xr.get("decision_trace") if isinstance(xr.get("decision_trace"), dict) else None - if isinstance(xr.get("cost_usd"), (int, float)): - cost_usd = float(xr["cost_usd"]) - cost_basis = xr.get("cost_basis") - if isinstance(xr.get("tokens_cached"), int): - tokens_cached = xr["tokens_cached"] - usage = meta.get("usage") - if isinstance(usage, dict): - tokens_in = int(usage.get("prompt_tokens") or 0) - tokens_out = int(usage.get("completion_tokens") or 0) - tokens_total = int(usage.get("total_tokens") or (tokens_in + tokens_out)) - err = meta.get("error") - if isinstance(err, dict): - error_type = err.get("type") - error_code = err.get("code") - error_message = err.get("message") - except Exception: - pass + _account(bytes(tail)) _finish() + pump_finished = True + if client_gone_at is None: + with contextlib.suppress(asyncio.TimeoutError): # None = end of stream + await asyncio.wait_for(queue.put(failure), STREAM_DRAIN_TIMEOUT_S) + + pump = asyncio.create_task(_pump()) + _stream_pumps.add(pump) + pump.add_done_callback(_stream_pumps.discard) + + async def _passthrough(): + try: + while (item := await queue.get()) is not None: + if isinstance(item, BaseException): + raise item + yield item + finally: + _client_left() async def _stream_background() -> None: - # Starlette normally closes the async generator on disconnect. - # The background fallback also covers a client that disconnects - # before the first iteration, so neither the upstream socket nor - # the admission permit can leak indefinitely. - await _close_upstream_stream() - _finish() + # Also covers a client that disconnects before the first + # iteration: the pump switches to bounded draining, and it — + # not the client — releases the upstream socket and the permit. + _client_left() return StreamingResponse( _passthrough(), status_code=status, media_type=content_type, @@ -4421,6 +4609,11 @@ def _read_usage_history(*, since: int | None = None, caller: str | None = None, _SSE_TAIL_HARD_CAP = 4 * 1024 * 1024 # 4 MiB backstop if no event boundary appears +STREAM_DRAIN_TIMEOUT_S = max(1.0, float(os.getenv("STREAM_DRAIN_TIMEOUT_S", "120"))) +STREAM_DRAIN_MAX_BYTES = max(1, int(os.getenv("STREAM_DRAIN_MAX_BYTES", str(32 * 1024 * 1024)))) +_stream_pumps: set[asyncio.Task] = set() # strong refs: a drain outlives its request + + def _trim_sse_tail(tail: bytearray) -> None: """Bound the streamed-response tail buffer WITHOUT splitting an SSE event. @@ -4886,8 +5079,7 @@ def _usage_events_for_key(digest: str, caller: str | None, prefix = digest[:12] with _stats_lock: memory_rows = [dict(r) for r in _stats["recent"] if r.get("key_sha256_prefix") == prefix] - history_rows = [dict(r) for r in _read_usage_history(since=since, caller=caller) - if r.get("key_sha256") == digest or r.get("key_sha256_prefix") == prefix] + history_rows = host_store.key_usage_rows(digest, since_ts=since, caller=caller) rows_by_id: dict[str, dict[str, Any]] = {} anonymous_rows: list[dict[str, Any]] = [] for row in history_rows + memory_rows: @@ -5283,7 +5475,9 @@ def _aggregate_usage_rows(rows: list[dict[str, Any]], *, selected: str | None = def _stats_snapshot(*, viewer: str, upstream_status: int, upstream_health: dict[str, Any], consumer: str | None = None, timeframe: str = "recent", key_sha256: str | None = None, viewer_role: str = "admin", provider: str | None = None, model: str | None = None) -> dict[str, Any]: - selected = consumer if consumer in _consumers() else None + # A consumer viewer is always scoped to itself, even if the consumer store + # read failed and its name is missing from _consumers(). + selected = consumer if (viewer_role != "admin" or consumer in _consumers()) else None provider = (provider or "").strip() or None model = (model or "").strip() or None key_sha256 = str(key_sha256 or "").strip().lower() diff --git a/byo_http.py b/byo_http.py index f66b9c5..dd53c1a 100644 --- a/byo_http.py +++ b/byo_http.py @@ -10,6 +10,19 @@ import httpx +# Prefixes that embed (and route to) an IPv4 address: NAT64 well-known/local-use, +# IPv4-compatible, 6to4. `is_global` judges the IPv6 wrapper, not the target. +_EMBEDDED_V4 = tuple(ipaddress.ip_network(n) for n in + ('64:ff9b::/96', '64:ff9b:1::/48', '::/96', '2002::/16')) + + +def public_address(ip): + if ip.version == 6 and ip.ipv4_mapped is not None: + ip = ip.ipv4_mapped + if ip.version == 6 and any(ip in net for net in _EMBEDDED_V4): + return False + return ip.is_global + class PublicHTTPS(httpx.AsyncBaseTransport): def __init__(self): @@ -25,7 +38,7 @@ async def handle_async_request(self, request): rows = await asyncio.wait_for(asyncio.get_running_loop().getaddrinfo( url.host, url.port or 443, type=socket.SOCK_STREAM), 5) addresses = [ipaddress.ip_address(row[4][0]) for row in rows] - if not addresses or any(not ip.is_global for ip in addresses): + if not addresses or not all(public_address(ip) for ip in addresses): raise ValueError('non-public endpoint') except (ValueError, OSError, TimeoutError) as exc: raise httpx.ConnectError('Buyer gateway is not a public HTTPS endpoint', request=request) from exc diff --git a/codex_backend.py b/codex_backend.py index 018a49a..e4b12dd 100644 --- a/codex_backend.py +++ b/codex_backend.py @@ -374,7 +374,9 @@ async def call(request: dict) -> dict: return _err("auth_error", 0, 0, "no codex access token (run `codex login`)") body = build_codex_body(request) headers = build_codex_headers(token, acct.account_id(), extra_headers) - url = (request.get("base_url") or base_url).rstrip("/") + "/responses" + # The OAuth token only ever goes to the configured upstream: a request's + # base_url (broker /v1/call bodies, policy output) is never trusted here. + url = base_url.rstrip("/") + "/responses" timeout = (request.get("timeout_ms") or int(timeout_s * 1000)) / 1000.0 t0 = time.monotonic() diff --git a/codex_broker.py b/codex_broker.py index 73fe838..656aa8d 100644 --- a/codex_broker.py +++ b/codex_broker.py @@ -64,7 +64,7 @@ def authorized(request: Request) -> JSONResponse | None: return JSONResponse(status_code=503, content={ "error": "CODEX_BROKER_TOKEN is not configured"}) supplied = _bearer(request) - if not supplied or not hmac.compare_digest(supplied, expected_token): + if not supplied or not hmac.compare_digest(supplied.encode(), expected_token.encode()): return JSONResponse(status_code=401, content={"error": "unauthorized"}) return None diff --git a/compose.yml b/compose.yml index 72ed74f..83b7f37 100644 --- a/compose.yml +++ b/compose.yml @@ -20,19 +20,28 @@ services: # compose runs antseed on its own network; socat re-exposes the # localhost-only buyer proxy on :8378 (k8s leaves this unset -> localhost) ANTSEED_BASE_URL: http://antseed:8378/v1 + # Bearer for the sidecar's funded buyer proxy on :8378 (same value on the + # antseed service). Unset -> the sidecar falls back to unauthenticated socat. + ANTSEED_PROXY_TOKEN: ${ANTSEED_PROXY_TOKEN:-} # wallet control server in the sidecar (deposit/withdraw/refresh from the # dashboard). Unset token -> the wallet endpoints return 503. k8s: same pod # -> http://127.0.0.1:8379. ANTSEED_CONTROL_URL: http://antseed:8379 ANTSEED_CONTROL_TOKEN: ${ANTSEED_CONTROL_TOKEN:-} # Operational store (router reads operator config from it). Prod = RDS. - DATABASE_URL: ${DATABASE_URL:-postgresql://hoststore:hoststore@postgres:5432/hoststore} + DATABASE_URL: ${DATABASE_URL:-postgresql://hoststore:${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in .env}@postgres:5432/hoststore} # 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:-} + # Router /x/* admin endpoints (wallet, providers, keys, reloads, calls, + # sessions) require x-internal-secret = CONTROL_PLANE_INTERNAL_SECRET; with + # no secret only a loopback caller may use them. The ingress reaches the + # router across the compose network, so set the secret (both services) or, + # for a trusted local stack only, opt in to the old unauthenticated mode. + ROUTER_ADMIN_ALLOW_UNAUTHENTICATED: ${ROUTER_ADMIN_ALLOW_UNAUTHENTICATED:-0} SAAS_SHARED_PROVIDERS: ${SAAS_SHARED_PROVIDERS:-} depends_on: postgres: @@ -88,7 +97,7 @@ services: ROUTER_UPSTREAM: http://router:18080 # Operational store: the ingress is the writer (consumer keys, the ledger, # operator-config writes). Prod = RDS. - DATABASE_URL: ${DATABASE_URL:-postgresql://hoststore:hoststore@postgres:5432/hoststore} + DATABASE_URL: ${DATABASE_URL:-postgresql://hoststore:${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in .env}@postgres:5432/hoststore} RATE_PER_MIN: ${RATE_PER_MIN:-600} BURST: ${BURST:-200} # Optional external control plane: consumer-key resolve fallback plus the @@ -157,10 +166,15 @@ services: # Wallet control server (deposit/withdraw/status from the dashboard). # Shared secret with the router; unset -> control.js does not start. ANTSEED_CONTROL_PORT: ${ANTSEED_CONTROL_PORT:-8379} + # control.js binds 127.0.0.1 by default (k8s: same pod as the router); + # here the router reaches it across the compose network. + ANTSEED_CONTROL_HOST: 0.0.0.0 ANTSEED_CONTROL_TOKEN: ${ANTSEED_CONTROL_TOKEN:-} + # >= 32 chars; the router sends it as a bearer (see router env). + ANTSEED_PROXY_TOKEN: ${ANTSEED_PROXY_TOKEN:-} # The sidecar writes the marketplace book (peer_offers) into the shared # host store; same DATABASE_URL as router/ingress. Prod = RDS. - DATABASE_URL: ${DATABASE_URL:-postgresql://hoststore:hoststore@postgres:5432/hoststore} + DATABASE_URL: ${DATABASE_URL:-postgresql://hoststore:${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in .env}@postgres:5432/hoststore} volumes: - antseed-data:/data depends_on: @@ -196,7 +210,10 @@ services: restart: unless-stopped environment: POSTGRES_USER: hoststore - POSTGRES_PASSWORD: hoststore + # No default: a well-known password on the shared host store is a + # credential anyone who reaches the network already has. Set it in .env + # (an existing postgres-data volume keeps its old password: set that one). + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in .env} POSTGRES_DB: hoststore volumes: - postgres-data:/var/lib/postgresql/data diff --git a/control_plane_client.py b/control_plane_client.py index 136509a..38634b7 100644 --- a/control_plane_client.py +++ b/control_plane_client.py @@ -71,8 +71,13 @@ def internal_secret_ok(headers: Mapping[str, str]) -> bool: secret is unconfigured — callers must treat that as 'feature hidden'.""" if not CONTROL_PLANE_INTERNAL_SECRET: return False - presented = headers.get("x-internal-secret") or "" - return hmac.compare_digest(presented, CONTROL_PLANE_INTERNAL_SECRET) + return secret_equal(headers.get("x-internal-secret") or "", CONTROL_PLANE_INTERNAL_SECRET) + + +def secret_equal(presented: str, expected: str) -> bool: + """Constant-time compare over UTF-8 bytes: hmac.compare_digest raises + TypeError on non-ASCII str, which would turn a junk header into a 500.""" + return hmac.compare_digest(str(presented).encode(), str(expected).encode()) @dataclass diff --git a/core b/core index d963280..2f43c47 160000 --- a/core +++ b/core @@ -1 +1 @@ -Subproject commit d963280211092e149051d1b200f85a40bc946563 +Subproject commit 2f43c47bf3e0c8de6df35cebaa0d19bcdcf28992 diff --git a/env_secrets.py b/env_secrets.py index 7bf5c5c..d4d7cef 100644 --- a/env_secrets.py +++ b/env_secrets.py @@ -24,6 +24,21 @@ DEFAULT_ENV_SECRETS_PATH = "/run/llm-router/.env.secrets" +_FORBIDDEN_ENV_PREFIXES = ("CONTROL_PLANE_", "ANTSEED_", "CODEX_", "AWS_", "DASHBOARD_", + "ROUTER_", "POSTGRES", "DATABASE", "CP_", "SAAS_", "GITHUB_") +_FORBIDDEN_ENV_PARTS = ("SECRET", "PASSWORD", "INTERNAL", "ADMIN", "SESSION") + + +def is_forbidden_auth_env(name: str) -> bool: + """Infrastructure/security variables a dashboard write must never set.""" + name = str(name or "") + return name.startswith(_FORBIDDEN_ENV_PREFIXES) or any(p in name for p in _FORBIDDEN_ENV_PARTS) + + +def _protected(key: str) -> bool: + return is_forbidden_auth_env(key) or "URL" in key + + def load_env_secrets(path: str | os.PathLike | None = None) -> list[str]: """Merge KEY=value lines from the env-secrets file into os.environ. @@ -42,7 +57,10 @@ def load_env_secrets(path: str | os.PathLike | None = None) -> list[str]: continue key, value = line.split("=", 1) key = key.strip() - if not key: + # The file is dashboard-writable: it may carry provider keys and the + # dashboard's own key maps, never override infrastructure/security + # variables (DB DSN, internal secrets, wallet/control tokens). + if not key or (key in os.environ and _protected(key)): continue os.environ[key] = value loaded.append(key) diff --git a/host_store.py b/host_store.py index a62b78e..7f7bd69 100644 --- a/host_store.py +++ b/host_store.py @@ -1259,7 +1259,19 @@ def tool_incapable_routes(window_ms: int = 1_800_000, min_samples: int = 20) -> "tokens_cached": 0, "cost_usd": 0.0} -def hot_route(session: "str | None") -> "str | None": +def _session_where(session: str, owner: "str | None") -> "tuple[str, list]": + """A session id is client-chosen, so two callers can collide on one: with an + owner (the ingress-authenticated caller) every read sees only that caller's + calls. Bounded to the retention horizon (older rows are pruned anyway).""" + where = " WHERE session_id = %s AND ts >= %s" + params: list = [session, int(time.time()) - _retention_days() * 86_400] + if owner is not None: + where += " AND caller = %s" + params.append(owner) + return where, params + + +def hot_route(session: "str | None", owner: "str | None" = None) -> "str | None": """The route (provider|family|served_by) that most recently served this session SUCCESSFULLY — its prompt-cache prefix is hot there. None when unknown. Matches route_reliability.route_key; resolved per request into the cache_hot @@ -1267,29 +1279,33 @@ def hot_route(session: "str | None") -> "str | None": if not session: return None try: + where, params = _session_where(session, owner) with _get_pool().connection() as conn: + _set_dashboard_statement_timeout(conn) row = conn.execute( - "SELECT provider_id, model_family, served_by FROM calls" - " WHERE session_id = %s AND status < 400 AND served_by IS NOT NULL" - " ORDER BY ts DESC, id DESC LIMIT 1", (session,)).fetchone() + "SELECT provider_id, model_family, served_by FROM calls" + where + + " AND status < 400 AND served_by IS NOT NULL" + " ORDER BY ts DESC, id DESC LIMIT 1", params).fetchone() return f"{row[0]}|{row[1]}|{row[2]}" if row else None except Exception as exc: # noqa: BLE001 _log.warning("host_store hot_route failed: %s", exc) return None -def session_totals(session: "str | None") -> dict[str, Any]: +def session_totals(session: "str | None", owner: "str | None" = None) -> dict[str, Any]: """The session's running totals (calls, tokens_in/out/cached, cost_usd) summed over its committed `calls`. The caller adds the in-flight call on top. Fail-soft -> zeros.""" if not session: return dict(_SESSION_TOTALS_ZERO) try: + where, params = _session_where(session, owner) with _get_pool().connection() as conn: + _set_dashboard_statement_timeout(conn) r = conn.execute( "SELECT count(*), coalesce(sum(tokens_in),0), coalesce(sum(tokens_out),0)," " coalesce(sum(tokens_cached),0), coalesce(sum(cost_usd),0)" - " FROM calls WHERE session_id = %s", (session,)).fetchone() + " FROM calls" + where, params).fetchone() return {"calls": int(r[0]), "tokens_in": int(r[1]), "tokens_out": int(r[2]), "tokens_cached": int(r[3]), "cost_usd": round(float(r[4]), 6)} except Exception as exc: # noqa: BLE001 @@ -1297,18 +1313,20 @@ def session_totals(session: "str | None") -> dict[str, Any]: return dict(_SESSION_TOTALS_ZERO) -def session_warm(session: "str | None") -> list[dict[str, Any]]: +def session_warm(session: "str | None", owner: "str | None" = None) -> list[dict[str, Any]]: """The session's warm routes for DISPLAY: per family, the most recent route that served it successfully ({family, provider, served_by}). Fail-soft -> [].""" if not session: return [] try: + where, params = _session_where(session, owner) with _get_pool().connection() as conn: + _set_dashboard_statement_timeout(conn) cur = conn.execute( "SELECT DISTINCT ON (model_family) model_family, provider_id, served_by" - " FROM calls WHERE session_id = %s AND status < 400" + " FROM calls" + where + " AND status < 400" " AND model_family IS NOT NULL" - " ORDER BY model_family, ts DESC, id DESC", (session,)) + " ORDER BY model_family, ts DESC, id DESC LIMIT 64", params) return [{"family": f, "provider": p, "served_by": s or p} for f, p, s in cur.fetchall()] except Exception as exc: # noqa: BLE001 @@ -1322,10 +1340,12 @@ def session_owner(session: "str | None") -> "str | None": if not session: return None try: + where, params = _session_where(session, None) with _get_pool().connection() as conn: + _set_dashboard_statement_timeout(conn) row = conn.execute( - "SELECT caller FROM calls WHERE session_id = %s" - " ORDER BY ts ASC, id ASC LIMIT 1", (session,)).fetchone() + "SELECT caller FROM calls" + where + + " ORDER BY ts ASC, id ASC LIMIT 1", params).fetchone() return row[0] if row else None except Exception as exc: # noqa: BLE001 _log.warning("host_store session_owner failed: %s", exc) @@ -1336,11 +1356,14 @@ def all_session_totals() -> dict[str, dict[str, Any]]: """Every session's totals (operator /x/sessions view), derived from calls.""" try: with _get_pool().connection() as conn: + _set_dashboard_statement_timeout(conn) cur = conn.execute( "SELECT session_id, count(*), coalesce(sum(tokens_in),0)," " coalesce(sum(tokens_out),0), coalesce(sum(tokens_cached),0)," " coalesce(sum(cost_usd),0) FROM calls" - " WHERE session_id IS NOT NULL GROUP BY session_id") + " WHERE session_id IS NOT NULL AND ts >= %s GROUP BY session_id" + " ORDER BY max(ts) DESC LIMIT 5000", + [int(time.time()) - _retention_days() * 86_400]) return {s: {"calls": int(c), "tokens_in": int(ti), "tokens_out": int(to), "tokens_cached": int(tc), "cost_usd": round(float(cu), 6)} for s, c, ti, to, tc, cu in cur.fetchall()} @@ -2634,3 +2657,22 @@ def truncate_all_for_tests() -> None: " consumer_budget_usage, analytics_hourly, analytics_rollup_state, peer_offers, buyer_status, route_observations," " login_history, provider_prices, wallet_ops," " subject_budget_usage, subject_budget_reservations") + + +def key_usage_rows(consumer_sha: str, since_ts: "int | None" = None, + caller: "str | None" = None, limit: int = 20_000) -> list[dict[str, Any]]: + """One key's usage rows for the consumer-facing /v1/usage: filtered by + consumer_sha IN SQL (idx_calls_consumer), newest `limit` in the window, under + the dashboard statement timeout — never the caller's whole 30-day history + loaded and filtered in Python. Fail-soft -> [].""" + try: + where, params = _usage_where(since_ts=since_ts, caller=caller, consumer_sha=consumer_sha) + with _get_pool().connection() as conn: + _set_dashboard_statement_timeout(conn) + cur = conn.execute( + f"SELECT {', '.join(_USAGE_COLS)} FROM calls{where}" + " ORDER BY ts DESC, id ASC LIMIT %s", params + [max(1, int(limit))]) + return [_map_usage_row(r) for r in cur.fetchall()] + except Exception as exc: # noqa: BLE001 + _log.warning("host_store key_usage_rows failed: %s", exc) + return [] diff --git a/internal_api.py b/internal_api.py index 8ef4808..2d83089 100644 --- a/internal_api.py +++ b/internal_api.py @@ -39,6 +39,7 @@ _BUDGET_SUBJECTS_MAX = 500 _SUBJECT_RE = re.compile(r"[gk]:[1-9][0-9]{0,17}") _PERIOD_RE = re.compile(r"[0-9]{4}-(0[1-9]|1[0-2])") +_MAX_TS = 253402300799 # 9999-12-31T23:59:59Z: beyond it the ledger's to_timestamp overflows def _gate(request: Request) -> JSONResponse | None: @@ -74,6 +75,9 @@ async def internal_usage(request: Request, caller: str = "", return JSONResponse({"error": "caller_required"}, status_code=400) if group_by is not None and group_by not in host_store._USAGE_GROUPS: return JSONResponse({"error": "invalid_group_by"}, status_code=400) + if any(ts is not None and not 0 <= ts <= _MAX_TS for ts in (since_ts, until_ts)) or ( + since_ts is not None and until_ts is not None and until_ts < since_ts): + return JSONResponse({"error": "invalid_window"}, status_code=400) window = {"since_ts": since_ts, "until_ts": until_ts} try: totals = await asyncio.to_thread(host_store.usage_totals, caller=caller, strict=True, **window, **scope) diff --git a/llm_router_host.py b/llm_router_host.py index 1a23751..f75a446 100644 --- a/llm_router_host.py +++ b/llm_router_host.py @@ -24,6 +24,7 @@ import host_store import route_reliability as _route_reliability +from byo_http import is_byo_buyer from provider_adapters.anthropic import make_anthropic_async_call_provider from provider_adapters.bedrock import make_bedrock_async_call_provider from provider_adapters.common import ( @@ -84,9 +85,29 @@ Clock = Callable[[], int] _AUTH_UNCONFIGURED = "auth_unconfigured" +# Host-side backstop for the engine's per-execution cap (core DEFAULTS). +_MAX_PROVIDER_ATTEMPTS = 32 _SECRET_PLACEHOLDERS = {"", "CHANGE_ME", "TODO", "TODO_CHANGE_ME", "PLACEHOLDER"} +def _operator_local(provider: dict) -> bool: + """Whether a provider endpoint is on the operator's own network.""" + import ipaddress + from urllib.parse import urlsplit + try: + parts = urlsplit(str(provider.get("base_url") or "")) + host = (parts.hostname or "").rstrip(".").lower() + except ValueError: + return True + if parts.scheme not in ("http", "https") or not host: + return False # no HTTP endpoint: the adapter's fixed public API (bedrock://, ...) + try: + return not ipaddress.ip_address(host).is_global + except ValueError: + return ("." not in host or host.endswith((".localhost", ".local", ".internal", ".svc")) + or ".svc." in host) + + def _secret_configured(value: str | None) -> bool: if value is None: return False @@ -131,6 +152,7 @@ def __init__( self._source_paths = (router_path, config_path, metrics_path) self._tenant_id = None self._tenant_allowed = None + self._tenant_managed = None self.lua = LuaRuntime(unpack_returned_tuples=True) self._custom_call_hook = call_provider is not None @@ -275,7 +297,9 @@ def for_tenant(self, tenant_id: int, env: dict, managed_providers=(), connection allowed = set(managed) for pid, provider in (catalog.get('providers') or {}).items(): key = auth_env(provider) - if key and env.get(key): + # A tenant key never unlocks an operator-local endpoint (e.g. a + # keyless local Ollama): only an explicit operator share can. + if key and env.get(key) and not _operator_local(provider): allowed.add(pid) managed.discard(pid) # BYO credentials take precedence for this provider. elif key and pid in managed and self._env.get(key): @@ -640,6 +664,10 @@ async def execute_async(self, contract: dict, call_override=None) -> dict: engine_contract = {**contract, 'decision': json.dumps(contract.get('decision'), allow_nan=False)} step = self.router.execute_step(None, _to_lua(self.lua, engine_contract), None) while True: + # A provider hook that fails without awaiting (and a zero-backoff + # retry) would otherwise spin here without ever yielding, so the + # request deadline could never cancel it. + await asyncio.sleep(0) status = step["status"] if status == "done": result = _to_py(step["result"]) @@ -650,7 +678,10 @@ async def execute_async(self, contract: dict, call_override=None) -> dict: handle = step["state_handle"] if status == "call": provider_attempt += 1 - req = _to_py(step["request"]) or {} + if provider_attempt > _MAX_PROVIDER_ATTEMPTS: + return {"ok": False, "error": "exhausted: attempt_cap", + "trace": {"provider_diagnostics": provider_diagnostics}} + req = self._pin_route(_to_py(step["request"]) or {}) if req.get('protocol') == 'decisions': req['decision'] = json.loads(req['decision']) if (contract.get("first_token_timeout_ms") is not None @@ -676,6 +707,19 @@ async def execute_async(self, contract: dict, call_override=None) -> dict: else: return {"ok": False, "error": f"internal: bad step status {status}", "trace": {}} + def _pin_route(self, req: dict) -> dict: + """Where a call goes and which credential it carries come from the + catalog, never from the engine's (policy-transformed) request.""" + provider = self.config.providers[req["provider_id"]] if req.get("provider_id") else None + if provider is None: + return req # per-call extra candidate: no catalog row to pin to + for key in ("auth_env", "auth", "api_kind", "aws_region"): + req[key] = _to_py(provider[key]) + offer = req.get("offer") + req["base_url"] = ((offer.get("seller_endpoint") if isinstance(offer, dict) else None) + if provider.discovery == "marketplace" else None) or provider.base_url + return req + async def _resolve_call_async(self, request: dict, call_override=None, session: "str | None" = None) -> dict: """Resolve one provider call for the async driver: mock first (so the @@ -695,8 +739,14 @@ async def _resolve_call_async(self, request: dict, call_override=None, # observation too, the host-owned perf the algebra reads (derived) and the # market view surfaces (#15/#4a). Mocks record as well, so a mocked call # is measured exactly like a live one. - if not (self._tenant_id is not None and result.get("error_kind") in - {"auth_error", "rate_limit", "payment_required"}): + # Shared route_observations are operator evidence. A tenant outcome counts + # only on an operator-managed route (operator credential and endpoint): + # BYO gateways/keys are tenant-controlled and could forge peer health, + # cooldowns or tool capability for everyone. + if self._tenant_id is None or ( + request.get("provider_id") in (self._tenant_managed or ()) + and not is_byo_buyer(request, self._env.get) + and result.get("error_kind") not in {"auth_error", "rate_limit", "payment_required"}): _fold_route_outcome(request, result, session=session) return result @@ -862,7 +912,7 @@ def _h_env(self, key): return self._env.get(key) def _h_call_provider(self, request): - py_req = _to_py(request) or {} + py_req = self._pin_route(_to_py(request) or {}) provider = py_req.get("provider_id") model = py_req.get("model_family") if (provider, model) in self._mock_responses: diff --git a/provider_adapters/decisions.py b/provider_adapters/decisions.py index 930ba1c..b7a5b31 100644 --- a/provider_adapters/decisions.py +++ b/provider_adapters/decisions.py @@ -67,7 +67,7 @@ async def send(http): try: async with asyncio.timeout(timeout): - slot, gate_error = await _acquire_peer_capacity(request, timeout) + slot, gate_error = await _acquire_peer_capacity(request, timeout, env_get) if gate_error: return _peer_capacity_error(str(offer.get('peer_id') or ''), int(offer.get('max_concurrency') or 0), gate_error, started) diff --git a/provider_adapters/openai_compatible.py b/provider_adapters/openai_compatible.py index c5b72e4..4fd52b2 100644 --- a/provider_adapters/openai_compatible.py +++ b/provider_adapters/openai_compatible.py @@ -180,6 +180,9 @@ def _prepare_openai_call( auth_headers = {} headers = {"Content-Type": "application/json", **auth_headers, **extra} + if provider_id.startswith("antseed") and os.getenv("ANTSEED_PROXY_TOKEN"): + # The operator's funded buyer proxy (antseed/public-proxy.js) requires it. + headers["Authorization"] = "Bearer " + os.environ["ANTSEED_PROXY_TOKEN"] from byo_http import is_byo_buyer if is_byo_buyer(request, env_get): # Never trust an offer's endpoint to choose where a tenant secret goes. @@ -354,7 +357,7 @@ async def release(self) -> None: async def _acquire_peer_capacity( - request: dict, call_timeout_s: float) \ + request: dict, call_timeout_s: float, env_get=None) \ -> tuple[_PeerCapacitySlot | None, str | None]: """Acquire the selected seller's cap locally or across all replicas. @@ -367,6 +370,11 @@ async def _acquire_peer_capacity( if not peer_id or not isinstance(cap, int) or isinstance(cap, bool) or cap <= 0: return None, None peer_id = str(peer_id).strip().lower() + from byo_http import is_byo_buyer + if env_get is not None and is_byo_buyer(request, env_get): + # A BYO offer's peer_id/max_concurrency come from the tenant's gateway; + # never let them size or starve the operator's shared gate for that peer. + peer_id = f"tenant:{env_get('SAAS_TENANT_SCOPE')}:{peer_id}" wait_s = _peer_gate_wait_s(call_timeout_s) if not _enabled("DISTRIBUTED_PEER_GATES"): gate = await _peer_gate(peer_id, cap) @@ -447,7 +455,7 @@ async def call(request: dict) -> dict: timing = ProviderTiming(request, timeout, "buffered_sse" if uses_streaming_backend else "buffered") slot = None if not uses_streaming_backend: - slot, gate_error = await _acquire_peer_capacity(request, timeout) + slot, gate_error = await _acquire_peer_capacity(request, timeout, _env_get) if gate_error: return timing.attach(_peer_capacity_error( str(peer_id or ""), int(cap or 0), gate_error, t0)) @@ -480,7 +488,7 @@ async def _ignore_delta(_delta: str) -> None: if is_byo_buyer(request, _env_get): async with buyer_client() as buyer: timing.data["phase"] = "connection_pool" - resp = await buyer.post(url, json=body, headers=headers, timeout=timeout, **timing.http_options(buyer)) + resp = await _post_bounded(buyer, url, json=body, headers=headers, timeout=timeout, **timing.http_options(buyer)) elif client is not None: timing.data["phase"] = "connection_pool" resp = await client.post( @@ -499,6 +507,8 @@ async def _ignore_delta(_delta: str) -> None: f"POST {url} timed out") except (httpx.NetworkError, httpx.RequestError) as e: result = _err("network_error", 0, _elapsed_ms(t0), str(e)) + except ResponseTooLarge as e: + result = _err("bad_response", 200, _elapsed_ms(t0), str(e)) return timing.attach(result) finally: if slot is not None: @@ -507,6 +517,56 @@ async def _ignore_delta(_delta: str) -> None: return call +class ResponseTooLarge(Exception): + """A tenant (BYO) gateway sent more than the configured byte caps.""" + + +def _byo_caps() -> tuple[int, int]: + return (int(os.getenv("BYO_MAX_RESPONSE_BYTES", str(32 << 20))), + int(os.getenv("BYO_MAX_SSE_LINE_BYTES", str(1 << 20)))) + + +async def _post_bounded(http, url: str, **kwargs): + """POST and buffer at most BYO_MAX_RESPONSE_BYTES of the body.""" + import httpx + cap = _byo_caps()[0] + async with http.stream("POST", url, **kwargs) as resp: + raw = bytearray() + async for chunk in resp.aiter_bytes(): + raw.extend(chunk) + if len(raw) > cap: + raise ResponseTooLarge(f"response exceeds {cap} bytes") + return httpx.Response(resp.status_code, headers=resp.headers, + content=bytes(raw), request=resp.request) + + +async def _read_head(resp, limit: int = 4096) -> bytes: + raw = b"" + async for chunk in resp.aiter_bytes(): + raw += chunk + if len(raw) >= limit: + break + return raw[:limit] + + +async def _bounded_lines(resp): + """aiter_lines with a per-line and a total byte cap.""" + total_cap, line_cap = _byo_caps() + buf, total = b"", 0 + async for chunk in resp.aiter_bytes(): + total += len(chunk) + if total > total_cap: + raise ResponseTooLarge(f"stream exceeds {total_cap} bytes") + buf += chunk + *lines, buf = buf.split(b"\n") + if len(buf) > line_cap or any(len(line) > line_cap for line in lines): + raise ResponseTooLarge(f"SSE line exceeds {line_cap} bytes") + for line in lines: + yield line.rstrip(b"\r").decode("utf-8", "replace") + if buf: + yield buf.rstrip(b"\r").decode("utf-8", "replace") + + def _classify_from_map(err_msg: str, error_map: dict | None) -> str | None: """Provider-declared body-substring -> canonical kind.""" if not error_map: @@ -571,7 +631,7 @@ async def _stream_openai_compatible_impl( offer = request.get("offer") or {} peer_id = offer.get("peer_id") cap = offer.get("max_concurrency") - slot, gate_error = await _acquire_peer_capacity(request, timeout) + slot, gate_error = await _acquire_peer_capacity(request, timeout, env_get or os.environ.get) if gate_error: return _peer_capacity_error( str(peer_id or ""), int(cap or 0), gate_error, t0) @@ -615,12 +675,13 @@ def _timeout_err() -> dict: except (asyncio.TimeoutError, TimeoutError): return _timeout_err() if not (200 <= resp.status_code < 300): - raw = (await resp.aread()).decode("utf-8", "replace")[:500] + raw = (await _read_head(resp) if _byo else await resp.aread() + ).decode("utf-8", "replace")[:500] kind = _classify_from_map(raw, rules.get("error_map")) \ or _classify_status(resp.status_code, raw) return _err(kind, resp.status_code, _latency(), raw) - lines = resp.aiter_lines().__aiter__() + lines = (_bounded_lines(resp) if _byo else resp.aiter_lines()).__aiter__() while True: try: line = await before_first_output( @@ -675,6 +736,8 @@ def _timeout_err() -> dict: acc["function"]["name"] = fn["name"] if fn.get("arguments"): acc["function"]["arguments"] += fn["arguments"] + except ResponseTooLarge as exc: + return _err("stream_interrupted" if emitted else "bad_response", 200, _latency(), str(exc)) except Exception as exc: # noqa: BLE001 — classified below partial = "".join(text_parts) if emitted: diff --git a/provider_overlay.py b/provider_overlay.py index 8f6ec69..557828c 100644 --- a/provider_overlay.py +++ b/provider_overlay.py @@ -24,12 +24,26 @@ """ from __future__ import annotations +import ipaddress import re +from urllib.parse import urlsplit import host_store +from env_secrets import is_forbidden_auth_env # noqa: F401 (re-exported) _ID_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{1,39}$") # 2-40 chars (1 + up to 39) -_ENV_RE = re.compile(r"^[A-Z][A-Z0-9_]{1,60}$") +# A runtime-added key names a provider credential, never infrastructure: the +# value is written to .env.secrets, which load_env_secrets() applies at boot. +AUTH_ENV_RE = re.compile(r"^[A-Z][A-Z0-9_]{0,60}_(API_KEY|TOKEN)\Z") + + +def _private_host(host: str) -> bool: + host = (host or "").rstrip(".").lower() + try: + return not ipaddress.ip_address(host).is_global + except ValueError: + return (not host or host == "localhost" or "." not in host + or host.endswith((".localhost", ".local", ".internal", ".svc")) or ".svc." in host) def load_overlay() -> dict: @@ -45,20 +59,35 @@ def save_overlay(overlay: dict) -> bool: return host_store.set_provider_overlays((overlay or {}).get("providers") or {}) -def validate_entry(pid: str, entry: dict, catalog: dict) -> list[str]: - """Validation errors for one overlay provider against the loaded catalog.""" +def validate_entry(pid: str, entry: dict, catalog: dict, *, key_supplied: bool = False) -> list[str]: + """Validation errors for one overlay provider against the loaded catalog. + An existing provider's auth_env may be named only together with a new key: + otherwise the operator's credential would be sent to this base_url.""" errors: list[str] = [] if not _ID_RE.match(pid or ""): errors.append("id must be lowercase [a-z0-9_-], 2-40 chars") if pid in (catalog.get("providers") or {}): errors.append(f"provider {pid!r} already exists") base_url = str(entry.get("base_url") or "") - if not base_url.startswith(("http://", "https://")): - errors.append("base_url must be http(s)") + try: + parts = urlsplit(base_url) + host = parts.hostname or "" + except ValueError: + parts, host = None, "" + if parts is None or parts.scheme != "https" or parts.username or parts.password: + errors.append("base_url must be https") + elif _private_host(host): + errors.append("base_url must be a public host") if entry.get("api_kind", "openai_compatible") != "openai_compatible": errors.append("only api_kind=openai_compatible can be added at runtime") - if not _ENV_RE.match(str(entry.get("auth_env") or "")): - errors.append("auth_env must be UPPER_SNAKE_CASE") + auth_env = str(entry.get("auth_env") or "") + if not AUTH_ENV_RE.match(auth_env) or is_forbidden_auth_env(auth_env): + errors.append("auth_env must be an UPPER_SNAKE_CASE *_API_KEY/*_TOKEN provider " + "credential name, not an infrastructure variable") + elif not key_supplied and any( + isinstance(p, dict) and auth_env in (p.get("auth_env"), (p.get("auth") or {}).get("env")) + for p in (catalog.get("providers") or {}).values()): + errors.append(f"auth_env {auth_env!r} belongs to an existing provider; supply a key") served = entry.get("served_models") if not isinstance(served, list) or not served: errors.append("served_models must be a non-empty list") diff --git a/remote_codex.py b/remote_codex.py index 67cbf19..fead762 100644 --- a/remote_codex.py +++ b/remote_codex.py @@ -7,7 +7,9 @@ """ from __future__ import annotations +import ipaddress import json +import os import time from typing import Any, Awaitable, Callable @@ -18,6 +20,24 @@ Emit = Callable[[str], Awaitable[None]] +def _broker_transport_ok(url: str) -> bool: + """The broker bearer rides every call: plain http never leaves the pod or + cluster network (loopback, single-label service names, *.svc/*.internal).""" + parsed = httpx.URL(url) + host = (parsed.host or "").rstrip(".").lower() + if parsed.scheme == "https": + return bool(host) + if parsed.scheme != "http" or not host: + return False + if os.getenv("CODEX_BROKER_ALLOW_HTTP", "0").lower() in {"1", "true", "yes"}: + return True + try: + return ipaddress.ip_address(host).is_loopback + except ValueError: + return (host == "localhost" or "." not in host + or host.endswith((".svc", ".svc.cluster.local", ".internal"))) + + class RemoteCodexClient: def __init__(self, base_url: str, token: str, *, timeout_s: float = 120.0, client: Any = None): @@ -25,6 +45,9 @@ def __init__(self, base_url: str, token: str, *, token = str(token or "").strip() if not base_url: raise ValueError("CODEX_BROKER_URL must not be empty") + if not _broker_transport_ok(base_url): + raise ValueError("CODEX_BROKER_URL must be https (plain http only to a " + "loopback/cluster-internal host, or CODEX_BROKER_ALLOW_HTTP=1)") if not token: raise ValueError("CODEX_BROKER_TOKEN must not be empty") self.base_url = base_url diff --git a/shim.py b/shim.py index 3a3c57c..d1defa5 100644 --- a/shim.py +++ b/shim.py @@ -28,7 +28,10 @@ from __future__ import annotations import asyncio +import hmac import logging +import os +import re import time import uuid from typing import Any @@ -321,7 +324,20 @@ async def _execute_with_deadline(awaitable): from saas_routes import ScopedHost, automatic_trace, install as install_saas host = ScopedHost(host) - app = FastAPI(title="llm-router shim", docs_url=None, redoc_url=None) + app = FastAPI(title="llm-router shim", docs_url=None, redoc_url=None, openapi_url=None) + + @app.middleware("http") + async def router_admin_auth(request: Request, call_next): + # Operator /x/* surfaces (wallet moves funds, providers/keys choose where + # credentials go, calls/sessions expose every caller's traffic) require + # the internal secret. With none configured only a same-pod (loopback) + # ingress may call them, unless ROUTER_ADMIN_ALLOW_UNAUTHENTICATED=1. + path = re.sub(r"/{2,}", "/", request.scope.get("path") or "") + if path.startswith(_ROUTER_ADMIN_PATHS) and not _router_admin_ok(request): + return JSONResponse(status_code=403, content={"error": { + "message": "router admin endpoint requires x-internal-secret", + "type": "auth_error", "code": "router_admin_forbidden"}}) + return await call_next(request) # subscription backends (codex) are billed $0 per request — their ranking # price is a scarcity shadow price, not a cost @@ -382,8 +398,8 @@ def session_meter(sid: str, request: Request): return JSONResponse(status_code=404, content={"error": { "message": "session not found", "type": "not_found", "code": "session_not_found"}}) - acc = host_store.session_totals(sid) - return {**acc, "warm": host_store.session_warm(sid)} + acc = host_store.session_totals(sid, caller or None) + return {**acc, "warm": host_store.session_warm(sid, caller or None)} @app.get("/x/sessions") def session_meters(): @@ -931,7 +947,7 @@ def add_provider(body: dict): "auth_env": body.get("auth_env"), "served_models": body.get("served_models") or [], } - errors = validate_entry(pid, entry, host.catalog()) + errors = validate_entry(pid, entry, host.catalog(), key_supplied=bool(body.get("key"))) if any("already exists" in e for e in errors): return JSONResponse(status_code=409, content={"error": { "message": f"provider {pid!r} already exists", @@ -1128,8 +1144,8 @@ async def compact(req: CompactRequest): summary_policy=req.policy_ir or _DEFAULT_COMPACT_POLICY, max_tokens=req.max_tokens or 512) try: # The same admitted flow executor exposed through flow_ir on chat. - res = await host.execute_flow_async(prepared['flow_ir'], - {'flow_input': prepared['flow_input'], 'messages': []}) + res = await _execute_with_deadline(host.execute_flow_async(prepared['flow_ir'], + {'flow_input': prepared['flow_input'], 'messages': []})) except Exception as exc: admission = _flow_admission_error(exc) if admission is not None: @@ -1399,7 +1415,7 @@ async def _handle_chat(req: ChatRequest, profile_name: str | None = None): if req.flow_ir is not None: if not req.stream: try: - result = await host.execute_flow_async(req.flow_ir, contract) + result = await _execute_with_deadline(host.execute_flow_async(req.flow_ir, contract)) except Exception as exc: admission = _flow_admission_error(exc) if admission is None: @@ -1416,7 +1432,7 @@ async def _handle_chat(req: ChatRequest, profile_name: str | None = None): # JSON error. Anything slower commits to SSE and HEARTBEATs while the # flow runs (flows have no token stream, so without this they sat # silent ~60s and the idle timeout cut them into an empty 200). - task = asyncio.create_task(host.execute_flow_async(req.flow_ir, contract)) + task = asyncio.create_task(_execute_with_deadline(host.execute_flow_async(req.flow_ir, contract))) done, _ = await asyncio.wait({task}, timeout=_EARLY_FAIL_S, return_when=asyncio.FIRST_COMPLETED) if task in done: @@ -1533,11 +1549,15 @@ async def _flow_stream(task: "asyncio.Task", req: ChatRequest): # Activity and to the client instead of nothing. stream_id = _streaming.new_stream_id() model = req.model or "" - yield _streaming.encode_role_chunk(stream_id, model) - while not task.done(): - await asyncio.wait({task}, timeout=_HEARTBEAT_S) - if not task.done(): - yield _streaming.HEARTBEAT + try: + yield _streaming.encode_role_chunk(stream_id, model) + while not task.done(): + await asyncio.wait({task}, timeout=_HEARTBEAT_S) + if not task.done(): + yield _streaming.HEARTBEAT + finally: + if not task.done(): # client went away: stop spending on its flow + task.cancel() try: result = task.result() except Exception as exc: @@ -1563,21 +1583,25 @@ async def _flow_stream(task: "asyncio.Task", req: ChatRequest): async def _sse_gen(queue: "asyncio.Queue", task: "asyncio.Task", req: ChatRequest): stream_id = _streaming.new_stream_id() model = req.model or "" - yield _streaming.encode_role_chunk(stream_id, model) # first byte, now streamed_any = False - while True: - getter = asyncio.create_task(queue.get()) - kind, payload = await _await_delta_or_beat(getter, task, _HEARTBEAT_S) - if kind == "delta": - streamed_any = True - yield _streaming.encode_text_chunk(stream_id, model, payload) - continue - if kind == "done": - while not queue.empty(): + try: + yield _streaming.encode_role_chunk(stream_id, model) # first byte, now + while True: + getter = asyncio.create_task(queue.get()) + kind, payload = await _await_delta_or_beat(getter, task, _HEARTBEAT_S) + if kind == "delta": streamed_any = True - yield _streaming.encode_text_chunk(stream_id, model, queue.get_nowait()) - break - yield _streaming.HEARTBEAT # 'beat' — keep the line warm + yield _streaming.encode_text_chunk(stream_id, model, payload) + continue + if kind == "done": + while not queue.empty(): + streamed_any = True + yield _streaming.encode_text_chunk(stream_id, model, queue.get_nowait()) + break + yield _streaming.HEARTBEAT # 'beat' — keep the line warm + finally: + if not task.done(): # client went away: cancel the run + task.cancel() result = task.result() if result.get("ok"): resp, usage, x_router = _final_chunk_parts(result, req.session, req.caller) @@ -1670,13 +1694,28 @@ def _request_to_contract( # reads off ctx.request. A brand-new session has no hot route -> the key # is simply absent -> cache_hot is false for everyone (no phantom pin). contract["session"] = req.session - hot = host_store.hot_route(req.session) + hot = host_store.hot_route(req.session, req.caller) if hot is not None: contract["cache_hot_route"] = hot return contract +_ROUTER_ADMIN_PATHS = ("/x/wallet", "/x/providers", "/x/provider-key", "/x/config/", + "/x/codex/", "/x/calls", "/x/sessions", "/x/session/") + + +def _router_admin_ok(request: Request) -> bool: + secret = control_plane_client.CONTROL_PLANE_INTERNAL_SECRET + if secret: + presented = request.headers.get("x-internal-secret") or "" + return hmac.compare_digest(presented.encode(), secret.encode()) + if os.getenv("ROUTER_ADMIN_ALLOW_UNAUTHENTICATED", "0").lower() in {"1", "true", "yes"}: + return True + host = request.client.host if request.client else "" + return host in {"127.0.0.1", "::1", "localhost"} + + def _invalid_policy_response(message: str) -> JSONResponse: return JSONResponse(status_code=400, content={"error": { "message": f"policy_ir rejected at admission: {message}", @@ -1953,7 +1992,7 @@ def _build_x_router(result: dict, subscription_providers=frozenset(), # #4b: derive the running total from the committed `calls` and add THIS # in-flight call (not yet in the ledger). Owner is derived from the # session's earliest call, so no explicit binding is needed here. - prior = host_store.session_totals(session) + prior = host_store.session_totals(session, owner) x_router["session_acc"] = { "calls": prior["calls"] + 1, "tokens_in": prior["tokens_in"] + (resp.get("tokens_in") or 0), diff --git a/sources/antseed.py b/sources/antseed.py index 6681b67..fd005c8 100644 --- a/sources/antseed.py +++ b/sources/antseed.py @@ -356,7 +356,9 @@ async def _refresh_decisions(self): try: url = cfg.get('base_url', '').rstrip('/') + '/models?type=decisions' async def fetch(client): - response = await client.get(url, timeout=5, follow_redirects=False) + token = os.getenv('ANTSEED_PROXY_TOKEN') # sidecar public-proxy.js + response = await client.get(url, timeout=5, follow_redirects=False, + headers={'Authorization': 'Bearer ' + token} if token else None) response.raise_for_status() return response.json() if self._decision_client is not None: diff --git a/streaming.py b/streaming.py index 04646ad..8136c3a 100644 --- a/streaming.py +++ b/streaming.py @@ -73,7 +73,7 @@ def _notify(status: int, headers=None) -> None: body = build_codex_body(request) headers = build_codex_headers(token, auth.account_id(), extra_headers) - url = (request.get("base_url") or base_url or CODEX_BASE_URL).rstrip("/") + "/responses" + url = (base_url or CODEX_BASE_URL).rstrip("/") + "/responses" # never request["base_url"] timeout = (request.get("timeout_ms") or int(timeout_s * 1000)) / 1000.0 owns_client = client is None diff --git a/tests/conftest.py b/tests/conftest.py index 4c52a38..6c4d4bc 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -44,6 +44,9 @@ # Test HTTP endpoints are local/fake; production requires TLS by default. os.environ["CP_ALLOW_INSECURE_HTTP"] = "1" +# TestClient peers are not loopback; the router /x/* admin gate is exercised +# explicitly in test_security_router.py. +os.environ.setdefault("ROUTER_ADMIN_ALLOW_UNAUTHENTICATED", "1") import host_store # noqa: E402 diff --git a/tests/test_antseed_node.py b/tests/test_antseed_node.py index 0ac37cb..960a93c 100644 --- a/tests/test_antseed_node.py +++ b/tests/test_antseed_node.py @@ -45,6 +45,18 @@ def test_antseed_market_protocol_writer(): _run_node_test('antseed/write-market.test.js') +def test_antseed_control_token_and_bind(): + """The wallet control server compares its token in constant time and binds + pod-local unless the deployment opts in (compose).""" + _run_node_test('antseed/auth.test.js') + + +def test_antseed_public_proxy_requires_bearer(): + """The network face of the funded buyer proxy (:8378) demands a bearer when + ANTSEED_PROXY_TOKEN is configured instead of forwarding anyone's traffic.""" + _run_node_test('antseed/public-proxy.test.js') + + def test_antseed_control_amount_cap(): """The control server's deposit-amount guard (antseed/amount.js). /deposit is now called autonomously by the router's wallet keeper, so the per-deposit diff --git a/tests/test_provider_overlay.py b/tests/test_provider_overlay.py index 21f5d41..951d93e 100644 --- a/tests/test_provider_overlay.py +++ b/tests/test_provider_overlay.py @@ -174,11 +174,13 @@ class FakeResp: status_code = 200 class FakeClient: - async def post(self, url, json=None, timeout=None): - calls.append((url, json)) + async def post(self, url, json=None, timeout=None, headers=None): + calls.append((url, json, headers)) return FakeResp() monkeypatch.setattr(auth_proxy, "_client", FakeClient()) + import control_plane_client + monkeypatch.setattr(control_plane_client, "CONTROL_PLANE_INTERNAL_SECRET", "router-admin") payload = {"id": "groq", "base_url": "https://api.groq.com/openai/v1", "tier": "partner", "auth_env": "GROQ_API_KEY", "key": "gsk-raw", @@ -201,8 +203,9 @@ async def post(self, url, json=None, timeout=None): saved = po.load_overlay() assert saved["providers"]["groq"]["auth_env"] == "GROQ_API_KEY" assert "key" not in saved["providers"]["groq"] # never in overlay - url, body = calls[0] + url, body, headers = calls[0] assert url.endswith("/x/providers") and body["key"] == "gsk-raw" + assert headers == {"x-internal-secret": "router-admin"} # router /x/* admin credential # second add of the same id is rejected (overlay merged into catalog view) monkeypatch.setattr(auth_proxy, "_load_policy_config", diff --git a/tests/test_security_ingress.py b/tests/test_security_ingress.py new file mode 100644 index 0000000..7a50d88 --- /dev/null +++ b/tests/test_security_ingress.py @@ -0,0 +1,459 @@ +"""Ingress (auth_proxy) security regressions: path traversal to router +internals, streaming budget bypass, fail-closed consumer store reads, router +admin credential, usage/session endpoint hardening, tenant/consumer collision, +and constant-time compares that must not raise on non-ASCII input.""" +from __future__ import annotations + +import asyncio +import hashlib +import time + +import httpx +import pytest +from fastapi.testclient import TestClient + +import auth_proxy +import control_plane_client +import host_store +from conftest import require_host_store + + +def _authed(monkeypatch, caller="tester", meta=None): + monkeypatch.setattr(auth_proxy, "_caller_auth", + lambda token: {"ok": True, "caller": caller, "digest": "d" * 64, "meta": meta or {}}) + monkeypatch.setattr(auth_proxy, "_rate_ok", lambda caller, meta=None: (True, True, 0.0)) + + +# ---- I1: path traversal ------------------------------------------------------ + +_TRAVERSALS = [ + "/x/calls", "/v1/%2e%2e/x/calls", "/v1/%2E%2E/x/providers", "/%2e/x/calls", + "/v1/..%2fx/calls", "/v1/..%2Fx/calls", "/v1/%2e%2e%2fx/calls", "/./x/calls", + "/a/../x/calls", "/v1/..%5cx/calls", "//x/calls", "/v1//chat/completions", + "/v1/%252e%252e/x/calls", "/%2e%2e/x/calls", "/X/calls", "/dashboard/../x/calls", +] + + +@pytest.mark.parametrize("raw", [ + b"/./x/calls", b"/a/../x/calls", b"/v1/../x/calls", b"/v1/./chat/completions", + b"/v1/chat%2Fcompletions", b"/v1/..\\x/calls", b"//x/calls", b"/v1/chat/completions/..", +]) +def test_proxy_path_refuses_raw_dot_and_encoded_segments(raw): + # httpx clients normalize dot segments before sending; a raw socket does not. + from starlette.requests import Request + from urllib.parse import unquote + scope = {"type": "http", "method": "POST", "headers": [], "query_string": b"", + "raw_path": raw, "path": unquote(raw.decode())} + assert auth_proxy._proxy_path(Request(scope)) is None + + +@pytest.mark.parametrize("path", _TRAVERSALS) +def test_traversal_never_reaches_router_internals(monkeypatch, path): + _authed(monkeypatch) + seen = [] + + def handler(req: httpx.Request): + seen.append(req.url.raw_path.decode()) + return httpx.Response(200, json={"reached": req.url.path}) + + monkeypatch.setattr(auth_proxy, "_client", httpx.AsyncClient(transport=httpx.MockTransport(handler))) + r = TestClient(auth_proxy.app).post(path, headers={"Authorization": "Bearer k"}, json={"id": "evil"}) + assert r.status_code in (400, 404), (path, r.status_code, r.text) + assert seen == [], (path, seen) + + +def test_proxy_forwards_normalized_allowed_paths(monkeypatch): + _authed(monkeypatch) + monkeypatch.setattr(host_store, "consumer_spend_usd", lambda caller: (0.0, True)) + seen = [] + + def handler(req: httpx.Request): + seen.append(req.url.raw_path.decode()) + return httpx.Response(200, json={"ok": True}) + + monkeypatch.setattr(auth_proxy, "_client", httpx.AsyncClient(transport=httpx.MockTransport(handler))) + c = TestClient(auth_proxy.app) + for p in ("/v1/chat/completions", "/v1/models", "/default/v1/chat/completions", "/api/tags"): + assert c.post(p, headers={"Authorization": "Bearer k"}, json={}).status_code == 200, p + assert seen == ["/v1/chat/completions", "/v1/models", "/default/v1/chat/completions", "/api/tags"] + + +# ---- I3/I4: consumer store read failures fail closed -------------------------- + +class _Up: + async def get(self, url, timeout=None, headers=None): + return httpx.Response(200, json={"ok": True}, request=httpx.Request("GET", url)) + + +def test_consumer_dashboard_fails_closed_when_consumer_store_unreadable(monkeypatch, tmp_path): + require_host_store() + monkeypatch.setattr(auth_proxy, "DASHBOARD_SESSION_SECRET", "s") + monkeypatch.setattr(auth_proxy, "_client", _Up()) + monkeypatch.setattr(auth_proxy, "DASHBOARD_KEY_ENV_PATH", str(tmp_path / ".env")) + tok = "llmr_alice" + dig = hashlib.sha256(tok.encode()).hexdigest() + assert host_store.set_consumer_keys({"alice": {"status": "active", "keys": [ + {"sha256_prefix": dig[:12], "status": "active"}]}}, key_digests={dig: "alice"}) + c = TestClient(auth_proxy.app) + c.cookies.set(auth_proxy.DASHBOARD_COOKIE_NAME, auth_proxy._make_dashboard_session( + "alice", role="consumer", consumer="alice", key_sha256=dig)) + auth_proxy._record_request(caller="victim-co", status=200, requested_model="profile:secret", + error_message="victim prompt", key_sha256="b" * 64, method="POST", + path="/v1/chat/completions", latency_ms=1) + d = c.get("/dashboard/api/stats?timeframe=runtime").json() + assert d["selected_consumer"] == "alice" + assert "victim-co" not in {x.get("caller") for x in d["recent"]} + monkeypatch.setattr(host_store, "get_consumer_keys", lambda: ({}, False)) + for path in ("/dashboard/api/stats?timeframe=runtime", "/dashboard/api/full"): + r = c.get(path) + assert r.status_code == 503, (path, r.status_code, r.text[:200]) + assert "victim" not in r.text + # Even if the pre-check races a later failure, the snapshot stays scoped. + snap = auth_proxy._stats_snapshot(viewer="consumer:alice", upstream_status=200, upstream_health={}, + consumer="alice", timeframe="runtime", key_sha256=dig, + viewer_role="consumer") + assert snap["selected_consumer"] == "alice" + assert "victim-co" not in {x.get("caller") for x in snap["recent"]} + + +def test_consumer_record_writes_abort_on_store_read_failure(monkeypatch, tmp_path): + require_host_store() + monkeypatch.setattr(auth_proxy, "DASHBOARD_SESSION_SECRET", "s") + monkeypatch.setattr(auth_proxy, "DASHBOARD_KEY_ENV_PATH", str(tmp_path / ".env")) + old = "llmr_old" + od = hashlib.sha256(old.encode()).hexdigest() + assert host_store.set_consumer_keys({"bob": { + "status": "inactive", "budget_usd": 1.0, "allowed_routes": ["profile:cheap"], + "keys": [{"sha256_prefix": od[:12], "status": "active", "expires_at": int(time.time()) - 10}]}}, + key_digests={od: "bob"}) + real = host_store.get_consumer_keys + c = TestClient(auth_proxy.app) + c.cookies.set(auth_proxy.DASHBOARD_COOKIE_NAME, auth_proxy._make_dashboard_session("admin")) + writes = [ + ("/dashboard/api/keys", {"consumer": "carol"}), + ("/dashboard/api/keys/batch", {"batch": "b1", "count": 2, "budget_usd": 1}), + ("/dashboard/api/consumers/bob", {"status": "active"}), + ("/dashboard/api/keys/revoke", {"consumer": "bob", "sha256_prefix": od[:12]}), + ] + for path, body in writes: + monkeypatch.setattr(host_store, "get_consumer_keys", lambda: ({}, False)) + r = c.post(path, json=body) + monkeypatch.setattr(host_store, "get_consumer_keys", real) + assert r.status_code == 503, (path, r.status_code, r.text[:200]) + records = real()[0] + assert set(records) == {"bob"}, (path, records) + assert records["bob"]["status"] == "inactive" and records["bob"]["budget_usd"] == 1.0 + assert auth_proxy._caller_auth(old).get("ok") is not True + + +# ---- M2 (client side): operator /x/* calls carry the router admin secret ------ + +def test_router_admin_calls_send_internal_secret_and_proxy_traffic_does_not(monkeypatch): + monkeypatch.setattr(control_plane_client, "CONTROL_PLANE_INTERNAL_SECRET", "sek") + monkeypatch.setattr(auth_proxy, "DASHBOARD_SESSION_SECRET", "s") + _authed(monkeypatch) + monkeypatch.setattr(host_store, "consumer_spend_usd", lambda caller: (0.0, True)) + seen = {} + + def handler(req: httpx.Request): + seen[req.url.path] = req.headers.get("x-internal-secret") + return httpx.Response(200, json={"ok": True}) + + monkeypatch.setattr(auth_proxy, "_client", httpx.AsyncClient(transport=httpx.MockTransport(handler))) + admin = TestClient(auth_proxy.app) + admin.cookies.set(auth_proxy.DASHBOARD_COOKIE_NAME, auth_proxy._make_dashboard_session("admin")) + admin.get("/dashboard/api/wallet") + admin.post("/dashboard/api/wallet/refresh", json={}) + TestClient(auth_proxy.app).post("/v1/chat/completions", headers={ + "Authorization": "Bearer k", "x-internal-secret": "forged"}, json={}) + assert seen["/x/wallet"] == "sek" + assert seen["/x/wallet/refresh"] == "sek" + assert seen["/v1/chat/completions"] is None + assert auth_proxy._router_admin_headers({"a": "b"}) == {"a": "b", "x-internal-secret": "sek"} + monkeypatch.setattr(control_plane_client, "CONTROL_PLANE_INTERNAL_SECRET", "") + assert auth_proxy._router_admin_headers() == {} + + +# ---- S7 + S1: /v1/session and /v1/usage -------------------------------------- + +@pytest.mark.parametrize("sid", ["..", "%2e%2e", "a b", "x" * 129, ".hidden"]) +def test_session_view_rejects_malformed_sid(monkeypatch, sid): + _authed(monkeypatch) + seen = [] + monkeypatch.setattr(auth_proxy, "_client", httpx.AsyncClient(transport=httpx.MockTransport( + lambda req: seen.append(req.url.raw_path) or httpx.Response(200, json={})))) + r = TestClient(auth_proxy.app).get(f"/v1/session/{sid}", headers={"Authorization": "Bearer k"}) + assert r.status_code in (400, 404), (sid, r.status_code) + assert seen == [] + + +def test_session_view_forwards_valid_sid_with_caller_and_secret(monkeypatch): + monkeypatch.setattr(control_plane_client, "CONTROL_PLANE_INTERNAL_SECRET", "sek") + _authed(monkeypatch, caller="alice") + seen = [] + + def handler(req): + seen.append((req.url.raw_path.decode(), req.headers.get("x-llm-router-caller"), + req.headers.get("x-internal-secret"))) + return httpx.Response(200, json={"calls": 1}) + + monkeypatch.setattr(auth_proxy, "_client", httpx.AsyncClient(transport=httpx.MockTransport(handler))) + r = TestClient(auth_proxy.app).get("/v1/session/ses_01J:ab-c.d", headers={"Authorization": "Bearer k"}) + assert r.status_code == 200 + assert seen == [("/x/session/ses_01J%3Aab-c.d", "alice", "sek")] + + +def test_usage_and_session_endpoints_are_rate_limited(monkeypatch): + _authed(monkeypatch) + monkeypatch.setattr(auth_proxy, "_rate_ok", lambda caller, meta=None: (False, True, 3.0)) + monkeypatch.setattr(auth_proxy, "_client", httpx.AsyncClient(transport=httpx.MockTransport( + lambda req: httpx.Response(200, json={})))) + c = TestClient(auth_proxy.app) + for path in ("/v1/usage", "/api/usage", "/v1/session/ses_1"): + r = c.get(path, headers={"Authorization": "Bearer k"}) + assert r.status_code == 429, (path, r.status_code) + assert r.headers["Retry-After"] == "3" + monkeypatch.setattr(auth_proxy, "_rate_ok", lambda caller, meta=None: (True, False, 0.0)) + assert c.get("/v1/usage", headers={"Authorization": "Bearer k"}).status_code == 503 + + +def test_key_usage_reads_only_the_key_in_sql_off_the_event_loop(monkeypatch): + import threading + require_host_store() + mine, other = "a" * 64, "b" * 64 + now = int(time.time()) + for i, sha in enumerate([mine, other, other, mine]): + host_store.insert_call({"ts": now - 10 + i, "usage_event_id": f"e{i}", "key_sha256": sha, + "caller": "tester", "status": 200, "tokens_in": 1, "tokens_out": 1, + "provider": "p", "model_family": "f"}) + _authed(monkeypatch) + monkeypatch.setattr(auth_proxy, "_caller_auth", lambda token: { + "ok": True, "caller": "tester", "digest": mine, "meta": {}}) + + def boom(*a, **k): + raise AssertionError("whole-caller history read") + + monkeypatch.setattr(host_store, "usage_rows", boom) + threads = [] + real = auth_proxy._key_usage_snapshot + monkeypatch.setattr(auth_proxy, "_key_usage_snapshot", + lambda **kw: threads.append(threading.current_thread()) or real(**kw)) + r = TestClient(auth_proxy.app).get("/v1/usage", headers={"Authorization": "Bearer k"}) + assert r.status_code == 200, r.text + assert r.json()["totals"]["requests"] == 2 + assert threads and threads[0] is not threading.main_thread() + + +# ---- S3 (ingress side): env-secrets writes ----------------------------------- + +def _admin(monkeypatch): + monkeypatch.setattr(auth_proxy, "DASHBOARD_SESSION_SECRET", "s") + c = TestClient(auth_proxy.app) + c.cookies.set(auth_proxy.DASHBOARD_COOKIE_NAME, auth_proxy._make_dashboard_session("admin")) + return c + + +def test_provider_key_writes_refuse_injection_and_infra_names(monkeypatch, tmp_path): + monkeypatch.setattr(auth_proxy, "_load_policy_config", lambda: {"providers": { + "heurist": {"auth_env": "HEURIST_API_KEY"}, + "evil": {"auth_env": "CONTROL_PLANE_INTERNAL_SECRET"}}}) + env_file = tmp_path / ".env.secrets" + monkeypatch.setattr(auth_proxy, "DASHBOARD_KEY_ENV_PATH", str(env_file)) + admin = _admin(monkeypatch) + for payload in ({"provider": "heurist", "key": "sk\nCONTROL_PLANE_INTERNAL_SECRET=owned"}, + {"provider": "heurist", "key": "sk\rX=1"}, {"provider": "heurist", "key": "a\x00b"}, + {"provider": "evil", "key": "sk"}): + r = admin.post("/dashboard/api/provider-keys/update", json=payload) + assert r.status_code == 400, (payload, r.status_code, r.text) + assert not env_file.exists() + for auth_env, key in (("CONTROL_PLANE_INTERNAL_SECRET", "x"), ("DATABASE_URL", "x"), + ("NEW_API_KEY", "a\nDATABASE_URL=postgres://evil")): + r = admin.post("/dashboard/api/provider-keys/add", json={ + "id": "newprov", "base_url": "https://api.example.com/v1", "auth_env": auth_env, "key": key, + "served_models": [{"model_family": "m", "served_model_id": "m"}]}) + assert r.status_code == 400, (auth_env, r.status_code, r.text) + assert not env_file.exists() + with pytest.raises(ValueError): + auth_proxy._upsert_env_line(env_file, "HEURIST_API_KEY", "a\nB=c") + assert admin.post("/dashboard/api/provider-keys/update", + json={"provider": "heurist", "key": "sk-ok"}).status_code == 200 + assert env_file.read_text() == "HEURIST_API_KEY=sk-ok\n" + + +# ---- S8: constant-time compares never 500 on non-ASCII input ------------------ + +def test_secret_compares_reject_non_ascii_without_500(monkeypatch): + monkeypatch.setattr(control_plane_client, "CONTROL_PLANE_INTERNAL_SECRET", "sek") + monkeypatch.setattr(auth_proxy, "DASHBOARD_SESSION_SECRET", "s") + monkeypatch.setattr(auth_proxy, "DASHBOARD_TRUSTED_USER_HEADER", "x-user") + monkeypatch.setattr(auth_proxy, "DASHBOARD_TRUSTED_USER_SECRET", "trusted") + c = TestClient(auth_proxy.app, raise_server_exceptions=False) + r = c.get("/internal/usage?caller=a", headers={"x-internal-secret": "s\xe9k".encode("latin-1")}) + assert r.status_code == 403 + r = c.get("/dashboard/api/stats", headers={"x-user": "op", "x-dashboard-trusted-secret": "\xe9".encode("latin-1")}) + assert r.status_code == 401 + r = c.get("/dashboard/api/stats", headers={ + "cookie": f"{auth_proxy.DASHBOARD_COOKIE_NAME}=e30.\xe9\xe9".encode("latin-1")}) + assert r.status_code == 401 + assert control_plane_client.secret_equal("é", "é") and not control_plane_client.secret_equal("é", "e") + + +# ---- S11: /internal/usage window validation ---------------------------------- + +@pytest.mark.parametrize("query", ["since_ts=200&until_ts=100", "since_ts=-1", + "until_ts=99999999999999999999", "since_ts=253402300800"]) +def test_internal_usage_rejects_invalid_windows(monkeypatch, query): + monkeypatch.setattr(control_plane_client, "CONTROL_PLANE_INTERNAL_SECRET", "sek") + called = [] + monkeypatch.setattr(host_store, "usage_totals", lambda **kw: called.append(kw) or {}) + r = TestClient(auth_proxy.app).get(f"/internal/usage?caller=a&{query}", headers={"x-internal-secret": "sek"}) + assert r.status_code == 400, (query, r.status_code, r.text) + assert called == [] + + +# ---- S2: tenant slug colliding with a local consumer fails closed ------------- + +@pytest.mark.parametrize("local", ["issued", "env_plain", "env_hash"]) +def test_control_plane_caller_colliding_with_local_consumer_is_rejected(monkeypatch, local): + require_host_store() + monkeypatch.setattr(control_plane_client, "enabled", lambda: True) + + async def resolve(digest): + return control_plane_client.ResolvedKey(active=True, consumer="acme", tenant_id=7, + rate_per_min=None, burst=None, fetched_at=0.0) + + monkeypatch.setattr(control_plane_client, "resolve_key", resolve) + if local == "issued": + dig = hashlib.sha256(b"llmr_local").hexdigest() + host_store.set_consumer_keys({"acme": {"status": "active", "keys": [ + {"sha256_prefix": dig[:12], "status": "active"}]}}, key_digests={dig: "acme"}) + elif local == "env_plain": + monkeypatch.setattr(auth_proxy, "CALLER_KEYS", {"llmr_local": "acme"}) + else: + monkeypatch.setattr(auth_proxy, "CALLER_KEY_HASHES", {"f" * 64: "acme"}) + auth = asyncio.run(auth_proxy._caller_auth_async("tok-tenant")) + assert auth["ok"] is False and auth["error_code"] == "caller_name_collision" + + +def test_control_plane_caller_without_local_keys_still_resolves(monkeypatch): + require_host_store() + monkeypatch.setattr(control_plane_client, "enabled", lambda: True) + + async def resolve(digest): + return control_plane_client.ResolvedKey(active=True, consumer="acme", tenant_id=7, + rate_per_min=5, burst=5, fetched_at=0.0) + + monkeypatch.setattr(control_plane_client, "resolve_key", resolve) + host_store.set_consumer_keys({"acme": {"status": "active", "rate_per_min": 9}}) # override, no keys + auth = asyncio.run(auth_proxy._caller_auth_async("tok-tenant")) + assert auth["ok"] is True and auth["caller"] == "acme" and auth["meta"]["rate_per_min"] == 9 + + +# ---- I2: a client that disconnects mid-stream still pays --------------------- + +def _serve(app): + import socket + import threading + import uvicorn + s = socket.socket() + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + s.close() + srv = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error", lifespan="off")) + t = threading.Thread(target=srv.run, daemon=True) + t.start() + while not srv.started: + time.sleep(0.02) + return srv, t, port + + +def _usage_sse(filler: bytes = b""): + async def body(): + yield b'data: {"choices":[{"delta":{"content":"THE ANSWER <>"}}]}\n\n' + await asyncio.sleep(0.8) # the provider is still finishing; usage arrives last + if filler: + yield filler + yield (b'data: {"choices":[{"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":100000,' + b'"completion_tokens":5000,"total_tokens":105000},"x_router":{"provider":"p",' + b'"model_family":"f","cost_usd":7.5}}\n\n') + yield b"data: [DONE]\n\n" + return body + + +def _disconnect_after_answer(port, tok): + with httpx.Client(timeout=10) as c: + with c.stream("POST", f"http://127.0.0.1:{port}/v1/chat/completions", + headers={"Authorization": f"Bearer {tok}"}, + json={"model": "profile:default", "stream": True, "messages": []}) as r: + for chunk in r.iter_raw(): + if b"<>" in chunk: + break + + +def _budgeted_consumer(name, tok, budget=1.0): + dig = hashlib.sha256(tok.encode()).hexdigest() + assert host_store.set_consumer_keys({name: {"status": "active", "budget_usd": budget, "keys": [ + {"sha256_prefix": dig[:12], "status": "active"}]}}, key_digests={dig: name}) + + +def _wait_for(pred, timeout=5.0): + end = time.time() + timeout + while time.time() < end: + if pred(): + return True + time.sleep(0.05) + return False + + +def test_stream_disconnect_before_usage_is_still_metered(monkeypatch): + require_host_store() + tok = "llmr_budgeted" + _budgeted_consumer("hack-001", tok) + body = _usage_sse() + monkeypatch.setattr(auth_proxy, "_client", httpx.AsyncClient(transport=httpx.MockTransport( + lambda req: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body())))) + srv, t, port = _serve(auth_proxy.app) + try: + _disconnect_after_answer(port, tok) + assert _wait_for(lambda: host_store.consumer_spend_usd("hack-001")[0] >= 7.5), \ + host_store.consumer_spend_usd("hack-001") + with httpx.Client(timeout=10) as c: + r = c.post(f"http://127.0.0.1:{port}/v1/chat/completions", headers={"Authorization": f"Bearer {tok}"}, + json={"model": "profile:default", "stream": True, "messages": []}) + assert r.status_code == 402, r.text + finally: + srv.should_exit = True + t.join(5) + + +def test_stream_cut_off_by_drain_cap_books_an_estimate_not_zero(monkeypatch): + require_host_store() + monkeypatch.setattr(auth_proxy, "STREAM_DRAIN_MAX_BYTES", 16) + tok = "llmr_budgeted2" + _budgeted_consumer("hack-002", tok) + body = _usage_sse(filler=b": " + b"x" * 4096 + b"\n\n") + monkeypatch.setattr(auth_proxy, "_client", httpx.AsyncClient(transport=httpx.MockTransport( + lambda req: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body())))) + srv, t, port = _serve(auth_proxy.app) + try: + _disconnect_after_answer(port, tok) + assert _wait_for(lambda: host_store.recent_calls(limit=5, caller="hack-002")) + row = host_store.recent_calls(limit=5, caller="hack-002")[0] + assert row["cost_usd"] == pytest.approx(auth_proxy.CLOUD_BUDGET_RESERVATION_USD) + assert host_store.consumer_spend_usd("hack-002")[0] > 0 + finally: + srv.should_exit = True + t.join(5) + + +def test_stream_completed_normally_is_forwarded_whole_and_metered(monkeypatch): + require_host_store() + tok = "llmr_budgeted3" + _budgeted_consumer("hack-003", tok, budget=100.0) + body = _usage_sse() + monkeypatch.setattr(auth_proxy, "_client", httpx.AsyncClient(transport=httpx.MockTransport( + lambda req: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body())))) + r = TestClient(auth_proxy.app).post("/v1/chat/completions", headers={"Authorization": f"Bearer {tok}"}, + json={"model": "profile:default", "stream": True, "messages": []}) + assert r.status_code == 200 and r.text.endswith("data: [DONE]\n\n") and "<>" in r.text + assert _wait_for(lambda: host_store.consumer_spend_usd("hack-003")[0] == 7.5) diff --git a/tests/test_security_router.py b/tests/test_security_router.py new file mode 100644 index 0000000..500b720 --- /dev/null +++ b/tests/test_security_router.py @@ -0,0 +1,528 @@ +"""Regression tests for the 2026-09-26 router security audit (router side). + +Each test encodes an attack that worked before the fix: caller Xforms rewriting +route keys, retry loops freezing the event loop, caller-controlled breaker +state, tenant outcomes poisoning shared observations, untimed flows, session +metering across callers, unauthenticated /x/* admin endpoints, unbounded BYO +reads, tenant-sized peer gates, NAT64 egress, tenant keys unlocking local +providers, and codex/antseed credential routing. +""" +from __future__ import annotations + +import asyncio +import ipaddress +import json +import os +import sys +import threading +import time +from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path + +import httpx +import pytest +from fastapi.testclient import TestClient + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +import llm_router_host as L # noqa: E402 +from llm_router_host import LLMRouterHost # noqa: E402 +from shim import create_app # noqa: E402 + +from conftest import seed_call # noqa: E402 + +MSG = [{"role": "user", "content": "hi"}] + + +def _enveloped_config(tmp_path): + cfg = tmp_path / "c.lua" + cfg.write_text(f'local cfg = dofile("{ROOT}/core/config.example.lua")\n' + 'cfg.policy_envelope = { "and", { "meets_req" }, { "not", { "is", "disabled" } } }\n' + 'return cfg\n') + return cfg + + +def _host(tmp_path, **kw): + h = LLMRouterHost(router_path=ROOT / "core" / "router.lua", config_path=_enveloped_config(tmp_path), + metrics_path=ROOT / "core" / "metrics.example.lua", enforce_provider_auth=False, **kw) + h.init() + return h + + +@pytest.fixture +def no_fold(monkeypatch): + monkeypatch.setattr(L, "_fold_route_outcome", lambda *a, **k: None) + + +# ---- C1: caller Xforms cannot move a call or swap its credential ------------ + +def test_set_param_cannot_send_internal_secret_to_attacker(tmp_path, monkeypatch, no_fold): + from provider_adapters.openai_compatible import make_async_call_provider + monkeypatch.setenv("CONTROL_PLANE_INTERNAL_SECRET", "TOP-SECRET-INTERNAL") + captured = {} + + class H(BaseHTTPRequestHandler): + def do_POST(self): + captured["auth"] = self.headers.get("Authorization") + self.rfile.read(int(self.headers.get("content-length", 0))) + body = json.dumps({"choices": [{"message": {"content": "pwned"}, + "finish_reason": "stop"}], "usage": {}}).encode() + self.send_response(200) + self.send_header("content-length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *a): + pass + + srv = HTTPServer(("127.0.0.1", 0), H) + threading.Thread(target=srv.serve_forever, daemon=True).start() + try: + h = _host(tmp_path, call_provider_async=make_async_call_provider(env_get=os.environ.get)) + pol = ["policy", ["top"], ["zero"], ["argmax"], + ["seq", ["set_param", "base_url", f"http://127.0.0.1:{srv.server_port}/v1"], + ["set_param", "api_kind", "openai_compatible"], ["set_param", "auth", False], + ["set_param", "auth_env", "CONTROL_PLANE_INTERNAL_SECRET"]], + ["always", {"action": "abort"}]] + r = TestClient(create_app(h, default_profile="default")).post( + "/v1/chat/completions", json={"messages": MSG, "policy_ir": pol}) + assert r.status_code == 400 + assert captured == {} + finally: + srv.shutdown() + + +def test_host_pins_route_keys_from_catalog(tmp_path): + h = _host(tmp_path) + pid, prov = next((p, v) for p, v in h.catalog()["providers"].items() + if v.get("discovery") == "static" and v.get("auth_env")) + req = h._pin_route({"provider_id": pid, "base_url": "https://evil.example/v1", + "auth_env": "CONTROL_PLANE_INTERNAL_SECRET", "auth": False, + "api_kind": "openai_codex"}) + assert req["base_url"] == prov["base_url"] + assert req["auth_env"] == prov["auth_env"] + assert req["api_kind"] == prov["api_kind"] + assert req["auth"] == prov.get("auth") + + +# ---- H2: a retry storm cannot freeze the loop past the deadline ------------- + +def test_huge_zero_backoff_retry_is_refused_and_request_returns(tmp_path, no_fold): + from provider_adapters.openai_compatible import make_async_call_provider + h = _host(tmp_path, call_provider_async=make_async_call_provider(env_get=lambda _k: None)) + app = create_app(h, default_profile="default", request_deadline_ms=2000) + pol = ["policy", ["meets_req"], ["zero"], ["argmax"], ["id"], + ["always", {"action": "retry_same", "attempts": 1e15, "backoff_ms": 0}]] + out = {} + + def run(): + async def main(): + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), + base_url="http://x") as c: + r = await c.post("/v1/chat/completions", json={"messages": MSG, "policy_ir": pol}) + out["status"] = r.status_code + asyncio.run(main()) + + t = threading.Thread(target=run, daemon=True) + t.start() + t.join(10) + assert not t.is_alive(), "event loop frozen by a zero-backoff retry storm" + assert out["status"] == 400 + + +def test_execute_async_yields_between_attempts(tmp_path, no_fold): + ticks = [] + + async def failing(_req): # returns without awaiting, like an auth_error + return {"ok": False, "error_kind": "server_error", "http_status": 500, "latency_ms": 0} + + h = _host(tmp_path, call_provider_async=failing) + pol = ["policy", ["top"], ["zero"], ["argmax"], ["id"], + ["always", {"action": "retry_same", "attempts": 10, "backoff_ms": 0}]] + + async def main(): + async def ticker(): + while True: + ticks.append(1) + await asyncio.sleep(0) + t = asyncio.create_task(ticker()) + await asyncio.sleep(0) + before = len(ticks) + res = await h.execute_async({"messages": MSG, "policy_ir": pol}) + t.cancel() + return res, len(ticks) - before + + res, during = asyncio.run(main()) + assert res["ok"] is False + assert during > 5 + + +# ---- H1: caller policies cannot hold shared breakers / disables ------------- + +def test_caller_open_breaker_ms_is_clamped(tmp_path, no_fold): + async def timeout(_req): + return {"ok": False, "error_kind": "timeout", "http_status": 0, "latency_ms": 1} + + h = _host(tmp_path, call_provider_async=timeout) + pol = ["policy", ["top"], ["zero"], ["argmax"], ["id"], + ["always", {"action": "next_candidate", "open_breaker_ms": 3e12}]] + asyncio.run(h.execute_async({"messages": MSG, "policy_ir": pol})) + now = h._now_ms() + until = [b.get("open_until_ms") for b in (h.dump_state().get("circuit_breakers") or {}).values() + if b.get("open")] + assert until and max(until) <= now + 5 * 60 * 1000 + 5_000 + + +def test_caller_disable_provider_does_not_disable_shared_provider(tmp_path, no_fold): + async def fail(req): + return {"ok": False, "error_kind": req.get("_kind", "server_error"), "http_status": 500, + "latency_ms": 1} + + h = _host(tmp_path, call_provider_async=fail) + pol = ["policy", ["top"], ["zero"], ["argmax"], ["id"], + ["always", {"action": "disable_provider"}]] + asyncio.run(h.execute_async({"messages": MSG, "policy_ir": pol})) + assert not (h.dump_state().get("disabled_providers") or {}) + + +# ---- H3: tenant BYO outcomes never land in operator route observations ------ + +def test_tenant_byo_outcome_not_recorded_under_operator_route(host_store_clean, monkeypatch): + import host_store + rows = [] + monkeypatch.setattr(host_store, "observe_route_call_async", rows.append) + base = LLMRouterHost(ROOT / "core/router.lua", ROOT / "tests/fixtures/managed.lua") + base.init() + child = base.for_tenant(7, {}, connections={ + "antseed": {"gateway_url": "https://gw.attacker.example", "token": "t" * 40}}) + victim = "operator-real-peer-0xabc" + child._tenant_offers["antseed"] = [{ + "model_family": "shared-model", "wire_model_id": "x", "peer_id": victim, + "price_in_usd_per_mtok": 0.1, "price_out_usd_per_mtok": 0.1, + "capabilities": {"context": 128000, "supports_tools": True}}] + child.update_metrics("__credits", "antseed", {"free_credits_remaining_usd": 5}) + calls = [] + + async def gw(request): + calls.append(request) + return {"ok": False, "error_kind": "server_error", "http_status": 500, "latency_ms": 1} + + child.set_async_call_hook(gw) + asyncio.run(child.execute_async({"messages": MSG, "policy_ir": [ + "policy", ["and", ["meets_req"], ["provider_eq", "antseed"]], ["zero"], ["ordered"], ["id"], + ["always", {"action": "next_candidate"}]]})) + assert calls, "the BYO route was not exercised" + assert not [r for r in rows if r["served_by"] == victim] + + +# ---- M4: flows honour the request deadline ----------------------------------- + +def test_untyped_flow_is_cut_at_request_deadline(tmp_path, no_fold): + async def slow(_req): + await asyncio.sleep(5) + return {"ok": True, "response": {"text": "late"}, "latency_ms": 5000} + + h = _host(tmp_path, call_provider_async=slow) + app = create_app(h, default_profile="default", request_deadline_ms=300) + pol = ["policy", ["top"], ["zero"], ["argmax"], ["id"], ["always", {"action": "abort"}]] + flow = ["flow", {"u": {"kind": "input"}, + "a": {"kind": "llm", "system": "s", "policy": pol, "inputs": ["u"]}, + "out": {"kind": "output", "inputs": ["a"]}}] + t0 = time.monotonic() + r = TestClient(app).post("/v1/chat/completions", json={"messages": MSG, "flow_ir": flow}) + assert time.monotonic() - t0 < 3 + assert r.status_code >= 400 + + +# ---- M6: session reads are scoped to the calling owner ----------------------- + +def test_session_views_are_per_caller(host_store_clean, tmp_path): + import host_store + seed_call(session="shared-sid", provider="p", family="f", served_by="peer-a", + caller="alice", tokens_in=10, cost_usd=1.0, ts=int(time.time()) - 5) + seed_call(session="shared-sid", provider="q", family="g", served_by="peer-b", + caller="bob", tokens_in=999, cost_usd=50.0) + assert host_store.session_totals("shared-sid", "alice")["cost_usd"] == 1.0 + assert host_store.session_totals("shared-sid", "alice")["tokens_in"] == 10 + assert host_store.hot_route("shared-sid", "alice") == "p|f|peer-a" + assert [w["served_by"] for w in host_store.session_warm("shared-sid", "alice")] == ["peer-a"] + r = TestClient(create_app(_host(tmp_path))).get( + "/x/session/shared-sid", headers={"x-llm-router-caller": "alice"}) + assert r.status_code == 200 + assert r.json()["calls"] == 1 and r.json()["cost_usd"] == 1.0 + + +# ---- M2 / S9: router admin endpoints ---------------------------------------- + +def test_router_admin_endpoints_require_internal_secret(tmp_path, monkeypatch): + import control_plane_client as cp + monkeypatch.delenv("ROUTER_ADMIN_ALLOW_UNAUTHENTICATED", raising=False) + client = TestClient(create_app(_host(tmp_path))) + monkeypatch.setattr(cp, "CONTROL_PLANE_INTERNAL_SECRET", "") + for path in ("/x/calls", "/x/sessions", "/x/wallet", "/x/session/abc"): + assert client.get(path).status_code == 403, path + for path in ("/x/providers", "/x/provider-key", "/x/config/reload", "/x/wallet/deposit"): + assert client.post(path, json={}).status_code == 403, path + monkeypatch.setattr(cp, "CONTROL_PLANE_INTERNAL_SECRET", "s3cret") + assert client.get("/x/calls").status_code == 403 + assert client.get("/x/calls", headers={"x-internal-secret": "wrong"}).status_code == 403 + assert client.get("/x/calls", headers={"x-internal-secret": "s3cret"}).status_code != 403 + assert client.get("/x/runtime").status_code == 200 # Cloud reads it without a secret + assert client.get("/openapi.json").status_code == 404 + + +# ---- M3: BYO gateway reads are byte-capped ----------------------------------- + +def _byo_env(monkeypatch, handler): + import byo_http + monkeypatch.setattr(byo_http, "buyer_client", + lambda: httpx.AsyncClient(transport=httpx.MockTransport(handler))) + env = {"SAAS_TENANT_SCOPE": "7", "ANTSEED_BYO_TOKEN": "t" * 40, + "ANTSEED_BYO_URL": "https://gw.example"} + return env.get + + +BYO_REQ = {"provider_id": "antseed", "served_model_id": "m", "base_url": "https://gw.example/v1", + "messages": MSG} + + +def test_byo_buffered_response_is_capped(monkeypatch): + from provider_adapters.openai_compatible import make_async_call_provider + monkeypatch.setenv("BYO_MAX_RESPONSE_BYTES", "1024") + big = {"choices": [{"message": {"content": "a" * 5000}, "finish_reason": "stop"}], "usage": {}} + env_get = _byo_env(monkeypatch, lambda _r: httpx.Response(200, json=big)) + res = asyncio.run(make_async_call_provider(env_get=env_get)(dict(BYO_REQ))) + assert res["ok"] is False and res["error_kind"] == "bad_response" + + +def test_byo_stream_line_is_capped(monkeypatch): + from provider_adapters.openai_compatible import stream_openai_compatible + monkeypatch.setenv("BYO_MAX_SSE_LINE_BYTES", "1024") + line = "data: " + json.dumps({"choices": [{"delta": {"content": "a" * 8000}}]}) + env_get = _byo_env(monkeypatch, lambda _r: httpx.Response( + 200, content=(line + "\n\ndata: [DONE]\n\n").encode())) + + async def emit(_d): + pass + + res = asyncio.run(stream_openai_compatible(dict(BYO_REQ), emit, env_get=env_get)) + assert res["ok"] is False and res["error_kind"] == "bad_response" + assert "exceeds" in res["error_message"] + + +# ---- L1: BYO offers cannot size the operator's shared peer gate -------------- + +def test_byo_peer_gate_is_tenant_scoped(monkeypatch): + from provider_adapters import openai_compatible as oc + monkeypatch.delenv("DISTRIBUTED_PEER_GATES", raising=False) + monkeypatch.setenv("PEER_GATE_WAIT_S", "0.05") + env = {"SAAS_TENANT_SCOPE": "7", "ANTSEED_BYO_TOKEN": "t" * 40, "ANTSEED_BYO_URL": "https://gw"} + peer = "shared-peer-" + str(time.time()) + + async def main(): + byo = {"provider_id": "antseed", "offer": {"peer_id": peer, "max_concurrency": 1}} + op = {"provider_id": "antseed", "offer": {"peer_id": peer, "max_concurrency": 1}} + tenant_slot, err1 = await oc._acquire_peer_capacity(byo, 1.0, env.get) + operator_slot, err2 = await oc._acquire_peer_capacity(op, 1.0, lambda _k: None) + for s in (tenant_slot, operator_slot): + if s is not None: + await s.release() + return err1, err2 + + assert asyncio.run(main()) == (None, None) + + +# ---- L2: NAT64 / IPv4-embedding addresses are not "public" ------------------- + +@pytest.mark.parametrize("addr", ["64:ff9b::7f00:1", "64:ff9b:1::a00:1", "::7f00:1", + "::ffff:127.0.0.1", "2002:7f00:1::1", "10.0.0.1"]) +def test_byo_rejects_embedded_private_addresses(addr, monkeypatch): + import byo_http + assert not byo_http.public_address(ipaddress.ip_address(addr)) + + async def fake_getaddrinfo(*_a, **_k): + return [(0, 0, 0, "", (addr, 443))] + + async def main(): + loop = asyncio.get_running_loop() + monkeypatch.setattr(loop, "getaddrinfo", fake_getaddrinfo) + async with byo_http.buyer_client() as c: + with pytest.raises(httpx.ConnectError): + await c.get("https://gw.example/") + + asyncio.run(main()) + + +def test_byo_accepts_public_addresses(): + import byo_http + assert byo_http.public_address(ipaddress.ip_address("8.8.8.8")) + assert byo_http.public_address(ipaddress.ip_address("2606:4700::1111")) + + +# ---- L3: a tenant key never unlocks an operator-local provider --------------- + +def test_tenant_ollama_key_does_not_unlock_local_ollama(tmp_path): + cfg = tmp_path / "c.lua" + cfg.write_text(f'local cfg = dofile("{ROOT}/tests/fixtures/managed.lua")\n' + 'cfg.providers.ollama = { discovery = "marketplace", discovery_id = "ollama",' + ' base_url = "http://localhost:11434/v1", api_kind = "openai_compatible",' + ' auth_env = "OLLAMA_API_KEY", tier = "partner" }\n' + 'cfg.providers.remote = { discovery = "static", base_url = "https://api.remote.example/v1",' + ' api_kind = "openai_compatible", auth_env = "REMOTE_API_KEY", tier = "partner" }\n' + 'return cfg\n') + base = LLMRouterHost(ROOT / "core/router.lua", cfg) + base.init() + child = base.for_tenant(9, {"OLLAMA_API_KEY": "tenant", "REMOTE_API_KEY": "tenant"}) + assert "ollama" not in child._tenant_allowed + assert "remote" in child._tenant_allowed + + +# ---- M1: codex OAuth tokens only go to the configured upstream --------------- + +class _Auth: + def access_token(self): + return "oauth-token" + + def account_id(self): + return "acct" + + def select_account(self): + return self + + +class _Resp: + status_code = 200 + headers = {} + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def aiter_lines(self): + if False: + yield "" + + async def aread(self): + return b"" + + +class _RecordingClient: + def __init__(self, *a, **k): + self.urls = [] + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + def stream(self, method, url, **_kw): + self.urls.append(url) + _RecordingClient.last = url + return _Resp() + + async def aclose(self): + pass + + +def test_codex_call_ignores_request_base_url(monkeypatch): + import codex_backend as cb + monkeypatch.setattr(httpx, "AsyncClient", _RecordingClient) + call = cb.make_codex_async_call_provider(_Auth()) + asyncio.run(call({"served_model_id": "gpt-5.5", "messages": MSG, + "base_url": "https://evil.example/codex"})) + assert _RecordingClient.last.startswith(cb.CODEX_BASE_URL) + + +def test_codex_stream_ignores_request_base_url(): + import codex_backend as cb + from streaming import stream_codex + client = _RecordingClient() + + async def emit(_d): + pass + + asyncio.run(stream_codex({"served_model_id": "gpt-5.5", "messages": MSG, + "base_url": "https://evil.example/codex"}, + emit, auth=_Auth(), client=client)) + assert client.urls and client.urls[0].startswith(cb.CODEX_BASE_URL) + + +# ---- L5: the codex broker bearer never rides plain http off-cluster --------- + +def test_remote_codex_requires_https_off_cluster(monkeypatch): + from remote_codex import RemoteCodexClient + monkeypatch.delenv("CODEX_BROKER_ALLOW_HTTP", raising=False) + with pytest.raises(ValueError): + RemoteCodexClient("http://broker.example.com", "tok") + with pytest.raises(ValueError): + RemoteCodexClient("http://8.8.8.8:8090", "tok") + for ok in ("https://broker.example.com", "http://127.0.0.1:8090", + "http://llm-router-providers:8090", "http://broker.ns.svc.cluster.local:8090"): + RemoteCodexClient(ok, "tok") + + +# ---- M5 (router side): the funded antseed proxy gets its bearer -------------- + +def test_antseed_proxy_token_sent_only_to_antseed(monkeypatch): + from provider_adapters.openai_compatible import _prepare_openai_call + monkeypatch.setenv("ANTSEED_PROXY_TOKEN", "p" * 40) + prep, _ = _prepare_openai_call({"provider_id": "antseed", "served_model_id": "m", + "base_url": "http://antseed:8378/v1", "auth": {"kind": "none"}}, + lambda _k: None, {}, 10) + assert prep[2]["Authorization"] == "Bearer " + "p" * 40 + prep, _ = _prepare_openai_call({"provider_id": "other", "served_model_id": "m", + "base_url": "https://x.example/v1", "auth": {"kind": "none"}}, + lambda _k: None, {}, 10) + assert "Authorization" not in prep[2] + + +# ---- S3: provider overlay / env-secrets cannot touch infrastructure ---------- + +def test_overlay_rejects_infra_auth_env_private_urls_and_key_reuse(tmp_path): + import provider_overlay as po + h = _host(tmp_path) + catalog = h.catalog() + fam = next(iter(catalog["models"])) + existing = next(p["auth_env"] for p in catalog["providers"].values() + if po.AUTH_ENV_RE.match(p.get("auth_env") or "") and not po.is_forbidden_auth_env(p["auth_env"])) + + def entry(**over): + return {"base_url": "https://api.new.example/v1", "api_kind": "openai_compatible", + "auth_env": "NEWPROV_API_KEY", "served_models": [{"family": fam}], **over} + + assert po.validate_entry("newprov", entry(), catalog) == [] + for bad in ("CONTROL_PLANE_INTERNAL_SECRET", "DATABASE_URL", "ANTSEED_CONTROL_TOKEN", + "AWS_SESSION_TOKEN", "CODEX_BROKER_TOKEN", "EVIL\nX_API_KEY", "X_API_KEY\n"): + assert any("auth_env" in e for e in po.validate_entry("newprov", entry(auth_env=bad), catalog)), bad + for url in ("http://api.new.example/v1", "https://127.0.0.1/v1", "https://localhost/v1", + "https://router:18080/v1", "https://169.254.169.254/v1"): + assert any("base_url" in e for e in po.validate_entry("newprov", entry(base_url=url), catalog)), url + assert any("existing provider" in e for e in po.validate_entry("newprov", entry(auth_env=existing), catalog)) + assert po.validate_entry("newprov", entry(auth_env=existing), catalog, key_supplied=True) == [] + + +def test_env_secrets_never_override_infrastructure(tmp_path, monkeypatch): + from env_secrets import load_env_secrets + monkeypatch.setenv("DATABASE_URL", "postgresql://real") + monkeypatch.setenv("CONTROL_PLANE_INTERNAL_SECRET", "real") + monkeypatch.setenv("GROQ_API_KEY", "old") + f = tmp_path / ".env.secrets" + f.write_text("DATABASE_URL=postgresql://evil\nCONTROL_PLANE_INTERNAL_SECRET=evil\n" + "GROQ_API_KEY=new\nCALLER_KEYS_JSON={}\n") + loaded = load_env_secrets(f) + assert os.environ["DATABASE_URL"] == "postgresql://real" + assert os.environ["CONTROL_PLANE_INTERNAL_SECRET"] == "real" + assert os.environ["GROQ_API_KEY"] == "new" + assert "CALLER_KEYS_JSON" in loaded + + +def test_codex_broker_non_ascii_bearer_is_401_not_500(): + import codex_broker + app = codex_broker.create_app(store=object(), token="internal-token", client=object()) + r = TestClient(app, raise_server_exceptions=False).post( + "/v1/call", json={}, headers={"authorization": "Bearer tést".encode()}) + assert r.status_code == 401 diff --git a/tests/test_security_sidecar.py b/tests/test_security_sidecar.py new file mode 100644 index 0000000..4a04e4d --- /dev/null +++ b/tests/test_security_sidecar.py @@ -0,0 +1,42 @@ +"""Image/compose hardening that no runtime test would notice regressing.""" +import fnmatch +import json +import re +from pathlib import Path + +_ROOT = Path(__file__).resolve().parent.parent + + +def _dockerignored(name: str) -> bool: + ignored = False + for line in (_ROOT / ".dockerignore").read_text().splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + neg = line.startswith("!") + if fnmatch.fnmatch(name, line.lstrip("!")): + ignored = not neg + return ignored + + +def test_build_context_never_ships_codex_oauth_tokens(): + # compose's default CODEX_AUTH_PATH is ./codex-auth.json in the repo root, + # and the router image does `COPY . .`. + for name in ("codex-auth.json", "codex-accounts", ".env.secrets", ".env", "secrets"): + assert _dockerignored(name), name + assert not _dockerignored(".env.example") + + +def test_router_image_defaults_to_the_authenticated_ingress(): + cmd = next(ln for ln in (_ROOT / "Dockerfile").read_text().splitlines() if ln.startswith("CMD ")) + argv = json.loads(cmd[4:]) + assert argv[:2] == ["uvicorn", "auth_proxy:app"], argv + assert "serve.py" not in argv + + +def test_compose_has_no_default_database_password(): + src = (_ROOT / "compose.yml").read_text() + assert "hoststore:hoststore@" not in src + assert re.search(r"POSTGRES_PASSWORD: \$\{POSTGRES_PASSWORD:\?", src) + for dsn in re.findall(r"DATABASE_URL: (.+)", src): + assert "${POSTGRES_PASSWORD:?" in dsn, dsn From 7b46445fd40ac55f3be842978792be27a8e41c21 Mon Sep 17 00:00:00 2001 From: jmlago Date: Sat, 26 Sep 2026 17:50:46 +0100 Subject: [PATCH 2/2] chore: pin core to merged engine 8900fcc (unhardcoded-engine#35) Co-Authored-By: Claude Opus 5.5 (1M context) --- core | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core b/core index 2f43c47..8900fcc 160000 --- a/core +++ b/core @@ -1 +1 @@ -Subproject commit 2f43c47bf3e0c8de6df35cebaa0d19bcdcf28992 +Subproject commit 8900fccbc47864a63d5f789685458fa2ec76c32d