From 3abbaa7cd2e380a2179ddaa8c77bd69459aa06a6 Mon Sep 17 00:00:00 2001 From: CMGS Date: Wed, 9 Sep 2026 17:50:29 +0900 Subject: [PATCH 1/2] security: harden the admin API and the MCP proxy A targeted review of the admin plane and /mcp/{server} against the recent LiteLLM advisories. Fixed here: - MCP replies the proxy must buffer are capped (max_reply_bytes), the client follows no redirects, and upstream or token-endpoint errors reach the caller as a generic 502 with the detail in the log. - resources/read and prompts/get results are reviewed like tools/call; multi-line SSE data and unparsable replies fail closed; a mask covers structuredContent; a reviewed tenant gets no Last-Event-ID resumption; a denial replaces the whole reply with one JSON-RPC error. - Mcp-Session-Id is bound to the key that first received it; a foreign session answers 404; tool calls are audited before the upstream call; a tools/call without a string params.name is a 400; an unentitled server answers like an unknown one. - Realtime sessions and MCP listen streams are capped per key (max_live_streams_per_key, default 64) through a guard released with the stream. - In-process monthly counters are swept to the current and previous month at the daily reset; x-gw-user is capped at 256 bytes. - A tenant admin token can no longer lift a ban or change an abuse suspension; admin listings cap `limit` at 10000; denial messages carry the key fingerprint instead of the key. - The fallback chain skips requests that pin signed thinking to the model that produced it, and is taken on gateway 502/503 as well as vendor 5xx/429. docs/security.md records the trust boundaries, the failure postures and the review ledger, fixed and accepted alike. --- Cargo.lock | 1 + README.md | 6 +- crates/config/src/lib.rs | 14 + crates/engines/src/http_transport.rs | 14 +- crates/handler/src/lib.rs | 16 +- crates/state/src/admission.rs | 31 +- crates/state/src/governance.rs | 6 + crates/state/src/lib.rs | 11 + crates/state/src/streams.rs | 68 +++ crates/task/src/lib.rs | 4 + crates/views/Cargo.toml | 1 + crates/views/src/lib.rs | 86 +++- crates/views/src/mcp.rs | 615 ++++++++++++++++++++------- crates/views/src/mcp_auth.rs | 38 +- docs/api.md | 59 ++- docs/architecture.md | 6 +- docs/configuration.md | 8 +- docs/deployment.md | 5 +- docs/governance.md | 88 ++-- docs/index.md | 5 +- docs/multi-instance.md | 3 + docs/observability.md | 8 +- docs/security.md | 140 ++++++ 23 files changed, 969 insertions(+), 264 deletions(-) create mode 100644 crates/state/src/streams.rs create mode 100644 docs/security.md diff --git a/Cargo.lock b/Cargo.lock index c5575622..bf967569 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1050,6 +1050,7 @@ dependencies = [ "gw-protocol", "gw-state", "metrics", + "moka", "opentelemetry", "reqwest 0.12.28", "serde_json", diff --git a/README.md b/README.md index d2ce0d9d..1f098939 100644 --- a/README.md +++ b/README.md @@ -10,14 +10,14 @@ key-based auth, quotas, rate limits, failover, and a billing ledger. - **OpenAI + Anthropic compatible surface** — `/v1/chat/completions`, `/v1/completions`, `/v1/responses`, `/v1/messages`, `/v1/embeddings`, `/v1/images/{generations,edits}`, `/v1/videos/generations` + `/v1/videos/{id}`, `/v1/audio/{speech,transcriptions,translations}`, `/v1/moderations`, `/v1/search`, `/v1/rerank`, `/v1/batches` + `/v1/files`, `/v1/models`, `/v1/realtime` (WebSocket) — streaming and non-streaming - **Cross-protocol conversion** — serve Anthropic-style `/v1/messages` on OpenAI-protocol models and vice versa, including streaming event mapping -- **MCP gateway** — `/mcp/{server}` proxies Model Context Protocol servers (Streamable HTTP) behind the same access keys: per-key server entitlement and tool allowlists, `tools/list` filtered to the allowlist, tool results reviewed by the tenant's moderator (mask or block), every tool call, denial and intervention audited, server credentials kept in the gateway's environment as static bearers or OAuth 2.0 client credentials the gateway refreshes itself +- **MCP gateway** — `/mcp/{server}` proxies Model Context Protocol servers (Streamable HTTP) behind the same access keys: per-key server entitlement and tool allowlists, `tools/list` filtered to the allowlist, tool results reviewed by the tenant's moderator (mask or block), every tool call, denial and intervention audited, sessions bound to the key that opened them, server credentials kept in the gateway's environment as static bearers or OAuth 2.0 client credentials the gateway refreshes itself ([Security model](docs/security.md)) - **Coding agents drop in** — Claude Code, Codex CLI, VS Code chat (Copilot), Cursor and opencode work with a base URL and an access key; the `anthropic-beta` header rides through to Anthropic-wire upstreams and each client's captured wire shape replays in the live matrix ([Examples](docs/examples.md#coding-agents)) - **Reasoning on every surface** — `reasoning_effort` / `reasoning{}` on `/v1/chat/completions` maps to each family's thinking dialect (Anthropic budget or adaptive by model generation, OpenAI effort, compatible vendors verbatim); reasoning comes back as `reasoning_content` + signed `reasoning_details` and replays into tool loops; `/v1/responses` forwards reasoning items and its native event stream verbatim; `/v1/messages` preserves signed `thinking`/`redacted_thinking` blocks end-to-end, pins reasoning traffic to its requested model, and audits tool-loop continuations against what was actually served (fail-open; tampering is a local 400) - **Staged request pipeline** — a 4-layer DAG per request: model resolve / quota / cache lookup → account selection (priority, PTU-first, round-robin or latency-ranked within a tier, failover) → rate limits + engine call (retry on upstream 5xx, then the model's `fallback_models` chain before any byte is sent) → usage extraction, billing, cache store - **Governance built in** — access-key auth, daily token quotas, QPS / QPM / TPM limits at key, product, and model level, request-level TTL cache, account cooldown and recovery, DLP redaction and blocklist plugins. Admission reserves then settles, so concurrent requests can't overshoot a quota - **Multi-tenant** — keys carry a tenant; tenants get a pooled QPS bucket, a model entitlement allowlist, per-(key, model) quota defaults with an optional fallback-model degrade, key lifecycle (expiry/ban), and tenant-scoped admin tokens. Billing records charged cost and (optionally) vendor cost per row, so margin is queryable per tenant × model - **Per-user billing & enterprise audit** — every ledger row attributes to an effective end user (the key's `owner`, else the request's `x-gw-user` / `user` hint) with a `request_id`, so cost rolls up per user (`/admin/usage/users`) and soft per-user daily budgets apply on every surface; daily and calendar-month cost budgets (charged micro-dollars) cap a tenant pool, each key and each end user, with optional month-to-month rollover of the unspent remainder, raising a webhook alert when reached. Per-tenant content policy adds blocklist action tiers (block / flag / shadow), regex recognizers, secret masking, and an external-moderation seam with an AWS Bedrock Guardrails backend (deny on blocked policies, mask anonymized PII); every hit is recorded without prompt text. An admin-operation trail (key CRUD / config / reload, with source IP) and optional at-rest content retention complete the audit surfaces (`/admin/audit/*`) -- **Fleet-ready** — run N instances behind a load balancer: Postgres shares config (versioned + a change feed), the access-key table, the ledger/files/batches store, and a distributed batch queue any instance drains; Redis shares rate/quota/TPM counters, account health, and optionally the response cache. Single-node stays zero-dependency +- **Fleet-ready** — run N instances behind a load balancer: Postgres shares config (versioned + a change feed), the access-key table, the ledger/files/batches store, and a distributed batch queue any instance drains; Redis shares rate/quota/TPM counters, monthly cost counters, account health, and optionally the response cache. Single-node stays zero-dependency - **Providers behind traits** — engines talk to upstreams through a `Transport` seam; accounts with a real endpoint go over HTTP (reqwest + rustls), accounts without one are served by a deterministic in-process mock; AWS Bedrock (Claude, Llama, Cohere natively; every model through Converse) via SigV4 or API key with EventStream streaming - **Fast** — the whole pipeline (auth, admission, DLP, engine, billing) costs ~25 µs per request in-process; over HTTP one node serves ~90k requests/s at p99 under 10 ms on small bodies and ~40k/s at 256 concurrency on 52 KB / 13k-token prompts, mock upstream ([numbers and method](docs/performance.md)) - **Observability built in** — Prometheus `/metrics` (per-route request/status counters, per-pipeline-stage latency, token counters), structured access logs, and one OTLP span per request (route, model, tenant, user, tokens, routing decisions; W3C `traceparent` joins the caller's trace) as soon as `OTEL_EXPORTER_OTLP_ENDPOINT` names a collector @@ -47,7 +47,7 @@ GW_CONFIG=conf/gateway.yaml cargo run -p gw-server # GW_TRANSPORT=mock forces zero egress; GW_TRANSPORT=http disables the mock. ``` -Guides: [Examples](docs/examples.md) · [API](docs/api.md) · [Providers](docs/providers.md) · [Governance](docs/governance.md) · [Observability](docs/observability.md) · [Deployment](docs/deployment.md) · [Configuration](docs/configuration.md) · [Architecture](docs/architecture.md) · [Development](docs/development.md) · [Performance](docs/performance.md) · [Roadmap](https://github.com/cocoonstack/gateway/issues/1) +Guides: [Examples](docs/examples.md) · [API](docs/api.md) · [Providers](docs/providers.md) · [Governance](docs/governance.md) · [Observability](docs/observability.md) · [Deployment](docs/deployment.md) · [Configuration](docs/configuration.md) · [Architecture](docs/architecture.md) · [Development](docs/development.md) · [Performance](docs/performance.md) · [Security](docs/security.md) · [Roadmap](https://github.com/cocoonstack/gateway/issues/1) ## Docker diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs index 37ba65e0..fcccd605 100644 --- a/crates/config/src/lib.rs +++ b/crates/config/src/lib.rs @@ -130,6 +130,9 @@ pub struct McpServerConf { pub oauth: Option, #[serde(default = "default_mcp_timeout")] pub timeout_seconds: u64, + /// Largest reply the proxy buffers for filtering or review; a larger one is refused. + #[serde(default = "default_mcp_max_reply_bytes")] + pub max_reply_bytes: usize, } impl McpServerConf { @@ -743,6 +746,9 @@ pub struct GatewayConfig { /// MCP servers reachable through `/mcp/{server}` by entitled keys. #[serde(default)] pub mcp_servers: Vec, + /// Concurrent realtime sessions plus MCP listen streams one key may hold; 0 = unlimited. + #[serde(default = "default_max_live_streams")] + pub max_live_streams_per_key: usize, /// Trust `x-real-ip` / `x-forwarded-for` for the audit source IP. Off by /// default: the audit records the real TCP peer, which a client can't forge. /// Enable only when a trusted proxy fronts the gateway and sets those headers. @@ -1436,6 +1442,14 @@ fn default_mcp_timeout() -> u64 { 60 } +fn default_mcp_max_reply_bytes() -> usize { + 16 * 1024 * 1024 +} + +fn default_max_live_streams() -> usize { + 64 +} + fn default_alert_dedup_seconds() -> u64 { 300 } diff --git a/crates/engines/src/http_transport.rs b/crates/engines/src/http_transport.rs index 1ab02d85..e22adde6 100644 --- a/crates/engines/src/http_transport.rs +++ b/crates/engines/src/http_transport.rs @@ -166,11 +166,15 @@ impl Transport for HttpTransport { tokio::time::sleep(RETRY_BACKOFF * attempt).await; } Err(e) => { - return Err(GatewayError::new( - upstream_fault_code(e.is_timeout()), - 502, - format!("upstream request failed: {e}"), - )); + let what = if e.is_timeout() { + "upstream request timed out" + } else { + "upstream request failed" + }; + return Err( + GatewayError::new(upstream_fault_code(e.is_timeout()), 502, what) + .with_source(e), + ); } } }; diff --git a/crates/handler/src/lib.rs b/crates/handler/src/lib.rs index fcf2ba1f..d4d47775 100644 --- a/crates/handler/src/lib.rs +++ b/crates/handler/src/lib.rs @@ -276,7 +276,8 @@ impl OnlineHandler { ctx.quota_at, ) .await; - if let Some((i, next)) = fallback_after(&e) + // signed thinking replays only against the model that produced it + if let Some((i, next)) = (is_upstream_fault(&e) && !ctx.request.pins_reasoning_route()) .then(|| next_fallback(&snap.cfg, &ctx, tried)) .flatten() { @@ -557,10 +558,13 @@ async fn note_abuse(ctx: &DagContext) { .emit("abuse_suspend", ctx.ak.ak.clone(), summary); } -/// Whether a pipeline error is the upstream's fault: a 5xx (vendor or -/// connection failure) or a vendor 429; gateway-side denials never fall back. -fn fallback_after(e: &GatewayError) -> bool { - e.http_status >= 500 || e.original_status() == Some(429) +/// Whether a pipeline error came from upstream: a vendor 5xx or 429, or a +/// 502/503 the gateway raised for a connection failure or an exhausted pool. +fn is_upstream_fault(e: &GatewayError) -> bool { + match e.original_status() { + Some(status) => status >= 500 || status == 429, + None => e.http_status >= 502, + } } /// The next entry of the requested model's fallback chain past `tried` that the caller's tenant may use. @@ -1021,7 +1025,9 @@ mod tests { async fn vendor_by_model() -> (String, Arc) { use std::sync::atomic::{AtomicU32, Ordering}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let hits = Arc::new(AtomicU32::new(0)); let seen = Arc::clone(&hits); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); diff --git a/crates/state/src/admission.rs b/crates/state/src/admission.rs index de1f223f..7341245e 100644 --- a/crates/state/src/admission.rs +++ b/crates/state/src/admission.rs @@ -77,7 +77,7 @@ enum BudgetScope { } impl BudgetScope { - fn per_user(self) -> bool { + fn is_per_user(self) -> bool { matches!(self, Self::UserTokens | Self::UserCost) } @@ -91,14 +91,7 @@ impl BudgetScope { /// The governance counter: user scopes carry the tenant, month counters their calendar month. fn key(self, month: Option<(i64, u32)>, ak: &AkInfo, user: &str) -> String { - let tag; - let prefix = match month { - Some((y, m)) => { - tag = format!("m:{y}{m:02}:"); - tag.as_str() - } - None => "", - }; + let prefix = month.map_or(String::new(), month_prefix); match self { Self::UserTokens => format!("{prefix}ub:{}:{user}", ak.tenant), Self::TenantCost => format!("{prefix}cb:tenant:{}", ak.tenant), @@ -404,7 +397,7 @@ pub async fn check_tenant_rate( /// Per-AK QPS. pub async fn check_ak_rate(gov: &dyn Governance, ak: &AkInfo) -> Result<(), String> { admit(gov.rate_allow(&ak.ak, ak.qps).await, || { - format!("rate limit exceeded for ak {} (qps {})", ak.ak, ak.qps) + format!("rate limit exceeded for key {} (qps {})", ak.ak_id, ak.qps) }) } @@ -450,7 +443,7 @@ pub async fn reserve_daily( admit( gov.quota_reserve(&ak.ak, amount, ak.daily_token_quota, at) .await, - || format!("daily token quota exhausted for ak {}", ak.ak), + || format!("daily token quota exhausted for key {}", ak.ak_id), ) } @@ -470,8 +463,8 @@ pub async fn reserve_tpm( Ok(Some(amount)) } else { Err(format!( - "token-per-minute limit exceeded for ak {} (tpm {tpm})", - ak.ak + "token-per-minute limit exceeded for key {} (tpm {tpm})", + ak.ak_id )) } } @@ -583,7 +576,7 @@ async fn budgets( let Some(mut limit) = limit else { continue; }; - if scope.per_user() && user.is_empty() { + if scope.is_per_user() && user.is_empty() { continue; } let key = match window { @@ -622,6 +615,16 @@ fn previous_month((y, m): (i64, u32)) -> (i64, u32) { if m == 1 { (y - 1, 12) } else { (y, m - 1) } } +fn month_prefix((y, m): (i64, u32)) -> String { + format!("m:{y}{m:02}:") +} + +/// The counter prefixes of the current and previous month: everything the rollover still reads. +pub fn month_prefixes() -> [String; 2] { + let month = civil_month(crate::epoch_secs()); + [month_prefix(month), month_prefix(previous_month(month))] +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/state/src/governance.rs b/crates/state/src/governance.rs index 72579599..874aa739 100644 --- a/crates/state/src/governance.rs +++ b/crates/state/src/governance.rs @@ -39,6 +39,8 @@ pub trait Governance: Send + Sync + std::fmt::Debug { async fn counter_get(&self, key: &str) -> i64; /// Add to a calendar-window counter, arming `ttl` on first use; returns the new total. async fn counter_add(&self, key: &str, amount: i64, ttl: Duration) -> i64; + /// Drop in-process counters outside the windows `prefixes` name; a TTL backend has nothing to do. + async fn counter_retain(&self, prefixes: &[String]); /// Fixed-window request limit (QPM): take one permit. async fn window_allow(&self, key: &str, limit: i64, window: Duration) -> bool; @@ -113,6 +115,9 @@ impl Governance for MemoryGovernance { async fn counter_add(&self, key: &str, amount: i64, _ttl: Duration) -> i64 { self.counters.consume(key, amount) } + async fn counter_retain(&self, prefixes: &[String]) { + self.counters.retain_prefixed(prefixes); + } async fn window_allow(&self, key: &str, limit: i64, window: Duration) -> bool { self.qpm.reserve(key, 1, limit, window) } @@ -296,6 +301,7 @@ impl Governance for RedisGovernance { async fn counter_add(&self, key: &str, amount: i64, ttl: Duration) -> i64 { self.incr_window(&counter_key(key), amount, ttl).await } + async fn counter_retain(&self, _prefixes: &[String]) {} async fn window_allow(&self, key: &str, limit: i64, window: Duration) -> bool { self.incr_window(&format!("gw:qpm:{key}"), 1, window).await <= limit } diff --git a/crates/state/src/lib.rs b/crates/state/src/lib.rs index 0ca7cfc5..01cb1983 100644 --- a/crates/state/src/lib.rs +++ b/crates/state/src/lib.rs @@ -24,6 +24,7 @@ pub mod health; pub mod keystore; pub mod latency; pub mod store; +pub mod streams; pub mod thinking_signature; pub use alerts::{AlertBus, AlertEvent}; @@ -391,6 +392,12 @@ impl QuotaStore { pub fn reset_all(&self) { self.used.clear(); } + + /// Drop every counter whose key starts with none of `prefixes`. + pub fn retain_prefixed(&self, prefixes: &[String]) { + self.used + .retain(|k, _| prefixes.iter().any(|p| k.starts_with(p.as_str()))); + } } /// Account pool: the highest-priority slot serving a model type, round-robin @@ -761,6 +768,8 @@ pub struct GatewayState { pub thinking_signatures: ThinkingSignatureAudit, /// Per-account call latency for `stability.latency_routing`; per instance. pub latency: latency::Latency, + /// Open long-lived streams per key, for `max_live_streams_per_key`. + pub streams: Arc, } impl Default for GatewayState { @@ -778,6 +787,7 @@ impl Default for GatewayState { alerts: Arc::new(alerts::AlertBus::default()), thinking_signatures: ThinkingSignatureAudit::new(), latency: latency::Latency::default(), + streams: Arc::default(), } } } @@ -896,6 +906,7 @@ impl GatewayState { alerts: prev.alerts.clone(), thinking_signatures: prev.thinking_signatures.clone(), latency: prev.latency.clone(), + streams: prev.streams.clone(), }) } } diff --git a/crates/state/src/streams.rs b/crates/state/src/streams.rs new file mode 100644 index 00000000..de0fa563 --- /dev/null +++ b/crates/state/src/streams.rs @@ -0,0 +1,68 @@ +//! Concurrent long-lived streams per access key — realtime sessions and MCP +//! listen streams — bounded by `max_live_streams_per_key`; a guard releases +//! the slot when the stream ends. + +use std::sync::Arc; + +use dashmap::DashMap; + +#[derive(Debug, Default)] +pub struct LiveStreams { + open: DashMap, +} + +impl LiveStreams { + /// Take a slot for `ak`; `None` at `cap` (0 = unlimited). + pub fn open(self: &Arc, ak: &str, cap: usize) -> Option { + let mut n = crate::slot_mut(&self.open, ak, || 0); + if cap > 0 && *n >= cap { + return None; + } + *n += 1; + drop(n); + Some(StreamGuard { + streams: Arc::clone(self), + ak: ak.to_owned(), + }) + } +} + +/// One held stream slot; dropping it frees the slot. +#[derive(Debug)] +pub struct StreamGuard { + streams: Arc, + ak: String, +} + +impl Drop for StreamGuard { + fn drop(&mut self) { + // the entry lock is released before the removal, which would deadlock on it + let left = self.streams.open.get_mut(&self.ak).map(|mut n| { + *n = n.saturating_sub(1); + *n + }); + if left == Some(0) { + self.streams.open.remove_if(&self.ak, |_, n| *n == 0); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn slots_are_capped_and_released_on_drop() { + let streams = Arc::new(LiveStreams::default()); + let a = streams.open("k", 2).expect("first slot"); + let b = streams.open("k", 2).expect("second slot"); + assert!(streams.open("k", 2).is_none(), "cap reached"); + assert!(streams.open("other", 2).is_some(), "caps are per key"); + assert_eq!(streams.open.get("k").as_deref(), Some(&2)); + drop(a); + assert_eq!(streams.open.get("k").as_deref(), Some(&1)); + assert!(streams.open("k", 2).is_some()); + drop(b); + assert!(streams.open("k", 0).is_some(), "0 = unlimited"); + } +} diff --git a/crates/task/src/lib.rs b/crates/task/src/lib.rs index b89ccc75..1f443fa4 100644 --- a/crates/task/src/lib.rs +++ b/crates/task/src/lib.rs @@ -31,6 +31,10 @@ pub fn spawn_quota_reset( loop { tick.tick().await; state.governance.quota_reset_all().await; + state + .governance + .counter_retain(&gw_state::admission::month_prefixes()) + .await; tracing::info!(target: "task", "quota_reset: all AK daily counters cleared"); } }) diff --git a/crates/views/Cargo.toml b/crates/views/Cargo.toml index f42a354f..35a5f510 100644 --- a/crates/views/Cargo.toml +++ b/crates/views/Cargo.toml @@ -20,6 +20,7 @@ tokio-tungstenite = { workspace = true } futures = { workspace = true } serde_json = { workspace = true } bytes = { workspace = true } +moka = { workspace = true } tracing = { workspace = true } tracing-opentelemetry = { workspace = true } opentelemetry = { workspace = true } diff --git a/crates/views/src/lib.rs b/crates/views/src/lib.rs index 0a8a1a35..d7c78856 100644 --- a/crates/views/src/lib.rs +++ b/crates/views/src/lib.rs @@ -9,7 +9,7 @@ use std::fmt::Write as _; use std::mem::take; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; -use std::time::Instant; +use std::time::{Duration, Instant}; use axum::extract::{Path, Query, State}; use axum::http::{HeaderMap, StatusCode}; @@ -58,6 +58,11 @@ const STREAM_CHANNEL_CAP: usize = 64; /// Target of the per-request span; a layer exporting it is installed only with an OTLP collector. pub const TRACE_TARGET: &str = "gw::trace"; const NO_OUTCOME: &str = "pipeline produced no outcome"; +const ADMIN_PAGE_MAX: usize = 10_000; +/// Longest `x-gw-user` accepted: the hint keys governance counters. +const USER_HINT_MAX_BYTES: usize = 256; +const MCP_SESSION_CAP: u64 = 100_000; +const MCP_SESSION_TTL: Duration = Duration::from_secs(24 * 3_600); /// Per-turn token reserve against the AK daily quota; settled to actuals at billing. const REALTIME_TURN_RESERVE: i64 = 1_000; @@ -76,6 +81,8 @@ pub struct AppState { pub mcp: reqwest::Client, /// Upstream MCP credentials, OAuth tokens cached per server. pub mcp_auth: Arc, + /// MCP session id → the fingerprint of the key that opened it. + pub mcp_sessions: moka::sync::Cache>, /// Reloads config from its source; `None` = reload not wired (tests). pub loader: Option, /// Fleet config store; enables `PUT /admin/config`. `None` = file-based. @@ -101,8 +108,9 @@ impl AppState { Self { handler, offline, - mcp: reqwest::Client::new(), + mcp: mcp_client(), mcp_auth: Arc::default(), + mcp_sessions: mcp_sessions(), loader, config_store: None, } @@ -124,6 +132,25 @@ impl AppState { } } +/// The MCP proxy's client: no redirects, so a server or token endpoint cannot +/// steer a credentialed request elsewhere. +fn mcp_client() -> reqwest::Client { + reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .build() + .unwrap_or_else(|e| { + tracing::error!(error = %e, "mcp client fell back to the default client"); + reqwest::Client::new() + }) +} + +fn mcp_sessions() -> moka::sync::Cache> { + moka::sync::Cache::builder() + .max_capacity(MCP_SESSION_CAP) + .time_to_live(MCP_SESSION_TTL) + .build() +} + pub fn app(state: AppState) -> Router { Router::new() .route("/health", get(health)) @@ -348,6 +375,13 @@ async fn realtime_ws( } } }; + let Some(stream_guard) = snap + .state + .streams + .open(&ak.ak, snap.cfg.max_live_streams_per_key) + else { + return error_response(429, "too many open realtime sessions for this key"); + }; let Some(model) = q.remove("model") else { return error_response(400, "model query param is required"); }; @@ -414,13 +448,14 @@ async fn realtime_ws( }; // select "realtime" so subprotocol-offering clients get a valid handshake let ws = ws.protocols(["realtime"]); - if account.endpoint.is_empty() { - ws.on_upgrade(move |socket| { - realtime_session(socket, s, ak, m, mt, account.name.clone(), hint) - }) - } else { - ws.on_upgrade(move |socket| realtime_bridge(socket, s, ak, m, mt, account, hint)) - } + ws.on_upgrade(move |socket| async move { + let _held = stream_guard; + if account.endpoint.is_empty() { + realtime_session(socket, s, ak, m, mt, account.name.clone(), hint).await + } else { + realtime_bridge(socket, s, ak, m, mt, account, hint).await + } + }) } /// A realtime session's model identity: entitlement judges `requested`, pricing and routing follow `served`. @@ -499,7 +534,7 @@ async fn realtime_gate( _ => { return Err(( ErrClass::AccessDenied, - format!("access key {} is no longer valid", ak.ak), + format!("access key {} is no longer valid", ak.ak_id), )); } }; @@ -1255,7 +1290,7 @@ async fn ledger( if let Err(r) = require_global_admin(&s, &headers) { return r; } - let limit = q_num(&q, "limit", LEDGER_PAGE_DEFAULT); + let limit = q_num(&q, "limit", LEDGER_PAGE_DEFAULT).min(ADMIN_PAGE_MAX); match s.handler.state().store.ledger_snapshot(limit).await { Ok((count, records)) => Json(json!({ "count": count, "records": records })).into_response(), Err(e) => gateway_error(e), @@ -1319,6 +1354,12 @@ async fn authenticate( "missing api key (Authorization: Bearer or x-api-key)", )); }; + if headers + .get("x-gw-user") + .is_some_and(|v| v.len() > USER_HINT_MAX_BYTES) + { + return Err((400, "x-gw-user exceeds 256 bytes")); + } let info = s .handler .state() @@ -1942,6 +1983,15 @@ async fn admin_key_patch( banned: body["banned"].as_bool(), suspended_until_epoch_secs: tri("suspended_until_epoch_secs"), }; + // a ban or an abuse suspension is a platform sanction; a tenant may add one, never lift one + if matches!(scope, AdminScope::Tenant(_)) + && (patch.banned == Some(false) || patch.suspended_until_epoch_secs.is_some()) + { + return error_response( + 403, + "lifting a ban or changing a suspension requires the global admin token", + ); + } let patched = s.handler.state().auth.patch(&ak, &patch).await; match patched { Err(e) => gateway_error(e), @@ -2033,7 +2083,7 @@ async fn admin_config_versions( Ok(v) => v, Err(r) => return r, }; - let limit = q_num(&q, "limit", CONFIG_VERSION_PAGE_DEFAULT); + let limit = q_num(&q, "limit", CONFIG_VERSION_PAGE_DEFAULT).min(ADMIN_PAGE_MAX); match store.list_versions(limit).await { Ok(versions) => Json(json!({ "versions": versions })).into_response(), Err(e) => gateway_error(e), @@ -2193,7 +2243,7 @@ async fn admin_key_list( return Json(resp).into_response(); } let offset = q_num(&q, "offset", 0); - let limit = q_num(&q, "limit", KEY_PAGE_DEFAULT); + let limit = q_num(&q, "limit", KEY_PAGE_DEFAULT).min(ADMIN_PAGE_MAX); // the scope filters in the store before paging, or a tenant admin's page could come back empty let tenant = scope.tenant_filter(&q); let listed = match s.handler.state().auth.list(tenant, offset, limit).await { @@ -2409,7 +2459,7 @@ async fn admin_security_events( Query(q): Query>, ) -> Response { let tenant = scope.tenant_filter(&q); - let limit = q_num(&q, "limit", LEDGER_PAGE_DEFAULT); + let limit = q_num(&q, "limit", LEDGER_PAGE_DEFAULT).min(ADMIN_PAGE_MAX); match s.handler.state().store.security_events(tenant, limit).await { Ok(events) => Json(json!({ "events": events })).into_response(), Err(e) => gateway_error(e), @@ -2426,7 +2476,7 @@ async fn admin_audit_ops( if let Err(r) = require_global_admin(&s, &headers) { return r; } - let limit = q_num(&q, "limit", LEDGER_PAGE_DEFAULT); + let limit = q_num(&q, "limit", LEDGER_PAGE_DEFAULT).min(ADMIN_PAGE_MAX); match s.handler.state().store.admin_audit_list(limit).await { Ok(entries) => Json(json!({ "entries": entries })).into_response(), Err(e) => gateway_error(e), @@ -5715,8 +5765,9 @@ mod tests { let app = AppState { handler, offline, - mcp: reqwest::Client::new(), + mcp: mcp_client(), mcp_auth: Arc::default(), + mcp_sessions: mcp_sessions(), loader: None, config_store: None, }; @@ -5780,8 +5831,9 @@ mod tests { let app = AppState { handler, offline, - mcp: reqwest::Client::new(), + mcp: mcp_client(), mcp_auth: Arc::default(), + mcp_sessions: mcp_sessions(), loader: None, config_store: None, }; diff --git a/crates/views/src/mcp.rs b/crates/views/src/mcp.rs index e90b018d..b2cbcfdc 100644 --- a/crates/views/src/mcp.rs +++ b/crates/views/src/mcp.rs @@ -1,8 +1,9 @@ //! `/mcp/{server}`: the Model Context Protocol proxy. A key reaches only the //! servers it is entitled to and the tools its allowlist names; `tools/list` -//! is filtered to that allowlist, a tenant under `security.moderate` has its -//! tool results reviewed, and every call, denial and intervention is a -//! security event. +//! is filtered to that allowlist, a tenant under `security.moderate` has the +//! results of `tools/call`, `resources/read` and `prompts/get` reviewed before +//! they leave, sessions are bound to the key that opened them, and every +//! call, denial and intervention is a security event. use std::time::Duration; @@ -10,7 +11,8 @@ use axum::body::{Body, Bytes}; use axum::extract::{Path, State}; use axum::http::{HeaderMap, Method, StatusCode}; use axum::response::{IntoResponse, Response}; -use gw_config::McpServerConf; +use futures::StreamExt as _; +use gw_config::{McpServerConf, SecurityConf}; use gw_handler::{RtModeration, plugins}; use gw_state::{AkInfo, GatewayState, SecurityEvent, Snapshot, admission}; use serde_json::{Value, json}; @@ -25,8 +27,13 @@ const FORWARDED_HEADERS: [&str; 5] = [ "last-event-id", ]; const RETURNED_HEADERS: [&str; 2] = ["content-type", "mcp-session-id"]; +/// Methods whose results carry prose an agent reads; reviewed under `security.moderate`. +const REVIEWED_METHODS: [&str; 3] = ["tools/call", "resources/read", "prompts/get"]; +/// Result fields that carry identifiers or binary, never prose. +const OPAQUE_KEYS: [&str; 5] = ["blob", "mimeType", "name", "type", "uri"]; const JSONRPC_TOOL_DENIED: i64 = -32000; const JSONRPC_RESULT_BLOCKED: i64 = -32001; +const UNREVIEWABLE: &str = "the result could not be reviewed"; /// The JSON-RPC envelope of one POST, as far as the proxy needs it. #[derive(Default)] @@ -36,9 +43,11 @@ struct Call { tool: Option, } -/// One piece of a buffered reply: a JSON-RPC message the proxy may rewrite, or bytes it passes through. +/// One piece of a buffered reply: a JSON-RPC message the proxy may rewrite, a +/// `data` payload it could not parse, or framing it passes through. enum Segment { Message(Value), + Opaque(String), Raw(String), } @@ -54,18 +63,29 @@ pub(crate) async fn proxy( Err((status, msg)) => return error_response(status, msg), }; let snap = s.handler.config.load(); - let Some(conf) = snap.cfg.find_mcp_server(&server) else { + // an unentitled server answers like an unknown one, so names cannot be probed + let Some(conf) = snap + .cfg + .find_mcp_server(&server) + .filter(|_| ak.mcp.reaches(&server)) + else { return error_response(404, format!("unknown mcp server: {server}")); }; - if !ak.mcp.reaches(&server) { - return error_response( - 403, - format!("mcp server `{server}` is not entitled for this key"), - ); + let gov = snap.state.governance.as_ref(); + if let Err(e) = admission::check_tenant_rate(gov, &snap.cfg, &ak.tenant).await { + return error_response(429, e); } - if let Err(e) = admission::check_ak_rate(snap.state.governance.as_ref(), &ak).await { + if let Err(e) = admission::check_ak_rate(gov, &ak).await { return error_response(429, e); } + let session = headers.get("mcp-session-id").and_then(|v| v.to_str().ok()); + if let Some(sid) = session + && s.mcp_sessions + .get(sid) + .is_some_and(|owner| owner != ak.ak_id) + { + return error_response(404, "unknown mcp session"); + } let call = if method == Method::POST { match parse_call(&body) { Ok(call) => call, @@ -93,25 +113,19 @@ pub(crate) async fn proxy( format!("tool `{tool}` is not permitted for this key"), ); } - let label = method_label(&call.method); - let mut reply = match send(&s, conf, &method, &headers, &body).await { - Ok(reply) => reply, - Err(e) => { - count(&server, label, "upstream_error"); - return error_response(502, format!("mcp server `{server}`: {e}")); - } - }; - // a token the server stopped honoring is fetched anew once - if reply.status() == StatusCode::UNAUTHORIZED && conf.oauth.is_some() { - s.mcp_auth.invalidate(&conf.name); - reply = match send(&s, conf, &method, &headers, &body).await { - Ok(reply) => reply, - Err(e) => { - count(&server, label, "upstream_error"); - return error_response(502, format!("mcp server `{server}`: {e}")); - } + let stream_guard = if method == Method::GET { + let Some(guard) = snap + .state + .streams + .open(&ak.ak, snap.cfg.max_live_streams_per_key) + else { + return error_response(429, "too many open mcp streams for this key"); }; - } + Some(guard) + } else { + None + }; + let label = method_label(&call.method); if let Some(tool) = call.tool.as_deref() { audit( &snap.state, @@ -122,6 +136,23 @@ pub(crate) async fn proxy( ) .await; } + let sec = snap.cfg.security_for(&ak.tenant); + // a reviewed tenant gets no stream resumption: a replayed result would skip the review + let resumable = !sec.moderate; + let mut sent = send(&s, conf, &method, &headers, &body, resumable).await; + // a token the server stopped honoring is fetched anew once + if conf.oauth.is_some() && matches!(&sent, Ok(r) if r.status() == StatusCode::UNAUTHORIZED) { + s.mcp_auth.invalidate(&conf.name); + sent = send(&s, conf, &method, &headers, &body, resumable).await; + } + let reply = match sent { + Ok(reply) => reply, + Err(e) => { + tracing::warn!(server, error = %e, "mcp upstream request failed"); + count(&server, label, "upstream_error"); + return error_response(502, format!("mcp server `{server}` is unavailable")); + } + }; let status = reply.status(); count(&server, label, crate::status_label(status)); let mut out = HeaderMap::new(); @@ -130,23 +161,48 @@ pub(crate) async fn proxy( out.insert(name, v.clone()); } } + if let Some(sid) = out.get("mcp-session-id").and_then(|v| v.to_str().ok()) + && s.mcp_sessions.get(sid).is_none() + { + s.mcp_sessions.insert(sid.to_owned(), ak.ak_id.clone()); + } let sse = out .get("content-type") .and_then(|v| v.to_str().ok()) .is_some_and(|ct| ct.starts_with("text/event-stream")); - let sec = snap.cfg.security_for(&ak.tenant); let filtered = status.is_success() && call.method == "tools/list" && allowed.is_some(); - let reviewed = status.is_success() && call.tool.is_some() && sec.moderate; + let reviewed = + status.is_success() && sec.moderate && REVIEWED_METHODS.contains(&call.method.as_str()); if !filtered && !reviewed { - return (status, out, Body::from_stream(reply.bytes_stream())).into_response(); + let stream = reply.bytes_stream().map(move |chunk| { + let _held = &stream_guard; + chunk + }); + return (status, out, Body::from_stream(stream)).into_response(); } - let bytes = match reply.bytes().await { + let bytes = match read_capped(reply, conf.max_reply_bytes).await { Ok(bytes) => bytes, - Err(e) => return error_response(502, format!("mcp server `{server}`: {e}")), + Err(e) => { + tracing::warn!(server, error = %e, "mcp reply not read"); + count(&server, label, "reply_unreadable"); + return error_response( + 502, + format!("mcp server `{server}` reply could not be read"), + ); + } }; let body = match allowed { - Some(list) if filtered => filter_tool_list(&bytes, sse, list), - _ => moderate_result(&s, &snap, &ak, &server, call.id, &bytes, sse).await, + Some(list) if filtered => match filter_tool_list(&bytes, sse, list) { + Some(body) => body, + None => { + count(&server, label, "reply_unreadable"); + return error_response( + 502, + format!("mcp server `{server}` reply could not be filtered"), + ); + } + }, + _ => moderate_result(&s, &snap, sec, &ak, &server, label, call.id, &bytes, sse).await, }; (status, out, Body::from(body)).into_response() } @@ -157,6 +213,7 @@ async fn send( method: &Method, headers: &HeaderMap, body: &Bytes, + resumable: bool, ) -> Result { let bearer = s .mcp_auth @@ -168,7 +225,9 @@ async fn send( upstream = upstream.timeout(Duration::from_secs(conf.timeout_seconds)); } for name in FORWARDED_HEADERS { - if let Some(v) = headers.get(name) { + if let Some(v) = headers.get(name) + && (resumable || name != "last-event-id") + { upstream = upstream.header(name, v); } } @@ -181,6 +240,20 @@ async fn send( upstream.send().await.map_err(|e| e.to_string()) } +/// Collect a reply body up to `cap` bytes; a larger one is refused rather than held. +async fn read_capped(reply: reqwest::Response, cap: usize) -> Result, String> { + let mut stream = reply.bytes_stream(); + let mut out = Vec::new(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(|e| e.to_string())?; + if out.len() + chunk.len() > cap { + return Err(format!("reply exceeds max_reply_bytes ({cap})")); + } + out.extend_from_slice(&chunk); + } + Ok(out) +} + fn parse_call(body: &[u8]) -> Result { let v: Value = serde_json::from_slice(body).map_err(|e| format!("body is not JSON-RPC: {e}"))?; @@ -192,13 +265,13 @@ fn parse_call(body: &[u8]) -> Result { .and_then(Value::as_str) .unwrap_or_default() .to_owned(); - let tool = (method == "tools/call") - .then(|| { - obj.get("params") - .and_then(|p| p["name"].as_str()) - .map(str::to_owned) - }) - .flatten(); + let tool = match obj.get("params").and_then(|p| p.get("name")) { + Some(Value::String(name)) if method == "tools/call" => Some(name.clone()), + _ if method == "tools/call" => { + return Err("tools/call needs a string params.name".to_owned()); + } + _ => None, + }; Ok(Call { method, id: obj.remove("id").unwrap_or(Value::Null), @@ -206,15 +279,18 @@ fn parse_call(body: &[u8]) -> Result { }) } -/// Keep only the allowlisted tools in every `tools/list` result the reply carries. -fn filter_tool_list(bytes: &[u8], sse: bool, allowed: &[String]) -> Vec { +/// Keep only the allowlisted tools in every `tools/list` result; `None` when a message could not be parsed. +fn filter_tool_list(bytes: &[u8], sse: bool, allowed: &[String]) -> Option> { let mut segments = parse_segments(bytes, sse); + if segments.iter().any(|seg| matches!(seg, Segment::Opaque(_))) { + return None; + } for tools in segments.iter_mut().filter_map(|seg| match seg { Segment::Message(msg) => msg .get_mut("result") .and_then(|r| r.get_mut("tools")) .and_then(Value::as_array_mut), - Segment::Raw(_) => None, + _ => None, }) { tools.retain(|t| { t["name"] @@ -222,110 +298,151 @@ fn filter_tool_list(bytes: &[u8], sse: bool, allowed: &[String]) -> Vec { .is_some_and(|n| allowed.iter().any(|a| a == n)) }); } - serialize_segments(segments, sse) + Some(serialize_segments(segments, sse, bytes.len())) } -/// Review a `tools/call` result's text: deny → JSON-RPC error, mask → in-place rewrite, both recorded. +/// Review a reply's prose: deny → one JSON-RPC error replaces the reply, mask → in-place rewrite, both recorded. +#[allow(clippy::too_many_arguments)] async fn moderate_result( s: &AppState, snap: &Snapshot, + sec: &SecurityConf, ak: &AkInfo, server: &str, + label: &'static str, id: Value, bytes: &[u8], sse: bool, ) -> Vec { let mut segments = parse_segments(bytes, sse); - let texts: Vec<&mut String> = segments.iter_mut().flat_map(tool_texts).collect(); + if segments.iter().any(|seg| matches!(seg, Segment::Opaque(_))) { + return blocked(snap, ak, server, label, id, UNREVIEWABLE, sse).await; + } + let texts: Vec<&mut String> = segments.iter_mut().flat_map(review_slots).collect(); let review = plugins::slot_text(texts.iter().map(|s| s.as_str())); if review.is_empty() { - return serialize_segments(segments, sse); + return serialize_segments(segments, sse, bytes.len()); } - let sec = snap.cfg.security_for(&ak.tenant); match s.handler.moderate_rt(sec, &review).await { RtModeration::Allow => {} RtModeration::Mask(spans) => { let hits = plugins::apply_mask_slots(&spans, texts); - if hits > 0 { - audit( - &snap.state, - ak, - "moderation".to_owned(), - "mask".to_owned(), - hits as i64, - ) - .await; - count(server, "tools/call", "masked"); + if hits == 0 { + return blocked(snap, ak, server, label, id, UNREVIEWABLE, sse).await; } + audit(&snap.state, ak, "moderation", "mask", hits as i64).await; + count(server, label, "masked"); } RtModeration::Deny(reason) => { - audit( - &snap.state, - ak, - "moderation".to_owned(), - "block".to_owned(), - 1, - ) - .await; - count(server, "tools/call", "blocked"); - for seg in &mut segments { - if let Segment::Message(msg) = seg - && msg.get("result").is_some() - { - *msg = jsonrpc_error_value(id.clone(), JSONRPC_RESULT_BLOCKED, &reason); - } - } + return blocked(snap, ak, server, label, id, &reason, sse).await; } } - serialize_segments(segments, sse) + serialize_segments(segments, sse, bytes.len()) } -/// The text items of a `tools/call` result, in wire order; nothing for other messages. -fn tool_texts(seg: &mut Segment) -> impl Iterator { - let content = match seg { - Segment::Message(msg) => msg - .get_mut("result") - .and_then(|r| r.get_mut("content")) - .and_then(Value::as_array_mut), - Segment::Raw(_) => None, - }; - content - .into_iter() - .flatten() - .filter(|c| c["type"] == "text") - .filter_map(|c| match c.get_mut("text") { - Some(Value::String(s)) => Some(s), - _ => None, - }) +/// The whole reply becomes one JSON-RPC error, so nothing unreviewed leaves. +async fn blocked( + snap: &Snapshot, + ak: &AkInfo, + server: &str, + label: &'static str, + id: Value, + reason: &str, + sse: bool, +) -> Vec { + audit(&snap.state, ak, "moderation", "block", 1).await; + count(server, label, "blocked"); + let error = jsonrpc_error_value(id, JSONRPC_RESULT_BLOCKED, reason); + let mut segments = vec![Segment::Message(error)]; + if sse { + segments.push(Segment::Raw("\n".to_owned())); + } + serialize_segments(segments, sse, 0) } -/// A bare JSON body is one message; an event stream is its `data:` lines, everything else verbatim. +/// Every prose slot of a message: string leaves under `result` (a notification's `params`), identifiers and binary skipped. +fn review_slots(seg: &mut Segment) -> Vec<&mut String> { + let mut slots = Vec::new(); + if let Segment::Message(msg) = seg { + let root = if msg.get("result").is_some() { + msg.get_mut("result") + } else { + msg.get_mut("params") + }; + if let Some(root) = root { + collect_prose(root, &mut slots); + } + } + slots +} + +fn collect_prose<'a>(v: &'a mut Value, out: &mut Vec<&'a mut String>) { + match v { + Value::String(s) => out.push(s), + Value::Array(items) => items.iter_mut().for_each(|x| collect_prose(x, out)), + Value::Object(map) => map + .iter_mut() + .filter(|(k, _)| !OPAQUE_KEYS.contains(&k.as_str())) + .for_each(|(_, x)| collect_prose(x, out)), + _ => {} + } +} + +/// A bare JSON body is one message; an event stream is its events, each event's `data` lines joined by newlines, framing kept verbatim. fn parse_segments(bytes: &[u8], sse: bool) -> Vec { let text = String::from_utf8_lossy(bytes); + let text = text.strip_prefix('\u{feff}').unwrap_or(&text); if !sse { - return vec![match serde_json::from_str(&text) { + return vec![match serde_json::from_str(text) { Ok(msg) => Segment::Message(msg), - Err(_) => Segment::Raw(text.into_owned()), + Err(_) => Segment::Opaque(text.to_owned()), }]; } - text.split_inclusive('\n') - .map(|line| { - match line - .strip_prefix("data:") - .and_then(|data| serde_json::from_str(data.trim()).ok()) - { - Some(msg) => Segment::Message(msg), - None => Segment::Raw(line.to_owned()), + let mut segments = Vec::new(); + let mut data: Option = None; + let flush = |data: &mut Option, segments: &mut Vec| { + if let Some(payload) = data.take() { + segments.push(match serde_json::from_str(&payload) { + Ok(msg) => Segment::Message(msg), + Err(_) => Segment::Opaque(payload), + }); + } + }; + for line in text.split_inclusive('\n') { + let field = line.trim_end_matches(['\r', '\n']); + if let Some(d) = field.strip_prefix("data:") { + let d = d.strip_prefix(' ').unwrap_or(d); + match &mut data { + Some(acc) => { + acc.push('\n'); + acc.push_str(d); + } + None => data = Some(d.to_owned()), } - }) - .collect() + continue; + } + if field.is_empty() { + flush(&mut data, &mut segments); + } + segments.push(Segment::Raw(line.to_owned())); + } + flush(&mut data, &mut segments); + segments } -fn serialize_segments(segments: Vec, sse: bool) -> Vec { - let mut out = Vec::new(); +fn serialize_segments(segments: Vec, sse: bool, hint: usize) -> Vec { + let mut out = Vec::with_capacity(hint); for seg in segments { match seg { Segment::Raw(s) => out.extend_from_slice(s.as_bytes()), + Segment::Opaque(payload) if sse => { + for line in payload.split('\n') { + out.extend_from_slice(b"data: "); + out.extend_from_slice(line.as_bytes()); + out.push(b'\n'); + } + } + Segment::Opaque(payload) => out.extend_from_slice(payload.as_bytes()), Segment::Message(msg) => { if sse { out.extend_from_slice(b"data: "); @@ -363,7 +480,13 @@ fn count(server: &str, method: &'static str, result: impl Into, + action: impl Into, + hits: i64, +) { SecurityEvent { created_at_epoch_secs: gw_state::epoch_secs(), request_id: gw_handler::new_request_id(), @@ -371,8 +494,8 @@ async fn audit(state: &GatewayState, ak: &AkInfo, rule: String, action: String, user_id: ak.owner.clone().unwrap_or_default(), tenant: ak.tenant.clone(), surface: "mcp".to_owned(), - rule, - action, + rule: rule.into(), + action: action.into(), hits, } .record(state.store.as_ref()) @@ -409,6 +532,7 @@ mod tests { fetches: AtomicUsize, expires_in: u64, reject_first_token: bool, + multiline: bool, } async fn stub( @@ -430,9 +554,14 @@ mod tests { Some("tools/call") => { json!({"jsonrpc":"2.0","id":req["id"],"result":{"content":[{"type":"text","text": format!("called {} with {}", req["params"]["name"], req["params"]["arguments"])}, - {"type":"image","data":"AAAA"}]}}) + {"type":"image","data":"AAAA"}],"structuredContent":{"echoed":req["params"]["arguments"]["s"]}}}) } - Some("ping") => json!({"jsonrpc":"2.0","id":req["id"],"result":{"auth":auth}}), + Some("resources/read") => { + json!({"jsonrpc":"2.0","id":req["id"],"result":{"contents":[{"uri":req["params"]["uri"],"text":"contact bob@example.com"}]}}) + } + Some("ping") => json!({"jsonrpc":"2.0","id":req["id"],"result":{"auth":auth, + "session":headers.get("mcp-session-id").and_then(|v| v.to_str().ok()), + "last_event_id":headers.get("last-event-id").and_then(|v| v.to_str().ok())}}), _ => { json!({"jsonrpc":"2.0","id":req["id"],"result":{"protocolVersion":"2025-06-18","capabilities":{}}}) } @@ -445,12 +574,34 @@ mod tests { out.insert("mcp-session-id", "sess-1".parse().unwrap()); if sse { out.insert("content-type", "text/event-stream".parse().unwrap()); - let body = format!("event: message\ndata: {reply}\n\n"); + let body = if st.multiline { + let text = serde_json::to_string_pretty(&reply).unwrap(); + let data: String = text + .lines() + .enumerate() + .map(|(i, l)| { + if i == 0 { + format!("data: {l}\r\n") + } else { + format!("data:{l}\r\n") + } + }) + .collect(); + format!("event: message\r\nid: 7\r\n{data}\r\n") + } else { + format!("event: message\ndata: {reply}\n\n") + }; return (StatusCode::OK, out, body).into_response(); } (StatusCode::OK, out, axum::Json(reply)).into_response() } + async fn stub_listen() -> Response { + let mut out = HeaderMap::new(); + out.insert("content-type", "text/event-stream".parse().unwrap()); + (StatusCode::OK, out, "event: message\ndata: {\"jsonrpc\":\"2.0\",\"method\":\"notifications/message\",\"params\":{\"data\":\"hello\"}}\n\n").into_response() + } + async fn token(State(st): State>, body: String) -> Response { assert!(body.contains("grant_type=client_credentials"), "{body}"); assert!(body.contains("client_id=gw"), "{body}"); @@ -460,15 +611,24 @@ mod tests { } async fn spawn_stub_with(expires_in: u64, reject_first_token: bool) -> (String, Arc) { + spawn_stub_full(expires_in, reject_first_token, false).await + } + + async fn spawn_stub_full( + expires_in: u64, + reject_first_token: bool, + multiline: bool, + ) -> (String, Arc) { let st = Arc::new(Stub { fetches: AtomicUsize::new(0), expires_in, reject_first_token, + multiline, }); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let app = Router::new() - .route("/mcp", post(stub)) + .route("/mcp", post(stub).get(stub_listen)) .route("/token", post(token)) .with_state(st.clone()); tokio::spawn(axum::serve(listener, app).into_future()); @@ -481,7 +641,7 @@ mod tests { fn app_yaml(base: &str) -> String { format!( - "listen: {{host: h, port: 1}}\nmcp_servers: [{{name: tools, endpoint: {base}/mcp, api_key_env: GW_TEST_MCP_TOKEN}}, {{name: locked, endpoint: {base}/mcp}}, {{name: oauth, endpoint: {base}/mcp, oauth: {{token_url: {base}/token, client_id: gw, client_secret_env: GW_TEST_MCP_SECRET}}}}]\ntenants: [{{name: reviewed, security: {{moderate: true}}}}]\naccess_keys: [{{ak: k-add, product: p, qps: 100, daily_token_quota: 1000, mcp_servers: [tools], mcp_tools: {{tools: [add]}}}}, {{ak: k-all, product: p, qps: 100, daily_token_quota: 1000, mcp_servers: [tools]}}, {{ak: k-oauth, product: p, qps: 100, daily_token_quota: 1000, mcp_servers: [oauth]}}, {{ak: k-mod, tenant: reviewed, product: p, qps: 100, daily_token_quota: 1000, mcp_servers: [tools]}}]" + "listen: {{host: h, port: 1}}\nmax_live_streams_per_key: 2\nmcp_servers: [{{name: tools, endpoint: {base}/mcp, api_key_env: GW_TEST_MCP_TOKEN, max_reply_bytes: 4096}}, {{name: locked, endpoint: {base}/mcp}}, {{name: oauth, endpoint: {base}/mcp, oauth: {{token_url: {base}/token, client_id: gw, client_secret_env: GW_TEST_MCP_SECRET}}}}]\ntenants: [{{name: reviewed, security: {{moderate: true}}}}]\naccess_keys: [{{ak: k-add, product: p, qps: 100, daily_token_quota: 1000, mcp_servers: [tools], mcp_tools: {{tools: [add]}}}}, {{ak: k-all, product: p, qps: 100, daily_token_quota: 1000, mcp_servers: [tools]}}, {{ak: k-oauth, product: p, qps: 100, daily_token_quota: 1000, mcp_servers: [oauth]}}, {{ak: k-mod, tenant: reviewed, product: p, qps: 100, daily_token_quota: 1000, mcp_servers: [tools]}}]" ) } @@ -498,13 +658,24 @@ mod tests { } fn rpc(ak: &str, server: &str, body: &str, accept: &str) -> axum::http::Request { + rpc_session(ak, server, body, accept, &format!("sess-{ak}")) + } + + fn rpc_session( + ak: &str, + server: &str, + body: &str, + accept: &str, + session: &str, + ) -> axum::http::Request { axum::http::Request::builder() .method("POST") .uri(format!("/mcp/{server}")) .header("authorization", format!("Bearer {ak}")) .header("content-type", "application/json") .header("accept", accept) - .header("mcp-session-id", "sess-1") + .header("mcp-session-id", session) + .header("last-event-id", "41") .body(Body::from(body.to_owned())) .unwrap() } @@ -556,6 +727,28 @@ mod tests { assert_eq!(msg["result"]["tools"].as_array().unwrap().len(), 1); } + #[tokio::test] + async fn multi_line_data_events_are_assembled_before_filtering() { + let (base, _) = spawn_stub_full(3600, false, true).await; + let (app, _) = app_with(&format!("{base}/mcp")).await; + let list = r#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#; + let resp = app + .oneshot(rpc("k-add", "tools", list, "text/event-stream")) + .await + .unwrap(); + let body = text(resp).await; + assert!( + body.starts_with("event: message\r\nid: 7\r\ndata: "), + "{body}" + ); + let data = body.lines().find_map(|l| l.strip_prefix("data: ")).unwrap(); + let msg: Value = serde_json::from_str(data).unwrap(); + assert_eq!( + msg["result"]["tools"], + json!([{"name":"add","inputSchema":{"type":"object"}}]) + ); + } + #[tokio::test] async fn tool_calls_are_gated_and_audited() { let (app, state) = app_with(&spawn_stub().await).await; @@ -596,24 +789,28 @@ mod tests { } #[tokio::test] - async fn entitlement_unknown_server_and_batches_are_refused() { + async fn unentitled_unknown_and_malformed_requests_are_refused() { let (app, _) = app_with(&spawn_stub().await).await; let ping = r#"{"jsonrpc":"2.0","id":1,"method":"ping"}"#; + for server in ["locked", "nope"] { + let resp = app + .clone() + .oneshot(rpc("k-add", server, ping, "application/json")) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::NOT_FOUND, "{server}"); + assert!(text(resp).await.contains("unknown mcp server"), "{server}"); + } let resp = app .clone() - .oneshot(rpc("k-add", "locked", ping, "application/json")) - .await - .unwrap(); - assert_eq!(resp.status(), StatusCode::FORBIDDEN); - let resp = app - .clone() - .oneshot(rpc("k-add", "nope", ping, "application/json")) + .oneshot(rpc("k-add", "tools", "[]", "application/json")) .await .unwrap(); - assert_eq!(resp.status(), StatusCode::NOT_FOUND); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + let positional = r#"{"jsonrpc":"2.0","id":9,"method":"tools/call","params":["add",{}]}"#; let resp = app .clone() - .oneshot(rpc("k-add", "tools", "[]", "application/json")) + .oneshot(rpc("k-add", "tools", positional, "application/json")) .await .unwrap(); assert_eq!(resp.status(), StatusCode::BAD_REQUEST); @@ -625,14 +822,75 @@ mod tests { } #[tokio::test] - async fn a_tool_call_without_params_is_forwarded_not_panicked() { + async fn sessions_are_bound_to_the_key_that_opened_them() { let (app, _) = app_with(&spawn_stub().await).await; - let bare = r#"{"jsonrpc":"2.0","id":9,"method":"tools/call"}"#; + let ping = r#"{"jsonrpc":"2.0","id":1,"method":"ping"}"#; let resp = app - .oneshot(rpc("k-add", "tools", bare, "application/json")) + .clone() + .oneshot(rpc_session( + "k-all", + "tools", + ping, + "application/json", + "fresh", + )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); + let resp = app + .clone() + .oneshot(rpc_session( + "k-add", + "tools", + ping, + "application/json", + "sess-1", + )) + .await + .unwrap(); + assert_eq!( + resp.status(), + StatusCode::NOT_FOUND, + "another key's session" + ); + let resp = app + .oneshot(rpc_session( + "k-all", + "tools", + ping, + "application/json", + "sess-1", + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK, "the owner keeps using it"); + } + + #[tokio::test] + async fn listen_streams_are_capped_per_key() { + let (app, _) = app_with(&spawn_stub().await).await; + let listen = |ak: &str| { + axum::http::Request::builder() + .method("GET") + .uri("/mcp/tools") + .header("authorization", format!("Bearer {ak}")) + .header("accept", "text/event-stream") + .body(Body::empty()) + .unwrap() + }; + let first = app.clone().oneshot(listen("k-all")).await.unwrap(); + let second = app.clone().oneshot(listen("k-all")).await.unwrap(); + assert_eq!(first.status(), StatusCode::OK); + assert_eq!(second.status(), StatusCode::OK); + let third = app.clone().oneshot(listen("k-all")).await.unwrap(); + assert_eq!(third.status(), StatusCode::TOO_MANY_REQUESTS); + assert!(text(first).await.contains("notifications/message")); + let again = app.oneshot(listen("k-all")).await.unwrap(); + assert_eq!( + again.status(), + StatusCode::OK, + "a finished stream frees its slot" + ); } async fn ping_auth(app: &Router, server: &str) -> String { @@ -657,12 +915,16 @@ mod tests { } #[tokio::test] - async fn expiring_and_refused_oauth_tokens_are_fetched_anew() { + async fn short_lived_and_refused_oauth_tokens_are_fetched_anew() { let (base, st) = spawn_stub_with(1, false).await; let (app, _) = app_with(&format!("{base}/mcp")).await; assert_eq!(ping_auth(&app, "oauth").await, "Bearer tok-1"); - assert_eq!(ping_auth(&app, "oauth").await, "Bearer tok-2"); - assert_eq!(st.fetches.load(Ordering::Relaxed), 2); + assert_eq!( + ping_auth(&app, "oauth").await, + "Bearer tok-1", + "a 1 s token is cached for half its life" + ); + assert_eq!(st.fetches.load(Ordering::Relaxed), 1); let (base, st) = spawn_stub_with(3600, true).await; let (app, _) = app_with(&format!("{base}/mcp")).await; @@ -704,7 +966,7 @@ mod tests { } #[tokio::test] - async fn tool_results_are_masked_or_blocked_under_moderation() { + async fn reviewed_results_are_masked_everywhere_or_blocked_whole() { let (app, state) = reviewed_app(&spawn_stub().await).await; let echo = r#"{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"echo","arguments":{"s":"mail bob@example.com twice bob@example.com"}}}"#; let resp = app @@ -716,6 +978,11 @@ mod tests { let masked = v["result"]["content"][0]["text"].as_str().unwrap(); assert!(masked.contains("mail [MASKED] twice [MASKED]"), "{masked}"); assert_eq!(v["result"]["content"][1]["type"], "image"); + assert_eq!(v["result"]["content"][1]["data"], "AAAA"); + assert_eq!( + v["result"]["structuredContent"]["echoed"], "mail [MASKED] twice [MASKED]", + "structured output is reviewed too" + ); let resp = app .clone() @@ -728,17 +995,33 @@ mod tests { "{body}" ); - let leak = r#"{"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"echo","arguments":{"s":"my ssn is 123"}}}"#; + let read = r#"{"jsonrpc":"2.0","id":5,"method":"resources/read","params":{"uri":"file:///notes"}}"#; let resp = app .clone() - .oneshot(rpc("k-mod", "tools", leak, "application/json")) + .oneshot(rpc("k-mod", "tools", read, "application/json")) .await .unwrap(); let v: Value = serde_json::from_str(&text(resp).await).unwrap(); + assert_eq!(v["result"]["contents"][0]["text"], "contact [MASKED]"); + assert_eq!(v["result"]["contents"][0]["uri"], "file:///notes"); + + let leak = r#"{"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"echo","arguments":{"s":"my ssn is 123"}}}"#; + let resp = app + .clone() + .oneshot(rpc("k-mod", "tools", leak, "text/event-stream")) + .await + .unwrap(); + let body = text(resp).await; + assert_eq!( + body.lines().filter(|l| l.starts_with("data:")).count(), + 1, + "{body}" + ); + let v: Value = serde_json::from_str(body.trim().strip_prefix("data: ").unwrap()).unwrap(); assert_eq!(v["id"], 4); assert_eq!(v["error"]["code"], JSONRPC_RESULT_BLOCKED); assert_eq!(v["error"]["message"], "blocked by guardrail: SSN"); - assert!(v.get("result").is_none()); + assert!(!body.contains("ssn"), "{body}"); let resp = app .oneshot(rpc("k-all", "tools", leak, "application/json")) @@ -759,7 +1042,55 @@ mod tests { .filter(|e| e.surface == "mcp" && e.rule == "moderation") .map(|e| (e.action.as_str(), e.hits)) .collect(); - assert_eq!(mods, vec![("block", 1), ("mask", 2), ("mask", 2)]); + assert_eq!( + mods, + vec![("block", 1), ("mask", 1), ("mask", 4), ("mask", 4)] + ); + } + + #[tokio::test] + async fn reviewed_tenants_get_no_stream_resumption() { + let (app, _) = reviewed_app(&spawn_stub().await).await; + let ping = r#"{"jsonrpc":"2.0","id":1,"method":"ping"}"#; + let resp = app + .clone() + .oneshot(rpc("k-mod", "tools", ping, "application/json")) + .await + .unwrap(); + let v: Value = serde_json::from_str(&text(resp).await).unwrap(); + assert!(v["result"]["last_event_id"].is_null(), "{v}"); + let resp = app + .oneshot(rpc("k-all", "tools", ping, "application/json")) + .await + .unwrap(); + let v: Value = serde_json::from_str(&text(resp).await).unwrap(); + assert_eq!(v["result"]["last_event_id"], "41"); + } + + #[test] + fn unparsable_data_fails_closed_and_framing_survives() { + let sse = b"event: message\ndata: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"tools\":[\ndata: {\"name\":\"add\"},{\"name\":\"exfil\"}]}}\n\n: comment\nid: 9\ndata: not json\n\n"; + let segments = parse_segments(sse, true); + assert!( + matches!(segments[1], Segment::Message(_)), + "joined data lines parse" + ); + assert!( + segments + .iter() + .any(|s| matches!(s, Segment::Opaque(p) if p == "not json")) + ); + assert!(filter_tool_list(sse, true, &["add".to_owned()]).is_none()); + let bom = "\u{feff}{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"tools\":[{\"name\":\"add\"},{\"name\":\"x\"}]}}"; + let filtered = filter_tool_list(bom.as_bytes(), false, &["add".to_owned()]).unwrap(); + let v: Value = serde_json::from_slice(&filtered).unwrap(); + assert_eq!(v["result"]["tools"].as_array().unwrap().len(), 1); + let plain = + b"event: message\ndata: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"tools\":[]}}\n\n"; + assert_eq!( + String::from_utf8(filter_tool_list(plain, true, &[]).unwrap()).unwrap(), + "event: message\ndata: {\"id\":1,\"jsonrpc\":\"2.0\",\"result\":{\"tools\":[]}}\n\n" + ); } #[test] diff --git a/crates/views/src/mcp_auth.rs b/crates/views/src/mcp_auth.rs index fddb132f..58cb34dc 100644 --- a/crates/views/src/mcp_auth.rs +++ b/crates/views/src/mcp_auth.rs @@ -12,12 +12,16 @@ use serde_json::Value; // a token is renewed this long before its own expiry so an in-flight call never presents a stale one const EXPIRY_MARGIN: Duration = Duration::from_secs(30); const DEFAULT_EXPIRES_IN: u64 = 3_600; +// a token endpoint claiming more is capped: Instant arithmetic must not overflow +const MAX_EXPIRES_IN: u64 = 30 * 86_400; const TOKEN_TIMEOUT: Duration = Duration::from_secs(10); /// Per-server token cache; one entry per OAuth-configured server. #[derive(Debug, Default)] pub struct McpAuth { tokens: Mutex>, + /// One fetch in flight at a time, so a cold cache costs one token round trip, not one per request. + fetching: tokio::sync::Mutex<()>, } impl McpAuth { @@ -34,20 +38,27 @@ impl McpAuth { let Some(oauth) = &conf.oauth else { return Ok(None); }; - let refresh = { - let tokens = self.lock(); - match tokens.get(&conf.name) { - Some(t) if t.expires_at > Instant::now() => return Ok(Some(t.access.clone())), - Some(t) => t.refresh.clone(), - None => None, - } - }; + if let Some(access) = self.cached(&conf.name) { + return Ok(Some(access)); + } + let _one_at_a_time = self.fetching.lock().await; + if let Some(access) = self.cached(&conf.name) { + return Ok(Some(access)); + } + let refresh = self.lock().get(&conf.name).and_then(|t| t.refresh.clone()); let token = fetch(client, oauth, refresh.as_deref()).await?; let access = token.access.clone(); self.lock().insert(conf.name.clone(), token); Ok(Some(access)) } + fn cached(&self, server: &str) -> Option { + self.lock() + .get(server) + .filter(|t| t.expires_at > Instant::now()) + .map(|t| t.access.clone()) + } + /// Expire `server`'s token: the upstream refused it, so the next call fetches anew. pub fn invalidate(&self, server: &str) { if let Some(t) = self.lock().get_mut(server) { @@ -121,14 +132,21 @@ async fn fetch( let Some(Value::String(access)) = reply.get_mut("access_token").map(Value::take) else { return Err("token reply carries no access_token".to_owned()); }; - let lifetime = Duration::from_secs(reply["expires_in"].as_u64().unwrap_or(DEFAULT_EXPIRES_IN)); + let lifetime = Duration::from_secs( + reply["expires_in"] + .as_u64() + .unwrap_or(DEFAULT_EXPIRES_IN) + .min(MAX_EXPIRES_IN), + ); + // a short-lived token is still cached for half its life instead of refetched per call + let usable = lifetime - EXPIRY_MARGIN.min(lifetime / 2); let rotated = match reply.get_mut("refresh_token").map(Value::take) { Some(Value::String(t)) => Some(t), _ => refresh.map(str::to_owned), }; Ok(Token { access, - expires_at: Instant::now() + lifetime.saturating_sub(EXPIRY_MARGIN), + expires_at: Instant::now() + usable, refresh: rotated, }) } diff --git a/docs/api.md b/docs/api.md index 93feb49d..2210b410 100644 --- a/docs/api.md +++ b/docs/api.md @@ -25,10 +25,11 @@ shape, so its SDKs can dispatch on it (`code` is additive): {"type": "error", "error": {"type": "rate_limit_error", "code": "throttling_exception", "message": "..."}} ``` -A terminal upstream failure (failover exhausted) is `424` with +A terminal upstream failure (account failover and the model's +`fallback_models` chain exhausted) is `424` with `code: "model_error_exception"`, plus `original_status_code` (when the -upstream returned a status) and `resource_name` (the requested model) inside -the error object. Retry on 408/429/500/503 with backoff (honor +last upstream tried returned a status) and `resource_name` (the requested +model) inside the error object. Retry on 408/429/500/503 with backoff (honor `retry-after`); never on the rest. Mid-stream failures arrive as a terminal SSE error frame carrying the same `code` field. @@ -186,23 +187,33 @@ days. | Method | Path | Notes | |--------|------|-------| -| POST / GET / DELETE | `/mcp/{server}` | Model Context Protocol (Streamable HTTP) proxy to the configured `mcp_servers[]` entry: the JSON-RPC message goes up with `Accept`, `Mcp-Session-Id`, `MCP-Protocol-Version` and `Last-Event-ID`, the server's own bearer token is attached upstream, and `Content-Type` and `Mcp-Session-Id` come back; JSON and `text/event-stream` replies stream back as the server sends them; `timeout_seconds` bounds POST and DELETE, the GET listen stream is unbounded | - -The access key rides as usual (`Authorization: Bearer` or `x-api-key`) and -must be entitled to the server (`access_keys[].mcp_servers`); when the key has -an allowlist for that server (`access_keys[].mcp_tools`), `tools/list` results -are filtered to it and a `tools/call` outside it answers a JSON-RPC error -(`-32000`) without reaching the server. Every `tools/call` — served or denied — -is a `mcp` security event (`rule = mcp:`, `action = call:` / -`deny:`). When the key's tenant sets `security.moderate`, a served -`tools/call` result is buffered and its text content reviewed by the -configured moderator before it reaches the client: a mask rewrites the text -in place, a denial replaces the result with a JSON-RPC error (`-32001`, the -moderator's reason), and either lands as a `mcp` security event -(`rule = moderation`, `action = mask` / `block`). JSON-RPC batches are refused -(400); the key's QPS applies. A server declared with `oauth` is called with an -access token the gateway fetches from the server's token endpoint; a `401` -from the server fetches a fresh token and retries the call once. +| POST / GET / DELETE | `/mcp/{server}` | Model Context Protocol (Streamable HTTP) proxy to the configured `mcp_servers[]` entry: the JSON-RPC message goes up with `Accept`, `Mcp-Session-Id`, `MCP-Protocol-Version` and `Last-Event-ID`, the server's static bearer or an OAuth access token the gateway fetched (nothing when the server declares neither) is attached upstream, and `Content-Type` and `Mcp-Session-Id` come back; replies stream back as the server sends them unless the key's tool allowlist filters a `tools/list` or the tenant's `security.moderate` reviews a result, which buffer the reply whole (up to `max_reply_bytes`); `timeout_seconds` bounds POST and DELETE, the GET listen stream is unbounded in time and counted against `max_live_streams_per_key` | + +The access key rides as usual (`Authorization: Bearer` or `x-api-key`); a +server the key is not entitled to (`access_keys[].mcp_servers`) answers like an +unknown one (`404`), so server names cannot be probed. When the key has an +allowlist for that server (`access_keys[].mcp_tools`), `tools/list` results are +filtered to it and a `tools/call` outside it answers a JSON-RPC error (`-32000`) +without reaching the server; a `tools/call` without a string `params.name` is +a `400`. Every `tools/call` — served or denied — is a `mcp` security event +(`rule = mcp:`, `action = call:` / `deny:`), written before +the call is forwarded so a dropped connection cannot erase it. `Mcp-Session-Id` +is bound to the key that first received it; another key presenting it gets +`404`. When the key's tenant sets `security.moderate`, a served `tools/call`, +`resources/read` or `prompts/get` result is buffered (up to the server's +`max_reply_bytes`) and every prose field of it reviewed by the configured +moderator before it reaches the client: a mask rewrites the text in place, a +denial — or a reply the gateway could not parse — replaces the whole reply with +one JSON-RPC error (`-32001`, the moderator's reason), and either lands as a +`mcp` security event (`rule = moderation`, `action = mask` / `block`). The GET +listen stream is not reviewed, and for a reviewed tenant `Last-Event-ID` is +not forwarded, so a result cannot replay through it; listen streams count +against `max_live_streams_per_key`. JSON-RPC batches are refused (400); the +key's QPS and the tenant's pooled QPS apply. A server declared with `oauth` is +called with an access token the gateway fetches from the server's token +endpoint; a `401` from the server fetches a fresh token and retries the call +once. Upstream failures answer a generic `502`; the server's endpoint and the +identity provider's error text stay in the gateway log. ## Batch & files @@ -233,7 +244,9 @@ is rejected the same way. `Authorization: Bearer ` header, or — for browser clients that cannot set headers — a `gw-api-key.` entry in the `Sec-WebSocket-Protocol` list. -The session is refused at accept if the tenant is not entitled to the model. +The session is refused at accept if the tenant is not entitled to the model, +or with `429` when the key already holds `max_live_streams_per_key` realtime +sessions and MCP listen streams. A realtime model bound to an account with a real `endpoint` bridges the session to that vendor's realtime WebSocket: a transparent relay, with the gateway enforcing the same governance chain as the REST path per generation — tenant and @@ -292,9 +305,9 @@ regardless. | PUT | `/admin/config` | validate + publish a new config document to the fleet config store; every instance reloads via the change feed; `?expected_version=` publishes only while that is still the head — a moved head answers 409 (global token; needs `storage.postgres_url`) | | GET | `/admin/config/versions` | retained config versions, newest first; `?limit=` (default 20) (global token; needs `storage.postgres_url`) | | POST | `/admin/config/versions/{id}/rollback` | republish a retained document as a new head and reload (global token; needs `storage.postgres_url`) | -| GET | `/admin/keys` | list keys with computed `status` / `available`, `?offset=&limit=` paged (default 200; a tenant token sees only its own tenant's); `?ak=` exact lookup answers a 0/1-key page — a foreign key is an empty page, never a 404 oracle | +| GET | `/admin/keys` | list keys with computed `status` / `available`, `?offset=&limit=` paged (default 200, every listing caps `limit` at 10 000; a tenant token sees only its own tenant's); `?ak=` exact lookup answers a 0/1-key page — a foreign key is an empty page, never a 404 oracle | | POST | `/admin/keys` | create/replace a key: `{ak, product, tenant?, owner?, qps, daily_token_quota, tokens_per_minute?, expires_at_epoch_secs?, banned?, model_quotas?}` (`owner` binds the key to one end user — authoritative for attribution) | -| PATCH | `/admin/keys/{ak}` | update any of `qps` / `daily_token_quota` / `tokens_per_minute` / `expires_at_epoch_secs` (null clears) / `banned` / `suspended_until_epoch_secs` (null lifts an abuse suspension early) | +| PATCH | `/admin/keys/{ak}` | update any of `qps` / `daily_token_quota` / `tokens_per_minute` / `expires_at_epoch_secs` (null clears) / `banned` / `suspended_until_epoch_secs` (null lifts an abuse suspension early); a tenant token may set `banned: true` but can neither lift a ban nor touch `suspended_until_epoch_secs` (403) | | DELETE | `/admin/keys/{ak}` | revoke a key | | GET | `/admin/usage` | ledger rollup by tenant × model (requests, tokens, charged `cost_micros`, `vendor_cost_micros` for margin); `?tenant=` filter for the global token; tenant-scoped — a tenant token reads `vendor_cost_micros` as 0 | | GET | `/admin/usage/users` | per-user cost rollup (user × model) over a billing period: `?since=&until=` (unix secs), `?user=` filter, `?format=csv` export; tenant-scoped — a tenant token reads `vendor_cost_micros` as 0 (operator-only margin basis) | diff --git a/docs/architecture.md b/docs/architecture.md index b85ad691..2ae7a8b3 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -27,7 +27,7 @@ server → views → handler → {dag, engines} → {models, state} → {protoco client ──► views (auth, parse, protocol normalize) ──► handler (pre plugins: blocklist, moderation, then DLP redact) ──► dag: preprocess resolve model, quota check, cache lookup - account_select priority / PTU-first / cooldown-aware selection + account_select priority / PTU-first / cooldown-aware selection, latency-ranked tiers when enabled model_access rate limits, engine call, retry-on-5xx failover post_process usage → billing ledger, cache store ──► handler (post plugins) ──► views (JSON or SSE re-emit) @@ -43,7 +43,9 @@ The DAG executes four fixed layers; nodes within a layer run sequentially in declaration order. `account_select` and `model_access` form a retry loop: an upstream 5xx excludes the failed account and reselects once; a PTU→paygo switch is recorded as -`ptu_spillover` in the ledger. +`ptu_spillover` in the ledger. Around the DAG, the handler re-runs all four +layers for the next entry of the model's `fallback_models` chain when a run +ends in an upstream fault before any byte was sent. ## Seams (traits) diff --git a/docs/configuration.md b/docs/configuration.md index 06a0d4df..b4b5cae9 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -38,8 +38,8 @@ default (lost on restart); a SQLite path makes them durable on one node. config (versioned documents + a change feed every instance follows), the shared access-key table, the shared ledger/files/batches store, and a distributed batch queue (any instance claims and runs submitted batches). -`redis_url` shares rate/quota/TPM counters and account-health cooldowns across -instances; `shared_cache: true` additionally moves the request cache into +`redis_url` shares rate/quota/TPM counters, monthly cost counters and +account-health cooldowns across instances; `shared_cache: true` additionally moves the request cache into Redis so a hit on one instance serves the fleet (off = each instance caches in-process, a miss just recomputes). `ledger_max_rows` is not a hard cap: pruning spares rows not yet folded into the usage rollup, so the table can @@ -128,7 +128,7 @@ models: long_context: {threshold_tokens: 200000, prompt_weight: 2.0, completion_weight: 1.5} # optional tier past a prompt size batch_discount: 0.5 # optional: /v1/batches items at this fraction of the price; must be finite and in (0.0, 1.0] prompt_cache: true # anthropic-messages only: prompt-cache breakpoints - fallback_models: [gpt-4o-mini] # optional: tried in order on an upstream 5xx / connection failure / vendor 429 + fallback_models: [gpt-4o-mini] # optional: tried in order on an upstream 5xx / connection failure / vendor 429 (rejected at load when unknown, self or duplicate) variants: # optional weighted canary split, sticky per user - {model: gpt-4o, weight: 90} # self-reference keeps a share here - {model: gpt-4o-next, weight: 10} @@ -272,6 +272,7 @@ mcp_servers: # Model Context Protocol servers proxied at /mcp/ endpoint: http://tools.internal:3001/mcp # the server's Streamable HTTP endpoint api_key_env: TOOLS_TOKEN # optional bearer token for the server, from the env timeout_seconds: 60 + max_reply_bytes: 16777216 # largest reply buffered for allowlist filtering or result review - name: crm endpoint: https://mcp.crm.example/mcp oauth: # exclusive with api_key_env: the gateway fetches the bearer itself @@ -310,6 +311,7 @@ scoped to that tenant (see [API — Admin](api.md#admin-dynamic-config)). ```yaml trust_proxy_headers: false # audit source IP: false = the real TCP peer (unforgeable); +max_live_streams_per_key: 64 # concurrent realtime sessions + MCP listen streams one key may hold; 0 = unlimited # true = trust x-real-ip / rightmost x-forwarded-for hop # (only behind a proxy that sets them) ``` diff --git a/docs/deployment.md b/docs/deployment.md index 532717de..db602332 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -40,6 +40,7 @@ on jemalloc as its global allocator. | `OTEL_EXPORTER_OTLP_ENDPOINT` / `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` | either enables the per-request OTLP span export ([Observability](observability.md#traces)) | | `OTEL_SERVICE_NAME`, `OTEL_TRACES_SAMPLER`, `OTEL_TRACES_SAMPLER_ARG` | service name (default `gw`) and sampler of the exported spans (SDK defaults: parent-based, always on) | | provider key vars | named by each account's `api_key_env` | +| MCP credential vars | named by `mcp_servers[].api_key_env`, `oauth.client_secret_env` and `oauth.refresh_token_env`; read at each call or token fetch, never stored | The process drains on SIGINT/SIGTERM (graceful shutdown of in-flight requests). @@ -85,7 +86,9 @@ storage: awaited write, so an accepted request never loses its row under overload. - **Rate limits & quotas**: shared in Redis when `redis_url` is set (keys namespaced under `gw:`, windows self-expire), otherwise in-process. Without - Redis, each replica limits independently. + Redis, each replica limits independently. A configured Redis that is + unreachable **fails open**: every limit, quota and budget passes with a + warning until it returns ([Governance](governance.md#limits)). - `ledger_max_rows` is not a hard cap: pruning spares rows not yet folded into the usage rollup, so the table can briefly exceed the cap under rollup lag. diff --git a/docs/governance.md b/docs/governance.md index 8c9a475b..9412bc53 100644 --- a/docs/governance.md +++ b/docs/governance.md @@ -29,23 +29,31 @@ from `GET /v1/models`), per-model daily-token quota defaults (each key metered separately against the same value; per-key `model_quotas` override), and an optional `fallback_model` — an over-quota request degrades to it instead of failing (the response echoes the requested model name; the ledger records both -requested and served). The per-key daily cap stays the hard backstop, and +requested and served; a model's own `fallback_models` chain for upstream +failures is a separate mechanism, below). The per-key daily cap stays the hard backstop, and unconfigured (key, model) pairs never touch a counter. ## Model fallback A model may name `fallback_models`, tried in order when the request fails -upstream — a vendor 5xx, a connection failure, or a vendor `429` — after the -account-level failover within the model is exhausted, and only while no byte -has reached the client (a failure after a stream has begun stays a failure). -Gateway-side denials (quotas, rate limits, entitlement, bad requests) never -fall back. Each hop re-runs the pipeline for the next model: entitlement, -quota reservation and account selection apply to the model actually served, -a fallback the caller's tenant is not entitled to is skipped, the response -echoes the requested name, the ledger records both requested and served -(`served_model`), the decision trail carries `fallback: -> : `, -and `gateway_model_fallbacks_total{from, to}` counts the hops. A chain does -not recurse: only the requested model's list applies. +upstream — a vendor 5xx or `429`, a connection failure, or no healthy account +left for the model (`503`) — after the account-level failover within the model +is exhausted, and only while no byte has reached the client (a failure after a +stream has begun stays a failure). Gateway-side denials (quotas, rate limits, +entitlement, bad requests) and gateway-internal errors never fall back, and a +request carrying signed thinking stays pinned to its model (a signature only +replays against the model that produced it). Each hop re-runs the pipeline for +the next model: entitlement, quota reservation and account selection apply to +the model actually served, a fallback the caller's tenant is not entitled to +is skipped, the response echoes the requested name, the ledger records both +requested and served (`served_model`), the decision trail carries +`fallback: -> : `, and `gateway_model_fallbacks_total{from, to}` +counts the hops. A chain does not recurse: only the requested model's list +applies. The chain covers the REST surfaces and batch items; a realtime +session never switches models. It is distinct from a tenant's `fallback_model` +above, which is a single swap the gateway makes on its own over-quota or +moderation decision. A chain naming the model itself, an unknown model or a +duplicate entry is rejected at load. ```yaml models: @@ -108,11 +116,15 @@ delivered text. A disconnect *before* any bytes are sent bills nothing. `mcp_servers`. A key reaches only the servers named in its `mcp_servers` entitlement, and `mcp_tools` narrows a server to an allowlist: `tools/list` is filtered to it before the client sees the catalog, and a `tools/call` for any -other tool is answered with a JSON-RPC error inside the gateway. Every tool -call and every denial lands in the security-event stream (`surface = mcp`, +other tool is answered with a JSON-RPC error inside the gateway. A server the +key is not entitled to answers `404` like an unknown one. Every tool call and +every denial lands in the security-event stream (`surface = mcp`, `rule = mcp:`, `action = call:` or `deny:`, attributed to -the key's `owner`), and `gateway_mcp_requests_total{server, method, result}` -counts the traffic. Keys created through the admin API carry no MCP +the key's `owner`) before the call is forwarded, and +`gateway_mcp_requests_total{server, method, result}` counts the traffic. An +`Mcp-Session-Id` is bound to the key that first received it, so one customer +cannot resume or end another's session; the key's QPS and the tenant's pooled +QPS apply, and GET listen streams count against `max_live_streams_per_key`. Keys created through the admin API carry no MCP entitlement; the servers' own credentials stay in the gateway's environment (`mcp_servers[].api_key_env`), so an agent never holds them. @@ -124,17 +136,22 @@ before its `expires_in`, and on a `401` from the server fetches a fresh token and retries the call once. Tokens are dropped on config reload. The client secret and refresh-token seed are read from the environment at fetch time. -A tenant whose `security.moderate` is on has every served `tools/call` result -reviewed by the moderator behind `moderation:` — the same review the chat and -realtime surfaces apply to inbound text, now over the result's text content -(`result.content[].text`, joined by newlines; other content types pass -untouched). A mask rewrites the text in place (`[MASKED]`), a denial replaces -the result with a JSON-RPC error `-32001` carrying the moderator's reason, and -a moderator failure follows `moderation_fail_open`. Results are buffered for -the review, so a reviewed tenant trades streaming of tool results for the -guarantee that no unreviewed text reaches the agent; both outcomes are `mcp` -security events with `rule = moderation` and `action = mask` (hits = spans) or -`block`. Tenants without `moderate` stream results as before. +A tenant whose `security.moderate` is on has every served `tools/call`, +`resources/read` and `prompts/get` result reviewed by the moderator behind +`moderation:` — the same review the chat and realtime surfaces apply to +inbound text, here over every prose field of the result (string values under +`result`, joined by newlines; `blob`, `mimeType`, `name`, `type` and `uri` +fields are identifiers or binary and pass untouched). The reply is buffered +for the review, up to the server's `max_reply_bytes`. A mask rewrites the text +in place (`[MASKED]`); a denial, a degrade verdict, a mask that matches +nothing, or a reply the gateway cannot parse replaces the whole reply with one +JSON-RPC error `-32001` carrying the reason, so nothing unreviewed leaves; a +moderator failure follows `moderation_fail_open`. The GET listen stream carries +server notifications and is not reviewed; for a reviewed tenant +`Last-Event-ID` is not forwarded, so a result cannot replay through it. Both +outcomes are `mcp` security events with `rule = moderation` and +`action = mask` (hits = spans) or `block`. Tenants without `moderate` stream +results as before. ```yaml mcp_servers: @@ -220,13 +237,16 @@ and a response-cache hit, which bills nothing, counts nothing. The monthly counterparts — `monthly_cost_quota_micros`, `key_monthly_cost_quota_micros`, `user_monthly_cost_quota_micros` — meter the same charged cost over the UTC calendar month and can be set alongside the -daily caps. With `monthly_cost_rollover: true` a month's cap grows by whatever -the previous month left unspent, at most one further month's cap; the carry is -computed from the previous month's own cap, so it never compounds. Monthly -counters live outside the daily reset (in Redis, `gw:counter:m::…` -with a 62-day TTL) and the rollover read costs one counter read per configured -scope per request; without it the monthly check is the same single read as the -daily one. +daily caps. With `monthly_cost_rollover: true` a month's cap grows by the current +cap minus the previous month's recorded spend, at most one further cap — a +month with no recorded spend (the first month, or a flushed Redis) carries a +whole cap; the carry never compounds, and a cap changed between months applies +to the carry retroactively. Monthly counters live outside the daily reset: in +Redis under `gw:counter:m::…` with a 62-day TTL, in-process swept at +the daily reset down to the current and previous month. Rollover costs one +extra counter read per configured monthly scope at admission and one at +settlement; without it the monthly check is the same single read as the daily +one. Keys without a tenant take a declared `default` tenant's budgets. Reaching any budget raises a `budget_exhausted` alert on the webhook — subject `tenant:`, `key:` or `user:/`, detail the diff --git a/docs/index.md b/docs/index.md index 98a193ff..3a849afa 100644 --- a/docs/index.md +++ b/docs/index.md @@ -6,8 +6,8 @@ key-based auth, quotas, rate limits, failover, and a billing ledger. ``` client ──► /v1/* (OpenAI + Anthropic surfaces, streaming SSE, realtime WS) - ──► pipeline: resolve/quota/cache → account select (PTU, failover) - → rate limits → engine → usage → billing ledger + ──► pipeline: resolve/quota/cache → account select (PTU, latency-ranked tiers, failover) + → rate limits → engine (retry, then the fallback chain) → usage → billing ledger ──► providers: real endpoints over HTTP · in-process mock for the rest ``` @@ -24,6 +24,7 @@ client ──► /v1/* (OpenAI + Anthropic surfaces, streaming SSE, realtime WS) - [Architecture](architecture.md) — crate layout, pipeline, trait seams - [Performance](performance.md) — measured per-node throughput and latency, how to reproduce - [Development](development.md) — build, test, workspace map, contributing +- [Security model](security.md) — trust boundaries, what is checked and recorded, failure postures, hardening ## Repository diff --git a/docs/multi-instance.md b/docs/multi-instance.md index 0b856ee0..489639b9 100644 --- a/docs/multi-instance.md +++ b/docs/multi-instance.md @@ -14,6 +14,9 @@ shared, what stays local, and what the LB needs to do. | Billing ledger / files / batches / video jobs (`Store`) | Postgres (`storage.postgres_url`), else SQLite | ✅ with Postgres (a video poll may land on any instance; the settle claim is one atomic row update); SQLite stays per-node | | Request cache | in-process (moka), or Redis with `shared_cache: true` | ⚠️ per-instance by default; fleet-shared when `shared_cache` is set | | Thinking-signature audit | in-process only | ⚠️ per-instance; a continuation landing on another instance finds no anchor and fails open (forwarded, not rejected) | +| Monthly cost counters (`Governance`) | Redis (`storage.redis_url`) | ✅ when Redis is set (62-day keys); in-process they are swept at the daily reset | +| Per-account latency (`stability.latency_routing`) | in-process only | ⚠️ per-instance; each instance ranks on its own samples, so a cold instance re-probes accounts the rest already measured | +| MCP OAuth tokens, session binding, live-stream counts | in-process only | ⚠️ per-instance: up to N token fetches per server; a session id is bound on the instance that first saw it, so pin a key to one instance at the balancer to keep the binding fleet-wide | **A correct fleet = one Postgres (`storage.postgres_url`) + one Redis (`storage.redis_url`) shared by every instance.** Without them each instance diff --git a/docs/observability.md b/docs/observability.md index 931c4da2..069a587a 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -16,13 +16,15 @@ | `gateway_upstream_status_retries_total` | counter | `account`, `status` | | `gateway_thinking_signature_review_total` | counter | `result` (match/mismatch/miss/no_evidence) | | `gateway_thinking_signature_cache_events_total` | counter | `event` | -| `gateway_mcp_requests_total` | counter | `server`, `method`, `result` | +| `gateway_mcp_requests_total` | counter | `server` (configured name), `method` (initialize / tools/list / tools/call / resources / prompts / ping / stream / other), `result` (HTTP status, denied, upstream_error, reply_unreadable, masked, blocked) | | `gateway_model_fallbacks_total` | counter | `from`, `to` | `gateway_requests_total` is recorded by router middleware, so every response — including error statuses and the realtime WebSocket upgrade — is counted, which -makes error-rate dashboards possible. All labels are bounded (route templates, -status codes, protocol/stage names) — no per-key or per-model cardinality. +makes error-rate dashboards possible. All labels are bounded by the config, never by a +caller: route templates, status codes, protocol/stage names, configured MCP +server names and configured model names (`gateway_model_fallbacks_total`) — no +per-key or per-user cardinality. ## Traces diff --git a/docs/security.md b/docs/security.md new file mode 100644 index 00000000..64c48b6b --- /dev/null +++ b/docs/security.md @@ -0,0 +1,140 @@ +# Security model + +What the gateway trusts, what it checks on every request, what it records, +and what an operator must put around it. Every statement here names the +config or code path that enforces it; the review that produced this page and +its findings are summarized at the end. + +## Trust boundaries + +| Party | Holds | May | +|-------|-------|-----| +| API customer | an access key (`access_keys[].ak`, or one created through the admin API) | call the model, MCP and batch surfaces its key is entitled to, within its quotas | +| Tenant admin | the tenant's `admin_token_env` bearer | read and change its own tenant's keys, usage, content-safety events; nothing global | +| Operator | the global `admin.token_env` bearer, the process environment, the config source | everything: config publish and reload, every tenant, `/internal/*`, the upstream credentials | +| Upstream vendor / MCP server | its own endpoint | is trusted only to answer; its errors are mapped, its headers are filtered, its credentials never leave the process | + +The gateway never holds a customer's vendor credentials: every upstream key, +Bedrock token, MCP bearer and OAuth client secret is read from the +environment variable the config names, at call time, and appears in no +response, log line, audit row or trace. + +## Authentication + +- Access keys ride as `Authorization: Bearer ` or `x-api-key: `; a + missing or unknown key is a `401`. Keys have a lifecycle: `expires_at_epoch_secs`, + `banned`, and an automatic abuse suspension (`abuse.tiers`) each fail + authentication with a distinct `403`, on every surface (REST, realtime — where + the key is re-checked per turn — batches, MCP). Logs, traces and the ledger + carry `ak_id`, a SHA-256 fingerprint of the key, never the credential. +- Admin routes are absent (`404`) until an admin token is configured. Two + tiers apply: the global token (`admin.token_env`) manages everything; a + tenant's `admin_token_env` token manages only that tenant. A tenant token on + a global-only route (reload, config publish, `/internal/*`, the cross-tenant + `/admin/audit/ops` trail) is a `403`; a tenant token looking up a foreign key + gets an empty page or a `404`, never a confirmation the key exists. +- The web control plane keeps its own identity store and browser sessions and + reaches the gateway only through the admin API with the tokens above. + +## Authorization and tenant scoping + +- A key belongs to one tenant (the implicit `default` tenant when undeclared). + Model entitlement (`tenants[].models`) is enforced before the cache and the + engine, so a tenant cannot be served another tenant's cached answer or reach + a model it is not entitled to — including through a fallback chain, where + unentitled entries are skipped. +- Files, batches and retained content are tenant-owned: a foreign id answers + `404`, not `403`, so ids cannot be probed. +- MCP: a key reaches only the servers named in its `mcp_servers` entitlement + (any other name answers `404`), and only the tools in its `mcp_tools` + allowlist; `tools/list` is filtered to the allowlist and a `tools/call` + outside it is refused inside the gateway (JSON-RPC `-32000`). An MCP session + id is bound to the key that first received it. Keys created through the + admin API carry no MCP entitlement. Only a fixed set of MCP headers crosses + the proxy in either direction; the caller's `Authorization` never reaches + the server and the server's headers never reach the caller beyond + `Content-Type` and `Mcp-Session-Id`. The proxy's HTTP client follows no + redirects, so a server or token endpoint cannot steer a credentialed + request elsewhere. JSON-RPC batches are refused. + +## Content controls + +- Per-tenant `security` policy: blocklist (block / flag / shadow), regex + recognizers, secret detection, DLP redaction of emails and phone numbers in + both directions (a redacted stream is buffered and replayed so no unmasked + span leaves), and an external moderator (`moderation:`, AWS Bedrock + Guardrails) over inbound text and — for the MCP proxy — over tool results. + Signed thinking blocks are never rewritten; a mask that would land in one + fails the request instead. +- A tenant admin may ban its own keys but can neither lift a ban nor change an + abuse suspension: those are platform sanctions the global token owns. +- Every hit is a security event (`/admin/audit/events`) carrying the rule, + action and hit count — never the prompt text. Admin mutations land in + `/admin/audit/ops` with the source IP (the TCP peer; the rightmost + `x-forwarded-for` hop only under `trust_proxy_headers`). Retained content (`tenants[].retention`) + is sealed at rest with `GW_CONTENT_KEY`; without the key, `full` retention + degrades to redacted text. + +## Failure postures + +Stated so an operator can choose them deliberately: + +- Redis unreachable: rate limits, quotas and budgets **fail open** and a + warning is logged; account health treats every account as healthy. +- Moderator unreachable: `security.moderation_fail_open` decides — `false` + (default) denies the request or tool result, `true` admits it. +- Upstream failure: account failover within the model, then the model's + `fallback_models` chain, only while nothing has been sent to the client. + Upstream and identity-provider errors reach the customer as a generic + `502`; endpoints and error text stay in the gateway log. +- Abuse of long-lived connections: realtime sessions and MCP listen streams + are capped per key (`max_live_streams_per_key`); realtime turns and every + MCP request spend the key's QPS permits; replies the proxy must buffer are + capped per server (`max_reply_bytes`); `x-gw-user` is capped at 256 bytes, + since it keys governance counters. +- Billing store unreachable: rows queue in a bounded repair queue and apply + backpressure rather than being dropped. + +## Operating recommendations + +- Terminate TLS in front of the gateway; it listens on plain HTTP. +- Keep `/admin/*`, `/internal/*` and `/metrics` off the public load balancer + (the sample nginx config in [multi-instance](multi-instance.md) does this) + and leave `GW_ADMIN_TOKEN` unset on instances that do not need the admin API. +- Put MCP servers on a private network the gateway alone can reach; the + gateway is their only client and holds their credentials. +- Give each customer their own key with `qps`, `daily_token_quota`, + `tokens_per_minute` and a cost budget, and route alerts + (`alerts.webhook_url_env`) somewhere watched. +- Rotate upstream and MCP credentials by changing the environment and + reloading; OAuth tokens the gateway fetched are dropped on reload. + +## Review record + +A targeted adversarial review of the admin plane and the MCP proxy ran on +2026-09-09 against three lenses: authentication and tenant scoping; +injection, smuggling and protocol edge cases; resource exhaustion and +information leakage. Fixed in the same change: unbounded buffering of MCP +replies, redirect following by the MCP client (credential replay and SSRF), +upstream endpoints and identity-provider error text in customer-visible +errors, `resources/read` and `prompts/get` results escaping review, multi-line +SSE `data` and other unparsable replies passing through unreviewed, structured +tool output escaping a mask, stream resumption replaying results past the +review, unbound MCP session ids, tool calls audited only after the upstream +answered, a `tools/call` without a string tool name skipping every gate, +realtime sessions and listen streams with no per-key cap, in-process monthly +counters keyed by `x-gw-user` never evicted, a tenant token lifting bans and +suspensions, raw access keys in denial messages, admin listings without a page +ceiling, and MCP server names probe-able through 404-vs-403. + +Accepted as is, with the reason: the ledger, security-event and retained- +content tables store the access key itself (the ledger joins usage by it; +database read access sits inside the operator boundary — protect the database +like the config); the listener has no header-read or idle timeout of its own +(a proxy in front terminates slow clients, which the deployment guidance +requires anyway); a video download is held in memory for the clip's size; a +config reload can race one in-flight token fetch for at most one token +lifetime; duplicate JSON keys parse last-wins here and possibly first-wins on +a non-compliant upstream (a compliant server behaves identically); the +cross-tenant key guard reads a two-second key cache before mutating, which +would need a key to change tenant inside that window. From cba7b39711e8c392ef38477c66fb65fb8b224d3b Mon Sep 17 00:00:00 2001 From: CMGS Date: Wed, 9 Sep 2026 18:37:03 +0900 Subject: [PATCH 2/2] security: close second-pass review findings on the MCP proxy and admin plane An adversarial review round over the same surface turned up further gaps: - MCP result review no longer depends on the reply's HTTP status and now covers a JSON-RPC error's own message, so a non-2xx or error-shaped reply cannot carry unreviewed prose past a reviewed tenant. - Result review reads prose under every key except base64 binary (`blob`, `data`), so identifiers no longer shield text and an image is never rewritten. - A tenant under `security.moderate` may not open an MCP listen stream, whose server-pushed content the proxy cannot review. - A tenant admin can no longer lift a ban by re-creating the key: a platform ban or suspension on an existing key survives a tenant re-create. - The 256-byte attribution cap applies on the realtime subprotocol path and to the request-body and batch-item user fields, not only the header. - The model fallback chain is disabled only for a request replaying reasoning output, not for any request that merely engages reasoning. - An upstream connection failure's cause is logged again while the client still gets a generic 502. - max_reply_bytes must be greater than 0; a token endpoint reporting expires_in 0 no longer defeats the token cache; one server's slow token endpoint no longer stalls another server's first token fetch. --- crates/config/src/lib.rs | 6 ++ crates/engines/src/http_transport.rs | 1 + crates/handler/src/lib.rs | 7 +- crates/models/src/request.rs | 17 ++++- crates/views/src/lib.rs | 42 ++++++---- crates/views/src/mcp.rs | 110 +++++++++++++++++++++++++-- crates/views/src/mcp_auth.rs | 20 ++++- docs/api.md | 16 ++-- docs/configuration.md | 2 +- docs/governance.md | 2 +- docs/security.md | 28 +++++-- 11 files changed, 208 insertions(+), 43 deletions(-) diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs index fcccd605..1177a714 100644 --- a/crates/config/src/lib.rs +++ b/crates/config/src/lib.rs @@ -1106,6 +1106,12 @@ impl GatewayConfig { self.mcp_servers.iter().map(|m| m.name.as_str()), )?; for m in &self.mcp_servers { + if m.max_reply_bytes == 0 { + return Err(ConfigError::BadMcpServer { + server: m.name.clone(), + reason: "max_reply_bytes must be greater than 0", + }); + } let Some(o) = &m.oauth else { continue; }; diff --git a/crates/engines/src/http_transport.rs b/crates/engines/src/http_transport.rs index e22adde6..9653c5d4 100644 --- a/crates/engines/src/http_transport.rs +++ b/crates/engines/src/http_transport.rs @@ -171,6 +171,7 @@ impl Transport for HttpTransport { } else { "upstream request failed" }; + tracing::warn!(account = %req.account, error = %e, what); return Err( GatewayError::new(upstream_fault_code(e.is_timeout()), 502, what) .with_source(e), diff --git a/crates/handler/src/lib.rs b/crates/handler/src/lib.rs index d4d47775..615126f3 100644 --- a/crates/handler/src/lib.rs +++ b/crates/handler/src/lib.rs @@ -277,9 +277,10 @@ impl OnlineHandler { ) .await; // signed thinking replays only against the model that produced it - if let Some((i, next)) = (is_upstream_fault(&e) && !ctx.request.pins_reasoning_route()) - .then(|| next_fallback(&snap.cfg, &ctx, tried)) - .flatten() + if let Some((i, next)) = (is_upstream_fault(&e) + && !ctx.request.replays_reasoning_output()) + .then(|| next_fallback(&snap.cfg, &ctx, tried)) + .flatten() { tried = i + 1; switch_model(&mut ctx, next, &e.message); diff --git a/crates/models/src/request.rs b/crates/models/src/request.rs index adf3a40e..964f3c08 100644 --- a/crates/models/src/request.rs +++ b/crates/models/src/request.rs @@ -105,7 +105,22 @@ impl GatewayRequest { /// replays only against the model that produced it, so the request that /// engages reasoning and the continuation carrying its output both pin. pub fn pins_reasoning_route(&self) -> bool { - self.reasoning_engaged() + self.reasoning_engaged() || self.replays_reasoning_output() + } + + /// Whether this request carries reasoning output that must replay against + /// the model that produced it: prior signed thinking, an Anthropic + /// protected block, or a Responses `input` reasoning item. A bare reasoning + /// request that has produced nothing yet does not, so it may still fall back. + pub fn replays_reasoning_output(&self) -> bool { + let responses_input_replay = self + .model_param_v2 + .as_ref() + .filter(|p| p.protocol == gw_consts::Protocol::Responses) + .and_then(|p| p.raw.get("input")) + .and_then(serde_json::Value::as_array) + .is_some_and(|items| items.iter().any(|item| item["type"] == "reasoning")); + responses_input_replay || self.message.iter().any(|m| { m.reasoning_details.is_some() || m.parts diff --git a/crates/views/src/lib.rs b/crates/views/src/lib.rs index d7c78856..e0477849 100644 --- a/crates/views/src/lib.rs +++ b/crates/views/src/lib.rs @@ -1327,14 +1327,24 @@ fn user_header(headers: &HeaderMap) -> Option { headers .get("x-gw-user") .and_then(|v| v.to_str().ok()) - .map(str::to_owned) + .map(cap_user_hint) .filter(|s| !s.is_empty()) } /// The REST attribution precedence: `x-gw-user` header, else the dialect's own /// user field (batch items invert it — per-item `user` first). fn user_hint(headers: &HeaderMap, field: &Value) -> Option { - user_header(headers).or_else(|| field.as_str().map(str::to_owned)) + user_header(headers).or_else(|| field.as_str().map(cap_user_hint)) +} + +/// Bound an attribution hint at `USER_HINT_MAX_BYTES`: it keys governance +/// counters, so an unbounded value from any surface would grow the keyspace. +fn cap_user_hint(s: &str) -> String { + let mut end = s.len().min(USER_HINT_MAX_BYTES); + while !s.is_char_boundary(end) { + end -= 1; + } + s[..end].to_owned() } /// AK auth: `Authorization: Bearer ` or `x-api-key: `. The error is @@ -1906,9 +1916,12 @@ async fn admin_key_create( if !s.handler.cfg().is_known_tenant(tenant) { return error_response(400, format!("unknown tenant `{tenant}`")); } - if let Err(r) = scoped_key(&s, &scope, ak).await { - return r; - } + let existing = match scoped_key(&s, &scope, ak).await { + Ok(found) => found, + Err(r) => return r, + }; + // a platform sanction on an existing key survives a tenant re-create + let tenant_scoped = matches!(scope, AdminScope::Tenant(_)); let info = AkInfo { ak_id: gw_state::access_key_fingerprint(ak).into(), ak: ak.to_owned(), @@ -1919,8 +1932,12 @@ async fn admin_key_create( daily_token_quota: body["daily_token_quota"].as_i64().unwrap_or(0), tokens_per_minute: body["tokens_per_minute"].as_i64(), expires_at_epoch_secs: body["expires_at_epoch_secs"].as_i64(), - banned: body["banned"].as_bool().unwrap_or(false), - suspended_until_epoch_secs: None, + banned: body["banned"].as_bool().unwrap_or(false) + || (tenant_scoped && existing.as_ref().is_some_and(|e| e.banned)), + suspended_until_epoch_secs: existing + .as_ref() + .filter(|_| tenant_scoped) + .and_then(|e| e.suspended_until_epoch_secs), model_quotas: Arc::new( body["model_quotas"] .as_object() @@ -1983,7 +2000,7 @@ async fn admin_key_patch( banned: body["banned"].as_bool(), suspended_until_epoch_secs: tri("suspended_until_epoch_secs"), }; - // a ban or an abuse suspension is a platform sanction; a tenant may add one, never lift one + // platform sanctions: a tenant may add a ban but not lift one, and may not set or clear a suspension if matches!(scope, AdminScope::Tenant(_)) && (patch.banned == Some(false) || patch.suspended_until_epoch_secs.is_some()) { @@ -4620,13 +4637,8 @@ async fn batches_submit( let mut batch_items = Vec::new(); // batch-level attribution hint; a per-item body `user` overrides it let hint = user_header(&headers); - let item_user = |v: &Value| { - v["user"] - .as_str() - .or(hint.as_deref()) - .unwrap_or_default() - .to_owned() - }; + let item_user = + |v: &Value| cap_user_hint(v["user"].as_str().or(hint.as_deref()).unwrap_or_default()); if let Some(file_id) = body["input_file_id"].as_str() { let found = s.handler.state().store.file_get(file_id).await; diff --git a/crates/views/src/mcp.rs b/crates/views/src/mcp.rs index b2cbcfdc..2a123e20 100644 --- a/crates/views/src/mcp.rs +++ b/crates/views/src/mcp.rs @@ -29,8 +29,9 @@ const FORWARDED_HEADERS: [&str; 5] = [ const RETURNED_HEADERS: [&str; 2] = ["content-type", "mcp-session-id"]; /// Methods whose results carry prose an agent reads; reviewed under `security.moderate`. const REVIEWED_METHODS: [&str; 3] = ["tools/call", "resources/read", "prompts/get"]; -/// Result fields that carry identifiers or binary, never prose. -const OPAQUE_KEYS: [&str; 5] = ["blob", "mimeType", "name", "type", "uri"]; +/// Result fields that carry base64 binary, never prose: skipped so the review +/// neither reads nor rewrites an image, audio clip or blob resource. +const OPAQUE_KEYS: [&str; 2] = ["blob", "data"]; const JSONRPC_TOOL_DENIED: i64 = -32000; const JSONRPC_RESULT_BLOCKED: i64 = -32001; const UNREVIEWABLE: &str = "the result could not be reviewed"; @@ -86,6 +87,14 @@ pub(crate) async fn proxy( { return error_response(404, "unknown mcp session"); } + let sec = snap.cfg.security_for(&ak.tenant); + // a listen stream carries server-pushed content the proxy cannot review + if method == Method::GET && sec.moderate { + return error_response( + 403, + "mcp listen streams are unavailable under content review", + ); + } let call = if method == Method::POST { match parse_call(&body) { Ok(call) => call, @@ -136,7 +145,6 @@ pub(crate) async fn proxy( ) .await; } - let sec = snap.cfg.security_for(&ak.tenant); // a reviewed tenant gets no stream resumption: a replayed result would skip the review let resumable = !sec.moderate; let mut sent = send(&s, conf, &method, &headers, &body, resumable).await; @@ -170,9 +178,8 @@ pub(crate) async fn proxy( .get("content-type") .and_then(|v| v.to_str().ok()) .is_some_and(|ct| ct.starts_with("text/event-stream")); - let filtered = status.is_success() && call.method == "tools/list" && allowed.is_some(); - let reviewed = - status.is_success() && sec.moderate && REVIEWED_METHODS.contains(&call.method.as_str()); + let filtered = call.method == "tools/list" && allowed.is_some(); + let reviewed = sec.moderate && REVIEWED_METHODS.contains(&call.method.as_str()); if !filtered && !reviewed { let stream = reply.bytes_stream().map(move |chunk| { let _held = &stream_guard; @@ -366,6 +373,8 @@ fn review_slots(seg: &mut Segment) -> Vec<&mut String> { if let Segment::Message(msg) = seg { let root = if msg.get("result").is_some() { msg.get_mut("result") + } else if msg.get("error").is_some() { + msg.get_mut("error") } else { msg.get_mut("params") }; @@ -1048,6 +1057,95 @@ mod tests { ); } + async fn tricky(axum::Json(req): axum::Json) -> Response { + let id = req["id"].clone(); + let reply = match req["params"]["name"].as_str() { + Some("err_status") => { + let body = json!({"jsonrpc":"2.0","id":id,"result":{"content":[{"type":"text","text":"leak bob@example.com"}]}}); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + mcp_headers(), + axum::Json(body), + ) + .into_response(); + } + Some("err_body") => { + json!({"jsonrpc":"2.0","id":id,"error":{"code":-1,"message":"see bob@example.com"}}) + } + _ => { + json!({"jsonrpc":"2.0","id":id,"result":{"structuredContent":{"name":"contact bob@example.com"}}}) + } + }; + (StatusCode::OK, mcp_headers(), axum::Json(reply)).into_response() + } + + fn mcp_headers() -> HeaderMap { + let mut out = HeaderMap::new(); + out.insert("mcp-session-id", "sess-1".parse().unwrap()); + out + } + + async fn spawn_tricky() -> String { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let app = Router::new().route("/mcp", post(tricky)); + tokio::spawn(axum::serve(listener, app).into_future()); + format!("http://{addr}") + } + + async fn call_tool(app: &Router, name: &str) -> Value { + let body = format!( + r#"{{"jsonrpc":"2.0","id":9,"method":"tools/call","params":{{"name":"{name}","arguments":{{}}}}}}"# + ); + let resp = app + .clone() + .oneshot(rpc("k-mod", "tools", &body, "application/json")) + .await + .unwrap(); + serde_json::from_str(&text(resp).await).unwrap() + } + + #[tokio::test] + async fn review_covers_error_status_error_body_and_structured_prose() { + let (app, _) = reviewed_app(&format!("{}/mcp", spawn_tricky().await)).await; + + let v = call_tool(&app, "err_status").await; + assert_eq!( + v["result"]["content"][0]["text"], "leak [MASKED]", + "a non-2xx reply is still reviewed: {v}" + ); + + let v = call_tool(&app, "err_body").await; + assert_eq!( + v["error"]["message"], "see [MASKED]", + "a JSON-RPC error's prose is reviewed: {v}" + ); + + let v = call_tool(&app, "structured").await; + assert_eq!( + v["result"]["structuredContent"]["name"], "contact [MASKED]", + "structured prose under any key is reviewed: {v}" + ); + } + + #[tokio::test] + async fn a_reviewed_tenant_cannot_open_a_listen_stream() { + let (app, _) = reviewed_app(&spawn_stub().await).await; + let resp = app + .oneshot( + axum::http::Request::builder() + .method("GET") + .uri("/mcp/tools") + .header("authorization", "Bearer k-mod") + .header("accept", "text/event-stream") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::FORBIDDEN); + } + #[tokio::test] async fn reviewed_tenants_get_no_stream_resumption() { let (app, _) = reviewed_app(&spawn_stub().await).await; diff --git a/crates/views/src/mcp_auth.rs b/crates/views/src/mcp_auth.rs index 58cb34dc..f6fc16f1 100644 --- a/crates/views/src/mcp_auth.rs +++ b/crates/views/src/mcp_auth.rs @@ -3,7 +3,7 @@ //! or refresh-token grant and caches until it nears expiry. use std::collections::HashMap; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use gw_config::{McpGrant, McpOAuthConf, McpServerConf}; @@ -20,8 +20,9 @@ const TOKEN_TIMEOUT: Duration = Duration::from_secs(10); #[derive(Debug, Default)] pub struct McpAuth { tokens: Mutex>, - /// One fetch in flight at a time, so a cold cache costs one token round trip, not one per request. - fetching: tokio::sync::Mutex<()>, + /// One fetch in flight per server, so a cold cache costs one token round + /// trip and one server's slow token endpoint never stalls another's. + fetching: Mutex>>>, } impl McpAuth { @@ -41,7 +42,8 @@ impl McpAuth { if let Some(access) = self.cached(&conf.name) { return Ok(Some(access)); } - let _one_at_a_time = self.fetching.lock().await; + let gate = self.server_gate(&conf.name); + let _one_at_a_time = gate.lock().await; if let Some(access) = self.cached(&conf.name) { return Ok(Some(access)); } @@ -74,6 +76,15 @@ impl McpAuth { fn lock(&self) -> std::sync::MutexGuard<'_, HashMap> { self.tokens.lock().unwrap_or_else(|e| e.into_inner()) } + + fn server_gate(&self, server: &str) -> Arc> { + self.fetching + .lock() + .unwrap_or_else(|e| e.into_inner()) + .entry(server.to_owned()) + .or_default() + .clone() + } } #[derive(Debug)] @@ -135,6 +146,7 @@ async fn fetch( let lifetime = Duration::from_secs( reply["expires_in"] .as_u64() + .filter(|&n| n > 0) .unwrap_or(DEFAULT_EXPIRES_IN) .min(MAX_EXPIRES_IN), ); diff --git a/docs/api.md b/docs/api.md index 2210b410..0b99d0a2 100644 --- a/docs/api.md +++ b/docs/api.md @@ -187,7 +187,7 @@ days. | Method | Path | Notes | |--------|------|-------| -| POST / GET / DELETE | `/mcp/{server}` | Model Context Protocol (Streamable HTTP) proxy to the configured `mcp_servers[]` entry: the JSON-RPC message goes up with `Accept`, `Mcp-Session-Id`, `MCP-Protocol-Version` and `Last-Event-ID`, the server's static bearer or an OAuth access token the gateway fetched (nothing when the server declares neither) is attached upstream, and `Content-Type` and `Mcp-Session-Id` come back; replies stream back as the server sends them unless the key's tool allowlist filters a `tools/list` or the tenant's `security.moderate` reviews a result, which buffer the reply whole (up to `max_reply_bytes`); `timeout_seconds` bounds POST and DELETE, the GET listen stream is unbounded in time and counted against `max_live_streams_per_key` | +| POST / GET / DELETE | `/mcp/{server}` | Model Context Protocol (Streamable HTTP) proxy to the configured `mcp_servers[]` entry: the JSON-RPC message goes up with `Accept`, `Mcp-Session-Id`, `MCP-Protocol-Version` and `Last-Event-ID`, the server's static bearer or an OAuth access token the gateway fetched (nothing when the server declares neither) is attached upstream, and `Content-Type` and `Mcp-Session-Id` come back; replies stream back as the server sends them unless the key's tool allowlist filters a `tools/list` or the tenant's `security.moderate` reviews a result, which buffer the reply whole (up to `max_reply_bytes`); `timeout_seconds` bounds POST and DELETE, the GET listen stream is unbounded in time and counted against `max_live_streams_per_key` (and refused for a tenant under `security.moderate`) | The access key rides as usual (`Authorization: Bearer` or `x-api-key`); a server the key is not entitled to (`access_keys[].mcp_servers`) answers like an @@ -202,12 +202,14 @@ is bound to the key that first received it; another key presenting it gets `404`. When the key's tenant sets `security.moderate`, a served `tools/call`, `resources/read` or `prompts/get` result is buffered (up to the server's `max_reply_bytes`) and every prose field of it reviewed by the configured -moderator before it reaches the client: a mask rewrites the text in place, a -denial — or a reply the gateway could not parse — replaces the whole reply with -one JSON-RPC error (`-32001`, the moderator's reason), and either lands as a -`mcp` security event (`rule = moderation`, `action = mask` / `block`). The GET -listen stream is not reviewed, and for a reviewed tenant `Last-Event-ID` is -not forwarded, so a result cannot replay through it; listen streams count +moderator before it reaches the client — regardless of the reply's HTTP status, +and including a JSON-RPC error's own message. A mask rewrites the text in +place, a denial — or a reply the gateway could not parse — replaces the whole +reply with one JSON-RPC error (`-32001`, the moderator's reason), and either +lands as a `mcp` security event (`rule = moderation`, `action = mask` / +`block`). Because the GET listen stream carries server-pushed content the proxy +cannot review, a reviewed tenant may not open one (`403`); other tenants may, +and for them `Last-Event-ID` is still not forwarded. Listen streams count against `max_live_streams_per_key`. JSON-RPC batches are refused (400); the key's QPS and the tenant's pooled QPS apply. A server declared with `oauth` is called with an access token the gateway fetches from the server's token diff --git a/docs/configuration.md b/docs/configuration.md index b4b5cae9..a343c8a7 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -272,7 +272,7 @@ mcp_servers: # Model Context Protocol servers proxied at /mcp/ endpoint: http://tools.internal:3001/mcp # the server's Streamable HTTP endpoint api_key_env: TOOLS_TOKEN # optional bearer token for the server, from the env timeout_seconds: 60 - max_reply_bytes: 16777216 # largest reply buffered for allowlist filtering or result review + max_reply_bytes: 16777216 # largest reply buffered for allowlist filtering or result review; must be > 0 - name: crm endpoint: https://mcp.crm.example/mcp oauth: # exclusive with api_key_env: the gateway fetches the bearer itself diff --git a/docs/governance.md b/docs/governance.md index 9412bc53..63a88005 100644 --- a/docs/governance.md +++ b/docs/governance.md @@ -124,7 +124,7 @@ the key's `owner`) before the call is forwarded, and `gateway_mcp_requests_total{server, method, result}` counts the traffic. An `Mcp-Session-Id` is bound to the key that first received it, so one customer cannot resume or end another's session; the key's QPS and the tenant's pooled -QPS apply, and GET listen streams count against `max_live_streams_per_key`. Keys created through the admin API carry no MCP +QPS apply, and GET listen streams count against `max_live_streams_per_key` (a tenant under `security.moderate` may not open one). Keys created through the admin API carry no MCP entitlement; the servers' own credentials stay in the gateway's environment (`mcp_servers[].api_key_env`), so an agent never holds them. diff --git a/docs/security.md b/docs/security.md index 64c48b6b..b9a3557c 100644 --- a/docs/security.md +++ b/docs/security.md @@ -63,8 +63,11 @@ response, log line, audit row or trace. recognizers, secret detection, DLP redaction of emails and phone numbers in both directions (a redacted stream is buffered and replayed so no unmasked span leaves), and an external moderator (`moderation:`, AWS Bedrock - Guardrails) over inbound text and — for the MCP proxy — over tool results. - Signed thinking blocks are never rewritten; a mask that would land in one + Guardrails) over inbound text and — for the MCP proxy — over tool results + (`tools/call`, `resources/read`, `prompts/get`), reviewed whatever the + reply's HTTP status and including a JSON-RPC error's message; a tenant under + review cannot open an MCP listen stream, whose server-pushed content the + proxy cannot review. Signed thinking blocks are never rewritten; a mask that would land in one fails the request instead. - A tenant admin may ban its own keys but can neither lift a ban nor change an abuse suspension: those are platform sanctions the global token owns. @@ -124,8 +127,18 @@ review, unbound MCP session ids, tool calls audited only after the upstream answered, a `tools/call` without a string tool name skipping every gate, realtime sessions and listen streams with no per-key cap, in-process monthly counters keyed by `x-gw-user` never evicted, a tenant token lifting bans and -suspensions, raw access keys in denial messages, admin listings without a page -ceiling, and MCP server names probe-able through 404-vs-403. +suspensions — through the key-create route as well as PATCH — raw access keys +in denial messages, admin listings without a page ceiling, and MCP server names +probe-able through 404-vs-403. A second pass over the same surface fixed: +results escaping review on a non-2xx status or when carried as a JSON-RPC +error; prose in structured tool output under identifier-named keys, while +base64 image and blob payloads are left untouched; the attribution hint +uncapped on the realtime subprotocol path and in request and batch bodies; the +whole fallback chain disabled for any reasoning request rather than only one +replaying reasoning output; a listen stream handing unreviewed server-pushed +content to a tenant under review; a zero `max_reply_bytes` refusing every +reply and a zero token lifetime defeating the token cache; and one server's +slow token endpoint stalling another server's first token fetch. Accepted as is, with the reason: the ledger, security-event and retained- content tables store the access key itself (the ledger joins usage by it; @@ -137,4 +150,9 @@ config reload can race one in-flight token fetch for at most one token lifetime; duplicate JSON keys parse last-wins here and possibly first-wins on a non-compliant upstream (a compliant server behaves identically); the cross-tenant key guard reads a two-second key cache before mutating, which -would need a key to change tenant inside that window. +would need a key to change tenant inside that window; MCP session binding is +enforced per instance, so a multi-instance deployment must route a session's +requests to the instance that opened it (sticky by `Mcp-Session-Id`), which +Streamable HTTP already assumes; a server that returns one constant session id +locks its session to the first key to use it (a conformant server mints a +unique id per session).