diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 77150399c..879fcfa0d 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -1033,22 +1033,32 @@ fn client_error(error: &LlmClientError) -> Response { "upstream_error", "upstream_error", ), - LlmClientError::Transport { source } | LlmClientError::InvalidResponse { source } => { - error_response( - StatusCode::BAD_GATEWAY, - source.to_string(), - "upstream_error", - "upstream_error", - ) - } + // A transport source is reqwest's, and it renders the full request URL, including any + // credentials configured in its query string. + LlmClientError::Transport { source } => redacted_error_response( + StatusCode::BAD_GATEWAY, + "upstream transport error", + source.to_string(), + "upstream_error", + "upstream_error", + ), + // Decoding failures describe the response body, not the request target. + LlmClientError::InvalidResponse { source } => error_response( + StatusCode::BAD_GATEWAY, + source.to_string(), + "upstream_error", + "upstream_error", + ), LlmClientError::ResponseTranslation(message) => error_response( StatusCode::BAD_GATEWAY, message, "upstream_error", "upstream_error", ), - LlmClientError::Timeout { source } => error_response( + // A timeout carries the same request context as a transport error. + LlmClientError::Timeout { source } => redacted_error_response( StatusCode::GATEWAY_TIMEOUT, + "upstream request timed out", source.to_string(), "upstream_error", "upstream_timeout", @@ -1078,6 +1088,9 @@ struct ApiError { message: String, error_type: &'static str, code: &'static str, + // Detail for the request log only. Errors whose source can carry the configured upstream + // URL keep the full text here and send a fixed message on the wire. + log_detail: Option, } impl ApiError { @@ -1092,9 +1105,15 @@ impl ApiError { message: message.into(), error_type, code, + log_detail: None, } } + fn with_log_detail(mut self, detail: impl Into) -> Self { + self.log_detail = Some(detail.into()); + self + } + fn into_response(self, wire_format: WireFormat) -> Response { let body = match wire_format { WireFormat::AnthropicMessages => json!({ @@ -1113,9 +1132,11 @@ impl ApiError { }), }; let mut response = (self.status, Json(body)).into_response(); - response - .extensions_mut() - .insert(RequestLogError(self.message.clone())); + response.extensions_mut().insert(RequestLogError( + self.log_detail + .clone() + .unwrap_or_else(|| self.message.clone()), + )); response.extensions_mut().insert(self); response } @@ -1167,6 +1188,20 @@ fn error_response( ApiError::new(status, message, error_type, code).into_response(WireFormat::OpenAiChat) } +// For errors whose source text can contain the configured upstream URL: the client sees a fixed +// message, the request log keeps the source. +fn redacted_error_response( + status: StatusCode, + message: &'static str, + log_detail: impl Into, + error_type: &'static str, + code: &'static str, +) -> Response { + ApiError::new(status, message, error_type, code) + .with_log_detail(log_detail) + .into_response(WireFormat::OpenAiChat) +} + async fn models(State(state): State) -> Json { Json(model_list_payload( state @@ -1766,6 +1801,80 @@ mod tests { ); } + // Redaction is limited to the wire: the request log keeps the full transport source so + // operators can still diagnose the failure. + #[test] + fn transport_error_keeps_source_in_request_log() { + const UPSTREAM_URL: &str = "https://upstream.invalid/v1?key=CANARY_ADMIN_QUERY_KEY"; + let error = LlmClientError::Transport { + source: Box::new(std::io::Error::other(format!( + "error sending request for url ({UPSTREAM_URL})" + ))), + }; + + let response = client_error(&error); + let api_error = response + .extensions() + .get::() + .expect("client error metadata"); + assert_eq!(api_error.message, "upstream transport error"); + + let logged = response + .extensions() + .get::() + .map(|error| error.0.as_str()) + .expect("request log error"); + assert!( + logged.contains(UPSTREAM_URL), + "transport source dropped from the request log: {logged:?}" + ); + } + + // A timeout redacts the wire message and keeps its source for the log on the same terms. + #[test] + fn client_timeout_error_hides_source_details() { + const UPSTREAM_URL: &str = "https://upstream.invalid/v1?key=CANARY_ADMIN_QUERY_KEY"; + let error = LlmClientError::Timeout { + source: Box::new(std::io::Error::other(format!( + "request timed out for {UPSTREAM_URL}" + ))), + }; + + let response = client_error(&error); + let api_error = response + .extensions() + .get::() + .expect("client error metadata"); + assert_eq!(response.status(), StatusCode::GATEWAY_TIMEOUT); + assert_eq!(api_error.message, "upstream request timed out"); + assert_eq!(api_error.code, "upstream_timeout"); + assert!(!api_error.message.contains(UPSTREAM_URL)); + + let logged = response + .extensions() + .get::() + .map(|error| error.0.as_str()) + .expect("request log error"); + assert!(logged.contains(UPSTREAM_URL)); + } + + // Response-decoding detail describes the body, not the request target, and stays visible. + #[test] + fn client_invalid_response_error_preserves_source_detail() { + let error = LlmClientError::InvalidResponse { + source: Box::new(std::io::Error::other("response body was truncated")), + }; + + let response = client_error(&error); + let api_error = response + .extensions() + .get::() + .expect("client error metadata"); + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + assert_eq!(api_error.message, "response body was truncated"); + assert_eq!(api_error.code, "upstream_error"); + } + // Canonical error text remains available without consuming the response body. #[test] fn error_response_carries_request_log_error() { diff --git a/crates/switchyard-server/src/sse.rs b/crates/switchyard-server/src/sse.rs index 6354945ec..774826f63 100644 --- a/crates/switchyard-server/src/sse.rs +++ b/crates/switchyard-server/src/sse.rs @@ -8,6 +8,7 @@ use std::convert::Infallible; use axum::response::sse::{Event, Sse}; use futures_util::Stream; use serde_json::{Value, json}; +use switchyard_protocol::LlmClientError; use switchyard_translation::{RawEventStream, WireFormat}; /// Boxed stream type accepted by Axum's SSE response wrapper. @@ -32,9 +33,10 @@ pub(crate) fn frame_stream( } }, Err(error) => { + // The full text stays in the log; only the client-facing copy is redacted. tracing::warn!(error = %error, "stream iteration failed"); failed = true; - error_event(target_format, error.to_string()) + error_event(target_format, client_visible_stream_error(error.as_ref())) } }; yield Ok(event); @@ -65,6 +67,18 @@ fn frame_event(target_format: WireFormat, value: Value) -> Result String { + match error.downcast_ref::() { + Some(LlmClientError::Transport { .. }) => "upstream transport error".to_string(), + Some(LlmClientError::Timeout { .. }) => "upstream request timed out".to_string(), + _ => error.to_string(), + } +} + fn error_event(target_format: WireFormat, message: String) -> Event { match target_format { WireFormat::OpenAiChat => Event::default().data( @@ -121,4 +135,49 @@ mod tests { assert!(!body.contains("[DONE]")); Ok(()) } + + // A stream that fails after its response has begun is past the buffered `client_error` + // boundary, so the redaction has to be repeated here. + #[tokio::test] + async fn stream_transport_error_hides_credential_bearing_upstream_url() -> TestResult { + const UPSTREAM_URL: &str = "http://upstream.invalid/v1?key=CANARY_ADMIN_QUERY_KEY"; + let failure: Box = Box::new(LlmClientError::Transport { + source: Box::new(io::Error::other(format!( + "error sending request for url ({UPSTREAM_URL})" + ))), + }); + let stream: RawEventStream = Box::pin(stream::iter(vec![ + Ok(json!({"id": "before"})), + Err(failure), + ])); + + let response = frame_stream(stream, WireFormat::OpenAiChat).into_response(); + let body = String::from_utf8(to_bytes(response.into_body(), usize::MAX).await?.to_vec())?; + + assert!(body.contains("upstream transport error")); + assert!( + !body.contains("CANARY_ADMIN_QUERY_KEY"), + "credential leaked in {body:?}" + ); + assert!( + !body.contains(UPSTREAM_URL), + "upstream URL leaked in {body:?}" + ); + Ok(()) + } + + // Errors that do not describe the request target keep their text. + #[tokio::test] + async fn stream_translation_error_keeps_its_message() -> TestResult { + let failure: Box = Box::new(LlmClientError::ResponseTranslation( + "unrecognized event type".to_string(), + )); + let stream: RawEventStream = Box::pin(stream::iter(vec![Err(failure)])); + + let response = frame_stream(stream, WireFormat::OpenAiChat).into_response(); + let body = String::from_utf8(to_bytes(response.into_body(), usize::MAX).await?.to_vec())?; + + assert!(body.contains("unrecognized event type")); + Ok(()) + } } diff --git a/crates/switchyard-server/tests/server.rs b/crates/switchyard-server/tests/server.rs index 32290ea36..da117f69a 100644 --- a/crates/switchyard-server/tests/server.rs +++ b/crates/switchyard-server/tests/server.rs @@ -3368,3 +3368,45 @@ async fn responses_round_trips_codex_tool_namespaces() -> TestResult { assert_eq!(completed["response"]["output"][0]["namespace"], "mcp__b"); Ok(()) } + +// A transport error must not expose credentials from the configured upstream URL. +#[tokio::test] +async fn transport_error_hides_credential_bearing_upstream_url() -> TestResult { + const CANARY: &str = "CANARY_ADMIN_QUERY_KEY"; + + let listener = TcpListener::bind("127.0.0.1:0").await?; + let addr = listener.local_addr()?; + drop(listener); + let base_url = format!("http://{addr}/v1?key={CANARY}"); + let upstream_request_url = format!("{base_url}/chat/completions"); + let app = build_switchyard_router(random_state(&base_url, &[(ROUTE_MODEL, &["model/a"])])?); + + let response = send( + &app, + "POST", + "/v1/chat/completions", + Some(json!({ + "model": ROUTE_MODEL, + "messages": [{"role": "user", "content": "hello"}] + })), + ) + .await?; + + assert_eq!(response.status, StatusCode::BAD_GATEWAY); + let body = response.json()?; + assert_eq!(body["error"]["type"], "upstream_error"); + assert_eq!(body["error"]["code"], "upstream_error"); + let message = body["error"]["message"] + .as_str() + .ok_or("transport error message was not text")?; + assert_eq!(message, "upstream transport error"); + assert!( + !message.contains(CANARY), + "credential leaked in {message:?}" + ); + assert!( + !message.contains(&upstream_request_url), + "upstream URL leaked in {message:?}" + ); + Ok(()) +} diff --git a/crates/switchyard-translation/src/helpers.rs b/crates/switchyard-translation/src/helpers.rs index 7306aa983..c5fa9c284 100644 --- a/crates/switchyard-translation/src/helpers.rs +++ b/crates/switchyard-translation/src/helpers.rs @@ -864,3 +864,41 @@ mod tests { Ok(()) } } + +#[cfg(test)] +mod stream_error_boxing { + use futures::executor::block_on; + use futures::{StreamExt, stream}; + + use super::*; + use crate::{LlmResponseStream, WireFormat}; + + // `switchyard-server` redacts credential-bearing stream failures by downcasting the + // boxed error back to `LlmClientError`. That only works while `?` boxes the concrete + // type instead of wrapping it, so pin the behaviour the redaction depends on. + #[test] + fn encoded_stream_error_downcasts_to_llm_client_error() { + block_on(async { + let chunks: LlmResponseStream = + Box::pin(stream::iter(vec![Err(LlmClientError::Transport { + source: Box::new(std::io::Error::other("upstream is unreachable")), + })])); + + let mut stream = encode_stream(chunks, WireFormat::OpenAiChat, None) + .expect("the built-in codec resolves"); + let error = stream + .next() + .await + .expect("the stream yields the failed chunk") + .expect_err("the chunk was an error"); + + assert!( + matches!( + error.downcast_ref::(), + Some(LlmClientError::Transport { .. }) + ), + "boxed stream error no longer downcasts to LlmClientError: {error}" + ); + }) + } +}