diff --git a/codex-rs/http-client/Cargo.toml b/codex-rs/http-client/Cargo.toml index 46afa51ba5..d6a67d1840 100644 --- a/codex-rs/http-client/Cargo.toml +++ b/codex-rs/http-client/Cargo.toml @@ -16,15 +16,13 @@ rustls-native-certs = { workspace = true } rustls-pki-types = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } +sha2 = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true, features = ["macros", "rt", "time", "sync"] } tracing = { workspace = true } tracing-opentelemetry = { workspace = true } zstd = { workspace = true } -[target.'cfg(any(target_os = "windows", target_os = "macos"))'.dependencies] -sha2 = { workspace = true } - [target.'cfg(target_os = "macos")'.dependencies] system-configuration = { workspace = true } diff --git a/codex-rs/http-client/src/outbound_proxy.rs b/codex-rs/http-client/src/outbound_proxy.rs index 692da254fc..d3b8a81426 100644 --- a/codex-rs/http-client/src/outbound_proxy.rs +++ b/codex-rs/http-client/src/outbound_proxy.rs @@ -12,19 +12,21 @@ use std::sync::Mutex; use std::sync::OnceLock; use std::time::Duration; use std::time::Instant; +#[cfg(any(target_os = "windows", target_os = "macos"))] +use tokio::sync::Semaphore; use crate::custom_ca::BuildCustomCaTransportError; use crate::custom_ca::build_reqwest_client_with_custom_ca; use crate::default_client::HttpClient; -#[cfg(any(target_os = "windows", target_os = "macos"))] use sha2::Digest; -#[cfg(any(target_os = "windows", target_os = "macos"))] use sha2::Sha256; use thiserror::Error; const SYSTEM_PROXY_SUCCESS_CACHE_TTL: Duration = Duration::from_secs(60); const SYSTEM_PROXY_UNAVAILABLE_CACHE_TTL: Duration = Duration::from_secs(5); const SYSTEM_PROXY_CACHE_MAX_ENTRIES: usize = 256; +#[cfg(any(target_os = "windows", target_os = "macos"))] +static ASYNC_SYSTEM_PROXY_RESOLUTION_PERMIT: Semaphore = Semaphore::const_new(1); #[cfg(target_os = "macos")] mod macos; @@ -100,16 +102,22 @@ pub enum OutboundProxyPolicy { /// Resolved proxy route for a concrete outbound destination. /// -/// `TransportDefault` delegates environment-proxy handling to the underlying transport. Proxy -/// URLs are intentionally redacted from `Debug` output because they may contain credentials. -#[derive(Clone, PartialEq, Eq)] +/// `TransportDefault` preserves the underlying transport behavior only when system-proxy support +/// is disabled. When system resolution is enabled, environment and direct fallbacks are resolved +/// explicitly so the transport cannot repeat system discovery. Proxy URLs and no-proxy settings +/// are intentionally redacted from `Debug` output because they may contain credentials or private +/// hostnames. +#[derive(Clone, Hash, PartialEq, Eq)] pub enum OutboundProxyRoute { /// Preserve the underlying transport's existing proxy behavior. TransportDefault, /// Connect directly and bypass transport-level proxy discovery. Direct, /// Connect through the selected proxy URL. - Proxy { url: String }, + Proxy { + url: String, + no_proxy: Option, + }, } impl fmt::Debug for OutboundProxyRoute { @@ -117,7 +125,11 @@ impl fmt::Debug for OutboundProxyRoute { match self { Self::TransportDefault => f.write_str("TransportDefault"), Self::Direct => f.write_str("Direct"), - Self::Proxy { .. } => f.debug_struct("Proxy").field("url", &"").finish(), + Self::Proxy { .. } => f + .debug_struct("Proxy") + .field("url", &"") + .field("no_proxy", &"") + .finish(), } } } @@ -149,16 +161,64 @@ impl HttpClientFactory { /// /// WebSocket schemes are resolved through their HTTP equivalents so platform PAC and system /// proxy APIs apply the same policy to `ws`/`wss` and `http`/`https` destinations. When system - /// resolution is unavailable, the transport retains responsibility for environment-proxy - /// fallback. + /// resolution is unavailable, explicit environment settings are resolved before falling back + /// to a direct route. pub fn resolve_proxy_route(&self, request_url: &str) -> OutboundProxyRoute { resolve_proxy_route( + &ProcessEnv, request_url, self.outbound_proxy_policy, resolve_system_proxy, ) } + /// Resolves the proxy route for a concrete destination without blocking a Tokio worker. + pub async fn resolve_proxy_route_async( + &self, + request_url: String, + ) -> io::Result { + if matches!( + self.outbound_proxy_policy, + OutboundProxyPolicy::ReqwestDefault + ) { + return Ok(OutboundProxyRoute::TransportDefault); + } + + if let Some(route) = self.cached_proxy_route(&request_url) { + return Ok(route); + } + + #[cfg(not(any(target_os = "windows", target_os = "macos")))] + return Ok(self.resolve_proxy_route(&request_url)); + + #[cfg(any(target_os = "windows", target_os = "macos"))] + { + let permit = ASYNC_SYSTEM_PROXY_RESOLUTION_PERMIT + .acquire() + .await + .map_err(io::Error::other)?; + let factory = self.clone(); + tokio::task::spawn_blocking(move || { + // Keep the permit with the blocking task: cancelling the caller must not allow a + // second PAC/WinHTTP lookup to start while this one is still running. + let _permit = permit; + factory.resolve_proxy_route(&request_url) + }) + .await + .map_err(io::Error::other) + } + } + + fn cached_proxy_route(&self, request_url: &str) -> Option { + let env_proxy_kind = EnvProxyKind::from_request_url(request_url); + let request_url = proxy_resolution_url(request_url); + if RequestOrigin::parse(&request_url).is_none() { + return Some(OutboundProxyRoute::Direct); + } + cached_system_proxy_decision(&request_url) + .map(|decision| route_from_system_decision(&ProcessEnv, env_proxy_kind, decision)) + } + /// Builds an HTTP client for a concrete outbound route. pub fn build_client( &self, @@ -196,6 +256,7 @@ impl HttpClientFactory { } fn resolve_proxy_route( + env: &dyn EnvSource, request_url: &str, outbound_proxy_policy: OutboundProxyPolicy, resolve_system_proxy: impl FnOnce(&str, &RequestOrigin) -> SystemProxyDecision, @@ -204,15 +265,79 @@ fn resolve_proxy_route( return OutboundProxyRoute::TransportDefault; } + let env_proxy_kind = EnvProxyKind::from_request_url(request_url); let request_url = proxy_resolution_url(request_url); let Some(origin) = RequestOrigin::parse(&request_url) else { - return OutboundProxyRoute::TransportDefault; + return OutboundProxyRoute::Direct; }; - match resolve_system_proxy(&request_url, &origin) { + route_from_system_decision( + env, + env_proxy_kind, + resolve_system_proxy(&request_url, &origin), + ) +} + +fn route_from_system_decision( + env: &dyn EnvSource, + env_proxy_kind: EnvProxyKind, + decision: SystemProxyDecision, +) -> OutboundProxyRoute { + match decision { SystemProxyDecision::Direct => OutboundProxyRoute::Direct, - SystemProxyDecision::Proxy { url } => OutboundProxyRoute::Proxy { url }, - SystemProxyDecision::Unavailable { .. } => OutboundProxyRoute::TransportDefault, + SystemProxyDecision::Proxy { url } => OutboundProxyRoute::Proxy { + url, + no_proxy: None, + }, + SystemProxyDecision::Unavailable { .. } => resolve_env_proxy_route(env, env_proxy_kind), + } +} + +fn resolve_env_proxy_route( + env: &dyn EnvSource, + env_proxy_kind: EnvProxyKind, +) -> OutboundProxyRoute { + let proxy_url = match env_proxy_kind { + EnvProxyKind::Https => { + proxy_env_value(env, "HTTPS_PROXY").or_else(|| proxy_env_value(env, "ALL_PROXY")) + } + EnvProxyKind::SecureWebSocket => proxy_env_value(env, "HTTPS_PROXY") + .or_else(|| proxy_env_value(env, "HTTP_PROXY")) + .or_else(|| proxy_env_value(env, "ALL_PROXY")), + EnvProxyKind::Http => { + proxy_env_value(env, "HTTP_PROXY").or_else(|| proxy_env_value(env, "ALL_PROXY")) + } + EnvProxyKind::Other => proxy_env_value(env, "ALL_PROXY"), + }; + match proxy_url { + Some(url) => OutboundProxyRoute::Proxy { + url, + no_proxy: proxy_env_value(env, "NO_PROXY"), + }, + None => OutboundProxyRoute::Direct, + } +} + +#[derive(Clone, Copy)] +enum EnvProxyKind { + Http, + Https, + SecureWebSocket, + Other, +} + +impl EnvProxyKind { + fn from_request_url(request_url: &str) -> Self { + let scheme = request_url + .parse::() + .ok() + .and_then(|uri| uri.scheme_str().map(str::to_ascii_lowercase)); + match scheme.as_deref() { + Some("http" | "ws") => Self::Http, + Some("https") => Self::Https, + Some("wss") => Self::SecureWebSocket, + Some(_) | None => Self::Other, + } } } @@ -276,22 +401,26 @@ fn configure_proxy_for_route( outbound_proxy_policy: OutboundProxyPolicy, resolve_system_proxy: impl FnOnce(&str, &RequestOrigin) -> SystemProxyDecision, ) -> Result { - if matches!(outbound_proxy_policy, OutboundProxyPolicy::ReqwestDefault) { - return Ok(builder); - } - let origin = RequestOrigin::parse(request_url); + let route = resolve_proxy_route( + env, + request_url, + outbound_proxy_policy, + resolve_system_proxy, + ); + configure_builder_for_resolved_route(builder, route_class, &route) +} - let Some(origin) = origin.as_ref() else { - return configure_env_proxy_handling(env, builder, /*origin*/ None, route_class); - }; - - match resolve_system_proxy(request_url, origin) { - SystemProxyDecision::Direct => Ok(builder.no_proxy()), - SystemProxyDecision::Proxy { url } => { - configure_concrete_proxy(builder, route_class, &url, /*no_proxy*/ None) - } - SystemProxyDecision::Unavailable { .. } => { - configure_env_proxy_handling(env, builder, Some(origin), route_class) +fn configure_builder_for_resolved_route( + builder: reqwest::ClientBuilder, + route_class: ClientRouteClass, + route: &OutboundProxyRoute, +) -> Result { + match route { + OutboundProxyRoute::TransportDefault => Ok(builder), + OutboundProxyRoute::Direct => Ok(builder.no_proxy()), + OutboundProxyRoute::Proxy { url, no_proxy } => { + let no_proxy = no_proxy.as_deref().and_then(reqwest::NoProxy::from_string); + configure_concrete_proxy(builder, route_class, url, no_proxy) } } } @@ -311,31 +440,6 @@ fn configure_concrete_proxy( Ok(builder.proxy(proxy.no_proxy(no_proxy))) } -fn configure_env_proxy_handling( - env: &dyn EnvSource, - builder: reqwest::ClientBuilder, - origin: Option<&RequestOrigin>, - route_class: ClientRouteClass, -) -> Result { - if let Some(origin) = origin { - let proxy_url = match origin.scheme.as_str() { - "https" => { - proxy_env_value(env, "HTTPS_PROXY").or_else(|| proxy_env_value(env, "ALL_PROXY")) - } - "http" => { - proxy_env_value(env, "HTTP_PROXY").or_else(|| proxy_env_value(env, "ALL_PROXY")) - } - _ => proxy_env_value(env, "ALL_PROXY"), - }; - if let Some(proxy_url) = proxy_url { - let no_proxy = proxy_env_value(env, "NO_PROXY") - .and_then(|value| reqwest::NoProxy::from_string(&value)); - return configure_concrete_proxy(builder, route_class, &proxy_url, no_proxy); - } - } - Ok(builder.no_proxy()) -} - #[derive(Debug, Clone, PartialEq, Eq)] #[allow(dead_code)] struct RequestOrigin { @@ -431,7 +535,6 @@ struct CachedSystemProxyDecision { static SYSTEM_PROXY_CACHE: OnceLock>> = OnceLock::new(); -#[cfg(test)] fn cached_system_proxy_decision(request_url: &str) -> Option { let cache = SYSTEM_PROXY_CACHE.get_or_init(|| Mutex::new(HashMap::new())); let mut cache = cache.lock().ok()?; @@ -494,17 +597,11 @@ fn insert_system_proxy_cache_entry( } fn system_proxy_cache_key(request_url: &str) -> String { - #[cfg(any(target_os = "windows", target_os = "macos"))] - { - // Keep URL-specific PAC decisions without retaining the raw routed URL. - let mut hasher = Sha256::new(); - hasher.update(b"system-proxy-cache-v1\0"); - hasher.update(request_url.as_bytes()); - format!("{:x}", hasher.finalize()) - } - - #[cfg(not(any(target_os = "windows", target_os = "macos")))] - request_url.to_string() + // Keep URL-specific PAC decisions without retaining the raw routed URL. + let mut hasher = Sha256::new(); + hasher.update(b"system-proxy-cache-v1\0"); + hasher.update(request_url.as_bytes()); + format!("{:x}", hasher.finalize()) } #[cfg(any(test, target_os = "windows"))] diff --git a/codex-rs/http-client/src/outbound_proxy_tests.rs b/codex-rs/http-client/src/outbound_proxy_tests.rs index 387973eec4..956b03813e 100644 --- a/codex-rs/http-client/src/outbound_proxy_tests.rs +++ b/codex-rs/http-client/src/outbound_proxy_tests.rs @@ -12,7 +12,11 @@ struct MapEnv { #[test] fn websocket_route_uses_http_equivalent_for_system_resolution() { + let env = MapEnv { + values: HashMap::new(), + }; let route = resolve_proxy_route( + &env, "wss://api.openai.com/v1/responses", OutboundProxyPolicy::RespectSystemProxy, |request_url, origin| { @@ -30,13 +34,18 @@ fn websocket_route_uses_http_equivalent_for_system_resolution() { route, OutboundProxyRoute::Proxy { url: "http://proxy.example:8080".to_string(), + no_proxy: None, } ); } #[test] fn reqwest_default_route_preserves_transport_proxy_behavior() { + let env = MapEnv { + values: HashMap::new(), + }; let route = resolve_proxy_route( + &env, "wss://api.openai.com/v1/responses", OutboundProxyPolicy::ReqwestDefault, |_, _| panic!("default policy should not resolve system proxy settings"), @@ -79,12 +88,11 @@ fn environment_fallback_reads_injected_proxy_environment() { let env = MapEnv { values: HashMap::from([("HTTPS_PROXY".to_string(), "://invalid".to_string())]), }; - let origin = RequestOrigin::parse("https://auth.openai.com/oauth/token").expect("valid URL"); - let result = configure_env_proxy_handling( - &env, + let route = resolve_env_proxy_route(&env, EnvProxyKind::Https); + let result = configure_builder_for_resolved_route( reqwest::Client::builder(), - Some(&origin), ClientRouteClass::Auth, + &route, ); assert!(matches!( @@ -95,6 +103,95 @@ fn environment_fallback_reads_injected_proxy_environment() { )); } +#[test] +fn unavailable_system_route_resolves_environment_or_direct_explicitly() { + let env = MapEnv { + values: HashMap::from([ + ( + "HTTPS_PROXY".to_string(), + "http://proxy.example:8080".to_string(), + ), + ("NO_PROXY".to_string(), "localhost,.internal".to_string()), + ]), + }; + + assert_eq!( + route_from_system_decision( + &env, + EnvProxyKind::Https, + SystemProxyDecision::Unavailable { + failure: RouteFailureClass::ProxyResolutionUnavailable, + }, + ), + OutboundProxyRoute::Proxy { + url: "http://proxy.example:8080".to_string(), + no_proxy: Some("localhost,.internal".to_string()), + } + ); + assert_eq!( + route_from_system_decision( + &MapEnv { + values: HashMap::new(), + }, + EnvProxyKind::Https, + SystemProxyDecision::Unavailable { + failure: RouteFailureClass::ProxyResolutionUnavailable, + }, + ), + OutboundProxyRoute::Direct + ); +} + +#[test] +fn unavailable_system_route_preserves_wss_http_proxy_fallback() { + let env = MapEnv { + values: HashMap::from([( + "HTTP_PROXY".to_string(), + "http://proxy.example:8080".to_string(), + )]), + }; + + let route = resolve_proxy_route( + &env, + "wss://api.openai.com/v1/responses", + OutboundProxyPolicy::RespectSystemProxy, + |_, _| SystemProxyDecision::Unavailable { + failure: RouteFailureClass::ProxyResolutionUnavailable, + }, + ); + + assert_eq!( + route, + OutboundProxyRoute::Proxy { + url: "http://proxy.example:8080".to_string(), + no_proxy: None, + } + ); +} + +#[cfg(any(target_os = "windows", target_os = "macos"))] +#[tokio::test] +async fn async_resolution_uses_cached_route_before_global_permit() { + let request_url = "https://cached-fast-path.test/request"; + cache_system_proxy_decision(request_url, SystemProxyDecision::Direct); + let factory = HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy); + let permit = ASYNC_SYSTEM_PROXY_RESOLUTION_PERMIT + .acquire() + .await + .expect("global proxy permit should stay open"); + + let route = tokio::time::timeout( + Duration::from_secs(2), + factory.resolve_proxy_route_async(request_url.to_string()), + ) + .await + .expect("cached resolution should not wait for the global permit") + .expect("cached route should resolve"); + drop(permit); + + assert_eq!(route, OutboundProxyRoute::Direct); +} + #[tokio::test] async fn enabled_environment_proxy_routes_request_through_proxy() { let listener = @@ -282,6 +379,5 @@ fn system_proxy_cache_key_preserves_url_specific_pac_decisions() { cache_key, system_proxy_cache_key("https://auth.openai.com/oauth/revoke") ); - #[cfg(any(target_os = "windows", target_os = "macos"))] assert!(!cache_key.contains(request_url)); } diff --git a/codex-rs/websocket-client/src/dialer.rs b/codex-rs/websocket-client/src/dialer.rs index 29d4f42884..93fedd3e64 100644 --- a/codex-rs/websocket-client/src/dialer.rs +++ b/codex-rs/websocket-client/src/dialer.rs @@ -37,7 +37,7 @@ pub(crate) async fn connect( tls_config: Arc, proxy_route: OutboundProxyRoute, ) -> Result<(ConnectionInner, Response), WebSocketError> { - let stream: Box = match proxy_route { + let proxy_url = match proxy_route { OutboundProxyRoute::TransportDefault => { // The workspace enables tokio-tungstenite's `proxy` feature, so its default dialer // resolves HTTP_PROXY, HTTPS_PROXY, ALL_PROXY, and NO_PROXY before opening the socket. @@ -50,7 +50,37 @@ pub(crate) async fn connect( .await?; return Ok((ConnectionInner::TransportDefault(stream), response)); } - OutboundProxyRoute::Direct => { + OutboundProxyRoute::Direct => None, + OutboundProxyRoute::Proxy { + url, + no_proxy: None, + } => Some(url), + OutboundProxyRoute::Proxy { + url, + no_proxy: Some(_), + } => { + // Let Tungstenite apply its complete NO_PROXY semantics. Its environment parser does + // not accept HTTPS proxy URLs, but that error occurs only after it decides the target + // is not bypassed, so retry that case through the explicit TLS-to-proxy path below. + match connect_async_tls_with_config( + request.clone(), + Some(config), + false, // Preserve Tungstenite's recommended Nagle default. + Some(Connector::Rustls(Arc::clone(&tls_config))), + ) + .await + { + Ok((stream, response)) => { + return Ok((ConnectionInner::TransportDefault(stream), response)); + } + Err(WebSocketError::Url(UrlError::UnsupportedProxyScheme)) => Some(url), + Err(error) => return Err(error), + } + } + }; + + let stream: Box = match proxy_url { + None => { let host = websocket_host(&request)?; let port = websocket_port(&request)?; Box::new( @@ -59,7 +89,7 @@ pub(crate) async fn connect( .map_err(WebSocketError::Io)?, ) } - OutboundProxyRoute::Proxy { url } => { + Some(url) => { let proxy = ProxyEndpoint::parse(&url)?; let host = websocket_host(&request)?; let port = websocket_port(&request)?; diff --git a/codex-rs/websocket-client/src/dialer_tests.rs b/codex-rs/websocket-client/src/dialer_tests.rs index 4d7ae7731d..eae86d90a0 100644 --- a/codex-rs/websocket-client/src/dialer_tests.rs +++ b/codex-rs/websocket-client/src/dialer_tests.rs @@ -1,4 +1,5 @@ use std::net::SocketAddr; +use std::process::Command; use std::sync::Arc; use std::time::Duration; @@ -14,6 +15,7 @@ use rcgen::generate_simple_self_signed; use rustls::ClientConfig; use rustls::RootCertStore; use rustls::ServerConfig; +use rustls::pki_types::CertificateDer; use rustls::pki_types::PrivateKeyDer; use rustls::pki_types::PrivatePkcs8KeyDer; use tokio::io::AsyncReadExt; @@ -35,7 +37,7 @@ use crate::WebSocketConnector; #[tokio::test] async fn public_connector_uses_factory_and_exposes_stream_and_sink() { - let (target_addr, target_task) = start_plain_echo_websocket_server().await; + let (target_addr, target_task) = start_echo_websocket_server(/*acceptor*/ None).await; let request = format!("ws://localhost:{}/v1/responses", target_addr.port()) .into_client_request() .expect("websocket request should build"); @@ -63,7 +65,7 @@ async fn public_connector_uses_factory_and_exposes_stream_and_sink() { #[tokio::test] async fn direct_route_connects_secure_websocket() { - let (tls_config, acceptor) = test_tls_configs(); + let (tls_config, acceptor, _) = test_tls_configs(); let (target_addr, target_task) = start_tls_websocket_server(acceptor).await; let request = format!("wss://localhost:{}/v1/responses", target_addr.port()) .into_client_request() @@ -92,6 +94,92 @@ async fn https_proxy_tunnels_secure_websocket_before_handshake() { assert_proxy_tunnels_secure_websocket(/*proxy_tls*/ true).await; } +#[tokio::test] +async fn environment_proxy_route_honors_no_proxy_in_a_subprocess() { + assert_no_proxy_subprocess( + "127.0.0.1", + /*expect_proxy*/ false, + /*proxy_tls*/ false, + ) + .await; + assert_no_proxy_subprocess( + "unrelated.example", + /*expect_proxy*/ true, + /*proxy_tls*/ false, + ) + .await; + assert_no_proxy_subprocess( + "unrelated.example", + /*expect_proxy*/ true, + /*proxy_tls*/ true, + ) + .await; +} + +#[tokio::test] +async fn no_proxy_subprocess_probe() { + let Ok(url) = std::env::var("CODEX_WEBSOCKET_NO_PROXY_PROBE_URL") else { + return; + }; + let proxy_url = std::env::var("CODEX_WEBSOCKET_NO_PROXY_PROBE_PROXY") + .expect("parent test should provide a proxy URL"); + let no_proxy = std::env::var("NO_PROXY").expect("parent test should provide a no-proxy value"); + let request = url + .into_client_request() + .expect("websocket request should build"); + let tls_config = + if let Ok(certificate_hex) = std::env::var("CODEX_WEBSOCKET_NO_PROXY_PROBE_CA_DER") { + ensure_rustls_crypto_provider(); + assert_eq!( + certificate_hex.len() % 2, + 0, + "encoded certificate should contain complete bytes" + ); + let certificate = (0..certificate_hex.len()) + .step_by(2) + .map(|index| { + u8::from_str_radix(&certificate_hex[index..index + 2], 16) + .expect("encoded certificate should contain hexadecimal bytes") + }) + .collect::>(); + let mut roots = RootCertStore::empty(); + roots + .add(CertificateDer::from(certificate)) + .expect("proxy certificate should be trusted"); + Arc::new( + ClientConfig::builder() + .with_root_certificates(roots) + .with_no_client_auth(), + ) + } else { + test_tls_configs().0 + }; + let (inner, _) = connect( + request, + WebSocketConfig::default(), + tls_config, + OutboundProxyRoute::Proxy { + url: proxy_url, + no_proxy: Some(no_proxy), + }, + ) + .await + .expect("websocket handshake should succeed"); + let mut websocket = WebSocketConnection { inner }; + websocket + .send(Message::Text("probe".into())) + .await + .expect("probe should send"); + assert_eq!( + websocket + .next() + .await + .expect("probe should receive a message") + .expect("probe message should be valid"), + Message::Text("probe".into()) + ); +} + #[test] fn https_proxy_defaults_to_port_443_and_preserves_explicit_port() { let default_port = ProxyEndpoint::parse("https://proxy.example") @@ -150,7 +238,9 @@ async fn happy_eyeballs_does_not_wait_for_stalled_preferred_family() { assert_eq!(connected, reachable); } -async fn start_plain_echo_websocket_server() -> (SocketAddr, JoinHandle<()>) { +async fn start_echo_websocket_server( + acceptor: Option, +) -> (SocketAddr, JoinHandle<()>) { let listener = TcpListener::bind("127.0.0.1:0") .await .expect("target listener should bind"); @@ -159,6 +249,15 @@ async fn start_plain_echo_websocket_server() -> (SocketAddr, JoinHandle<()>) { .expect("target listener should have an address"); let task = tokio::spawn(async move { let (stream, _) = listener.accept().await.expect("target should accept"); + let stream: Box = match acceptor { + Some(acceptor) => Box::new( + acceptor + .accept(stream) + .await + .expect("target TLS handshake should succeed"), + ), + None => Box::new(stream), + }; let mut websocket = accept_async(stream) .await .expect("target websocket handshake should succeed"); @@ -175,8 +274,133 @@ async fn start_plain_echo_websocket_server() -> (SocketAddr, JoinHandle<()>) { (address, task) } +async fn assert_no_proxy_subprocess(no_proxy: &str, expect_proxy: bool, proxy_tls: bool) { + let (target_acceptor, proxy_acceptor, certificate) = if proxy_tls { + let (_, acceptor, certificate) = test_tls_configs(); + (Some(acceptor.clone()), Some(acceptor), Some(certificate)) + } else { + (None, None, None) + }; + let (target_addr, target_task) = start_echo_websocket_server(target_acceptor).await; + let proxy_listener = Arc::new( + TcpListener::bind("127.0.0.1:0") + .await + .expect("proxy listener should bind"), + ); + let proxy_addr = proxy_listener + .local_addr() + .expect("proxy listener should have an address"); + let proxy_task = if expect_proxy { + let proxy_listener = Arc::clone(&proxy_listener); + Some(tokio::spawn(async move { + let (client, _) = proxy_listener.accept().await.expect("proxy should accept"); + let mut client: Box = match proxy_acceptor { + Some(acceptor) => Box::new( + acceptor + .accept(client) + .await + .expect("proxy TLS handshake should succeed"), + ), + None => Box::new(client), + }; + let mut request = Vec::new(); + let mut byte = [0_u8; 1]; + while !request.ends_with(b"\r\n\r\n") { + client + .read_exact(&mut byte) + .await + .expect("proxy should read CONNECT request"); + request.push(byte[0]); + } + let mut target = tokio::net::TcpStream::connect(target_addr) + .await + .expect("proxy should connect to target"); + client + .write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n") + .await + .expect("proxy should acknowledge CONNECT"); + let _ = tokio::io::copy_bidirectional(&mut client, &mut target).await; + String::from_utf8(request).expect("CONNECT request should be UTF-8") + })) + } else { + None + }; + let executable = std::env::current_exe().expect("test executable should be available"); + let target_scheme = if proxy_tls { "wss" } else { "ws" }; + let proxy_scheme = if proxy_tls { "https" } else { "http" }; + let target_host = if proxy_tls { "localhost" } else { "127.0.0.1" }; + let target_url = format!( + "{target_scheme}://{target_host}:{}/v1/responses", + target_addr.port() + ); + let proxy_url = format!("{proxy_scheme}://localhost:{}", proxy_addr.port()); + let no_proxy = no_proxy.to_string(); + let output = tokio::task::spawn_blocking(move || { + let mut command = Command::new(executable); + command.args([ + "--exact", + "dialer::tests::no_proxy_subprocess_probe", + "--nocapture", + ]); + for key in [ + "HTTP_PROXY", + "http_proxy", + "HTTPS_PROXY", + "https_proxy", + "ALL_PROXY", + "all_proxy", + "NO_PROXY", + "no_proxy", + ] { + command.env_remove(key); + } + command + .env( + if proxy_tls { + "HTTPS_PROXY" + } else { + "HTTP_PROXY" + }, + &proxy_url, + ) + .env("NO_PROXY", no_proxy) + .env("CODEX_WEBSOCKET_NO_PROXY_PROBE_URL", target_url) + .env("CODEX_WEBSOCKET_NO_PROXY_PROBE_PROXY", proxy_url); + command.env_remove("CODEX_WEBSOCKET_NO_PROXY_PROBE_CA_DER"); + if let Some(certificate) = certificate { + let certificate_hex = certificate + .as_ref() + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + command.env("CODEX_WEBSOCKET_NO_PROXY_PROBE_CA_DER", certificate_hex); + } + command + .output() + .expect("WebSocket no-proxy subprocess should run") + }) + .await + .expect("WebSocket no-proxy subprocess should join"); + assert!( + output.status.success(), + "WebSocket no-proxy subprocess failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + + target_task.await.expect("target task should finish"); + // In the bypass case no task services the proxy listener, so the successful child connection + // above also proves that the matching NO_PROXY value selected the target directly. + if let Some(proxy_task) = proxy_task { + let request = proxy_task.await.expect("proxy task should finish"); + let expected_request_line = + format!("CONNECT {target_host}:{} HTTP/1.1", target_addr.port()); + assert_eq!(request.lines().next(), Some(expected_request_line.as_str())); + } +} + async fn assert_proxy_tunnels_secure_websocket(proxy_tls: bool) { - let (tls_config, acceptor) = test_tls_configs(); + let (tls_config, acceptor, _) = test_tls_configs(); let (target_addr, target_task) = start_tls_websocket_server(acceptor.clone()).await; let proxy_listener = TcpListener::bind("127.0.0.1:0") @@ -232,6 +456,7 @@ async fn assert_proxy_tunnels_secure_websocket(proxy_tls: bool) { tls_config, OutboundProxyRoute::Proxy { url: format!("{proxy_scheme}://localhost:{}", proxy_addr.port()), + no_proxy: None, }, ) .await @@ -270,7 +495,7 @@ async fn start_tls_websocket_server(acceptor: TlsAcceptor) -> (SocketAddr, JoinH (address, task) } -fn test_tls_configs() -> (Arc, TlsAcceptor) { +fn test_tls_configs() -> (Arc, TlsAcceptor, CertificateDer<'static>) { ensure_rustls_crypto_provider(); let CertifiedKey { cert, signing_key } = generate_simple_self_signed(vec!["localhost".to_string()]) @@ -284,7 +509,7 @@ fn test_tls_configs() -> (Arc, TlsAcceptor) { let mut roots = RootCertStore::empty(); roots - .add(certificate) + .add(certificate.clone()) .expect("test certificate should be trusted"); let client_config = ClientConfig::builder() .with_root_certificates(roots) @@ -293,5 +518,6 @@ fn test_tls_configs() -> (Arc, TlsAcceptor) { ( Arc::new(client_config), TlsAcceptor::from(Arc::new(server_config)), + certificate, ) }