diff --git a/lib/bindings/python/tests/test_http_server.py b/lib/bindings/python/tests/test_http_server.py index 15c9acedb38e..e048622625a1 100644 --- a/lib/bindings/python/tests/test_http_server.py +++ b/lib/bindings/python/tests/test_http_server.py @@ -199,8 +199,15 @@ async def test_chat_completion_http_error(http_server, msg_to_code: tuple[str, i assert response.status == msg_to_code[1] error_json = await response.json() if msg_to_code[0] == MSG_CONTAINS_ERROR: + # 4xx HTTP protocol contract: backend message is forwarded + # to the client so callers can react to validation errors. assert MSG_CONTAINS_ERROR in str(error_json) elif msg_to_code[0] == MSG_CONTAINS_STATUS_ERROR: + # Same 4xx contract via the duck-typed `.status` path. assert MSG_CONTAINS_STATUS_ERROR in str(error_json) elif msg_to_code[0] == MSG_CONTAINS_INTERNAL_ERROR: - assert "simulated internal error" in str(error_json).lower() + # 5xx is sanitized: the client receives a static message + # and never the raw backend error text; the underlying + # detail is logged server-side. + assert "simulated internal error" not in str(error_json).lower() + assert "internal server error" in str(error_json).lower() diff --git a/lib/llm/src/http/service/anthropic.rs b/lib/llm/src/http/service/anthropic.rs index 1a977fe54050..697f2a5c649d 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,10 +179,9 @@ async fn handler_anthropic_messages( tokio::spawn(anthropic_messages(state, template, request, stream_handle).in_current_span()) .await .map_err(|e| { - anthropic_error( - StatusCode::INTERNAL_SERVER_ERROR, - "api_error", - &format!("Failed to await messages task: {:?}", e), + anthropic_sanitized_error_with_details( + SanitizedError::Internal, + format!("Failed to await Anthropic messages task: {e:?}"), ) })?; @@ -333,17 +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); - return anthropic_error( - StatusCode::from_u16(499).unwrap(), - "request_cancelled", - &format!("Request cancelled: {}", e), + return anthropic_sanitized_error_with_details( + SanitizedError::Cancelled, + format!("{e:#}"), ); } inflight_guard.mark_error(super::metrics::ErrorType::Internal); - anthropic_error( - StatusCode::INTERNAL_SERVER_ERROR, - "api_error", - &format!("Failed to generate completions: {}", e), + anthropic_sanitized_error_with_details( + SanitizedError::Internal, + format!("Failed to generate Anthropic completions: {e}"), ) })?; @@ -446,13 +444,30 @@ 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; 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); @@ -468,11 +483,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", - &format!("Failed to fold messages stream: {}", e), + anthropic_sanitized_error_with_details( + SanitizedError::Internal, + format!("Failed to fold messages stream: {e:?}"), ) })?; @@ -734,6 +747,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 4196d72028f8..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,13 +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(); - // DIS-1768: emit structured OpenAI-style error frame + `data: [DONE]` - // so naive `data:`-line parsers see both the error and a stream terminator. + 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. 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": err.to_string(), - "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()); @@ -508,7 +513,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 +555,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}") }); @@ -582,25 +589,32 @@ 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(expected_message), - "[{case}] structured error `message` mismatch. Body:\n{text}" + 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!( !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 +623,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 +637,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/error.rs b/lib/llm/src/http/service/error.rs index b1d1c0ba8504..830dc55dfc23 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,170 @@ pub struct HttpError { pub code: u16, pub message: String, } + +/// 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. + 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. + /// + /// 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) => { + debug_assert!( + code.is_server_error(), + "PreserveServerError requires a 5xx status; got {code}" + ); + code + } + } + } + + /// 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 => "api_error", + SanitizedError::PreserveServerError(status) => match status.as_u16() { + 503 | 529 => "overloaded_error", + _ => "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 => "internal_server_error", + SanitizedError::PreserveServerError(status) => match status.as_u16() { + 503 | 529 => "service_unavailable", + _ => "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 => f.write_str("Service temporarily overloaded"), + SanitizedError::Internal | SanitizedError::PreserveServerError(_) => { + f.write_str("Internal server error") + } + } + } +} + +#[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 21054e025604..dd603cacf469 100644 --- a/lib/llm/src/http/service/openai.rs +++ b/lib/llm/src/http/service/openai.rs @@ -76,8 +76,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. @@ -137,22 +137,6 @@ fn extract_error_type_from_response(response: &ErrorResponse) -> ErrorType { classify_error_for_metrics(response.0, &response.1.message) } -fn find_dynamo_error_in_chain<'a>( - err: &'a (dyn std::error::Error + 'static), - error_type: dynamo_runtime::error::ErrorType, -) -> Option<&'a dynamo_runtime::error::DynamoError> { - let mut current = Some(err); - while let Some(e) = current { - if let Some(dynamo_err) = e.downcast_ref::() - && dynamo_err.error_type() == error_type - { - return Some(dynamo_err); - } - current = e.source(); - } - None -} - /// Match `InvalidArgument` at top-level OR under `Backend()`. /// `py_err_to_dynamo` wraps Python `ValueError`/`TypeError` as /// `Backend(InvalidArgument)`; both variants are 400-worthy. @@ -262,6 +246,53 @@ 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(), + }), + ) + } + + /// 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. @@ -299,20 +330,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()) { - 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()); - - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ErrorMessage { - message: format!("{}. {}", message, 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:#}"), ); } @@ -330,29 +350,33 @@ 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}"); - return ( - code, - Json(ErrorMessage { - message: err.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), - Err(err) => ErrorMessage::internal_server_error(&format!("{alt_msg}: {err:#}")), + 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 { + // 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(&err.message); + return ErrorMessage::sanitized_with_details(SanitizedError::Internal, err.message); } match StatusCode::from_u16(err.code) { Ok(code) => ( @@ -363,7 +387,7 @@ impl ErrorMessage { code: code.as_u16(), }), ), - Err(_) => ErrorMessage::internal_server_error(&err.message), + Err(_) => ErrorMessage::sanitized_with_details(SanitizedError::Internal, err.message), } } } @@ -548,10 +572,10 @@ 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, - )) + 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 @@ -725,8 +749,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 @@ -920,8 +943,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 @@ -1138,10 +1160,10 @@ 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, - )) + 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 @@ -1242,8 +1264,33 @@ 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") +} + +/// 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 +/// order before yielding the remaining items. pub(super) async fn check_for_backend_error( mut stream: impl futures::Stream> + Send @@ -1255,27 +1302,33 @@ 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) { - return Err(( - status_code, - Json(ErrorMessage { - message: error_msg, - error_type: map_error_code_to_error_type(status_code), - code: status_code.as_u16(), - }), - )); + let mut buffered: Vec> = Vec::new(); + while let Some(event) = stream.next().await { + 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) { + 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(), + }), + ), + }); } - // 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)) } #[derive(Serialize)] @@ -1677,10 +1730,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 })?; @@ -1858,10 +1910,10 @@ 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, - )) + 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 @@ -2136,10 +2188,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 })?; @@ -2873,7 +2923,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}"), + ) }) } @@ -3066,34 +3119,48 @@ 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"); + // 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" + ); + } + } - 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"); + #[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 + // 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" ); } @@ -3113,9 +3180,10 @@ mod tests { .into(); let response = ErrorMessage::from_anyhow(err, BACKUP_ERROR_MESSAGE); assert_eq!(response.0, StatusCode::SERVICE_UNAVAILABLE); - assert_eq!( - response.1.message, - format!("All workers are busy, please retry later. {ADMISSION_CONTROL_REJECTION_HINT}") + assert_eq!(response.1.message, "Service temporarily overloaded"); + assert!( + !response.1.message.contains("All workers are busy"), + "client response must not include the underlying engine message" ); } @@ -3194,7 +3262,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] @@ -3855,7 +3927,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") + ); } } @@ -3882,11 +3961,187 @@ 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")); + } + } + + #[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; @@ -3968,7 +4223,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 87bcf339eca9..886daccb7825 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::{ @@ -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; } @@ -171,12 +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 { + 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: err.to_string(), + message: sanitized.to_string(), param: None, event_id: None, }, @@ -469,7 +471,7 @@ where send_error_event( ws_tx, "server_error", - &err.to_string(), + &SanitizedError::Internal.to_string(), Some("session.model"), ) .await;