diff --git a/crates/aisix-core/src/lib.rs b/crates/aisix-core/src/lib.rs index 1d34bcc3..5ca08d34 100644 --- a/crates/aisix-core/src/lib.rs +++ b/crates/aisix-core/src/lib.rs @@ -49,11 +49,10 @@ pub use models::{ GuardrailExecution, GuardrailHookPoint, GuardrailKind, GuardrailMetricsSink, GuardrailMonitorHit, KeywordConfig, KeywordPattern, McpAuthType, McpProtocolVersion, McpRateLimit, McpServer, McpServerType, McpTransport, Model, ObservabilityExporter, - ParamConstraints, PassthroughAuthMode, PassthroughCredentialMode, PassthroughProtocol, - PassthroughRoute, PolicyScope, PolicyWindow, ProviderKey, RateLimit, RateLimitPolicy, - RequestOverrides, ResponseOverrides, Routing, RoutingStrategy, RoutingTarget, SchemaError, - StreamDoneMarker, TelemetryKind, TelemetryTags, WhenAllUnavailablePolicy, - DEFAULT_COOLDOWN_TRIGGER_STATUSES, + ParamConstraints, PassthroughAuthMode, PassthroughCredentialMode, PassthroughRoute, + PolicyScope, PolicyWindow, ProviderKey, RateLimit, RateLimitPolicy, RequestOverrides, + ResponseOverrides, Routing, RoutingStrategy, RoutingTarget, SchemaError, StreamDoneMarker, + TelemetryKind, TelemetryTags, WhenAllUnavailablePolicy, DEFAULT_COOLDOWN_TRIGGER_STATUSES, }; pub use resource::{Resource, ResourceEntry}; pub use snapshot::{ResourceTable, SnapshotHandle}; diff --git a/crates/aisix-core/src/models/mod.rs b/crates/aisix-core/src/models/mod.rs index 84469328..f723fa55 100644 --- a/crates/aisix-core/src/models/mod.rs +++ b/crates/aisix-core/src/models/mod.rs @@ -62,9 +62,7 @@ pub use observability_exporter::{ ObjectStoreProvider, ObservabilityExporter, OtlpHttpConfig, SlsContentMode, }; pub use oidc_provider::{BoundClaimExpect, OidcProvider}; -pub use passthrough_route::{ - PassthroughAuthMode, PassthroughCredentialMode, PassthroughProtocol, PassthroughRoute, -}; +pub use passthrough_route::{PassthroughAuthMode, PassthroughCredentialMode, PassthroughRoute}; pub use policy_conditions::{ eval_condition_nodes, validate_condition_nodes, ConditionGroup, ConditionInput, ConditionLogic, ConditionNode, ConditionOperator, ConditionValue, GroupByDimension, PolicyAction, diff --git a/crates/aisix-core/src/models/passthrough_route.rs b/crates/aisix-core/src/models/passthrough_route.rs index 9064232c..5456547f 100644 --- a/crates/aisix-core/src/models/passthrough_route.rs +++ b/crates/aisix-core/src/models/passthrough_route.rs @@ -113,17 +113,6 @@ pub struct PassthroughRoute { #[schemars(length(min = 1))] pub provider_key_id: Option, - /// Body-shape hint for auditing, guardrails and usage extraction. - /// Parsing is best-effort: a body that does not match the declared - /// shape degrades to `raw` handling, it is never rejected for shape. - #[serde(default)] - pub protocol: PassthroughProtocol, - - /// Relay `text/event-stream` upstream responses incrementally. When - /// `false` streaming responses are fully buffered like any other body. - #[serde(default = "default_true")] - pub streaming: bool, - /// Optional header carrying the end-user identity injected by the /// upstream network device (e.g. `x-aisix-user`). Its value is recorded /// on the usage event for per-employee audit attribution and stripped @@ -135,12 +124,11 @@ pub struct PassthroughRoute { #[schemars(regex(pattern = "^[!#$%&'*+.^_`|~0-9a-z-]+$"), length(min = 1))] pub identity_header: Option, - /// Maximum time, in milliseconds, for a non-streaming upstream - /// exchange. On a streaming route it bounds the response-header phase - /// and any non-SSE body read, but never a healthy SSE relay (which - /// ends with the upstream stream or the client hanging up). When - /// omitted, the gateway default request timeout applies to - /// non-streaming exchanges only. + /// Maximum time, in milliseconds, for the upstream exchange. Bounds + /// the response-header phase and any non-SSE body read, but never a + /// healthy SSE relay (which ends with the upstream stream or the + /// client hanging up). When omitted, the gateway default request + /// timeout applies the same way. #[serde(default, skip_serializing_if = "Option::is_none")] #[schemars(range(min = 1))] pub timeout_ms: Option, @@ -193,30 +181,6 @@ pub enum PassthroughCredentialMode { ForwardClient, } -/// Body-shape hint for a passthrough route. -#[derive( - Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema, -)] -#[serde(rename_all = "snake_case")] -pub enum PassthroughProtocol { - /// No parsing: bodies are opaque byte blobs (guardrails scan them as - /// one lossy-UTF-8 text). - #[default] - Raw, - /// OpenAI-compatible chat envelope (`messages`, streamed - /// `choices[].delta.content`, final-chunk / response `usage`). - OpenaiChat, - /// OpenAI-compatible legacy completions / FIM envelope (`prompt` [+ - /// `suffix`], streamed `choices[].text`, `usage`). - OpenaiCompletions, - /// OpenAI Responses API envelope: `input` items on the request, - /// `output` items on the response, `response.output_text.delta` - /// events while streaming, and `usage` in the - /// `input_tokens`/`output_tokens` spelling — carried on the terminal - /// `response.completed` event when the response streams. - OpenaiResponses, -} - impl Resource for PassthroughRoute { fn id(&self) -> &str { &self.runtime_id @@ -476,8 +440,6 @@ mod tests { let r = minimal(); assert_eq!(r.auth_mode, PassthroughAuthMode::GatewayKey); assert_eq!(r.credential_mode, PassthroughCredentialMode::Inject); - assert_eq!(r.protocol, PassthroughProtocol::Raw); - assert!(r.streaming); assert!(r.enabled); assert!(!r.preserve_host); } @@ -608,6 +570,30 @@ mod coupling_tests { assert!(validate_passthrough_route_lenient(&doc).is_ok()); } + #[test] + fn removed_protocol_and_streaming_fields_are_unknown() { + // The pre-0.10.0 dev cycle carried `protocol` / `streaming` route + // fields; both were removed before the kind ever shipped (the + // envelope is now detected per request, SSE always relays + // incrementally). The strict write path rejects them like any + // unknown field; the lenient etcd path tolerates-and-strips. + for (field, value) in [ + ("protocol", json!("openai_responses")), + ("streaming", json!(false)), + ] { + let mut doc = base(); + doc[field] = value; + assert!( + validate_passthrough_route(&doc).is_err(), + "strict must reject unknown field {field}" + ); + assert!( + validate_passthrough_route_lenient(&doc).is_ok(), + "lenient must tolerate unknown field {field}" + ); + } + } + #[test] fn cross_mode_leftover_companions_are_rejected() { // A companion outside its mode is never consulted at runtime, so diff --git a/crates/aisix-core/src/models/schema.rs b/crates/aisix-core/src/models/schema.rs index 42c3b25f..088961c6 100644 --- a/crates/aisix-core/src/models/schema.rs +++ b/crates/aisix-core/src/models/schema.rs @@ -715,7 +715,7 @@ pub fn claim_mapping_root_schema() -> Value { /// the [`PassthroughRoute`](crate::models::PassthroughRoute) struct. Uses the /// nullable `Option` representation (`true`) so unset optional fields accept /// an explicit `null` as well as being absent. The `auth_mode` / -/// `credential_mode` / `protocol` closed sets come from their enums; every +/// `credential_mode` closed sets come from their enums; every /// cross-field invariant (match dimensions, target shape, per-mode required /// companions) is injected as an `allOf` (see /// [`super::passthrough_route::passthrough_route_coupling`]) so the strict @@ -755,16 +755,6 @@ pub fn passthrough_route_root_schema() -> Value { ("forward_client", "Forward the caller's own credential"), ], ); - title_single_value_enum_variants( - defs, - "PassthroughProtocol", - &[ - ("raw", "Opaque body"), - ("openai_chat", "OpenAI-compatible chat"), - ("openai_completions", "OpenAI-compatible completions / FIM"), - ("openai_responses", "OpenAI Responses API"), - ], - ); } schema } diff --git a/crates/aisix-proxy/src/passthrough_route.rs b/crates/aisix-proxy/src/passthrough_route.rs index d9c13fbb..b1690612 100644 --- a/crates/aisix-proxy/src/passthrough_route.rs +++ b/crates/aisix-proxy/src/passthrough_route.rs @@ -2,10 +2,22 @@ //! //! Replaces the removed implicit `/passthrough/:provider/*rest` tunnel: a //! route binds a gateway entry (path prefix and/or inbound `Host`) to ONE -//! upstream target with its own gateway-auth mode, credential handling, -//! protocol hint, and streaming behavior. There is no implicit -//! provider→Model credential borrowing (AISIX-Cloud#1127) and no forced -//! `Authorization` replacement (AISIX-Cloud#1312). +//! upstream target with its own gateway-auth mode and credential handling. +//! There is no implicit provider→Model credential borrowing +//! (AISIX-Cloud#1127) and no forced `Authorization` replacement +//! (AISIX-Cloud#1312). +//! +//! ## Envelope detection +//! +//! The request body's envelope is detected once per exchange from its +//! top-level keys ([`detect_protocol`]) and drives guardrail text +//! extraction, content capture and usage extraction for both the request +//! and the response (buffered or streamed). Detection never affects the +//! relay itself — bodies are forwarded verbatim regardless — and every +//! extraction degrades to the whole lossy-UTF-8 body when the detected +//! shape yields no text, so a mis-detected envelope loses no audit +//! coverage. SSE upstream responses are always relayed incrementally; +//! anything else is buffered (guardrails and usage need the whole body). //! //! ## Entry points //! @@ -51,9 +63,7 @@ use axum::middleware::Next; use axum::response::{IntoResponse, Response}; use aisix_core::resource::ResourceEntry; -use aisix_core::{ - PassthroughAuthMode, PassthroughCredentialMode, PassthroughProtocol, PassthroughRoute, -}; +use aisix_core::{PassthroughAuthMode, PassthroughCredentialMode, PassthroughRoute}; use crate::auth::AuthenticatedKey; use crate::error::ProxyError; @@ -577,9 +587,13 @@ async fn dispatch( let resolved_chain = state.guardrail_index.resolve(&guardrail_ctx); let mut monitor_hits: Vec = Vec::new(); - // INPUT guardrails on the (protocol-extracted) request text. + // Envelope detection: once per exchange, from the request body's + // top-level keys; the response and stream frames reuse it. + let protocol = detect_protocol(&body_bytes); + + // INPUT guardrails on the (envelope-extracted) request text. if !resolved_chain.is_empty() { - let text = request_guardrail_text(route.protocol, &body_bytes); + let text = request_guardrail_text(protocol, &body_bytes); let chat = aisix_gateway::ChatFormat::new( route.name.clone(), vec![aisix_gateway::ChatMessage::user(text)], @@ -619,7 +633,7 @@ async fn dispatch( .iter() .map(|e| &e.value), ); - let captured_prompt = content_cap.map(|_| request_guardrail_text(route.protocol, &body_bytes)); + let captured_prompt = content_cap.map(|_| request_guardrail_text(protocol, &body_bytes)); // Rate limits AFTER the input guardrail so a content block doesn't burn // an RPM slot (matching the typed endpoints). The body's `model` field @@ -720,33 +734,27 @@ async fn dispatch( builder = builder.body(body_bytes.clone()); } - // Exchange bound. A non-streaming route carries a plain total-exchange - // timeout (route override, else the gateway default). A streaming - // route must not bound the relay itself — a healthy long-lived SSE + // Exchange bound. The timeout (route override, else the gateway + // default) must not bound the relay itself — a healthy long-lived SSE // stream is the point — but a blackholed upstream still can't pin the // connection: the header phase (and, below, a non-SSE body read) get - // the same bound via an explicit timer. + // the bound via an explicit timer. let exchange_timeout = route .timeout_ms .map(Duration::from_millis) .or(state.default_timeouts.request); - if !route.streaming { - if let Some(d) = exchange_timeout { - builder = builder.timeout(d); - } - } let bridge_timeout = |d: Duration| aisix_gateway::BridgeError::Timeout { elapsed_ms: d.as_millis().min(u64::MAX as u128) as u64, cause: "passthrough route upstream exchange".into(), }; let send_fut = builder.send(); - let sent = match (route.streaming, exchange_timeout) { - (true, Some(d)) => match tokio::time::timeout(d, send_fut).await { + let sent = match exchange_timeout { + Some(d) => match tokio::time::timeout(d, send_fut).await { Ok(r) => r, Err(_) => return Err(RouteError::of(ProxyError::Bridge(bridge_timeout(d)), &auth)), }, - _ => send_fut.await, + None => send_fut.await, }; let upstream_resp = sent.map_err(|e| { RouteError::of( @@ -799,9 +807,9 @@ async fn dispatch( emitted: false, }; - if route.streaming && is_sse { + if is_sse { return Ok(stream_response( - route.protocol, + protocol, resolved_chain, upstream_resp, resp_headers, @@ -813,19 +821,19 @@ async fn dispatch( // ----- buffered response ----- - // A streaming route reaching this branch got a non-SSE answer; its - // reqwest request carries no built-in timeout, so the body read gets - // the exchange bound explicitly (same blackhole guard as the send). + // A non-SSE answer: the reqwest request carries no built-in timeout, + // so the body read gets the exchange bound explicitly (same blackhole + // guard as the send). let body_fut = upstream_resp.bytes(); - let read = match (route.streaming, exchange_timeout) { - (true, Some(d)) => match tokio::time::timeout(d, body_fut).await { + let read = match exchange_timeout { + Some(d) => match tokio::time::timeout(d, body_fut).await { Ok(r) => r, Err(_) => { telemetry.emitted = true; return Err(RouteError::of(ProxyError::Bridge(bridge_timeout(d)), &auth)); } }, - _ => body_fut.await, + None => body_fut.await, }; let resp_body = read.map_err(|e| { telemetry.emitted = true; @@ -835,9 +843,9 @@ async fn dispatch( ) })?; - // OUTPUT guardrails on the (protocol-extracted) response text. + // OUTPUT guardrails on the (envelope-extracted) response text. if !resolved_chain.is_empty() { - let text = response_guardrail_text(route.protocol, &resp_body); + let text = response_guardrail_text(protocol, &resp_body); let synth = aisix_gateway::ChatResponse { id: String::new(), model: route.name.clone(), @@ -873,12 +881,12 @@ async fn dispatch( } } - if let Some((p, c)) = response_usage(route.protocol, &resp_body) { + if let Some((p, c)) = response_usage(protocol, &resp_body) { telemetry.prompt_tokens = p; telemetry.completion_tokens = c; } if telemetry.content_cap.is_some() { - telemetry.response_text = response_guardrail_text(route.protocol, &resp_body); + telemetry.response_text = response_guardrail_text(protocol, &resp_body); } let mut response = Response::builder() @@ -1007,43 +1015,92 @@ fn content_text(v: &serde_json::Value) -> String { } } -/// The request text a guardrail scans, per the route's protocol hint. -/// Parsing is best-effort: anything that doesn't match the declared shape -/// degrades to the raw lossy-UTF-8 body. +/// The body envelope detected for one exchange. Not configuration: +/// detected per request from the body's top-level keys +/// ([`detect_protocol`]) and sticky for the exchange — the buffered +/// response and every stream frame are read with the same detection. It +/// drives extraction (guardrail text, capture, usage) only; the relay +/// forwards bytes verbatim regardless. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum PassthroughProtocol { + /// No recognized envelope: bodies are opaque (guardrails scan them as + /// one lossy-UTF-8 text; buffered responses are not probed for usage). + Raw, + /// OpenAI-compatible chat envelope (`messages`, streamed + /// `choices[].delta.content`, final-chunk / response `usage`). + OpenaiChat, + /// OpenAI-compatible legacy completions / FIM envelope (`prompt` [+ + /// `suffix`], streamed `choices[].text`, `usage`). + OpenaiCompletions, + /// OpenAI Responses API envelope: `input` on the request, `output` + /// items on the response, `response.output_text.delta` events while + /// streaming, and `usage` in the `input_tokens`/`output_tokens` + /// spelling — carried on the terminal `response.completed` event when + /// the response streams. + OpenaiResponses, +} + +/// Detect the request envelope from the body's top-level keys. The three +/// LLM envelopes are structurally exclusive — `messages`, `input` and +/// `prompt` are each the required carrier field of exactly one API — so +/// real LLM traffic detects unambiguously, and everything else (JSON-RPC, +/// REST, non-JSON, empty/GET bodies) is `Raw`. An unknown API colliding +/// with a carrier key costs nothing: detection drives extraction only, +/// and extraction degrades to the whole body when the detected shape +/// yields no text. +fn detect_protocol(body: &[u8]) -> PassthroughProtocol { + let Ok(v) = serde_json::from_slice::(body) else { + return PassthroughProtocol::Raw; + }; + if v.get("messages").is_some_and(serde_json::Value::is_array) { + PassthroughProtocol::OpenaiChat + } else if v + .get("input") + .is_some_and(|i| i.is_string() || i.is_array()) + { + PassthroughProtocol::OpenaiResponses + } else if v + .get("prompt") + .is_some_and(|p| p.is_string() || p.is_array()) + { + PassthroughProtocol::OpenaiCompletions + } else { + PassthroughProtocol::Raw + } +} + +/// The request text a guardrail scans, per the detected envelope. +/// Extraction is best-effort: a shape that yields no text degrades to the +/// raw lossy-UTF-8 body, so detection never loses audit coverage. fn request_guardrail_text(protocol: PassthroughProtocol, body: &[u8]) -> String { let raw = || String::from_utf8_lossy(body).into_owned(); let Ok(v) = serde_json::from_slice::(body) else { return raw(); }; - match protocol { - PassthroughProtocol::Raw => raw(), - PassthroughProtocol::OpenaiChat => match v.get("messages").and_then(|m| m.as_array()) { - Some(msgs) => msgs - .iter() - .filter_map(|m| m.get("content").map(content_text)) - .filter(|t| !t.is_empty()) - .collect::>() - .join("\n"), - None => raw(), - }, + let extracted = match protocol { + PassthroughProtocol::Raw => return raw(), + PassthroughProtocol::OpenaiChat => v + .get("messages") + .and_then(|m| m.as_array()) + .map(|msgs| { + msgs.iter() + .filter_map(|m| m.get("content").map(content_text)) + .filter(|t| !t.is_empty()) + .collect::>() + .join("\n") + }) + .unwrap_or_default(), // Responses API: `input` is either a bare string or an array of // items whose `content` parts carry the text. PassthroughProtocol::OpenaiResponses => match v.get("input") { Some(serde_json::Value::String(t)) => t.clone(), - Some(serde_json::Value::Array(items)) => { - let joined = items - .iter() - .filter_map(|i| i.get("content").map(content_text)) - .filter(|t| !t.is_empty()) - .collect::>() - .join("\n"); - if joined.is_empty() { - raw() - } else { - joined - } - } - _ => raw(), + Some(serde_json::Value::Array(items)) => items + .iter() + .filter_map(|i| i.get("content").map(content_text)) + .filter(|t| !t.is_empty()) + .collect::>() + .join("\n"), + _ => String::new(), }, PassthroughProtocol::OpenaiCompletions => { let prompt = v.get("prompt").map(|p| match p { @@ -1055,20 +1112,20 @@ fn request_guardrail_text(protocol: PassthroughProtocol, body: &[u8]) -> String other => content_text(other), }); let suffix = v.get("suffix").and_then(|s| s.as_str()); - match (prompt, suffix) { - (None, None) => raw(), - (p, s) => { - let mut out = p.unwrap_or_default(); - if let Some(s) = s { - if !out.is_empty() { - out.push('\n'); - } - out.push_str(s); - } - out + let mut out = prompt.unwrap_or_default(); + if let Some(s) = suffix { + if !out.is_empty() { + out.push('\n'); } + out.push_str(s); } + out } + }; + if extracted.is_empty() { + raw() + } else { + extracted } } @@ -1566,9 +1623,17 @@ fn stream_response( telemetry.emit(); }; + // Re-attach the request span (the body is polled after the request-id + // middleware returns, so end-of-stream telemetry would otherwise log + // without a request_id) and heartbeat silence gaps — this branch is + // SSE-only, where a comment frame is protocol-legal and identical to + // what the typed endpoints emit; relayed frames are untouched. let mut response = Response::builder() .status(status) - .body(Body::from_stream(stream)) + .body(Body::from_stream(crate::sse_keepalive::with_heartbeat( + crate::request_id::in_request_span(stream), + crate::sse_keepalive::interval(), + ))) .unwrap(); copy_safe_headers(&resp_headers, response.headers_mut()); // The relay re-chunks the body; a stale upstream length must not ride @@ -2250,6 +2315,59 @@ mod tests { request_guardrail_text(PassthroughProtocol::OpenaiChat, not_chat), r#"{"input":"x"}"# ); + // A detected envelope whose items carry no text ALSO degrades to + // the raw body — detection must never scan less than raw would. + let empty_chat = br#"{"messages":[{"role":"tool","tool_call_id":"1"}]}"#; + assert_eq!( + request_guardrail_text(PassthroughProtocol::OpenaiChat, empty_chat), + String::from_utf8_lossy(empty_chat) + ); + } + + #[test] + fn detect_protocol_from_request_envelope() { + // The real Copilot CLI surface, one shape per endpoint family. + let cases: [(&[u8], PassthroughProtocol); 8] = [ + // Chat: `messages` array. + ( + br#"{"model":"m","messages":[{"role":"user","content":"hi"}]}"#, + PassthroughProtocol::OpenaiChat, + ), + // Responses API: `input` as items or a bare string. + ( + br#"{"model":"m","input":[{"role":"user","content":"hi"}]}"#, + PassthroughProtocol::OpenaiResponses, + ), + ( + br#"{"model":"m","input":"hi"}"#, + PassthroughProtocol::OpenaiResponses, + ), + // FIM / legacy completions: `prompt`. + ( + br#"{"prompt":"def f(","suffix":"return"}"#, + PassthroughProtocol::OpenaiCompletions, + ), + // MCP JSON-RPC relays as raw. + ( + br#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#, + PassthroughProtocol::Raw, + ), + // Plain REST / unrecognized JSON relays as raw. + (br#"{"ref":"main","inputs":{}}"#, PassthroughProtocol::Raw), + // Carrier keys of the wrong TYPE stay raw: only the API's own + // shape (array/string) counts as that envelope. + (br#"{"messages":"not-an-array"}"#, PassthroughProtocol::Raw), + // Non-JSON / empty (GET) bodies are raw. + (b"", PassthroughProtocol::Raw), + ]; + for (body, want) in cases { + assert_eq!( + detect_protocol(body), + want, + "body {:?}", + String::from_utf8_lossy(body) + ); + } } #[test] diff --git a/schemas/resources/passthrough_route.schema.json b/schemas/resources/passthrough_route.schema.json index 0223fdb6..1b319d81 100644 --- a/schemas/resources/passthrough_route.schema.json +++ b/schemas/resources/passthrough_route.schema.json @@ -437,43 +437,6 @@ "type": "string" } ] - }, - "PassthroughProtocol": { - "description": "Body-shape hint for a passthrough route.", - "oneOf": [ - { - "description": "No parsing: bodies are opaque byte blobs (guardrails scan them as one lossy-UTF-8 text).", - "enum": [ - "raw" - ], - "title": "Opaque body", - "type": "string" - }, - { - "description": "OpenAI-compatible chat envelope (`messages`, streamed `choices[].delta.content`, final-chunk / response `usage`).", - "enum": [ - "openai_chat" - ], - "title": "OpenAI-compatible chat", - "type": "string" - }, - { - "description": "OpenAI-compatible legacy completions / FIM envelope (`prompt` [+ `suffix`], streamed `choices[].text`, `usage`).", - "enum": [ - "openai_completions" - ], - "title": "OpenAI-compatible completions / FIM", - "type": "string" - }, - { - "description": "OpenAI Responses API envelope: `input` items on the request, `output` items on the response, `response.output_text.delta` events while streaming, and `usage` in the `input_tokens`/`output_tokens` spelling — carried on the terminal `response.completed` event when the response streams.", - "enum": [ - "openai_responses" - ], - "title": "OpenAI Responses API", - "type": "string" - } - ] } }, "properties": { @@ -559,15 +522,6 @@ "description": "Derive the target from the request's own `Host` header (`https://`), for forward-proxy routes that fan one route out over several official hosts. Only legal when `hosts` is set — the matched allowlist is what makes the derived target non-attacker- controlled (SSRF guard).", "type": "boolean" }, - "protocol": { - "allOf": [ - { - "$ref": "#/definitions/PassthroughProtocol" - } - ], - "default": "raw", - "description": "Body-shape hint for auditing, guardrails and usage extraction. Parsing is best-effort: a body that does not match the declared shape degrades to `raw` handling, it is never rejected for shape." - }, "provider_key_id": { "description": "ProviderKey whose secret is injected upstream when `credential_mode` is `inject` (per-provider auth shape: `x-api-key` + `anthropic-version` for Anthropic, `Authorization: Bearer` otherwise; its `strip_headers` and TLS settings apply). Required for `inject`; forbidden for `forward_client`.", "minLength": 1, @@ -586,11 +540,6 @@ "null" ] }, - "streaming": { - "default": true, - "description": "Relay `text/event-stream` upstream responses incrementally. When `false` streaming responses are fully buffered like any other body.", - "type": "boolean" - }, "target_url": { "description": "Explicit upstream base URL, e.g. `https://api.openai.com`. The matched request's remainder path and query are appended. Exactly one of `target_url` / `preserve_host` must be configured.", "minLength": 1, @@ -600,7 +549,7 @@ ] }, "timeout_ms": { - "description": "Maximum time, in milliseconds, for a non-streaming upstream exchange. On a streaming route it bounds the response-header phase and any non-SSE body read, but never a healthy SSE relay (which ends with the upstream stream or the client hanging up). When omitted, the gateway default request timeout applies to non-streaming exchanges only.", + "description": "Maximum time, in milliseconds, for the upstream exchange. Bounds the response-header phase and any non-SSE body read, but never a healthy SSE relay (which ends with the upstream stream or the client hanging up). When omitted, the gateway default request timeout applies the same way.", "format": "uint64", "minimum": 1.0, "type": [ diff --git a/tests/e2e/src/cases/passthrough-route-e2e.test.ts b/tests/e2e/src/cases/passthrough-route-e2e.test.ts index 7e4af06c..e267bb82 100644 --- a/tests/e2e/src/cases/passthrough-route-e2e.test.ts +++ b/tests/e2e/src/cases/passthrough-route-e2e.test.ts @@ -10,6 +10,7 @@ import { type OpenAiUpstream, type SpawnedApp, } from "../harness/index.js"; +import { startMockOtlp, type MockOtlp } from "../harness/otlp-mock.js"; // E2E: explicit PassthroughRoute resources — the successor of the removed // implicit `/passthrough/{provider}/*rest` tunnel. A route binds a gateway @@ -36,6 +37,10 @@ import { // gated by `source_cidrs` (real TCP, so 127.0.0.1 resolves). // 6. SSE relay: a streaming upstream is forwarded as SSE with the // frames intact. +// 7. Envelope auto-detection: a route has NO protocol/streaming +// configuration — usage extraction follows the request body's own +// envelope (chat buffered, Responses streamed) and a non-LLM +// exchange is never probed for phantom usage. const CALLER_PLAINTEXT = "sk-ptr-e2e-caller"; const CALLER_KEY_HASH = createHash("sha256") @@ -47,6 +52,7 @@ describe("passthrough-route e2e: explicit routes, BYO credentials, 410 tombstone let seed: SeedClient | undefined; let etcdReachable = false; const upstreams: OpenAiUpstream[] = []; + const otlps: MockOtlp[] = []; beforeAll(async () => { const etcd = new EtcdClient(); @@ -65,6 +71,7 @@ describe("passthrough-route e2e: explicit routes, BYO credentials, 410 tombstone afterAll(async () => { await app?.exit(); await Promise.all(upstreams.map((u) => u.close())); + await Promise.all(otlps.map((o) => o.close())); }); test("inject route on the legacy prefix: /v1 dedup, verbatim body, Bearer injection", async (ctx) => { @@ -451,7 +458,6 @@ describe("passthrough-route e2e: explicit routes, BYO credentials, 410 tombstone path_prefix: "/sse-tunnel", target_url: upstream.baseUrl, provider_key_id: pk.id, - protocol: "openai_chat", }); const headers = { @@ -462,7 +468,11 @@ describe("passthrough-route e2e: explicit routes, BYO credentials, 410 tombstone fetch(`${app!.proxyUrl}/sse-tunnel/chat/completions`, { method: "POST", headers, - body: JSON.stringify({ model: "gpt-4o", stream: true }), + body: JSON.stringify({ + model: "gpt-4o", + messages: [{ role: "user", content: "hi" }], + stream: true, + }), }); await waitConfigPropagation(async () => { @@ -487,4 +497,151 @@ describe("passthrough-route e2e: explicit routes, BYO credentials, 410 tombstone expect(text).toContain('"content":"lo"'); expect(text).toContain("[DONE]"); }); + + test("envelope auto-detection: usage follows the request body, never the config", async (ctx) => { + if (!etcdReachable || !app || !seed) { + ctx.skip(); + return; + } + + // A route carries NO protocol or streaming configuration — the + // envelope is detected per exchange from the request body's top-level + // keys. Three exchanges through three identical-config routes: + // + // - chat-shaped request, buffered response → `usage` extracted + // - Responses-shaped request, SSE response → nested usage on the + // terminal `response.completed` event extracted (the shape the + // GitHub Copilot CLI streams) + // - JSON-RPC request (MCP) → raw: a usage-looking + // object in the response is NOT probed (no phantom tokens) + const otlp = await startMockOtlp(); + otlps.push(otlp); + await seed.createObservabilityExporter({ + name: "ptr-auto-otlp", + enabled: true, + kind: "otlp_http", + endpoint: otlp.url, + }); + + const chatUpstream = await startOpenAiUpstream({ + nonStreamBody: { + id: "chatcmpl-auto", + choices: [{ message: { role: "assistant", content: "ok" } }], + usage: { prompt_tokens: 7, completion_tokens: 3 }, + }, + }); + const responsesUpstream = await startOpenAiUpstream({ + streamEvents: [ + JSON.stringify({ type: "response.output_text.delta", delta: "he" }), + JSON.stringify({ type: "response.output_text.delta", delta: "y" }), + JSON.stringify({ + type: "response.completed", + response: { usage: { input_tokens: 11, output_tokens: 4 } }, + }), + ], + }); + const rpcUpstream = await startOpenAiUpstream({ + nonStreamBody: { + jsonrpc: "2.0", + id: 1, + result: { usage: { prompt_tokens: 99, completion_tokens: 99 } }, + }, + }); + upstreams.push(chatUpstream, responsesUpstream, rpcUpstream); + + const pk = await seed.createProviderKey({ + display_name: "ptr-auto-pk", + secret: "sk-mock", + api_base: "http://unused-on-routes", + }); + for (const [name, prefix, upstream] of [ + ["ptr-auto-chat", "/auto-chat", chatUpstream], + ["ptr-auto-resp", "/auto-resp", responsesUpstream], + ["ptr-auto-rpc", "/auto-rpc", rpcUpstream], + ] as const) { + await seed.createPassthroughRoute({ + name, + path_prefix: prefix, + target_url: upstream.baseUrl, + provider_key_id: pk.id, + }); + } + + // Readiness sentinel, seeded LAST: the snapshot watch applies etcd + // revisions in order, so this route serving proves every resource + // seeded before it (exporter, the three routes under test) is live — + // without the probe exercising any exchange the test asserts on. + await seed.createPassthroughRoute({ + name: "ptr-auto-ready", + path_prefix: "/auto-ready", + target_url: rpcUpstream.baseUrl, + provider_key_id: pk.id, + }); + + const headers = { + authorization: `Bearer ${CALLER_PLAINTEXT}`, + "content-type": "application/json", + }; + const post = (path: string, body: unknown) => + fetch(`${app!.proxyUrl}${path}`, { + method: "POST", + headers, + body: JSON.stringify(body), + }); + + await waitConfigPropagation(async () => { + try { + const r = await post("/auto-ready/ping", {}); + await r.text(); + return r.status === 200; + } catch { + return false; + } + }); + + const chatRes = await post("/auto-chat/chat/completions", { + model: "m", + messages: [{ role: "user", content: "hi" }], + }); + expect(chatRes.status).toBe(200); + await chatRes.text(); + const responsesRes = await post("/auto-resp/responses", { + model: "m", + input: [{ role: "user", content: "hi" }], + stream: true, + }); + expect(responsesRes.status).toBe(200); + await responsesRes.text(); + const rpcRes = await post("/auto-rpc/mcp", { + jsonrpc: "2.0", + id: 1, + method: "tools/list", + }); + expect(rpcRes.status).toBe(200); + await rpcRes.text(); + + const spanFor = async (route: string) => { + const deadline = Date.now() + 10_000; + while (Date.now() < deadline) { + const hit = otlp.spans.find( + (s) => s.attributes["aisix.passthrough.route_name"] === route, + ); + if (hit) return hit; + await new Promise((r) => setTimeout(r, 50)); + } + throw new Error(`no OTLP span for route ${route}`); + }; + + const chatSpan = await spanFor("ptr-auto-chat"); + expect(chatSpan.attributes["gen_ai.usage.input_tokens"]).toBe(7); + expect(chatSpan.attributes["gen_ai.usage.output_tokens"]).toBe(3); + + const respSpan = await spanFor("ptr-auto-resp"); + expect(respSpan.attributes["gen_ai.usage.input_tokens"]).toBe(11); + expect(respSpan.attributes["gen_ai.usage.output_tokens"]).toBe(4); + + const rpcSpan = await spanFor("ptr-auto-rpc"); + expect(rpcSpan.attributes["gen_ai.usage.input_tokens"] ?? 0).toBe(0); + expect(rpcSpan.attributes["gen_ai.usage.output_tokens"] ?? 0).toBe(0); + }); });