diff --git a/codex-rs/http-client/Cargo.toml b/codex-rs/http-client/Cargo.toml index 77cf8168a3..41f84bd709 100644 --- a/codex-rs/http-client/Cargo.toml +++ b/codex-rs/http-client/Cargo.toml @@ -11,7 +11,7 @@ futures = { workspace = true } http = { workspace = true } native-tls = "0.2" opentelemetry = { workspace = true } -opentelemetry-http = { workspace = true, features = ["reqwest"] } +opentelemetry-http = { workspace = true, features = ["reqwest", "reqwest-blocking"] } reqwest = { workspace = true, features = ["blocking", "json", "rustls-tls-native-roots", "stream"] } rustls = { workspace = true } rustls-native-certs = { workspace = true } @@ -39,6 +39,7 @@ workspace = true [dev-dependencies] codex-utils-cargo-bin = { workspace = true } +futures = { workspace = true, features = ["executor"] } opentelemetry_sdk = { workspace = true } pretty_assertions = { workspace = true } rcgen = { workspace = true } diff --git a/codex-rs/http-client/src/custom_ca.rs b/codex-rs/http-client/src/custom_ca.rs index d3217a1a2f..f43f0dc687 100644 --- a/codex-rs/http-client/src/custom_ca.rs +++ b/codex-rs/http-client/src/custom_ca.rs @@ -182,6 +182,13 @@ pub fn build_reqwest_client_with_custom_ca( build_reqwest_client_with_env(&ProcessEnv, builder) } +/// Applies the same custom-CA policy to blocking exporter transports. +pub(crate) fn build_blocking_reqwest_client_with_custom_ca( + builder: reqwest::blocking::ClientBuilder, +) -> Result { + build_reqwest_client_with_env(&ProcessEnv, builder) +} + /// Builds a rustls client config when a Codex custom CA bundle is configured. /// /// This is the websocket-facing sibling of [`build_reqwest_client_with_custom_ca`]. When diff --git a/codex-rs/http-client/src/telemetry_client.rs b/codex-rs/http-client/src/telemetry_client.rs index 7dedf620b3..6d358d82f5 100644 --- a/codex-rs/http-client/src/telemetry_client.rs +++ b/codex-rs/http-client/src/telemetry_client.rs @@ -4,7 +4,9 @@ use crate::BuildRouteAwareHttpClientError; use crate::ClientRouteClass; use crate::HttpClientFactory; use crate::OutboundProxyPolicy; +use crate::OutboundProxyRoute; use crate::custom_ca::BuildCustomCaTransportError; +use crate::custom_ca::build_blocking_reqwest_client_with_custom_ca; use opentelemetry_http::HttpClient; use std::fs; use std::io; @@ -91,6 +93,46 @@ impl HttpClientFactory { .map_err(Into::into) } + /// Builds a blocking exporter client for one fixed collector endpoint. + /// + /// Callers inside a Tokio runtime must construct this on a blocking-capable + /// thread, just as they would for any other blocking reqwest client. + pub fn build_blocking_telemetry_client( + &self, + endpoint: &str, + timeout: Duration, + tls: &TelemetryClientTlsConfig, + ) -> Result, BuildTelemetryHttpClientError> { + let tls = prepare_tls_config(tls)?; + let mut builder = reqwest::blocking::Client::builder().timeout(timeout); + if let Some(certificate) = tls.root_certificate { + builder = builder + .tls_built_in_root_certs(false) + .add_root_certificate(certificate); + } + if let Some(identity) = tls.client_identity { + builder = builder.identity(identity).https_only(true); + } + if self.outbound_proxy_policy() == OutboundProxyPolicy::RespectSystemProxy { + builder = builder.redirect(reqwest::redirect::Policy::none()); + } + + builder = match self.resolve_proxy_route(endpoint) { + OutboundProxyRoute::TransportDefault => builder, + OutboundProxyRoute::Direct => builder.no_proxy(), + OutboundProxyRoute::Proxy { url, no_proxy } => { + let proxy = reqwest::Proxy::all(&url).map_err(|_| { + BuildRouteAwareHttpClientError::InvalidProxyConfig { + route_class: ClientRouteClass::Other, + } + })?; + let no_proxy = no_proxy.as_deref().and_then(reqwest::NoProxy::from_string); + builder.proxy(proxy.no_proxy(no_proxy)) + } + }; + + build_blocking_reqwest_client_with_custom_ca(builder).map_err(Into::into) + } } fn prepare_tls_config( diff --git a/codex-rs/http-client/src/telemetry_client_tests.rs b/codex-rs/http-client/src/telemetry_client_tests.rs index eb501a560a..2e6d3496f9 100644 --- a/codex-rs/http-client/src/telemetry_client_tests.rs +++ b/codex-rs/http-client/src/telemetry_client_tests.rs @@ -41,6 +41,30 @@ async fn async_telemetry_client_uses_resolved_system_proxy() { ); } +#[test] +fn blocking_telemetry_client_uses_resolved_system_proxy() { + let (proxy_address, proxy) = spawn_proxy(); + let endpoint = "http://blocking-telemetry-proxy.test/v1/metrics"; + cache_system_proxy_route_for_test(endpoint, format!("http://{proxy_address}")); + let client = HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy) + .build_blocking_telemetry_client( + endpoint, + Duration::from_secs(2), + &TelemetryClientTlsConfig::default(), + ) + .expect("blocking telemetry client should build"); + + let response = futures::executor::block_on(client.send_bytes(telemetry_request(endpoint))) + .expect("telemetry request should use proxy"); + let request = proxy.join().expect("proxy should complete"); + + assert_eq!(response.status(), http::StatusCode::OK); + assert_eq!( + request.lines().next(), + Some("POST http://blocking-telemetry-proxy.test/v1/metrics HTTP/1.1") + ); +} + #[tokio::test] async fn async_system_proxy_telemetry_client_does_not_follow_redirects() { let (proxy_address, proxy) = spawn_proxy_with_response( @@ -65,6 +89,28 @@ async fn async_system_proxy_telemetry_client_does_not_follow_redirects() { assert_eq!(response.status(), http::StatusCode::TEMPORARY_REDIRECT); } +#[test] +fn blocking_system_proxy_telemetry_client_does_not_follow_redirects() { + let (proxy_address, proxy) = spawn_proxy_with_response( + "HTTP/1.1 307 Temporary Redirect\r\nLocation: http://different-route.test/v1/metrics\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + ); + let endpoint = "http://blocking-telemetry-redirect.test/v1/metrics"; + cache_system_proxy_route_for_test(endpoint, format!("http://{proxy_address}")); + let client = HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy) + .build_blocking_telemetry_client( + endpoint, + Duration::from_secs(2), + &TelemetryClientTlsConfig::default(), + ) + .expect("blocking telemetry client should build"); + + let response = futures::executor::block_on(client.send_bytes(telemetry_request(endpoint))) + .expect("redirect should be returned to the exporter"); + proxy.join().expect("proxy should complete"); + + assert_eq!(response.status(), http::StatusCode::TEMPORARY_REDIRECT); +} + #[test] fn telemetry_client_rejects_incomplete_client_identity() { let result = HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault)