diff --git a/codex-rs/http-client/src/route_aware_client_pool.rs b/codex-rs/http-client/src/route_aware_client_pool.rs index c2a0de7853..94088fda22 100644 --- a/codex-rs/http-client/src/route_aware_client_pool.rs +++ b/codex-rs/http-client/src/route_aware_client_pool.rs @@ -6,6 +6,8 @@ use std::sync::Arc; use std::sync::Mutex; use std::time::Duration; +use bytes::Bytes; +use futures::TryStream; use http::HeaderMap; use http::HeaderName; use http::HeaderValue; @@ -100,6 +102,24 @@ impl RouteAwareRequestError { pub fn is_connect(&self) -> bool { matches!(self, Self::Request(error) if error.is_connect()) } + + pub fn is_body(&self) -> bool { + matches!(self, Self::Request(error) if error.is_body()) + } + + pub fn is_request(&self) -> bool { + matches!(self, Self::Request(error) if error.is_request()) + } + + /// Removes a request URL from the underlying transport error before it is logged or returned. + /// + /// Use this for requests whose URL can contain credentials, such as signed blob uploads. + pub fn without_url(self) -> Self { + match self { + Self::Request(error) => Self::Request(error.without_url()), + other => other, + } + } } #[must_use = "requests are not sent unless `send` is awaited"] @@ -211,6 +231,19 @@ impl RouteAwareRequestBuilder { self } + /// Sets a streaming request body without exposing the underlying HTTP implementation. + pub fn body_stream(mut self, stream: S) -> Self + where + S: TryStream + Send + 'static, + S::Error: Into>, + Bytes: From, + { + if let Ok(request) = &mut self.request { + *request.body_mut() = Some(reqwest::Body::wrap_stream(stream)); + } + self + } + pub async fn send(self) -> Result { self.pool.send(self.request?).await } diff --git a/codex-rs/http-client/src/route_aware_client_pool_tests.rs b/codex-rs/http-client/src/route_aware_client_pool_tests.rs index 46a989b309..7cce63f90f 100644 --- a/codex-rs/http-client/src/route_aware_client_pool_tests.rs +++ b/codex-rs/http-client/src/route_aware_client_pool_tests.rs @@ -10,6 +10,8 @@ use std::sync::atomic::Ordering; use std::time::Duration; use std::time::Instant; +use bytes::Bytes; +use futures::stream; use pretty_assertions::assert_eq; use tracing_subscriber::Layer; use tracing_subscriber::layer::SubscriberExt; @@ -38,6 +40,47 @@ fn request_builder_debug_redacts_url_secrets() { ); } +#[tokio::test] +async fn streams_request_bodies_without_exposing_reqwest_body() { + let (address, server) = spawn_response_server(vec![ + "HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_string(), + ]); + let pool = RouteAwareClientPool::new( + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + ClientRouteClass::Api, + ); + let response = pool + .put(format!("http://{address}/upload")) + .header(http::header::CONTENT_LENGTH, /*value*/ 5) + .body_stream(stream::iter(vec![Ok::<_, io::Error>(Bytes::from_static( + b"hello", + ))])) + .send() + .await + .expect("streaming request should succeed"); + + assert_eq!(response.status(), StatusCode::OK); + let requests = server.join().expect("response server should finish"); + assert_eq!(requests.len(), 1); + assert!(requests[0].ends_with("\r\n\r\nhello")); +} + +#[tokio::test] +async fn without_url_redacts_transport_error_urls() { + let listener = TcpListener::bind(("127.0.0.1", 0)).expect("listener should bind"); + let address = listener.local_addr().expect("listener should have address"); + drop(listener); + let secret = "signed-secret"; + let error = reqwest::Client::new() + .get(format!("http://{address}/upload?sig={secret}")) + .send() + .await + .expect_err("closed listener should reject request"); + let error = RouteAwareRequestError::from(error).without_url(); + + assert!(!error.to_string().contains(secret)); +} + #[tokio::test] async fn forwards_exact_urls_and_caches_clients_by_resolved_route() { let pool = RouteAwareClientPool::with_builder(