From 5887e834be9c8fa710ebd98bb675b6c75c862e96 Mon Sep 17 00:00:00 2001 From: Pei Li Date: Thu, 28 May 2026 15:40:34 -0700 Subject: [PATCH 1/4] fix(http): sanitize HTTP error responses to avoid leaking internals Multiple paths in the HTTP service formatted dynamic error content (anyhow chains, JoinError debug, backend-supplied messages from SSE events) directly into the client response. These can expose internal file paths, library versions, panic stack info, and raw internal error JSON. Affected paths: - ErrorMessage::from_anyhow fallback ({err:#}) - ResourceExhausted branch (raw err.to_string) - Three JoinError sites in completions/chat_completions/responses handlers ({e:?}) - ErrorMessage::from_http_error for 5xx / out-of-range codes - completions/chat/responses fold and parse error handlers - MJPEG response builder - check_for_backend_error for non-4xx events (panic text, raw JSON, comment payloads, or backend-asserted 1xx/2xx/3xx codes) - Streaming SSE error frame in monitor_for_disconnects, which emitted the raw backend error across every streaming endpoint (OpenAI chat/completions/videos and Anthropic messages) - Anthropic 500 handlers: await-task JoinError, generate failure, and fold-stream failure - Cancellation (HTTP 499) handlers in from_anyhow and the Anthropic generate path, which echoed raw backend cancellation text (context IDs, "stopped or killed") - Realtime WebSocket server_error frames: the outbound engine-error frame and the engine-lookup failure handler echoed err.to_string() Add ErrorMessage::internal_server_error_with_details(public_msg, details) which logs the detail server-side and returns only the sanitized public_msg to the client. Route the affected paths through it, or trim the dynamic content from the client message where a structured tracing::error! already captures it. The ResourceExhausted branch now returns a static "Service temporarily overloaded" + admission-control hint. from_http_error only forwards backend messages in the documented 4xx range; everything else gets "Internal server error". check_for_backend_error mirrors that contract: only is_client_error() statuses are forwarded unchanged, everything else is sanitized. The streaming SSE frame and the Anthropic 500 handlers now log the full detail server-side and return a static "Internal server error"; client-facing 4xx messages are intentionally preserved. The 499 cancellation paths now return a static "Request cancelled" and log the detail server-side. The realtime WebSocket server_error frames now return a static "Internal server error" and log the detail server-side; the realtime model-not-found / model-unavailable frames (which only echo the client-supplied model name) are preserved. Update the unit tests that previously codified the leaky behavior to assert the sanitized client messages and that the underlying detail is absent (test_other_error_response_from_anyhow, test_resource_exhausted_error_response_from_anyhow, test_error_response_from_anyhow_out_of_range, the three test_check_for_backend_error_* 500-path tests, the two monitor_for_disconnects fault-contract tests, and test_cancelled_error_response_from_anyhow). Add test_check_for_backend_error_with_non_client_error_code and test_mid_stream_error_does_not_leak_internal_details to lock in sanitization for backend-asserted non-4xx codes and for streaming path/panic/exception fragments. Ref: DYN-2636 Signed-off-by: Pei Li --- lib/llm/src/http/service/anthropic.rs | 11 +- lib/llm/src/http/service/disconnect.rs | 62 +++++-- lib/llm/src/http/service/openai.rs | 243 +++++++++++++++++-------- lib/llm/src/http/service/realtime.rs | 8 +- 4 files changed, 232 insertions(+), 92 deletions(-) diff --git a/lib/llm/src/http/service/anthropic.rs b/lib/llm/src/http/service/anthropic.rs index 1a977fe54050..16bd35d192b7 100644 --- a/lib/llm/src/http/service/anthropic.rs +++ b/lib/llm/src/http/service/anthropic.rs @@ -178,10 +178,11 @@ async fn handler_anthropic_messages( tokio::spawn(anthropic_messages(state, template, request, stream_handle).in_current_span()) .await .map_err(|e| { + tracing::error!("Failed to await Anthropic messages task: {e:?}"); anthropic_error( StatusCode::INTERNAL_SERVER_ERROR, "api_error", - &format!("Failed to await messages task: {:?}", e), + "Internal server error", ) })?; @@ -333,17 +334,19 @@ async fn anthropic_messages( // Check for cancelled request (client disconnected before response was sent) if super::metrics::request_was_cancelled(e.as_ref()) { inflight_guard.mark_error(super::metrics::ErrorType::Cancelled); + tracing::debug!("Anthropic request cancelled before response: {e:#}"); return anthropic_error( StatusCode::from_u16(499).unwrap(), "request_cancelled", - &format!("Request cancelled: {}", e), + "Request cancelled", ); } inflight_guard.mark_error(super::metrics::ErrorType::Internal); + tracing::error!("Failed to generate Anthropic completions: {e}"); anthropic_error( StatusCode::INTERNAL_SERVER_ERROR, "api_error", - &format!("Failed to generate completions: {}", e), + "Internal server error", ) })?; @@ -472,7 +475,7 @@ async fn anthropic_messages( anthropic_error( StatusCode::INTERNAL_SERVER_ERROR, "api_error", - &format!("Failed to fold messages stream: {}", e), + "Internal server error", ) })?; diff --git a/lib/llm/src/http/service/disconnect.rs b/lib/llm/src/http/service/disconnect.rs index 4196d72028f8..51fe213777c3 100644 --- a/lib/llm/src/http/service/disconnect.rs +++ b/lib/llm/src/http/service/disconnect.rs @@ -226,11 +226,17 @@ pub fn monitor_for_disconnects( // doesn't later record this as ClosedUnexpectedly (which // would mis-attribute the fault as a client disconnect). stream_handle.disarm(); - // DIS-1768: emit structured OpenAI-style error frame + `data: [DONE]` - // so naive `data:`-line parsers see both the error and a stream terminator. + // Log the underlying error server-side; the client only + // receives a sanitized message so streaming responses don't + // leak internal paths, panic text, or backend exception + // details. + tracing::error!("Streaming error: {err}"); + // Emit a structured OpenAI-style error frame + `data: [DONE]` + // so naive `data:`-line parsers see both the error and a + // stream terminator. let err_json = serde_json::json!({ "error": { - "message": err.to_string(), + "message": "Internal server error", "type": "internal_server_error", "code": 500, } @@ -508,7 +514,7 @@ mod tests { } // ───────────────────────────────────────────────────────────────────────────── - // DIS-1768: mid-stream fault SSE contract + // mid-stream fault SSE contract // // When the upstream stream yields `Err(_)` mid-stream — e.g. an upstream // worker dies and the mpsc channel reports @@ -550,9 +556,11 @@ mod tests { String::from_utf8(body.to_vec()).expect("utf8 body") } - /// Assert the post-fix SSE fault contract: parsed structured error frame with exact - /// message/type/code, positioned before `[DONE]`, and no bare `event: error` trailer. - fn assert_fault_contract(case: &str, text: &str, expected_message: &str) { + /// Assert the post-fix SSE fault contract: a parsed structured error frame + /// carrying the sanitized static message/type/code, positioned before + /// `[DONE]`, with no bare `event: error` trailer, and crucially with no trace + /// of `leaked_detail` (the raw backend error) anywhere in the body. + fn assert_fault_contract(case: &str, text: &str, leaked_detail: &str) { let done_pos = text.find("data: [DONE]").unwrap_or_else(|| { panic!("[{case}] body does not terminate with `data: [DONE]`. Body:\n{text}") }); @@ -584,8 +592,8 @@ mod tests { .unwrap_or_else(|| panic!("[{case}] `error` field is not an object. Body:\n{text}")); assert_eq!( error.get("message").and_then(|v| v.as_str()), - Some(expected_message), - "[{case}] structured error `message` mismatch. Body:\n{text}" + Some("Internal server error"), + "[{case}] structured error `message` must be the sanitized static string. Body:\n{text}" ); assert_eq!( error.get("type").and_then(|v| v.as_str()), @@ -601,6 +609,11 @@ mod tests { !text.contains("event: error\n: "), "[{case}] body contains bare `event: error\\n: ` trailer (pre-fix bug). Body:\n{text}" ); + assert!( + !text.contains(leaked_detail), + "[{case}] SSE body leaked raw backend error detail to the client. \ + Expected `{leaked_detail}` to be absent. Body:\n{text}" + ); } /// Upstream worker killed mid-stream → mpsc channel reports `Disconnected` to the @@ -609,12 +622,12 @@ mod tests { #[serial] async fn test_simulate_worker_kill_emits_structured_error_and_done() { let (_metrics, guard, ctx, handle) = setup_test("worker-kill-model", "req-wk", "0"); - let expected_message = "Disconnected: Stream ended before generation completed"; - let stream = simulate_mid_stream_error(3, expected_message); + let backend_detail = "Disconnected: Stream ended before generation completed"; + let stream = simulate_mid_stream_error(3, backend_detail); let monitored = monitor_for_disconnects(stream, ctx, guard, handle); let body = collect_sse_body(monitored).await; cleanup_env(); - assert_fault_contract("worker_kill", &body, expected_message); + assert_fault_contract("worker_kill", &body, backend_detail); } /// Python chat-processor raises mid-stream → Rust→Python `tx.send()` fails with @@ -623,11 +636,30 @@ mod tests { #[serial] async fn test_simulate_python_consumer_drop_emits_structured_error_and_done() { let (_metrics, guard, ctx, handle) = setup_test("py-drop-model", "req-py", "0"); - let expected_message = "Failed to send response: SendError { .. }"; - let stream = simulate_mid_stream_error(3, expected_message); + let backend_detail = "Failed to send response: SendError { .. }"; + let stream = simulate_mid_stream_error(3, backend_detail); + let monitored = monitor_for_disconnects(stream, ctx, guard, handle); + let body = collect_sse_body(monitored).await; + cleanup_env(); + assert_fault_contract("python_consumer_drop", &body, backend_detail); + } + + /// A backend error carrying sensitive internals (file paths, panic text, + /// Python exception details) MUST NOT reach the streaming client. The client + /// receives only the sanitized static frame; the detail stays server-side. + #[tokio::test] + #[serial] + async fn test_mid_stream_error_does_not_leak_internal_details() { + let (_metrics, guard, ctx, handle) = setup_test("leak-model", "req-leak", "0"); + let backend_detail = "panicked at '/opt/dynamo/lib/python3.12/site-packages/engine/worker.py:512: ValueError: secret tensor shape mismatch'"; + let stream = simulate_mid_stream_error(2, backend_detail); let monitored = monitor_for_disconnects(stream, ctx, guard, handle); let body = collect_sse_body(monitored).await; cleanup_env(); - assert_fault_contract("python_consumer_drop", &body, expected_message); + assert_fault_contract("internal_detail_leak", &body, backend_detail); + // Spot-check the most damaging fragments explicitly. + assert!(!body.contains("site-packages"), "leaked a filesystem path"); + assert!(!body.contains("panicked at"), "leaked panic text"); + assert!(!body.contains("ValueError"), "leaked exception type"); } } diff --git a/lib/llm/src/http/service/openai.rs b/lib/llm/src/http/service/openai.rs index 096d56cc88b7..0d3124e4ab58 100644 --- a/lib/llm/src/http/service/openai.rs +++ b/lib/llm/src/http/service/openai.rs @@ -223,6 +223,28 @@ impl ErrorMessage { ) } + /// Internal Server Error with sanitized client message. + /// Logs `details` server-side and returns only `public_msg` to the client. + /// Use this whenever the detail could carry an anyhow chain, JoinError + /// debug output, or anything else that may leak file paths, library + /// versions, or other internal implementation details. + pub fn internal_server_error_with_details( + public_msg: &str, + details: impl std::fmt::Display, + ) -> ErrorResponse { + tracing::error!("Internal server error: {public_msg}: {details}"); + let code = StatusCode::INTERNAL_SERVER_ERROR; + let error_type = map_error_code_to_error_type(code); + ( + code, + Json(ErrorMessage { + message: public_msg.to_string(), + error_type, + code: code.as_u16(), + }), + ) + } + /// Not Implemented Error /// Return this error when the client requests a feature that is not yet implemented. /// This should be used for features that are planned but not available. @@ -260,17 +282,15 @@ impl ErrorMessage { pub fn from_anyhow(err: anyhow::Error, alt_msg: &str) -> ErrorResponse { // Check for ResourceExhausted anywhere in the error chain → HTTP 503 if super::metrics::request_was_rejected(err.as_ref()) { - let message = find_dynamo_error_in_chain( - err.as_ref(), - dynamo_runtime::error::ErrorType::ResourceExhausted, - ) - .map(|dynamo_err| dynamo_err.message().to_string()) - .unwrap_or_else(|| err.to_string()); - + // Log the full chain server-side; client only sees a sanitized hint + // so we don't leak internal paths or implementation details. + tracing::warn!("Request rejected by admission control: {err:#}"); return ( StatusCode::SERVICE_UNAVAILABLE, Json(ErrorMessage { - message: format!("{}. {}", message, ADMISSION_CONTROL_REJECTION_HINT), + message: format!( + "Service temporarily overloaded. {ADMISSION_CONTROL_REJECTION_HINT}" + ), error_type: map_error_code_to_error_type(StatusCode::SERVICE_UNAVAILABLE), code: StatusCode::SERVICE_UNAVAILABLE.as_u16(), }), @@ -299,11 +319,14 @@ impl ErrorMessage { // Check for Cancelled anywhere in the error chain → HTTP 499 (Client Closed Request) if super::metrics::request_was_cancelled(err.as_ref()) { let code = StatusCode::from_u16(499).unwrap(); - tracing::debug!("Request cancelled before response: {err}"); + // Log the detail server-side; the client gets only a static message so a + // 499 surfaced before the connection actually drops can't leak context + // IDs or backend cancellation internals. + tracing::debug!("Request cancelled before response: {err:#}"); return ( code, Json(ErrorMessage { - message: err.to_string(), + message: "Request cancelled".to_string(), error_type: map_error_code_to_error_type(code), code: code.as_u16(), }), @@ -313,14 +336,24 @@ impl ErrorMessage { // Then check for HttpError match err.downcast::() { Ok(http_error) => ErrorMessage::from_http_error(http_error), - Err(err) => ErrorMessage::internal_server_error(&format!("{alt_msg}: {err:#}")), + // Log the full anyhow chain server-side but only return `alt_msg` + // to the client so we don't leak internal paths. + Err(err) => { + ErrorMessage::internal_server_error_with_details(alt_msg, format!("{err:#}")) + } } } /// Implementers should only be able to throw 400-499 errors. pub fn from_http_error(err: HttpError) -> ErrorResponse { + // Backend-supplied messages are only forwarded for the documented 4xx + // range; for 5xx or codes outside the HTTP space the message may + // contain internal paths/details and is kept server-side only. if err.code < 400 || err.code >= 500 { - return ErrorMessage::internal_server_error(&err.message); + return ErrorMessage::internal_server_error_with_details( + "Internal server error", + err.message, + ); } match StatusCode::from_u16(err.code) { Ok(code) => ( @@ -331,7 +364,10 @@ impl ErrorMessage { code: code.as_u16(), }), ), - Err(_) => ErrorMessage::internal_server_error(&err.message), + Err(_) => ErrorMessage::internal_server_error_with_details( + "Internal server error", + err.message, + ), } } } @@ -515,10 +551,12 @@ async fn handler_completions( let response = tokio::spawn(completions(state, request, stream_handle).in_current_span()) .await .map_err(|e| { - ErrorMessage::internal_server_error(&format!( - "Failed to await chat completions task: {:?}", - e, - )) + // JoinError debug output can include panic messages with stack + // info — keep that server-side only. + ErrorMessage::internal_server_error_with_details( + "Failed to await chat completions task", + format!("{e:?}"), + ) })?; // if we got here, then we will return a response and the potentially long running task has completed successfully @@ -692,8 +730,7 @@ async fn completions_single( e ); let err_response = ErrorMessage::internal_server_error(&format!( - "Failed to fold completions stream for {}: {:?}", - request_id, e + "Failed to fold completions stream for {request_id}" )); inflight_guard.mark_error(extract_error_type_from_response(&err_response)); err_response @@ -887,8 +924,7 @@ async fn completions_batch( e ); let err_response = ErrorMessage::internal_server_error(&format!( - "Failed to fold completions stream for {}: {:?}", - request_id, e + "Failed to fold completions stream for {request_id}" )); inflight_guard.mark_error(extract_error_type_from_response(&err_response)); err_response @@ -1035,10 +1071,12 @@ async fn handler_chat_completions( tokio::spawn(chat_completions(state, template, request, stream_handle).in_current_span()) .await .map_err(|e| { - ErrorMessage::internal_server_error(&format!( - "Failed to await chat completions task: {:?}", - e, - )) + // JoinError debug output can include panic messages with stack + // info — keep that server-side only. + ErrorMessage::internal_server_error_with_details( + "Failed to await chat completions task", + format!("{e:?}"), + ) })?; // if we got here, then we will return a response and the potentially long running task has completed successfully @@ -1156,13 +1194,24 @@ pub(super) async fn check_for_backend_error( if let Some(first_event) = stream.next().await { // Check if it's an error event if let Some((error_msg, status_code)) = extract_backend_error_if_present(&first_event) { - return Err(( - status_code, - Json(ErrorMessage { - message: error_msg, - error_type: map_error_code_to_error_type(status_code), - code: status_code.as_u16(), - }), + // Mirror from_http_error: only the documented 4xx range is part + // of the protocol contract and forwarded unchanged. Anything + // else (5xx, or a 1xx/2xx/3xx asserted by the backend payload) + // may carry panic text, file paths, or raw internal error JSON + // and is kept server-side only. + if status_code.is_client_error() { + return Err(( + status_code, + Json(ErrorMessage { + message: error_msg, + error_type: map_error_code_to_error_type(status_code), + code: status_code.as_u16(), + }), + )); + } + return Err(ErrorMessage::internal_server_error_with_details( + "Internal server error", + error_msg, )); } @@ -1550,10 +1599,9 @@ async fn chat_completions( "Failed to parse chat completion response: {:?}", e ); - let err_response = ErrorMessage::internal_server_error(&format!( - "Failed to parse chat completion response: {}", - e - )); + let err_response = ErrorMessage::internal_server_error( + "Failed to parse chat completion response", + ); inflight_guard.mark_error(extract_error_type_from_response(&err_response)); err_response })?; @@ -1710,10 +1758,12 @@ async fn handler_responses( tokio::spawn(responses(state, template, request, stream_handle).in_current_span()) .await .map_err(|e| { - ErrorMessage::internal_server_error(&format!( - "Failed to await responses task: {:?}", - e, - )) + // JoinError debug output can include panic messages with stack + // info — keep that server-side only. + ErrorMessage::internal_server_error_with_details( + "Failed to await responses task", + format!("{e:?}"), + ) })?; // if we got here, then we will return a response and the potentially long running task has completed successfully @@ -1984,10 +2034,8 @@ async fn responses( .await .map_err(|e| { tracing::error!(request_id, "Failed to fold responses stream: {:?}", e); - let err_response = ErrorMessage::internal_server_error(&format!( - "Failed to fold responses stream: {}", - e - )); + let err_response = + ErrorMessage::internal_server_error("Failed to fold responses stream"); inflight_guard.mark_error(extract_error_type_from_response(&err_response)); err_response })?; @@ -2668,7 +2716,10 @@ async fn video_stream( .map_err(|e| { // inflight is already owned by the monitored_stream which handles // mark_ok (stream end) and mark_error (cancellation). - ErrorMessage::internal_server_error(&format!("Failed to build MJPEG response: {e}")) + ErrorMessage::internal_server_error_with_details( + "Failed to build MJPEG response", + format!("{e}"), + ) }) } @@ -2855,34 +2906,32 @@ mod tests { #[test] fn test_error_response_from_anyhow_out_of_range() { - let err = http_error_from_engine(399).unwrap_err(); - let response = ErrorMessage::from_anyhow(err, BACKUP_ERROR_MESSAGE); - assert_eq!(response.0, StatusCode::INTERNAL_SERVER_ERROR); - assert_eq!(response.1.message, "custom error message"); - - let err = http_error_from_engine(500).unwrap_err(); - let response = ErrorMessage::from_anyhow(err, BACKUP_ERROR_MESSAGE); - assert_eq!(response.0, StatusCode::INTERNAL_SERVER_ERROR); - assert_eq!(response.1.message, "custom error message"); - - let err = http_error_from_engine(501).unwrap_err(); - let response = ErrorMessage::from_anyhow(err, BACKUP_ERROR_MESSAGE); - assert_eq!(response.0, StatusCode::INTERNAL_SERVER_ERROR); - assert_eq!(response.1.message, "custom error message"); + // Backend-supplied messages outside the 4xx range must NOT be + // forwarded to the client — they may include internal paths. + for code in [399u16, 500, 501] { + let err = http_error_from_engine(code).unwrap_err(); + let response = ErrorMessage::from_anyhow(err, BACKUP_ERROR_MESSAGE); + assert_eq!(response.0, StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!(response.1.message, "Internal server error"); + assert!( + !response.1.message.contains("custom error message"), + "client response must not include the backend-supplied HttpError message" + ); + } } #[test] fn test_other_error_response_from_anyhow() { + // Non-HttpError anyhow chains must NOT be exposed to the client; only + // the static backup message should appear in the response. let err = other_error_from_engine().unwrap_err(); + let leaked_chain = format!("{err:#}"); let response = ErrorMessage::from_anyhow(err, BACKUP_ERROR_MESSAGE); assert_eq!(response.0, StatusCode::INTERNAL_SERVER_ERROR); - assert_eq!( - response.1.message, - format!( - "{}: {}", - BACKUP_ERROR_MESSAGE, - other_error_from_engine().unwrap_err() - ) + assert_eq!(response.1.message, BACKUP_ERROR_MESSAGE); + assert!( + !response.1.message.contains(&leaked_chain), + "client response must not contain the anyhow error chain" ); } @@ -2904,7 +2953,11 @@ mod tests { assert_eq!(response.0, StatusCode::SERVICE_UNAVAILABLE); assert_eq!( response.1.message, - format!("All workers are busy, please retry later. {ADMISSION_CONTROL_REJECTION_HINT}") + format!("Service temporarily overloaded. {ADMISSION_CONTROL_REJECTION_HINT}") + ); + assert!( + !response.1.message.contains("All workers are busy"), + "client response must not include the underlying engine message" ); } @@ -2963,7 +3016,11 @@ mod tests { ); assert_eq!(response.1.code, 499); assert_eq!(response.1.error_type, "Client Closed Request"); - assert!(response.1.message.contains("stopped or killed")); + // The client gets a static message; the backend detail (context id, + // cancellation internals) must not leak into the 499 body. + assert_eq!(response.1.message, "Request cancelled"); + assert!(!response.1.message.contains("abc-123")); + assert!(!response.1.message.contains("stopped or killed")); } #[test] @@ -3595,7 +3652,14 @@ mod tests { assert!(result.is_err()); if let Err(error_response) = result { assert_eq!(error_response.0, StatusCode::INTERNAL_SERVER_ERROR); - assert_eq!(error_response.1.message, "Backend service unavailable"); + // Backend-supplied 5xx text must not be forwarded to the client. + assert_eq!(error_response.1.message, "Internal server error"); + assert!( + !error_response + .1 + .message + .contains("Backend service unavailable") + ); } } @@ -3622,8 +3686,42 @@ mod tests { assert!(result.is_err()); if let Err(error_response) = result { assert_eq!(error_response.0, StatusCode::INTERNAL_SERVER_ERROR); - assert_eq!(error_response.1.message, "prompt > max_seq_len"); + // 500 backend JSON messages are sanitized to a static client + // message; the raw payload is only logged server-side. + assert_eq!(error_response.1.message, "Internal server error"); + assert_eq!(error_response.1.code, 500); + assert!(!error_response.1.message.contains("prompt > max_seq_len")); + } + } + + #[tokio::test] + async fn test_check_for_backend_error_with_non_client_error_code() { + use crate::types::openai::chat_completions::NvCreateChatCompletionStreamResponse; + use futures::stream; + + // A backend asserting a non-4xx code (here 399) must not be able to + // smuggle a sensitive message through with a non-error status: + // anything outside the 4xx range is sanitized to 500. + let error_json = + r#"{"message":"panic at /srv/model.py:42","type":"Backend Error","code":399}"#; + let error_event = Annotated:: { + data: None, + id: None, + event: Some("error".to_string()), + comment: Some(vec![error_json.to_string()]), + error: None, + }; + + let test_stream = stream::iter(vec![error_event]); + let result = check_for_backend_error(test_stream).await; + + assert!(result.is_err()); + if let Err(error_response) = result { + assert_eq!(error_response.0, StatusCode::INTERNAL_SERVER_ERROR); assert_eq!(error_response.1.code, 500); + assert_eq!(error_response.1.message, "Internal server error"); + assert!(!error_response.1.message.contains("/srv/model.py")); + assert!(!error_response.1.message.contains("panic")); } } @@ -3708,7 +3806,10 @@ mod tests { assert!(result.is_err()); if let Err(error_response) = result { assert_eq!(error_response.0, StatusCode::INTERNAL_SERVER_ERROR); - assert_eq!(error_response.1.message, "Connection timeout"); + // Backend comment text falls under the 5xx default — must be + // sanitized so it cannot leak internals to the client. + assert_eq!(error_response.1.message, "Internal server error"); + assert!(!error_response.1.message.contains("Connection timeout")); } } diff --git a/lib/llm/src/http/service/realtime.rs b/lib/llm/src/http/service/realtime.rs index ce6f1b018881..1103df2ce892 100644 --- a/lib/llm/src/http/service/realtime.rs +++ b/lib/llm/src/http/service/realtime.rs @@ -171,12 +171,16 @@ async fn handle_socket(socket: WebSocket, state: Arc) { let event = if let Some(event) = annotated.data { event } else if let Some(err) = annotated.error { + // Log the engine error server-side; the client only receives a + // static message so the realtime stream can't leak internal paths, + // panic text, or backend exception details. + tracing::error!("/v1/realtime engine error: {err}"); RealtimeServerEvent::Error(RealtimeServerEventError { event_id: format!("event_{}", Uuid::new_v4()), error: RealtimeAPIError { r#type: "server_error".to_string(), code: None, - message: err.to_string(), + message: "Internal server error".to_string(), param: None, event_id: None, }, @@ -469,7 +473,7 @@ where send_error_event( ws_tx, "server_error", - &err.to_string(), + "Internal server error", Some("session.model"), ) .await; From 612c541a07a65019df310ce213d22b3aecb9b178 Mon Sep 17 00:00:00 2001 From: Pei Li Date: Fri, 29 May 2026 17:02:06 -0700 Subject: [PATCH 2/4] test: update python http-error test for sanitized responses The HTTP error sanitization on this branch replaces raw backend error text in client responses with a static "Internal server error". Update test_chat_completion_http_error, which previously asserted the raw HttpError / ValueError text was forwarded to the client, to assert the sanitized message instead and that the underlying backend detail is absent. The status-code expectations are unchanged. Refs: DYN-2636 Signed-off-by: Pei Li --- lib/bindings/python/tests/test_http_server.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/lib/bindings/python/tests/test_http_server.py b/lib/bindings/python/tests/test_http_server.py index 75acaca8c526..472d94b27c76 100644 --- a/lib/bindings/python/tests/test_http_server.py +++ b/lib/bindings/python/tests/test_http_server.py @@ -187,7 +187,11 @@ async def test_chat_completion_http_error(http_server, msg_to_code: tuple[str, i async with session.post(url, json=data) as response: assert response.status == msg_to_code[1] error_json = await response.json() + # Error responses are sanitized: the client receives a static + # message and never the raw backend error text. The underlying + # detail is logged server-side instead. + assert "internal server error" in str(error_json).lower() if msg_to_code[0] == MSG_CONTAINS_ERROR: - assert MSG_CONTAINS_ERROR in str(error_json) + assert MSG_CONTAINS_ERROR not in str(error_json) elif msg_to_code[0] == MSG_CONTAINS_INTERNAL_ERROR: - assert "simulated internal error" in str(error_json).lower() + assert "simulated internal error" not in str(error_json).lower() From 78fd46fc0a95aa8f180adf16d12d70070591a1f2 Mon Sep 17 00:00:00 2001 From: Pei Li Date: Mon, 1 Jun 2026 12:16:21 -0700 Subject: [PATCH 3/4] refactor(http): consolidate sanitized errors + close two leak gaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review feedback by routing every canonical sanitized error response through one type instead of duplicating literals, and close two leak paths flagged by coderabbitai that the earlier patch missed. http::service::error::SanitizedError now owns: - Cancelled -> 499, "Request cancelled" - Overloaded -> 503, "Service temporarily overloaded. " - Internal -> 500, "Internal server error" - PreserveServerError(c) -> c, "Internal server error" (5xx pass-through) Each variant exposes status(), Display, anthropic_type(), openai_type_slug(), and log_as_error(); helpers ErrorMessage::sanitized_with_details (openai.rs) and anthropic_sanitized_error_with_details (anthropic.rs) build the response from a variant and log the supplied details server-side. Migrated all canonical sanitized sites across openai.rs, anthropic.rs, disconnect.rs, and realtime.rs to use the enum. Additional fixes: - from_http_error previously forwarded HttpError { code: 499, message: ... } verbatim because 499 sits inside the 4xx range. Added a 499 carve-out that sanitizes via SanitizedError::Cancelled. - check_for_backend_error inspected only the first stream item. The request_id annotation prepended by chat_completions/responses shadowed any immediate backend error event, dropping intended 4xx responses to the generic fold/parse 500 path. Now buffers leading annotation frames (no data/error, event tag != "error") via the new is_annotation_frame helper and inspects the first non- annotation event instead. Buffered frames are replayed in their original order when no error is found. Tests: - test_from_http_error_sanitizes_499_message - test_check_for_backend_error_skips_leading_annotation_frames - test_check_for_backend_error_replays_leading_annotation_frames Cancellation paths keep tracing::debug! (Cancelled.log_as_error() == false) to avoid client-disconnect noise dominating error logs. Per-site sanitization comments removed throughout — the helper docstrings carry the rationale. DYN-2636 Signed-off-by: Pei Li --- lib/llm/src/http/service/anthropic.rs | 80 ++++--- lib/llm/src/http/service/disconnect.rs | 23 +- lib/llm/src/http/service/error.rs | 82 +++++++ lib/llm/src/http/service/openai.rs | 311 ++++++++++++++++++++----- lib/llm/src/http/service/realtime.rs | 10 +- 5 files changed, 402 insertions(+), 104 deletions(-) diff --git a/lib/llm/src/http/service/anthropic.rs b/lib/llm/src/http/service/anthropic.rs index 16bd35d192b7..85cf2a0d8489 100644 --- a/lib/llm/src/http/service/anthropic.rs +++ b/lib/llm/src/http/service/anthropic.rs @@ -50,6 +50,7 @@ use crate::request_template::{RequestTemplate, resolve_request_model}; use crate::types::Annotated; // Re-use helpers from the openai module (sibling under service/) +use super::error::SanitizedError; use super::metadata::extract_metadata_from_http; use super::openai::{get_body_limit, get_or_create_request_id}; @@ -178,11 +179,9 @@ async fn handler_anthropic_messages( tokio::spawn(anthropic_messages(state, template, request, stream_handle).in_current_span()) .await .map_err(|e| { - tracing::error!("Failed to await Anthropic messages task: {e:?}"); - anthropic_error( - StatusCode::INTERNAL_SERVER_ERROR, - "api_error", - "Internal server error", + anthropic_sanitized_error_with_details( + SanitizedError::Internal, + format!("Failed to await Anthropic messages task: {e:?}"), ) })?; @@ -334,19 +333,15 @@ async fn anthropic_messages( // Check for cancelled request (client disconnected before response was sent) if super::metrics::request_was_cancelled(e.as_ref()) { inflight_guard.mark_error(super::metrics::ErrorType::Cancelled); - tracing::debug!("Anthropic request cancelled before response: {e:#}"); - return anthropic_error( - StatusCode::from_u16(499).unwrap(), - "request_cancelled", - "Request cancelled", + return anthropic_sanitized_error_with_details( + SanitizedError::Cancelled, + format!("{e:#}"), ); } inflight_guard.mark_error(super::metrics::ErrorType::Internal); - tracing::error!("Failed to generate Anthropic completions: {e}"); - anthropic_error( - StatusCode::INTERNAL_SERVER_ERROR, - "api_error", - "Internal server error", + anthropic_sanitized_error_with_details( + SanitizedError::Internal, + format!("Failed to generate Anthropic completions: {e}"), ) })?; @@ -449,12 +444,20 @@ async fn anthropic_messages( // Check first event for backend errors using the openai helper let stream_with_check = super::openai::check_for_backend_error(engine_stream) .await - .map_err(|(status, json_err)| { - tracing::error!(request_id, %status, ?json_err, "Backend error detected"); - anthropic_error( - StatusCode::INTERNAL_SERVER_ERROR, - "api_error", - "Backend error during generation", + .map_err(|(status, _json_err)| { + // check_for_backend_error has already sanitized the body and + // logged the backend detail; we just preserve its status when + // re-wrapping in Anthropic format. + let err = if status.as_u16() == 499 { + SanitizedError::Cancelled + } else if status.is_server_error() { + SanitizedError::PreserveServerError(status) + } else { + SanitizedError::Internal + }; + anthropic_sanitized_error_with_details( + err, + format!("backend error event (status {})", status.as_u16()), ) })?; @@ -471,11 +474,9 @@ async fn anthropic_messages( NvCreateChatCompletionResponse::from_annotated_stream(stream, parsing_options.clone()) .await .map_err(|e| { - tracing::error!(request_id, "Failed to fold messages stream: {:?}", e); - anthropic_error( - StatusCode::INTERNAL_SERVER_ERROR, - "api_error", - "Internal server error", + anthropic_sanitized_error_with_details( + SanitizedError::Internal, + format!("Failed to fold messages stream: {e:?}"), ) })?; @@ -737,6 +738,33 @@ fn estimate_input_tokens(req: &AnthropicCreateMessageRequest) -> u32 { count_req.estimate_tokens() } +/// Build an Anthropic-formatted error response from a canonical +/// [`SanitizedError`] variant. The status, public message, and Anthropic +/// `error_type` all come from the variant; `details` are logged +/// server-side but never reach the client. +fn anthropic_sanitized_error_with_details( + err: SanitizedError, + details: impl std::fmt::Display, +) -> Response { + let status = err.status(); + if err.log_as_error() { + tracing::error!(status = %status, "Anthropic {err}: {details}"); + } else { + tracing::debug!(status = %status, "Anthropic {err}: {details}"); + } + ( + status, + Json(AnthropicErrorResponse { + object_type: "error".to_string(), + error: AnthropicErrorBody { + error_type: err.anthropic_type().to_string(), + message: err.to_string(), + }, + }), + ) + .into_response() +} + /// Build an Anthropic-formatted error response. /// Maps HTTP status codes to Anthropic error types following the Anthropic API spec. fn anthropic_error(status: StatusCode, error_type: &str, message: &str) -> Response { diff --git a/lib/llm/src/http/service/disconnect.rs b/lib/llm/src/http/service/disconnect.rs index 51fe213777c3..6153a7e7ff11 100644 --- a/lib/llm/src/http/service/disconnect.rs +++ b/lib/llm/src/http/service/disconnect.rs @@ -34,6 +34,7 @@ use futures::{Stream, StreamExt}; use std::sync::Arc; use std::time::Duration; +use crate::http::service::error::SanitizedError; use crate::http::service::metrics::{CancellationLabels, ErrorType, InflightGuard, Metrics}; use dynamo_runtime::config::environment_names::llm::DYN_HTTP_BACKEND_STREAM_TIMEOUT_SECS as BACKEND_STREAM_TIMEOUT_ENV; @@ -226,19 +227,17 @@ pub fn monitor_for_disconnects( // doesn't later record this as ClosedUnexpectedly (which // would mis-attribute the fault as a client disconnect). stream_handle.disarm(); - // Log the underlying error server-side; the client only - // receives a sanitized message so streaming responses don't - // leak internal paths, panic text, or backend exception - // details. tracing::error!("Streaming error: {err}"); // Emit a structured OpenAI-style error frame + `data: [DONE]` // so naive `data:`-line parsers see both the error and a - // stream terminator. + // stream terminator. Body derived from SanitizedError so + // the sanitized message + status live in one place. + let sanitized = SanitizedError::Internal; let err_json = serde_json::json!({ "error": { - "message": "Internal server error", - "type": "internal_server_error", - "code": 500, + "message": sanitized.to_string(), + "type": sanitized.openai_type_slug(), + "code": sanitized.status().as_u16(), } }); yield Event::default().data(err_json.to_string()); @@ -590,19 +589,21 @@ mod tests { .get("error") .and_then(|v| v.as_object()) .unwrap_or_else(|| panic!("[{case}] `error` field is not an object. Body:\n{text}")); + let expected = SanitizedError::Internal; + let expected_message = expected.to_string(); assert_eq!( error.get("message").and_then(|v| v.as_str()), - Some("Internal server error"), + Some(expected_message.as_str()), "[{case}] structured error `message` must be the sanitized static string. Body:\n{text}" ); assert_eq!( error.get("type").and_then(|v| v.as_str()), - Some("internal_server_error"), + Some(expected.openai_type_slug()), "[{case}] structured error `type` mismatch. Body:\n{text}" ); assert_eq!( error.get("code").and_then(|v| v.as_i64()), - Some(500), + Some(i64::from(expected.status().as_u16())), "[{case}] structured error `code` mismatch. Body:\n{text}" ); assert!( diff --git a/lib/llm/src/http/service/error.rs b/lib/llm/src/http/service/error.rs index b1d1c0ba8504..c6078d3574eb 100644 --- a/lib/llm/src/http/service/error.rs +++ b/lib/llm/src/http/service/error.rs @@ -1,6 +1,7 @@ // SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +use axum::http::StatusCode; use thiserror::Error; /// Implementation of the Completion Engines served by the HTTP service should @@ -12,3 +13,84 @@ pub struct HttpError { pub code: u16, pub message: String, } + +/// Admission-control rejection hint appended to overload responses. +pub const ADMISSION_CONTROL_REJECTION_HINT: &str = + "If this rejection is not intended, consider passing --admission-control none to the frontend."; + +/// Canonical sanitized error responses returned at the HTTP boundary. +/// +/// Each variant fixes the `(status, public message, protocol error_type)` +/// triple so call sites stop duplicating literals. The protocol-specific +/// mappings (OpenAI `error_type` string, Anthropic `error_type`) and the +/// `Display` impl that produces the user-safe message all live on this +/// enum — clients see exactly what the enum says, never a backend error +/// chain, file path, or panic stack. +#[derive(Debug, Clone, Copy)] +pub enum SanitizedError { + /// 499 Client Closed Request. + Cancelled, + /// 503 Service Unavailable, with admission-control hint. + Overloaded, + /// 500 Internal Server Error. + Internal, + /// Preserve a backend-reported 5xx status code while replacing the + /// body with the generic internal-error message. Clients still see + /// the original status (so 503 retry semantics survive); only the + /// payload is sanitized. + PreserveServerError(StatusCode), +} + +impl SanitizedError { + pub fn status(self) -> StatusCode { + match self { + // 499 is not IANA-registered but is widely used (nginx). + SanitizedError::Cancelled => StatusCode::from_u16(499).unwrap(), + SanitizedError::Overloaded => StatusCode::SERVICE_UNAVAILABLE, + SanitizedError::Internal => StatusCode::INTERNAL_SERVER_ERROR, + SanitizedError::PreserveServerError(code) => code, + } + } + + /// Anthropic `error.type` for this category. + pub fn anthropic_type(self) -> &'static str { + match self { + SanitizedError::Cancelled => "request_cancelled", + SanitizedError::Overloaded => "overloaded_error", + SanitizedError::Internal | SanitizedError::PreserveServerError(_) => "api_error", + } + } + + /// OpenAI-style snake_case `type` field used in inline error frames. + pub fn openai_type_slug(self) -> &'static str { + match self { + SanitizedError::Cancelled => "request_cancelled", + SanitizedError::Overloaded => "service_unavailable", + SanitizedError::Internal | SanitizedError::PreserveServerError(_) => { + "internal_server_error" + } + } + } + + /// Whether to log this category at `error!` (true) or `debug!` (false). + /// Cancellations are client-driven and routinely fire on disconnect, so + /// they stay at debug to avoid drowning real errors. + pub fn log_as_error(self) -> bool { + !matches!(self, SanitizedError::Cancelled) + } +} + +impl std::fmt::Display for SanitizedError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + SanitizedError::Cancelled => f.write_str("Request cancelled"), + SanitizedError::Overloaded => write!( + f, + "Service temporarily overloaded. {ADMISSION_CONTROL_REJECTION_HINT}" + ), + SanitizedError::Internal | SanitizedError::PreserveServerError(_) => { + f.write_str("Internal server error") + } + } + } +} diff --git a/lib/llm/src/http/service/openai.rs b/lib/llm/src/http/service/openai.rs index 0d3124e4ab58..d3661ed4fbf1 100644 --- a/lib/llm/src/http/service/openai.rs +++ b/lib/llm/src/http/service/openai.rs @@ -75,8 +75,8 @@ const X_REQUEST_ID_HEADER: &str = "x-request-id"; pub const ANNOTATION_REQUEST_ID: &str = "request_id"; const VALIDATION_PREFIX: &str = "Validation: "; -const ADMISSION_CONTROL_REJECTION_HINT: &str = - "If this rejection is not intended, consider passing --admission-control none to the frontend."; + +use super::error::SanitizedError; // Default axum max body limit without configuring is 2MB: https://docs.rs/axum/latest/axum/extract/struct.DefaultBodyLimit.html /// Default body limit in bytes (45MB) to support 500k+ token payloads. @@ -245,6 +245,31 @@ impl ErrorMessage { ) } + /// Build a sanitized error response from a [`SanitizedError`] variant. + /// The status, public message, and protocol error_type all come from + /// the variant — call sites do not pass any of them as literals. + /// Server-side `details` are logged alongside the canonical category; + /// the client only ever sees the variant's public message. + pub fn sanitized_with_details( + err: SanitizedError, + details: impl std::fmt::Display, + ) -> ErrorResponse { + let status = err.status(); + if err.log_as_error() { + tracing::error!(status = %status, "{err}: {details}"); + } else { + tracing::debug!(status = %status, "{err}: {details}"); + } + ( + status, + Json(ErrorMessage { + message: err.to_string(), + error_type: map_error_code_to_error_type(status), + code: status.as_u16(), + }), + ) + } + /// Not Implemented Error /// Return this error when the client requests a feature that is not yet implemented. /// This should be used for features that are planned but not available. @@ -282,18 +307,9 @@ impl ErrorMessage { pub fn from_anyhow(err: anyhow::Error, alt_msg: &str) -> ErrorResponse { // Check for ResourceExhausted anywhere in the error chain → HTTP 503 if super::metrics::request_was_rejected(err.as_ref()) { - // Log the full chain server-side; client only sees a sanitized hint - // so we don't leak internal paths or implementation details. - tracing::warn!("Request rejected by admission control: {err:#}"); - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ErrorMessage { - message: format!( - "Service temporarily overloaded. {ADMISSION_CONTROL_REJECTION_HINT}" - ), - error_type: map_error_code_to_error_type(StatusCode::SERVICE_UNAVAILABLE), - code: StatusCode::SERVICE_UNAVAILABLE.as_u16(), - }), + return ErrorMessage::sanitized_with_details( + SanitizedError::Overloaded, + format!("{err:#}"), ); } @@ -318,26 +334,15 @@ impl ErrorMessage { // Check for Cancelled anywhere in the error chain → HTTP 499 (Client Closed Request) if super::metrics::request_was_cancelled(err.as_ref()) { - let code = StatusCode::from_u16(499).unwrap(); - // Log the detail server-side; the client gets only a static message so a - // 499 surfaced before the connection actually drops can't leak context - // IDs or backend cancellation internals. - tracing::debug!("Request cancelled before response: {err:#}"); - return ( - code, - Json(ErrorMessage { - message: "Request cancelled".to_string(), - error_type: map_error_code_to_error_type(code), - code: code.as_u16(), - }), + return ErrorMessage::sanitized_with_details( + SanitizedError::Cancelled, + format!("{err:#}"), ); } // Then check for HttpError match err.downcast::() { Ok(http_error) => ErrorMessage::from_http_error(http_error), - // Log the full anyhow chain server-side but only return `alt_msg` - // to the client so we don't leak internal paths. Err(err) => { ErrorMessage::internal_server_error_with_details(alt_msg, format!("{err:#}")) } @@ -346,14 +351,16 @@ impl ErrorMessage { /// Implementers should only be able to throw 400-499 errors. pub fn from_http_error(err: HttpError) -> ErrorResponse { + // 499 is part of the 4xx range but its body can carry cancellation + // context (queue paths, context IDs) — sanitize separately. + if err.code == 499 { + return ErrorMessage::sanitized_with_details(SanitizedError::Cancelled, err.message); + } // Backend-supplied messages are only forwarded for the documented 4xx // range; for 5xx or codes outside the HTTP space the message may // contain internal paths/details and is kept server-side only. if err.code < 400 || err.code >= 500 { - return ErrorMessage::internal_server_error_with_details( - "Internal server error", - err.message, - ); + return ErrorMessage::sanitized_with_details(SanitizedError::Internal, err.message); } match StatusCode::from_u16(err.code) { Ok(code) => ( @@ -364,10 +371,7 @@ impl ErrorMessage { code: code.as_u16(), }), ), - Err(_) => ErrorMessage::internal_server_error_with_details( - "Internal server error", - err.message, - ), + Err(_) => ErrorMessage::sanitized_with_details(SanitizedError::Internal, err.message), } } } @@ -551,8 +555,6 @@ async fn handler_completions( let response = tokio::spawn(completions(state, request, stream_handle).in_current_span()) .await .map_err(|e| { - // JoinError debug output can include panic messages with stack - // info — keep that server-side only. ErrorMessage::internal_server_error_with_details( "Failed to await chat completions task", format!("{e:?}"), @@ -1071,8 +1073,6 @@ async fn handler_chat_completions( tokio::spawn(chat_completions(state, template, request, stream_handle).in_current_span()) .await .map_err(|e| { - // JoinError debug output can include panic messages with stack - // info — keep that server-side only. ErrorMessage::internal_server_error_with_details( "Failed to await chat completions task", format!("{e:?}"), @@ -1177,8 +1177,25 @@ fn extract_backend_error_if_present( None } -/// Checks if the first event in the stream is a backend error. -/// Returns Err(ErrorResponse) if error detected, Ok(stream) otherwise. +/// Returns true for events that only carry an annotation tag (e.g. the +/// `request_id` frame prepended to every stream): no data, no error, and +/// an `event` field that is *not* the `"error"` marker. Annotations may +/// still carry a serialized value in `comment` (that is how +/// `Annotated::from_annotation` builds them), so the comment field is +/// not part of the check. These frames are stepped over by +/// `check_for_backend_error` so an immediate backend error in the *next* +/// slot is still caught instead of slipping through to the fold/parse +/// path. +fn is_annotation_frame(e: &Annotated) -> bool { + e.data.is_none() + && e.error.is_none() + && matches!(e.event.as_deref(), Some(tag) if tag != "error") +} + +/// Inspect the first non-annotation event in the stream for a backend error. +/// Returns Err(ErrorResponse) if error detected, Ok(stream) otherwise — the +/// returned stream replays any buffered annotation frames in their original +/// order before yielding the remaining items. pub(super) async fn check_for_backend_error( mut stream: impl futures::Stream> + Send @@ -1190,15 +1207,22 @@ pub(super) async fn check_for_backend_error( > { use futures::stream::StreamExt; - // Peek at the first event - if let Some(first_event) = stream.next().await { - // Check if it's an error event - if let Some((error_msg, status_code)) = extract_backend_error_if_present(&first_event) { - // Mirror from_http_error: only the documented 4xx range is part - // of the protocol contract and forwarded unchanged. Anything - // else (5xx, or a 1xx/2xx/3xx asserted by the backend payload) - // may carry panic text, file paths, or raw internal error JSON - // and is kept server-side only. + let mut buffered: Vec> = Vec::new(); + while let Some(event) = stream.next().await { + if is_annotation_frame(&event) { + buffered.push(event); + continue; + } + if let Some((error_msg, status_code)) = extract_backend_error_if_present(&event) { + // 499 is part of is_client_error() but its text can carry + // context IDs / cancellation internals — sanitize separately. + if status_code.as_u16() == 499 { + return Err(ErrorMessage::sanitized_with_details( + SanitizedError::Cancelled, + error_msg, + )); + } + // 4xx is the protocol contract — forward backend message as-is. if status_code.is_client_error() { return Err(( status_code, @@ -1209,19 +1233,27 @@ pub(super) async fn check_for_backend_error( }), )); } - return Err(ErrorMessage::internal_server_error_with_details( - "Internal server error", + // 5xx — preserve the status (clients distinguish 503 retry from 500) + // but replace the body with a sanitized message. + if status_code.is_server_error() { + return Err(ErrorMessage::sanitized_with_details( + SanitizedError::PreserveServerError(status_code), + error_msg, + )); + } + // 1xx/2xx/3xx asserted by the backend payload — coerce to 500. + return Err(ErrorMessage::sanitized_with_details( + SanitizedError::Internal, error_msg, )); } - // Not an error - reconstruct stream with first event - let reconstructed_stream = futures::stream::iter(vec![first_event]).chain(stream); - Ok(reconstructed_stream) - } else { - // Empty stream - this shouldn't happen but handle gracefully - Ok(futures::stream::iter(vec![]).chain(stream)) + // First non-annotation, non-error event — push it back and stop; + // downstream consumers see the original ordering. + buffered.push(event); + break; } + Ok(futures::stream::iter(buffered).chain(stream)) } /// Serialize `payload` and wrap it as an SSE event with the given name. @@ -1758,8 +1790,6 @@ async fn handler_responses( tokio::spawn(responses(state, template, request, stream_handle).in_current_span()) .await .map_err(|e| { - // JoinError debug output can include panic messages with stack - // info — keep that server-side only. ErrorMessage::internal_server_error_with_details( "Failed to await responses task", format!("{e:?}"), @@ -2861,6 +2891,7 @@ mod tests { use super::*; use crate::discovery::ModelManagerError; + use crate::http::service::error::ADMISSION_CONTROL_REJECTION_HINT; use crate::protocols::openai::chat_completions::NvCreateChatCompletionRequest; use crate::protocols::openai::common_ext::CommonExt; use crate::protocols::openai::completions::NvCreateCompletionRequest; @@ -2920,6 +2951,22 @@ mod tests { } } + #[test] + fn test_from_http_error_sanitizes_499_message() { + // Backend may construct HttpError { code: 499, message: "..." }; that + // message can carry context IDs / queue paths and must not leak. + let err = HttpError { + code: 499, + message: "session abc-123 cancelled at /srv/queue.py:42".to_string(), + }; + let response = ErrorMessage::from_http_error(err); + assert_eq!(response.0.as_u16(), 499); + assert_eq!(response.1.code, 499); + assert_eq!(response.1.message, "Request cancelled"); + assert!(!response.1.message.contains("abc-123")); + assert!(!response.1.message.contains("/srv/queue.py")); + } + #[test] fn test_other_error_response_from_anyhow() { // Non-HttpError anyhow chains must NOT be exposed to the client; only @@ -3725,6 +3772,148 @@ mod tests { } } + #[tokio::test] + async fn test_check_for_backend_error_with_503_preserves_status() { + use crate::types::openai::chat_completions::NvCreateChatCompletionStreamResponse; + use futures::stream; + + // Backend 5xx status codes must round-trip so clients can distinguish + // retryable overload (503) from generic 500; only the body is sanitized. + let error_json = r#"{"message":"engine pool exhausted at /srv/engine.py:88","code":503}"#; + let error_event = Annotated:: { + data: None, + id: None, + event: Some("error".to_string()), + comment: Some(vec![error_json.to_string()]), + error: None, + }; + + let test_stream = stream::iter(vec![error_event]); + let result = check_for_backend_error(test_stream).await; + + assert!(result.is_err()); + if let Err(error_response) = result { + assert_eq!(error_response.0, StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(error_response.1.code, 503); + assert_eq!(error_response.1.message, "Internal server error"); + assert!(!error_response.1.message.contains("engine pool")); + assert!(!error_response.1.message.contains("/srv/engine.py")); + } + } + + #[tokio::test] + async fn test_check_for_backend_error_with_499_sanitizes_cancellation() { + use crate::types::openai::chat_completions::NvCreateChatCompletionStreamResponse; + use futures::stream; + + // 499 falls inside is_client_error(); ensure cancellation text from + // the backend (e.g. context IDs) cannot reach the client. + let error_json = + r#"{"message":"Context id abc-123 cancelled at /srv/queue.py:42","code":499}"#; + let error_event = Annotated:: { + data: None, + id: None, + event: Some("error".to_string()), + comment: Some(vec![error_json.to_string()]), + error: None, + }; + + let test_stream = stream::iter(vec![error_event]); + let result = check_for_backend_error(test_stream).await; + + assert!(result.is_err()); + if let Err(error_response) = result { + assert_eq!(error_response.0.as_u16(), 499); + assert_eq!(error_response.1.code, 499); + assert_eq!(error_response.1.message, "Request cancelled"); + assert!(!error_response.1.message.contains("abc-123")); + assert!(!error_response.1.message.contains("/srv/queue.py")); + } + } + + #[tokio::test] + async fn test_check_for_backend_error_skips_leading_annotation_frames() { + use crate::types::openai::chat_completions::NvCreateChatCompletionStreamResponse; + use futures::stream; + + // Streams prepend a request_id annotation before forwarding engine + // events. An immediate backend error in the next slot must still be + // caught so a 4xx surfaces as a 4xx instead of falling through to + // the generic fold/parse 500. + let annotation = Annotated::::from_annotation( + ANNOTATION_REQUEST_ID, + &"req-123".to_string(), + ) + .expect("annotation construction should succeed"); + let error_event = Annotated:: { + data: None, + id: None, + event: Some("error".to_string()), + comment: Some(vec![ + r#"{"message":"bad input from client","code":400}"#.to_string(), + ]), + error: None, + }; + + let test_stream = stream::iter(vec![annotation, error_event]); + let result = check_for_backend_error(test_stream).await; + + assert!( + result.is_err(), + "annotation followed by an error event must still be detected as an error" + ); + if let Err(error_response) = result { + assert_eq!(error_response.0, StatusCode::BAD_REQUEST); + assert_eq!(error_response.1.code, 400); + assert_eq!(error_response.1.message, "bad input from client"); + } + } + + #[tokio::test] + async fn test_check_for_backend_error_replays_leading_annotation_frames() { + use crate::types::openai::chat_completions::NvCreateChatCompletionStreamResponse; + use dynamo_protocols::types::CreateChatCompletionStreamResponse; + use futures::stream::{self, StreamExt}; + + // A leading annotation followed by a normal data event must yield + // a stream that replays both, in their original order. + let annotation = Annotated::::from_annotation( + ANNOTATION_REQUEST_ID, + &"req-123".to_string(), + ) + .expect("annotation construction should succeed"); + let normal_event = Annotated:: { + data: Some(NvCreateChatCompletionStreamResponse { + inner: CreateChatCompletionStreamResponse { + id: "test-id".to_string(), + choices: vec![], + created: 0, + model: "test-model".to_string(), + system_fingerprint: None, + object: "chat.completion.chunk".to_string(), + service_tier: None, + usage: None, + }, + nvext: None, + }), + id: Some("msg-1".to_string()), + event: None, + comment: None, + error: None, + }; + + let test_stream = stream::iter(vec![annotation, normal_event]); + let result = check_for_backend_error(test_stream).await; + + assert!(result.is_ok()); + let mut returned: Vec<_> = result.unwrap().collect().await; + assert_eq!(returned.len(), 2, "annotation + data event must replay"); + let first = returned.remove(0); + assert_eq!(first.event.as_deref(), Some(ANNOTATION_REQUEST_ID)); + let second = returned.remove(0); + assert_eq!(second.id, Some("msg-1".to_string())); + } + #[tokio::test] async fn test_check_for_backend_error_with_normal_event() { use crate::types::openai::chat_completions::NvCreateChatCompletionStreamResponse; diff --git a/lib/llm/src/http/service/realtime.rs b/lib/llm/src/http/service/realtime.rs index 1103df2ce892..19d55b8c735e 100644 --- a/lib/llm/src/http/service/realtime.rs +++ b/lib/llm/src/http/service/realtime.rs @@ -59,7 +59,7 @@ const REQUEST_CHANNEL_CAPACITY: usize = 64; /// half-broken peer from parking the connection indefinitely. const CLOSE_DRAIN_TIMEOUT: Duration = Duration::from_secs(5); -use super::{RouteDoc, service_v2}; +use super::{RouteDoc, error::SanitizedError, service_v2}; use crate::discovery::ModelManagerError; use crate::types::RealtimeBidirectionalEngine; use dynamo_protocols::types::realtime::{ @@ -171,16 +171,14 @@ async fn handle_socket(socket: WebSocket, state: Arc) { let event = if let Some(event) = annotated.data { event } else if let Some(err) = annotated.error { - // Log the engine error server-side; the client only receives a - // static message so the realtime stream can't leak internal paths, - // panic text, or backend exception details. tracing::error!("/v1/realtime engine error: {err}"); + let sanitized = SanitizedError::Internal; RealtimeServerEvent::Error(RealtimeServerEventError { event_id: format!("event_{}", Uuid::new_v4()), error: RealtimeAPIError { r#type: "server_error".to_string(), code: None, - message: "Internal server error".to_string(), + message: sanitized.to_string(), param: None, event_id: None, }, @@ -473,7 +471,7 @@ where send_error_event( ws_tx, "server_error", - "Internal server error", + &SanitizedError::Internal.to_string(), Some("session.model"), ) .await; From f0d313943279759fe0bd8b76274dd6efcd862d46 Mon Sep 17 00:00:00 2001 From: Pei Li Date: Mon, 8 Jun 2026 18:39:08 -0700 Subject: [PATCH 4/4] fix(http): address review + self-review on sanitized error edges MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reviewer feedback (vvagias): - realtime.rs handle_socket: engine.generate() failures put the raw backend error into the WebSocket close reason; startup failures could leak internals. Use SanitizedError::Internal.to_string() for the close reason; the original err is still tracing::error!-logged. - anthropic.rs check_for_backend_error rewrap: non-499 4xx errors were collapsed into HTTP 500. Preserve client-error statuses via anthropic_error (whose existing table maps 401/403/404/429 to spec-correct types). Self-review: - SanitizedError::PreserveServerError(503).anthropic_type() returned "api_error" instead of "overloaded_error" (same for 529). Both anthropic_type() and openai_type_slug() now consult the inner status of PreserveServerError so the spec-correct overload type surfaces. Added unit tests in error.rs. - check_for_backend_error buffered leading annotation frames in an unbounded Vec. A backend that emits only annotations could pin per-request memory. Cap at MAX_LEADING_ANNOTATIONS = 16; past the cap the loop falls through. - PreserveServerError(StatusCode) had no enforcement of its 5xx invariant. Added debug_assert in status() and documented the invariant; construct via SanitizedError::for_backend_status to enforce by construction. - The four-way status-classification ladder (499 / 4xx / 5xx / else) was duplicated in check_for_backend_error (openai.rs) and the anthropic.rs rewrap. Both now go through the new SanitizedError::for_backend_status helper — one source of truth. DYN-2636 Signed-off-by: Pei Li --- lib/llm/src/http/service/anthropic.rs | 35 +++++---- lib/llm/src/http/service/error.rs | 105 ++++++++++++++++++++++++-- lib/llm/src/http/service/openai.rs | 42 ++++------- lib/llm/src/http/service/realtime.rs | 2 +- 4 files changed, 137 insertions(+), 47 deletions(-) diff --git a/lib/llm/src/http/service/anthropic.rs b/lib/llm/src/http/service/anthropic.rs index 85cf2a0d8489..697f2a5c649d 100644 --- a/lib/llm/src/http/service/anthropic.rs +++ b/lib/llm/src/http/service/anthropic.rs @@ -446,19 +446,28 @@ async fn anthropic_messages( .await .map_err(|(status, _json_err)| { // check_for_backend_error has already sanitized the body and - // logged the backend detail; we just preserve its status when - // re-wrapping in Anthropic format. - let err = if status.as_u16() == 499 { - SanitizedError::Cancelled - } else if status.is_server_error() { - SanitizedError::PreserveServerError(status) - } else { - SanitizedError::Internal - }; - anthropic_sanitized_error_with_details( - err, - format!("backend error event (status {})", status.as_u16()), - ) + // logged the backend detail; preserve its status when + // re-wrapping in Anthropic format. Status classification is + // delegated to SanitizedError::for_backend_status so the + // openai and anthropic surfaces stay aligned. + let details = format!("backend error event (status {})", status.as_u16()); + match SanitizedError::for_backend_status(status) { + Some(variant) => anthropic_sanitized_error_with_details(variant, details), + // 4xx (non-499): preserve the client-error status; the + // message is the canonical reason so we don't smuggle + // backend text through. The "invalid_request_error" + // argument is a fallback — anthropic_error remaps + // 401/403/404/429 to their spec-correct types from the + // status code itself. + None => { + tracing::error!(%status, "Anthropic backend error event"); + anthropic_error( + status, + "invalid_request_error", + status.canonical_reason().unwrap_or("Client error"), + ) + } + } })?; let mut http_queue_guard = Some(http_queue_guard); diff --git a/lib/llm/src/http/service/error.rs b/lib/llm/src/http/service/error.rs index c6078d3574eb..446d43ab4c16 100644 --- a/lib/llm/src/http/service/error.rs +++ b/lib/llm/src/http/service/error.rs @@ -38,26 +38,64 @@ pub enum SanitizedError { /// body with the generic internal-error message. Clients still see /// the original status (so 503 retry semantics survive); only the /// payload is sanitized. + /// + /// Invariant: the inner status MUST be in the 500–599 range. Construct + /// via [`SanitizedError::for_backend_status`] to enforce this. PreserveServerError(StatusCode), } impl SanitizedError { + /// Classify a backend-supplied HTTP status into the right sanitized + /// variant. Returns `None` to mean "forward this 4xx (non-499) + /// message as-is" — that case is the protocol contract for client + /// errors and is the caller's responsibility to handle. + /// + /// The single source of truth for the status → variant mapping; + /// every site that triages a backend status code should call this + /// instead of inlining the if-chain. + pub fn for_backend_status(status: StatusCode) -> Option { + if status.as_u16() == 499 { + Some(SanitizedError::Cancelled) + } else if status.is_client_error() { + // 4xx (non-499) is the protocol contract; caller forwards. + None + } else if status.is_server_error() { + Some(SanitizedError::PreserveServerError(status)) + } else { + // 1xx/2xx/3xx asserted by a backend payload — coerce to 500. + Some(SanitizedError::Internal) + } + } + pub fn status(self) -> StatusCode { match self { // 499 is not IANA-registered but is widely used (nginx). SanitizedError::Cancelled => StatusCode::from_u16(499).unwrap(), SanitizedError::Overloaded => StatusCode::SERVICE_UNAVAILABLE, SanitizedError::Internal => StatusCode::INTERNAL_SERVER_ERROR, - SanitizedError::PreserveServerError(code) => code, + SanitizedError::PreserveServerError(code) => { + debug_assert!( + code.is_server_error(), + "PreserveServerError requires a 5xx status; got {code}" + ); + code + } } } - /// Anthropic `error.type` for this category. + /// Anthropic `error.type` for this category. For `PreserveServerError` + /// the inner status is consulted so a backend 503/529 is reported as + /// `overloaded_error` (matching the Anthropic spec) rather than the + /// generic `api_error`. pub fn anthropic_type(self) -> &'static str { match self { SanitizedError::Cancelled => "request_cancelled", SanitizedError::Overloaded => "overloaded_error", - SanitizedError::Internal | SanitizedError::PreserveServerError(_) => "api_error", + SanitizedError::Internal => "api_error", + SanitizedError::PreserveServerError(status) => match status.as_u16() { + 503 | 529 => "overloaded_error", + _ => "api_error", + }, } } @@ -66,9 +104,11 @@ impl SanitizedError { match self { SanitizedError::Cancelled => "request_cancelled", SanitizedError::Overloaded => "service_unavailable", - SanitizedError::Internal | SanitizedError::PreserveServerError(_) => { - "internal_server_error" - } + SanitizedError::Internal => "internal_server_error", + SanitizedError::PreserveServerError(status) => match status.as_u16() { + 503 | 529 => "service_unavailable", + _ => "internal_server_error", + }, } } @@ -94,3 +134,56 @@ impl std::fmt::Display for SanitizedError { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn preserve_server_error_503_maps_to_overload_types() { + // Backend-asserted 503 must surface as the spec-correct overload + // type on both protocols, not as a generic api_error / + // internal_server_error. + let err = SanitizedError::PreserveServerError(StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(err.anthropic_type(), "overloaded_error"); + assert_eq!(err.openai_type_slug(), "service_unavailable"); + } + + #[test] + fn preserve_server_error_529_maps_to_overload_types() { + // Anthropic uses 529 as an alternative overload signal; mirror + // the 503 mapping so clients can apply the same backoff. + let err = SanitizedError::PreserveServerError(StatusCode::from_u16(529).unwrap()); + assert_eq!(err.anthropic_type(), "overloaded_error"); + assert_eq!(err.openai_type_slug(), "service_unavailable"); + } + + #[test] + fn preserve_server_error_500_remains_generic() { + let err = SanitizedError::PreserveServerError(StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!(err.anthropic_type(), "api_error"); + assert_eq!(err.openai_type_slug(), "internal_server_error"); + } + + #[test] + fn for_backend_status_classifies_correctly() { + // 499 → Cancelled + assert!(matches!( + SanitizedError::for_backend_status(StatusCode::from_u16(499).unwrap()), + Some(SanitizedError::Cancelled) + )); + // 5xx → PreserveServerError preserving the code + assert!(matches!( + SanitizedError::for_backend_status(StatusCode::SERVICE_UNAVAILABLE), + Some(SanitizedError::PreserveServerError(s)) if s == StatusCode::SERVICE_UNAVAILABLE + )); + // Non-499 4xx → None (forward as-is) + assert!(SanitizedError::for_backend_status(StatusCode::BAD_REQUEST).is_none()); + assert!(SanitizedError::for_backend_status(StatusCode::NOT_FOUND).is_none()); + // 1xx/2xx/3xx asserted by backend → Internal + assert!(matches!( + SanitizedError::for_backend_status(StatusCode::from_u16(399).unwrap()), + Some(SanitizedError::Internal) + )); + } +} diff --git a/lib/llm/src/http/service/openai.rs b/lib/llm/src/http/service/openai.rs index 5936ed8b4dc5..59090984a940 100644 --- a/lib/llm/src/http/service/openai.rs +++ b/lib/llm/src/http/service/openai.rs @@ -1278,6 +1278,14 @@ fn is_annotation_frame(e: &Annotated) -> bool { && matches!(e.event.as_deref(), Some(tag) if tag != "error") } +/// Cap on how many leading annotation frames `check_for_backend_error` +/// will buffer before giving up the inspection. A pathological backend +/// (or attacker who can influence the engine output) that emits only +/// annotation frames must not be able to pin unbounded memory per +/// request. The handful of real annotations a frontend prepends +/// (currently just `request_id`) fits well under this cap. +const MAX_LEADING_ANNOTATIONS: usize = 16; + /// Inspect the first non-annotation event in the stream for a backend error. /// Returns Err(ErrorResponse) if error detected, Ok(stream) otherwise — the /// returned stream replays any buffered annotation frames in their original @@ -1295,43 +1303,23 @@ pub(super) async fn check_for_backend_error( let mut buffered: Vec> = Vec::new(); while let Some(event) = stream.next().await { - if is_annotation_frame(&event) { + if is_annotation_frame(&event) && buffered.len() < MAX_LEADING_ANNOTATIONS { buffered.push(event); continue; } if let Some((error_msg, status_code)) = extract_backend_error_if_present(&event) { - // 499 is part of is_client_error() but its text can carry - // context IDs / cancellation internals — sanitize separately. - if status_code.as_u16() == 499 { - return Err(ErrorMessage::sanitized_with_details( - SanitizedError::Cancelled, - error_msg, - )); - } - // 4xx is the protocol contract — forward backend message as-is. - if status_code.is_client_error() { - return Err(( + return Err(match SanitizedError::for_backend_status(status_code) { + Some(variant) => ErrorMessage::sanitized_with_details(variant, error_msg), + // 4xx (non-499): protocol contract — forward backend message as-is. + None => ( status_code, Json(ErrorMessage { message: error_msg, error_type: map_error_code_to_error_type(status_code), code: status_code.as_u16(), }), - )); - } - // 5xx — preserve the status (clients distinguish 503 retry from 500) - // but replace the body with a sanitized message. - if status_code.is_server_error() { - return Err(ErrorMessage::sanitized_with_details( - SanitizedError::PreserveServerError(status_code), - error_msg, - )); - } - // 1xx/2xx/3xx asserted by the backend payload — coerce to 500. - return Err(ErrorMessage::sanitized_with_details( - SanitizedError::Internal, - error_msg, - )); + ), + }); } // First non-annotation, non-error event — push it back and stop; diff --git a/lib/llm/src/http/service/realtime.rs b/lib/llm/src/http/service/realtime.rs index 3ebb1e4dba93..886daccb7825 100644 --- a/lib/llm/src/http/service/realtime.rs +++ b/lib/llm/src/http/service/realtime.rs @@ -153,7 +153,7 @@ async fn handle_socket(socket: WebSocket, state: Arc) { tracing::error!(%err, "/v1/realtime engine.generate() failed"); *close_reason.lock() = Some(close_message( close_code::ERROR, - &format!("engine error: {err}"), + &SanitizedError::Internal.to_string(), )); return; }