openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-08-23 13:09:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Resolve outbound proxy routes explicitly (#34435)

Browse Source 
## Why

System proxy discovery can block, and delegating fallback to each transport can repeat discovery or apply inconsistent environment proxy behavior.

## What changed

- Resolve unavailable system proxy decisions to an explicit environment proxy or direct route, including `NO_PROXY` settings and WebSocket-specific fallbacks.
- Add asynchronous system proxy resolution that uses cached decisions first and serializes blocking platform lookups on Windows and macOS.
- Preserve `NO_PROXY` handling for WebSocket connections, including HTTP and HTTPS proxies, and hash proxy cache keys on every platform.

## Testing

- Add coverage for explicit environment fallback, cached asynchronous resolution, and proxied or bypassed WebSocket connections.

GitOrigin-RevId: 3b7cf170dcfe639eec53c3c1514f92bfcf13e7e9
This commit is contained in:
Michael Bolin
2026-07-21 00:27:10 +00:00
committed by copyberry
parent 2be7d3bcd9
commit c9ef7eff00
5 changed files with 529 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
codex-rs/http-client/Cargo.toml
View File

@@ -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 }

227
codex-rs/http-client/src/outbound_proxy.rs
View File

@@ -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<String>,
},
}
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", &"<redacted>").finish(),
Self::Proxy { .. } => f
.debug_struct("Proxy")
.field("url", &"<redacted>")
.field("no_proxy", &"<redacted>")
.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<OutboundProxyRoute> {
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<OutboundProxyRoute> {
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::<http::Uri>()
.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<reqwest::ClientBuilder, BuildRouteAwareHttpClientError> {
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<reqwest::ClientBuilder, BuildRouteAwareHttpClientError> {
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<reqwest::ClientBuilder, BuildRouteAwareHttpClientError> {
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<Mutex<HashMap<String, CachedSystemProxyDecision>>> =
OnceLock::new();
#[cfg(test)]
fn cached_system_proxy_decision(request_url: &str) -> Option<SystemProxyDecision> {
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"))]

106
codex-rs/http-client/src/outbound_proxy_tests.rs
View File

@@ -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));
}

36
codex-rs/websocket-client/src/dialer.rs
View File

@@ -37,7 +37,7 @@ pub(crate) async fn connect(
tls_config: Arc<ClientConfig>,
proxy_route: OutboundProxyRoute,
) -> Result<(ConnectionInner, Response), WebSocketError> {
let stream: Box<dyn AsyncIo> = 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<dyn AsyncIo> = 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)?;

238
codex-rs/websocket-client/src/dialer_tests.rs
View File

@@ -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::<Vec<_>>();
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<TlsAcceptor>,
) -> (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<dyn AsyncIo> = 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<dyn AsyncIo> = 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::<String>();
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<ClientConfig>, TlsAcceptor) {
fn test_tls_configs() -> (Arc<ClientConfig>, 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<ClientConfig>, 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<ClientConfig>, TlsAcceptor) {
(
Arc::new(client_config),
TlsAcceptor::from(Arc::new(server_config)),
certificate,
)
}