diff --git a/codex-rs/cli/src/doctor.rs b/codex-rs/cli/src/doctor.rs index f7dcede626..09e26875c9 100644 --- a/codex-rs/cli/src/doctor.rs +++ b/codex-rs/cli/src/doctor.rs @@ -30,6 +30,7 @@ use anyhow::Context; use clap::Parser; use codex_api::ApiError; use codex_api::ResponsesWebsocketClient; +use codex_api::TransportError; use codex_api::is_azure_responses_provider; use codex_arg0::Arg0DispatchPaths; use codex_config::ConfigLoadError; @@ -2373,7 +2374,10 @@ async fn websocket_reachability_check( )); if let Some(close) = probe.immediate_close { details.push(format!("immediate close code: {}", close.code)); - details.push(format!("immediate close reason: {}", close.reason)); + details.push(format!( + "immediate close reason: {}", + close_reason_detail(&close.reason) + )); return DoctorCheck::new( "network.websocket_reachability", "websocket", @@ -2396,7 +2400,7 @@ async fn websocket_reachability_check( Ok(Err(err)) => websocket_probe_warning( "Responses WebSocket failed; HTTPS fallback may still work", details, - websocket_error_detail(&err), + websocket_error_detail(&err.error), ), Err(_) => websocket_probe_warning( "Responses WebSocket timed out; HTTPS fallback may still work", @@ -2424,28 +2428,60 @@ fn websocket_probe_warning( fn websocket_error_detail(err: &ApiError) -> String { match err { - ApiError::Transport(transport) => format!("handshake transport error: {transport}"), - ApiError::Api { status, message } => { - format!("handshake API error: {status} {message}") + ApiError::Transport(transport) => { + let detail = match transport { + // Policy denials contain only fixed descriptions, without wire values. + TransportError::Policy(denied) => denied.to_string(), + TransportError::Http { status, .. } => format!("http {status}: [redacted]"), + TransportError::RetryLimit => "retry limit reached".to_string(), + TransportError::Timeout => "timeout".to_string(), + TransportError::Connection(_) => "connection failed: [redacted]".to_string(), + TransportError::Network(_) => "network error: [redacted]".to_string(), + TransportError::Build(_) => "request build error: [redacted]".to_string(), + TransportError::ResponseTooLarge { max_bytes } => { + format!("response body exceeds the {max_bytes} byte limit") + } + }; + format!("handshake transport error: {detail}") } - ApiError::Stream(message) => format!("handshake stream error: {message}"), + ApiError::Api { status, .. } => format!("handshake API error: {status} [redacted]"), + ApiError::Stream(_) => "handshake stream error: [redacted]".to_string(), + ApiError::ResponseIncomplete(_) => { + "handshake error: incomplete response: [redacted]".to_string() + } + // These variants format fixed text; retry metadata is typed and never printed. ApiError::ContentFilter - | ApiError::ResponseIncomplete(_) | ApiError::ContextWindowExceeded | ApiError::QuotaExceeded | ApiError::UsageNotIncluded - | ApiError::Retryable { .. } - | ApiError::RateLimitExceeded { .. } - | ApiError::RateLimit(_) - | ApiError::InvalidRequest { .. } - | ApiError::ModelNotFound { .. } - | ApiError::InvalidPrompt { .. } - | ApiError::CyberPolicy { .. } - | ApiError::BioPolicy { .. } - | ApiError::MisalignmentPolicyViolation { .. } | ApiError::FlexUnavailable | ApiError::ServerOverloaded { .. } | ApiError::PolicyCapacityUnavailable { .. } => format!("handshake error: {err}"), + ApiError::Retryable { .. } => "handshake error: retryable error: [redacted]".to_string(), + ApiError::RateLimitExceeded { .. } => { + "handshake error: rate limit exceeded: [redacted]".to_string() + } + ApiError::RateLimit(_) => "handshake error: rate limit: [redacted]".to_string(), + ApiError::InvalidRequest { .. } => { + "handshake error: invalid request: [redacted]".to_string() + } + ApiError::ModelNotFound { .. } => { + "handshake error: model not found: [redacted]".to_string() + } + ApiError::InvalidPrompt { .. } => "handshake error: invalid prompt: [redacted]".to_string(), + ApiError::CyberPolicy { .. } => "handshake error: cyber policy: [redacted]".to_string(), + ApiError::BioPolicy { .. } => "handshake error: bio policy: [redacted]".to_string(), + ApiError::MisalignmentPolicyViolation { .. } => { + "handshake error: misalignment policy violation: [redacted]".to_string() + } + } +} + +fn close_reason_detail(reason: &str) -> &'static str { + if reason.is_empty() { + "empty" + } else { + "present (redacted)" } } @@ -3144,6 +3180,163 @@ mod tests { assert_eq!(overall_status(&checks), CheckStatus::Fail); } + #[test] + fn websocket_probe_human_output() { + let report = |websocket: DoctorCheck| DoctorReport { + schema_version: 1, + generated_at: "0s since unix epoch".to_string(), + overall_status: websocket.status, + codex_version: "0.0.0".to_string(), + checks: vec![ + DoctorCheck::new( + "network.environment", + "network", + CheckStatus::Ok, + "network environment is readable", + ) + .detail("HTTPS proxy: not set"), + websocket, + DoctorCheck::new( + "network.provider_reachability", + "reachability", + CheckStatus::Ok, + "active provider endpoints are reachable over HTTP", + ) + .detail("OpenAI endpoint: HTTP 200"), + ], + }; + let options = HumanOutputOptions { + show_details: true, + show_all: true, + ascii: true, + color_enabled: false, + }; + let success = render_human_report( + &report( + DoctorCheck::new( + "network.websocket_reachability", + "websocket", + CheckStatus::Ok, + "Responses WebSocket handshake succeeded", + ) + .detail("handshake result: HTTP 101") + .detail("reasoning header: true") + .detail("server model present: true"), + ), + options, + ); + let immediate_close = render_human_report( + &report( + DoctorCheck::new( + "network.websocket_reachability", + "websocket", + CheckStatus::Warning, + "Responses WebSocket closed immediately after handshake", + ) + .detail("handshake result: HTTP 101") + .detail("immediate close code: 1008") + .detail(format!( + "immediate close reason: {}", + close_reason_detail("private-close-reason") + )) + .remediation( + "Check proxy, VPN, firewall, DNS, custom CA, and WebSocket policy support.", + ), + ), + options, + ); + let generic_failure = render_human_report( + &report(websocket_probe_warning( + "Responses WebSocket failed; HTTPS fallback may still work", + vec!["endpoint: wss://example.com/responses".to_string()], + websocket_error_detail(&ApiError::Stream("private-stream-error".to_string())), + )), + options, + ); + let rendered = format!( + "successful handshake\n{success}\nimmediate close\n{immediate_close}\nstream failure\n{generic_failure}" + ); + + assert!(!rendered.contains("private-close-reason")); + assert!(!rendered.contains("private-stream-error")); + insta::assert_snapshot!("websocket_probe_human_output", rendered); + } + + #[test] + fn websocket_error_detail_preserves_categories_without_wire_values() { + let secret = "private-wire-value"; + let message = || secret.to_string(); + let cases = [ + ApiError::Transport(TransportError::Http { + status: http::StatusCode::FORBIDDEN, + url: Some(format!("https://example.com/?token={secret}")), + headers: Some(HeaderMap::from_iter([( + http::header::SET_COOKIE, + HeaderValue::from_static(secret), + )])), + body: Some(message()), + retry_after: None, + }), + ApiError::Transport(TransportError::Network(message())), + ApiError::Transport(TransportError::Build(message())), + ApiError::Api { + status: http::StatusCode::UNAUTHORIZED, + message: message(), + }, + ApiError::Stream(message()), + ApiError::ResponseIncomplete(Box::new( + codex_protocol::error::ResponseIncompleteError { + reason: message(), + response_id: Some(message()), + token_usage: None, + }, + )), + ApiError::Retryable { + message: message(), + retry_after: None, + }, + ApiError::RateLimitExceeded { + message: message(), + retry_after: None, + }, + ApiError::RateLimit(message()), + ApiError::InvalidRequest { message: message() }, + ApiError::ModelNotFound { + message: message(), + status: Some(http::StatusCode::NOT_FOUND), + }, + ApiError::InvalidPrompt { message: message() }, + ApiError::CyberPolicy { message: message() }, + ApiError::BioPolicy { message: message() }, + ApiError::MisalignmentPolicyViolation { + message: message(), + misalignment: None, + }, + ]; + let details = cases.iter().map(websocket_error_detail).collect::>(); + assert!(!details.join("\n").contains(secret)); + assert_eq!( + details, + [ + "handshake transport error: http 403 Forbidden: [redacted]", + "handshake transport error: network error: [redacted]", + "handshake transport error: request build error: [redacted]", + "handshake API error: 401 Unauthorized [redacted]", + "handshake stream error: [redacted]", + "handshake error: incomplete response: [redacted]", + "handshake error: retryable error: [redacted]", + "handshake error: rate limit exceeded: [redacted]", + "handshake error: rate limit: [redacted]", + "handshake error: invalid request: [redacted]", + "handshake error: model not found: [redacted]", + "handshake error: invalid prompt: [redacted]", + "handshake error: cyber policy: [redacted]", + "handshake error: bio policy: [redacted]", + "handshake error: misalignment policy violation: [redacted]", + ] + ); + } + #[test] fn run_sync_check_notifies_progress() { let progress_impl = Arc::new(RecordingProgress::default()); diff --git a/codex-rs/cli/src/snapshots/codex__doctor__tests__websocket_probe_human_output.snap b/codex-rs/cli/src/snapshots/codex__doctor__tests__websocket_probe_human_output.snap new file mode 100644 index 0000000000..06b5cae1b6 --- /dev/null +++ b/codex-rs/cli/src/snapshots/codex__doctor__tests__websocket_probe_human_output.snap @@ -0,0 +1,67 @@ +--- +source: cli/src/doctor.rs +expression: rendered +--- +successful handshake +Codex Doctor v0.0.0 + +Connectivity + [ok] network network environment is readable + HTTPS proxy not set + [ok] websocket Responses WebSocket handshake succeeded + handshake result HTTP 101 + reasoning header true + server model present true + [ok] reachability active provider endpoints are reachable over HTTP + OpenAI endpoint HTTP 200 + +------------------------------------------------------------- +3 ok | 0 warn | 0 fail ok + +--summary compact output --all expand truncated lists +--json redacted report + +immediate close +Codex Doctor v0.0.0 + +Notes + [!!] websocket Responses WebSocket closed immediately after handshake - Check proxy, VPN, firewall, DNS, custom CA, and WebSocket policy support. +------------------------------------------------------------- + +Connectivity + [ok] network network environment is readable + HTTPS proxy not set + [!!] websocket Responses WebSocket closed immediately after handshake - Check proxy, VPN, firewall, DNS, custom CA, and WebSocket policy support. + handshake result HTTP 101 + immediate close code 1008 + immediate close reason present (redacted) + [ok] reachability active provider endpoints are reachable over HTTP + OpenAI endpoint HTTP 200 + +------------------------------------------------------------- +2 ok | 1 notes | 1 warn | 0 fail degraded + +--summary compact output --all expand truncated lists +--json redacted report + +stream failure +Codex Doctor v0.0.0 + +Notes + [!!] websocket Responses WebSocket failed; HTTPS fallback may still work - Check proxy, VPN, firewall, DNS, custom CA, and WebSocket policy support. +------------------------------------------------------------- + +Connectivity + [ok] network network environment is readable + HTTPS proxy not set + [!!] websocket Responses WebSocket failed; HTTPS fallback may still work - Check proxy, VPN, firewall, DNS, custom CA, and WebSocket policy support. + endpoint wss://example.com/responses + handshake stream error [redacted] + [ok] reachability active provider endpoints are reachable over HTTP + OpenAI endpoint HTTP 200 + +------------------------------------------------------------- +2 ok | 1 notes | 1 warn | 0 fail degraded + +--summary compact output --all expand truncated lists +--json redacted report diff --git a/codex-rs/codex-api/Cargo.toml b/codex-rs/codex-api/Cargo.toml index 82bec69218..90db05b840 100644 --- a/codex-rs/codex-api/Cargo.toml +++ b/codex-rs/codex-api/Cargo.toml @@ -16,6 +16,7 @@ codex-utils-rustls-provider = { workspace = true } codex-websocket-client = { workspace = true } futures = { workspace = true } http = { workspace = true } +rustls = { workspace = true } schemars = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true, features = ["raw_value"] } @@ -35,7 +36,6 @@ anyhow = { workspace = true } assert_matches = { workspace = true } pretty_assertions = { workspace = true } rcgen = { workspace = true } -rustls = { workspace = true } tempfile = { workspace = true } tokio = { workspace = true, features = ["test-util"] } tokio-test = { workspace = true } diff --git a/codex-rs/codex-api/src/endpoint/mod.rs b/codex-rs/codex-api/src/endpoint/mod.rs index 77b3054aee..f3d3c4fd3c 100644 --- a/codex-rs/codex-api/src/endpoint/mod.rs +++ b/codex-rs/codex-api/src/endpoint/mod.rs @@ -5,6 +5,7 @@ pub(crate) mod realtime_call; pub(crate) mod realtime_websocket; pub(crate) mod responses; pub(crate) mod responses_websocket; +mod responses_websocket_probe; pub(crate) mod search; mod session; @@ -28,7 +29,9 @@ pub use realtime_websocket::session_update_session_json; pub use responses::ResponsesClient; pub use responses::ResponsesOptions; pub use responses_websocket::ResponsesWebsocketClient; -pub use responses_websocket::ResponsesWebsocketClose; pub use responses_websocket::ResponsesWebsocketConnection; -pub use responses_websocket::ResponsesWebsocketProbe; +pub use responses_websocket_probe::ResponsesWebsocketClose; +pub use responses_websocket_probe::ResponsesWebsocketProbe; +pub use responses_websocket_probe::ResponsesWebsocketProbeError; +pub use responses_websocket_probe::ResponsesWebsocketProbeFailureStage; pub use search::SearchClient; diff --git a/codex-rs/codex-api/src/endpoint/responses_websocket.rs b/codex-rs/codex-api/src/endpoint/responses_websocket.rs index 12c157aa2c..27361aef0c 100644 --- a/codex-rs/codex-api/src/endpoint/responses_websocket.rs +++ b/codex-rs/codex-api/src/endpoint/responses_websocket.rs @@ -5,6 +5,10 @@ use crate::common::ResponseStream; use crate::common::ResponsesWsRequest; use crate::common::SafetyBufferingTreatment; use crate::common::WS_REQUEST_HEADER_TRACEPARENT_CLIENT_METADATA_KEY; +use crate::endpoint::responses_websocket_probe::ResponsesWebsocketClose; +use crate::endpoint::responses_websocket_probe::ResponsesWebsocketProbe; +use crate::endpoint::responses_websocket_probe::ResponsesWebsocketProbeError; +use crate::endpoint::responses_websocket_probe::ResponsesWebsocketProbeFailureStage; use crate::error::ApiError; use crate::provider::Provider; use crate::rate_limits::parse_rate_limit_event; @@ -37,6 +41,7 @@ use tokio::time::Instant; use tokio_tungstenite::tungstenite::Error as WsError; use tokio_tungstenite::tungstenite::Message; use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tokio_tungstenite::tungstenite::handshake::client::Response; use tokio_tungstenite::tungstenite::protocol::CloseFrame; use tracing::Instrument; use tracing::Span; @@ -395,30 +400,6 @@ pub struct ResponsesWebsocketClient { auth: SharedAuthProvider, } -/// Close frame information captured by a handshake probe. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ResponsesWebsocketClose { - /// WebSocket close code returned by the server. - pub code: String, - /// Human-readable close reason returned by the server. - pub reason: String, -} - -/// Result of a handshake-only Responses WebSocket probe. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ResponsesWebsocketProbe { - /// Redacted by callers before displaying or serializing support reports. - pub url: String, - /// HTTP status returned by the successful WebSocket upgrade. - pub status: StatusCode, - /// Whether the server reported reasoning support in the upgrade response. - pub reasoning_included: bool, - /// Whether the server returned a server-selected model in the upgrade response. - pub server_model_present: bool, - /// Close frame received immediately after upgrade, when one arrives quickly. - pub immediate_close: Option, -} - impl ResponsesWebsocketClient { /// Creates a Responses WebSocket client for an already-resolved provider and auth source. pub fn new(provider: Provider, auth: SharedAuthProvider) -> Self { @@ -458,35 +439,67 @@ impl ResponsesWebsocketClient { extra_headers: HeaderMap, default_headers: HeaderMap, immediate_close_timeout: Duration, - ) -> Result { + ) -> Result { let ws_url = self .provider .websocket_url_for_path("/responses") - .map_err(|err| ApiError::Stream(format!("failed to build websocket URL: {err}")))?; + .map_err(|err| { + ResponsesWebsocketProbeError::before_upgrade( + ApiError::Stream(format!("failed to build websocket URL: {err}")), + ResponsesWebsocketProbeFailureStage::Unknown, + ) + })?; let headers = merge_request_headers(&self.provider.headers, extra_headers, default_headers); - let headers = self.auth.apply_auth_headers(headers).await?; + let headers = self.auth.apply_auth_headers(headers).await.map_err(|err| { + ResponsesWebsocketProbeError::before_upgrade( + err.into(), + ResponsesWebsocketProbeFailureStage::Unknown, + ) + })?; - let connector = WebSocketConnector::new(http_client_factory) - .map_err(|err| ApiError::Stream(format!("failed to configure websocket TLS: {err}")))?; - let (mut stream, status, reasoning_included, server_model) = connect_websocket( + let connector = WebSocketConnector::new(http_client_factory).map_err(|err| { + ResponsesWebsocketProbeError::before_upgrade( + ApiError::Stream(format!("failed to configure websocket TLS: {err}")), + ResponsesWebsocketProbeFailureStage::Tls, + ) + })?; + let (mut stream, response, reasoning_included, server_model) = connect_websocket( ws_url.clone(), headers, &connector, /*turn_state*/ None, ) .await?; + let status = response.status(); + let mut headers = HeaderMap::new(); + for name in [ + "x-request-id", + "x-oai-request-id", + "cf-ray", + "x-ms-request-id", + ] { + for value in response.headers().get_all(name) { + headers.append(name, value.clone()); + } + } let immediate_close = tokio::time::timeout(immediate_close_timeout, stream.next()) .await .ok() .flatten() .transpose() - .map_err(map_ws_stream_error)? + .map_err(|error| ResponsesWebsocketProbeError { + error: Box::new(map_ws_stream_error(error)), + stage: ResponsesWebsocketProbeFailureStage::Unknown, + status: Some(status), + headers: Some(headers.clone()), + })? .and_then(immediate_close_from_message); Ok(ResponsesWebsocketProbe { url: ws_url.to_string(), status, + headers, reasoning_included, server_model_present: server_model.is_some(), immediate_close, @@ -498,7 +511,13 @@ fn immediate_close_from_message(message: Message) -> Option close_frame_to_probe(frame), + None => ResponsesWebsocketClose { + code: "1005".to_string(), + reason: String::new(), + }, + }) } fn close_frame_to_probe(frame: CloseFrame) -> ResponsesWebsocketClose { @@ -528,25 +547,46 @@ async fn connect_websocket( headers: HeaderMap, connector: &WebSocketConnector, turn_state: Option>>, -) -> Result<(WsStream, StatusCode, bool, Option), ApiError> { - info!("connecting to websocket: {url}"); +) -> Result<(WsStream, Response, bool, Option), ResponsesWebsocketProbeError> { + info!("connecting to Responses WebSocket"); - let mut request = url - .as_str() - .into_client_request() - .map_err(|err| ApiError::Stream(format!("failed to build websocket request: {err}")))?; + let mut request = url.as_str().into_client_request().map_err(|err| { + ResponsesWebsocketProbeError::before_upgrade( + ApiError::Stream(format!("failed to build websocket request: {err}")), + ResponsesWebsocketProbeFailureStage::Unknown, + ) + })?; request.headers_mut().extend(headers); let response = connector.connect(request, websocket_config()).await; let (stream, response) = match response { Ok((stream, response)) => { - info!("successfully connected to websocket: {url}"); + info!( + status = response.status().as_u16(), + "successfully connected to Responses WebSocket" + ); (stream, response) } Err(err) => { - error!("failed to connect to websocket: {err}, url: {url}"); - return Err(map_ws_error(err, &url)); + // Tokio-rustls wraps certificate failures in I/O errors. + let is_tls = match &err { + WsError::Tls(_) => true, + WsError::Io(error) => { + matches!(error.get_ref(), Some(source) if source.is::()) + } + _ => false, + }; + let stage = if is_tls { + ResponsesWebsocketProbeFailureStage::Tls + } else { + ResponsesWebsocketProbeFailureStage::Unknown + }; + error!(?stage, "failed to connect to Responses WebSocket"); + return Err(ResponsesWebsocketProbeError::before_upgrade( + map_ws_error(err, &url), + stage, + )); } }; @@ -566,7 +606,7 @@ async fn connect_websocket( } Ok(( WsStream::new(stream), - response.status(), + response, reasoning_included, server_model, )) diff --git a/codex-rs/codex-api/src/endpoint/responses_websocket/connector.rs b/codex-rs/codex-api/src/endpoint/responses_websocket/connector.rs index 802d613ade..4dffd362f1 100644 --- a/codex-rs/codex-api/src/endpoint/responses_websocket/connector.rs +++ b/codex-rs/codex-api/src/endpoint/responses_websocket/connector.rs @@ -35,8 +35,10 @@ impl ResponsesWebsocketClient { .map_err(|err| ApiError::Stream(format!("failed to build websocket URL: {err}")))?; let headers = merge_request_headers(&self.provider.headers, extra_headers, default_headers); let headers = self.auth.apply_auth_headers(headers).await?; - let (stream, _status, server_reasoning_included, server_model) = - connect_websocket(ws_url, headers, connector, turn_state).await?; + let (stream, _response, server_reasoning_included, server_model) = + connect_websocket(ws_url, headers, connector, turn_state) + .await + .map_err(|failure| *failure.error)?; Ok(ResponsesWebsocketConnection::new( stream, self.provider.stream_idle_timeout, diff --git a/codex-rs/codex-api/src/endpoint/responses_websocket_probe.rs b/codex-rs/codex-api/src/endpoint/responses_websocket_probe.rs new file mode 100644 index 0000000000..2140b54ae4 --- /dev/null +++ b/codex-rs/codex-api/src/endpoint/responses_websocket_probe.rs @@ -0,0 +1,101 @@ +//! Handshake probe evidence. Failure stages require typed evidence, and Debug output hides wire +//! values that may contain credentials; diagnostic callers must redact displayed values themselves. + +use crate::error::ApiError; +use http::HeaderMap; +use http::StatusCode; + +/// Close frame information captured by a handshake probe. +#[derive(Clone, PartialEq, Eq)] +pub struct ResponsesWebsocketClose { + /// WebSocket close code returned by the server. + pub code: String, + /// Human-readable close reason returned by the server. + pub reason: String, +} + +impl std::fmt::Debug for ResponsesWebsocketClose { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ResponsesWebsocketClose") + .field("code", &self.code) + .finish_non_exhaustive() + } +} + +/// Result of a handshake-only Responses WebSocket probe. +#[derive(Clone, PartialEq, Eq)] +pub struct ResponsesWebsocketProbe { + /// Redacted by callers before displaying or serializing support reports. + pub url: String, + /// HTTP status returned by the successful WebSocket upgrade. + pub status: StatusCode, + /// Allowlisted tracking headers from the successful upgrade. + pub headers: HeaderMap, + /// Whether the server reported reasoning support in the upgrade response. + pub reasoning_included: bool, + /// Whether the server returned a server-selected model in the upgrade response. + pub server_model_present: bool, + /// Close frame received immediately after upgrade, when one arrives quickly. + pub immediate_close: Option, +} + +impl std::fmt::Debug for ResponsesWebsocketProbe { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ResponsesWebsocketProbe") + .field("status", &self.status) + .field("reasoning_included", &self.reasoning_included) + .field("server_model_present", &self.server_model_present) + .field("immediate_close", &self.immediate_close) + .finish_non_exhaustive() + } +} + +/// A failure stage established by typed evidence at the handshake boundary. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ResponsesWebsocketProbeFailureStage { + /// TLS configuration or a typed TLS handshake failure. + Tls, + /// The transport did not expose enough evidence to identify a failure stage. + Unknown, +} + +/// Probe failure with any HTTP upgrade evidence received before the failure. +/// +/// Rejected upgrades retain status and headers in the original [`ApiError`]. An error after a +/// successful upgrade also retains that upgrade's status and allowlisted tracking headers here. +#[derive(thiserror::Error)] +#[error("Responses WebSocket probe failed")] +pub struct ResponsesWebsocketProbeError { + /// Original API error, preserving the production connection's error behavior. + #[source] + pub error: Box, + /// Failure stage, when typed evidence establishes one. + pub stage: ResponsesWebsocketProbeFailureStage, + /// Status received before a subsequent read failure. + pub status: Option, + /// Allowlisted tracking headers received before a subsequent read failure. + pub headers: Option, +} + +impl ResponsesWebsocketProbeError { + pub(crate) fn before_upgrade( + error: ApiError, + stage: ResponsesWebsocketProbeFailureStage, + ) -> Self { + Self { + error: Box::new(error), + stage, + status: None, + headers: None, + } + } +} + +impl std::fmt::Debug for ResponsesWebsocketProbeError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ResponsesWebsocketProbeError") + .field("stage", &self.stage) + .field("status", &self.status) + .finish_non_exhaustive() + } +} diff --git a/codex-rs/codex-api/src/lib.rs b/codex-rs/codex-api/src/lib.rs index 90752102d8..ad1a361a79 100644 --- a/codex-rs/codex-api/src/lib.rs +++ b/codex-rs/codex-api/src/lib.rs @@ -74,6 +74,8 @@ pub use crate::endpoint::ResponsesWebsocketClient; pub use crate::endpoint::ResponsesWebsocketClose; pub use crate::endpoint::ResponsesWebsocketConnection; pub use crate::endpoint::ResponsesWebsocketProbe; +pub use crate::endpoint::ResponsesWebsocketProbeError; +pub use crate::endpoint::ResponsesWebsocketProbeFailureStage; pub use crate::endpoint::SearchClient; pub use crate::endpoint::session_update_session_json; pub use crate::error::ApiError; diff --git a/codex-rs/codex-api/tests/responses_websocket_probe.rs b/codex-rs/codex-api/tests/responses_websocket_probe.rs new file mode 100644 index 0000000000..278f7ac995 --- /dev/null +++ b/codex-rs/codex-api/tests/responses_websocket_probe.rs @@ -0,0 +1,294 @@ +#![allow(clippy::expect_used)] +//! Exercise TLS, upgrade, rejection, and post-upgrade fault evidence with local wire fixtures. + +use codex_api::ApiError; +use codex_api::AuthProvider; +use codex_api::Provider; +use codex_api::ResponsesWebsocketClient; +use codex_api::ResponsesWebsocketProbeFailureStage; +use codex_api::RetryConfig; +use codex_api::TransportError; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; +use http::HeaderMap; +use http::HeaderValue; +use pretty_assertions::assert_eq; +use std::sync::Arc; +use std::time::Duration; +use tokio::io::AsyncReadExt; +use tokio::io::AsyncWriteExt; +use tokio::net::TcpListener; +use tokio_tungstenite::tungstenite::handshake::derive_accept_key; + +struct NoAuth; +impl AuthProvider for NoAuth { + fn add_auth_headers(&self, _headers: &mut HeaderMap) -> Result<(), codex_api::AuthError> { + Ok(()) + } +} + +fn probe_client(base_url: String) -> ResponsesWebsocketClient { + ResponsesWebsocketClient::new( + Provider { + name: "probe fixture".into(), + base_url, + query_params: None, + headers: HeaderMap::new(), + retry: RetryConfig { + max_attempts: 1, + base_delay: Duration::from_millis(/*millis*/ 1), + retry_429: false, + retry_5xx: false, + retry_transport: false, + }, + stream_idle_timeout: Duration::from_secs(/*secs*/ 5), + }, + Arc::new(NoAuth), + ) +} + +#[tokio::test] +async fn probe_distinguishes_untrusted_certificates_from_io_failures() -> anyhow::Result<()> { + codex_utils_rustls_provider::ensure_rustls_crypto_provider(); + let timeout = Duration::from_secs(/*secs*/ 5); + let certificate = rcgen::generate_simple_self_signed(vec!["localhost".to_string()])?; + let config = Arc::new( + rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert( + vec![certificate.cert.der().clone()], + certificate.signing_key.into(), + )?, + ); + for (serve_tls, expected_stage) in [ + (true, ResponsesWebsocketProbeFailureStage::Tls), + (false, ResponsesWebsocketProbeFailureStage::Unknown), + ] { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let address = listener.local_addr()?; + let config = Arc::clone(&config); + let server = tokio::spawn(async move { + let (socket, _) = tokio::time::timeout(timeout, listener.accept()).await??; + if serve_tls { + let mut socket = socket.into_std()?; + socket.set_nonblocking(/*nonblocking*/ false)?; + socket.set_read_timeout(Some(timeout))?; + socket.set_write_timeout(Some(timeout))?; + let mut connection = rustls::ServerConnection::new(config)?; + tokio::task::spawn_blocking(move || { + connection + .complete_io(&mut socket) + .expect_err("client must reject the untrusted certificate"); + }) + .await?; + } + // Closing before the handshake produces an ordinary I/O failure. + Ok::<_, anyhow::Error>(()) + }); + let client = probe_client(format!("https://localhost:{}", address.port())); + let error = tokio::time::timeout( + timeout, + client.probe_handshake( + &HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + HeaderMap::new(), + HeaderMap::new(), + Duration::from_millis(/*millis*/ 25), + ), + ) + .await + .expect("probe finishes") + .expect_err("handshake must fail"); + assert_eq!( + (error.stage, error.status, error.headers), + (expected_stage, None, None) + ); + server.await??; + } + Ok(()) +} + +#[derive(Clone, Copy, Eq, PartialEq)] +enum Reply { + EmptyClose, + CloseWithReason, + HoldOpen, + Rejected, + MalformedFrame, +} + +#[tokio::test] +async fn probe_preserves_evidence_and_redacts_wire_secrets() { + for (reply, expected_status, expected_close) in [ + (Reply::EmptyClose, 101, Some(("1005", ""))), + ( + Reply::CloseWithReason, + 101, + Some(("1008", "private-close-reason")), + ), + (Reply::HoldOpen, 101, None), + (Reply::Rejected, 403, None), + (Reply::MalformedFrame, 101, None), + ] { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind fixture"); + let address = listener.local_addr().expect("fixture address"); + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.expect("accept fixture"); + let mut request = Vec::new(); + while !request.ends_with(b"\r\n\r\n") { + request.push(socket.read_u8().await.expect("read handshake")); + assert!(request.len() < 8192); + } + let headers = concat!( + "x-request-id: fixture-request\r\n", + "x-oai-request-id: fixture-oai-request\r\n", + "cf-ray: fixture-ray\r\n", + "x-ms-request-id: fixture-ms-one\r\n", + "x-ms-request-id: fixture-ms-two\r\n", + "x-reasoning-included: true\r\n", + "openai-model: fixture-model\r\n", + "set-cookie: private-cookie\r\n", + ); + let response = match reply { + Reply::Rejected => format!( + "HTTP/1.1 403 Forbidden\r\n{headers}content-length: 12\r\n\r\nprivate-body" + ), + Reply::EmptyClose + | Reply::CloseWithReason + | Reply::HoldOpen + | Reply::MalformedFrame => { + let request = String::from_utf8(request).expect("HTTP request"); + let key = request + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("sec-websocket-key") + .then(|| value.trim()) + }) + .expect("WebSocket key"); + let accept = derive_accept_key(key.as_bytes()); + format!( + "HTTP/1.1 101 Switching Protocols\r\nconnection: Upgrade\r\nupgrade: websocket\r\nsec-websocket-accept: {accept}\r\n{headers}\r\n" + ) + } + }; + socket + .write_all(response.as_bytes()) + .await + .expect("write handshake"); + match reply { + Reply::EmptyClose => socket.write_all(&[0x88, 0]).await.expect("write frame"), + Reply::CloseWithReason => { + let mut frame = vec![0x88, 22, 0x03, 0xf0]; + frame.extend_from_slice(b"private-close-reason"); + socket.write_all(&frame).await.expect("write frame"); + } + Reply::HoldOpen => tokio::time::sleep(Duration::from_millis(/*millis*/ 50)).await, + // Unassigned opcode 3 does not depend on extension negotiation. + Reply::MalformedFrame => socket.write_all(&[0x83, 0]).await.expect("write frame"), + Reply::Rejected => (), + } + }); + let client = probe_client(format!("http://{address}")); + let immediate_close_timeout = if reply == Reply::HoldOpen { + Duration::from_millis(/*millis*/ 25) + } else { + Duration::from_secs(/*secs*/ 1) + }; + let result = tokio::time::timeout( + Duration::from_secs(/*secs*/ 5), + client.probe_handshake( + &HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + HeaderMap::new(), + HeaderMap::new(), + immediate_close_timeout, + ), + ) + .await + .expect("probe finishes"); + let (status, headers, close) = match result { + Ok(probe) => { + assert!(matches!( + reply, + Reply::EmptyClose | Reply::CloseWithReason | Reply::HoldOpen + )); + assert_eq!( + (probe.reasoning_included, probe.server_model_present), + (true, true) + ); + let debug = format!("{probe:?}"); + for secret in [ + "fixture-request", + "fixture-oai-request", + "fixture-ray", + "fixture-ms", + "private-cookie", + "private-close-reason", + ] { + assert!(!debug.contains(secret), "{debug}"); + } + let close = probe + .immediate_close + .map(|close| (close.code, close.reason)); + (probe.status, probe.headers, close) + } + Err(error) => { + assert_eq!(error.to_string(), "Responses WebSocket probe failed"); + assert!(!format!("{error:?}").contains("private-")); + assert_eq!(error.stage, ResponsesWebsocketProbeFailureStage::Unknown); + match *error.error { + ApiError::Transport(TransportError::Http { + status, + headers, + body, + .. + }) => { + assert!(matches!(reply, Reply::Rejected)); + assert_eq!(body.as_deref(), Some("private-body")); + (status, headers.expect("rejected headers"), None) + } + _ => { + assert!(matches!(reply, Reply::MalformedFrame)); + ( + error.status.expect("received 101"), + error.headers.expect("received headers"), + None, + ) + } + } + } + }; + let mut expected_headers = HeaderMap::new(); + for (name, value) in [ + ("x-request-id", "fixture-request"), + ("x-oai-request-id", "fixture-oai-request"), + ("cf-ray", "fixture-ray"), + ("x-ms-request-id", "fixture-ms-one"), + ("x-ms-request-id", "fixture-ms-two"), + ] { + expected_headers.append(name, HeaderValue::from_static(value)); + } + if matches!(reply, Reply::Rejected) { + for (name, value) in [ + ("x-reasoning-included", "true"), + ("openai-model", "fixture-model"), + ("set-cookie", "private-cookie"), + ("content-length", "12"), + ] { + expected_headers.append(name, HeaderValue::from_static(value)); + } + } + assert_eq!( + ( + status.as_u16(), + headers, + close + .as_ref() + .map(|(code, reason)| (code.as_str(), reason.as_str())), + ), + (expected_status, expected_headers, expected_close) + ); + server.await.expect("fixture completes"); + } +}