http-client: cover route-aware proxy and privacy behavior

This commit is contained in:
Michael Bolin
2026-07-09 15:34:38 -07:00
parent 42e741270b
commit 6cc150c02e
7 changed files with 652 additions and 21 deletions
+1
View File
@@ -6,6 +6,7 @@ mod error;
mod outbound_proxy;
mod request;
mod route_aware_client_pool;
mod route_aware_redirect;
mod transport;
pub use crate::chatgpt_cloudflare_cookies::with_chatgpt_cloudflare_cookie_store;
+167 -15
View File
@@ -1,6 +1,7 @@
//! Shared outbound proxy policy tests.
use super::*;
use http::header::COOKIE;
use pretty_assertions::assert_eq;
use std::io::Read;
use std::io::Write;
@@ -10,6 +11,76 @@ struct MapEnv {
values: HashMap<String, String>,
}
fn spawn_proxy_listener() -> (
std::net::SocketAddr,
std::thread::JoinHandle<Option<String>>,
) {
let listener =
std::net::TcpListener::bind(("127.0.0.1", 0)).expect("local proxy listener should bind");
let proxy_addr = listener
.local_addr()
.expect("local proxy listener should have an address");
listener
.set_nonblocking(true)
.expect("proxy listener should become nonblocking");
let proxy_thread = std::thread::spawn(move || {
let deadline = Instant::now() + Duration::from_secs(2);
loop {
match listener.accept() {
Ok((mut stream, _)) => {
stream
.set_read_timeout(Some(Duration::from_secs(2)))
.expect("proxy stream should get a read timeout");
let mut buffer = [0_u8; 4096];
let size = stream.read(&mut buffer).expect("proxy should read request");
stream
.write_all(
b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok",
)
.expect("proxy should write response");
break Some(String::from_utf8_lossy(&buffer[..size]).into_owned());
}
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
if Instant::now() >= deadline {
break None;
}
std::thread::sleep(Duration::from_millis(10));
}
Err(error) => panic!("proxy should accept a request: {error}"),
}
}
});
(proxy_addr, proxy_thread)
}
fn spawn_redirect_listener(
location: &str,
) -> (std::net::SocketAddr, std::thread::JoinHandle<String>) {
let listener =
std::net::TcpListener::bind(("127.0.0.1", 0)).expect("redirect listener should bind");
let address = listener
.local_addr()
.expect("redirect listener should have an address");
let location = location.to_string();
let thread = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().expect("redirect server should accept");
stream
.set_read_timeout(Some(Duration::from_secs(2)))
.expect("redirect stream should get a read timeout");
let mut buffer = [0_u8; 4096];
let size = stream
.read(&mut buffer)
.expect("redirect request should read");
write!(
stream,
"HTTP/1.1 302 Found\r\nLocation: {location}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
)
.expect("redirect response should write");
String::from_utf8_lossy(&buffer[..size]).into_owned()
});
(address, thread)
}
#[test]
fn websocket_route_uses_http_equivalent_for_system_resolution() {
let env = MapEnv {
@@ -194,20 +265,7 @@ async fn async_resolution_uses_cached_route_before_global_permit() {
#[tokio::test]
async fn enabled_environment_proxy_routes_request_through_proxy() {
let listener =
std::net::TcpListener::bind(("127.0.0.1", 0)).expect("local proxy listener should bind");
let proxy_addr = listener
.local_addr()
.expect("local proxy listener should have an address");
let proxy_thread = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().expect("proxy should accept a request");
let mut buffer = [0_u8; 4096];
let size = stream.read(&mut buffer).expect("proxy should read request");
stream
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")
.expect("proxy should write response");
String::from_utf8_lossy(&buffer[..size]).into_owned()
});
let (proxy_addr, proxy_thread) = spawn_proxy_listener();
let env = MapEnv {
values: HashMap::from([("HTTP_PROXY".to_string(), format!("http://{proxy_addr}"))]),
};
@@ -231,7 +289,10 @@ async fn enabled_environment_proxy_routes_request_through_proxy() {
.send()
.await
.expect("request should use local proxy");
let proxy_request = proxy_thread.join().expect("proxy thread should finish");
let proxy_request = proxy_thread
.join()
.expect("proxy thread should finish")
.expect("proxy should receive request before timeout");
assert_eq!(response.status(), reqwest::StatusCode::OK);
assert_eq!(
@@ -240,6 +301,97 @@ async fn enabled_environment_proxy_routes_request_through_proxy() {
);
}
#[tokio::test]
async fn route_aware_pool_uses_respect_system_proxy_route_for_exact_url() {
let (proxy_addr, proxy_thread) = spawn_proxy_listener();
let request_url = "http://route-aware-proxy.test/proxy-check?pac=exact";
cache_system_proxy_decision(
request_url,
SystemProxyDecision::Proxy {
url: format!("http://{proxy_addr}"),
},
);
let pool = crate::RouteAwareClientPool::new(
HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy),
ClientRouteClass::Api,
);
let response = tokio::time::timeout(Duration::from_secs(2), pool.get(request_url).send())
.await
.expect("proxy request should finish")
.expect("request should use local proxy");
let proxy_request = proxy_thread
.join()
.expect("proxy thread should finish")
.expect("proxy should receive request before timeout");
assert_eq!(response.status(), reqwest::StatusCode::OK);
assert_eq!(
proxy_request.lines().next(),
Some("GET http://route-aware-proxy.test/proxy-check?pac=exact HTTP/1.1")
);
}
#[tokio::test]
async fn route_aware_pool_resolves_each_redirect_hop_and_strips_credentials() {
let (proxy_addr, proxy_thread) = spawn_proxy_listener();
let redirected_url = "http://redirect-target.test/final?pac=redirect";
let (redirect_addr, redirect_thread) = spawn_redirect_listener(redirected_url);
let initial_url = format!("http://{redirect_addr}/start");
cache_system_proxy_decision(&initial_url, SystemProxyDecision::Direct);
cache_system_proxy_decision(
redirected_url,
SystemProxyDecision::Proxy {
url: format!("http://{proxy_addr}"),
},
);
let pool = crate::RouteAwareClientPool::new(
HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy),
ClientRouteClass::Api,
);
let response = tokio::time::timeout(
Duration::from_secs(2),
pool.get(&initial_url)
.bearer_auth("redirect-secret")
.header(COOKIE, "session=redirect-secret")
.send(),
)
.await
.expect("redirected request should finish")
.expect("redirected request should use selected routes");
let initial_request = redirect_thread
.join()
.expect("redirect thread should finish");
let proxy_request = proxy_thread
.join()
.expect("proxy thread should finish")
.expect("redirect target should reach proxy");
assert_eq!(response.status(), reqwest::StatusCode::OK);
assert_eq!(response.url().as_str(), redirected_url);
assert!(
initial_request
.to_ascii_lowercase()
.contains("authorization: bearer redirect-secret")
);
assert_eq!(
proxy_request.lines().next(),
Some("GET http://redirect-target.test/final?pac=redirect HTTP/1.1")
);
assert!(
!proxy_request
.to_ascii_lowercase()
.contains("authorization:")
);
assert!(
proxy_request
.to_ascii_lowercase()
.contains(&format!("referer: {initial_url}").to_ascii_lowercase())
);
assert!(!proxy_request.contains("redirect-secret"));
}
#[test]
fn parses_pac_proxy_tokens() {
assert_eq!(
@@ -22,6 +22,12 @@ use crate::HttpClient;
use crate::HttpClientFactory;
use crate::OutboundProxyPolicy;
use crate::OutboundProxyRoute;
use crate::route_aware_redirect::MAX_REDIRECTS;
use crate::route_aware_redirect::insert_referer;
use crate::route_aware_redirect::is_redirect;
use crate::route_aware_redirect::redirect_request;
use crate::route_aware_redirect::redirect_url;
use crate::route_aware_redirect::remove_sensitive_headers;
use crate::with_chatgpt_cloudflare_cookie_store;
const MAX_CACHED_ROUTES: usize = 16;
@@ -29,7 +35,8 @@ const MAX_CACHED_ROUTES: usize = 16;
/// Reuses transport clients by resolved route while selecting a route for every request URL.
///
/// Request creation stays on the pool so the URL used for PAC or system-proxy resolution cannot
/// differ from the URL that is sent.
/// differ from the URL that is sent. Redirects are followed through the pool as new requests, so
/// each hop gets its own route decision while connections are still reused by route.
#[derive(Clone)]
pub struct RouteAwareClientPool {
http_client_factory: HttpClientFactory,
@@ -74,18 +81,28 @@ pub enum RouteAwareRequestError {
Route(#[from] RouteAwareClientPoolError),
#[error("failed to build route-aware request: {0}")]
Build(String),
#[error("redirect target uses unsupported URL scheme: {0}")]
UnsupportedRedirectScheme(String),
#[error("too many redirects while requesting {0}")]
TooManyRedirects(reqwest::Url),
#[error("route-aware request timed out")]
Timeout,
}
impl RouteAwareRequestError {
pub fn status(&self) -> Option<StatusCode> {
match self {
Self::Request(error) => error.status(),
Self::Route(_) | Self::Build(_) => None,
Self::Route(_)
| Self::Build(_)
| Self::UnsupportedRedirectScheme(_)
| Self::TooManyRedirects(_)
| Self::Timeout => None,
}
}
pub fn is_timeout(&self) -> bool {
matches!(self, Self::Request(error) if error.is_timeout())
matches!(self, Self::Timeout) || matches!(self, Self::Request(error) if error.is_timeout())
}
pub fn is_connect(&self) -> bool {
@@ -326,10 +343,73 @@ impl RouteAwareClientPool {
async fn send(
&self,
request: reqwest::Request,
mut request: reqwest::Request,
) -> Result<reqwest::Response, RouteAwareRequestError> {
let client = self.client_for_url(request.url().as_str()).await?;
Ok(client.execute(request).await?)
let timeout_deadline = request
.timeout()
.copied()
.map(|timeout| tokio::time::Instant::now() + timeout);
let mut redirects = 0;
loop {
let current_url = request.url().clone();
let client = match timeout_deadline {
Some(timeout_deadline) => tokio::time::timeout_at(
timeout_deadline,
self.client_for_url(current_url.as_str()),
)
.await
.map_err(|_| RouteAwareRequestError::Timeout)??,
None => self.client_for_url(current_url.as_str()).await?,
};
if let Some(timeout_deadline) = timeout_deadline {
let remaining = timeout_deadline
.checked_duration_since(tokio::time::Instant::now())
.ok_or(RouteAwareRequestError::Timeout)?;
if remaining.is_zero() {
return Err(RouteAwareRequestError::Timeout);
}
*request.timeout_mut() = Some(remaining);
}
let method = request.method().clone();
let headers = request.headers().clone();
let version = request.version();
let timeout = request.timeout().copied();
let replay = request.try_clone();
let response = match timeout_deadline {
Some(timeout_deadline) => {
tokio::time::timeout_at(timeout_deadline, client.execute(request))
.await
.map_err(|_| RouteAwareRequestError::Timeout)??
}
None => client.execute(request).await?,
};
let status = response.status();
if !is_redirect(status) {
return Ok(response);
}
let Some(next_url) = redirect_url(&response) else {
return Ok(response);
};
if !matches!(next_url.scheme(), "http" | "https") {
return Err(RouteAwareRequestError::UnsupportedRedirectScheme(
next_url.scheme().to_string(),
));
}
if redirects >= MAX_REDIRECTS {
return Err(RouteAwareRequestError::TooManyRedirects(current_url));
}
let Some(mut next_request) =
redirect_request(status, method, headers, version, timeout, replay, next_url)
else {
return Ok(response);
};
let next_request_url = next_request.url().clone();
remove_sensitive_headers(next_request.headers_mut(), &current_url, &next_request_url);
insert_referer(next_request.headers_mut(), &current_url, &next_request_url);
request = next_request;
redirects += 1;
}
}
async fn client_for_url(
@@ -112,6 +112,23 @@ async fn bounds_cached_routes_and_rebuilds_an_evicted_route() {
);
}
#[tokio::test]
async fn request_timeout_covers_route_selection() {
let pool = RouteAwareClientPool::new(
HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
ClientRouteClass::Api,
);
let error = pool
.get("http://127.0.0.1:1")
.timeout(Duration::ZERO)
.send()
.await
.expect_err("expired request should time out before connecting");
assert!(error.is_timeout());
}
#[derive(Clone)]
struct FakeRouteResolver {
routes: Arc<HashMap<String, OutboundProxyRoute>>,
@@ -0,0 +1,120 @@
use std::time::Duration;
use http::HeaderMap;
use http::Method;
use http::StatusCode;
use http::header::AUTHORIZATION;
use http::header::CONTENT_ENCODING;
use http::header::CONTENT_LENGTH;
use http::header::CONTENT_TYPE;
use http::header::COOKIE;
use http::header::LOCATION;
use http::header::PROXY_AUTHORIZATION;
use http::header::REFERER;
use http::header::TRANSFER_ENCODING;
use http::header::WWW_AUTHENTICATE;
pub(super) const MAX_REDIRECTS: usize = 10;
pub(super) fn is_redirect(status: StatusCode) -> bool {
matches!(
status,
StatusCode::MOVED_PERMANENTLY
| StatusCode::FOUND
| StatusCode::SEE_OTHER
| StatusCode::TEMPORARY_REDIRECT
| StatusCode::PERMANENT_REDIRECT
)
}
pub(super) fn redirect_url(response: &reqwest::Response) -> Option<reqwest::Url> {
let location = response.headers().get(LOCATION)?.to_str().ok()?;
response.url().join(location).ok()
}
pub(super) fn redirect_request(
status: StatusCode,
mut method: Method,
mut headers: HeaderMap,
version: http::Version,
timeout: Option<Duration>,
replay: Option<reqwest::Request>,
next_url: reqwest::Url,
) -> Option<reqwest::Request> {
let drop_body = match status {
StatusCode::MOVED_PERMANENTLY | StatusCode::FOUND if method == Method::POST => {
method = Method::GET;
true
}
StatusCode::SEE_OTHER => {
if method != Method::HEAD {
method = Method::GET;
}
true
}
StatusCode::MOVED_PERMANENTLY
| StatusCode::FOUND
| StatusCode::TEMPORARY_REDIRECT
| StatusCode::PERMANENT_REDIRECT => false,
_ => return None,
};
if drop_body {
for header in [
CONTENT_TYPE,
CONTENT_LENGTH,
CONTENT_ENCODING,
TRANSFER_ENCODING,
] {
headers.remove(header);
}
let mut request = reqwest::Request::new(method, next_url);
*request.headers_mut() = headers;
*request.version_mut() = version;
*request.timeout_mut() = timeout;
Some(request)
} else {
replay.map(|mut request| {
*request.url_mut() = next_url;
request
})
}
}
pub(super) fn remove_sensitive_headers(
headers: &mut HeaderMap,
previous: &reqwest::Url,
next: &reqwest::Url,
) {
let cross_origin = previous.scheme() != next.scheme()
|| previous.host_str() != next.host_str()
|| previous.port_or_known_default() != next.port_or_known_default();
if cross_origin {
for header in [AUTHORIZATION, COOKIE, PROXY_AUTHORIZATION, WWW_AUTHENTICATE] {
headers.remove(header);
}
headers.remove("cookie2");
}
}
pub(super) fn insert_referer(
headers: &mut HeaderMap,
previous: &reqwest::Url,
next: &reqwest::Url,
) {
if next.scheme() == "http" && previous.scheme() == "https" {
return;
}
let mut referer = previous.clone();
let _ = referer.set_username("");
let _ = referer.set_password(None);
referer.set_fragment(None);
if let Ok(value) = referer.as_str().parse() {
headers.insert(REFERER, value);
}
}
#[cfg(test)]
#[path = "route_aware_redirect_tests.rs"]
mod tests;
@@ -0,0 +1,130 @@
use http::HeaderValue;
use http::header::CONTENT_LENGTH;
use http::header::CONTENT_TYPE;
use http::header::COOKIE;
use http::header::REFERER;
use pretty_assertions::assert_eq;
use super::*;
#[test]
fn redirects_match_reqwest_method_and_body_rules() {
let url = reqwest::Url::parse("https://example.com/next").expect("redirect URL should parse");
let mut original = reqwest::Request::new(Method::POST, url.clone());
original
.headers_mut()
.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
original
.headers_mut()
.insert(CONTENT_LENGTH, HeaderValue::from_static("2"));
*original.body_mut() = Some("{}".into());
let found = redirect_request(
StatusCode::FOUND,
original.method().clone(),
original.headers().clone(),
original.version(),
original.timeout().copied(),
original.try_clone(),
url.clone(),
)
.expect("POST redirect should be followed");
assert_eq!(
(
found.method(),
found.body().is_some(),
found.headers().contains_key(CONTENT_TYPE),
found.headers().contains_key(CONTENT_LENGTH),
),
(&Method::GET, false, false, false)
);
let temporary = redirect_request(
StatusCode::TEMPORARY_REDIRECT,
original.method().clone(),
original.headers().clone(),
original.version(),
original.timeout().copied(),
original.try_clone(),
url,
)
.expect("replayable temporary redirect should be followed");
assert_eq!(
(
temporary.method(),
temporary.body().is_some(),
temporary.headers().get(CONTENT_TYPE),
temporary.headers().get(CONTENT_LENGTH),
),
(
&Method::POST,
true,
Some(&HeaderValue::from_static("application/json")),
Some(&HeaderValue::from_static("2")),
)
);
}
#[test]
fn redirect_referer_matches_reqwest_defaults() {
let previous =
reqwest::Url::parse("https://user:password@example.com/start#fragment").expect("valid URL");
let next = reqwest::Url::parse("https://other.example/next").expect("valid URL");
let mut headers = HeaderMap::new();
insert_referer(&mut headers, &previous, &next);
assert_eq!(
headers.get(REFERER),
Some(&HeaderValue::from_static("https://example.com/start"))
);
let mut downgrade_headers = HeaderMap::new();
let downgrade = reqwest::Url::parse("http://other.example/next").expect("valid URL");
insert_referer(&mut downgrade_headers, &previous, &downgrade);
assert_eq!(downgrade_headers.get(REFERER), None);
}
#[test]
fn redirect_credentials_are_retained_only_for_the_same_origin() {
for (previous, next, retain_credentials) in [
(
"https://example.com:8080/start",
"https://example.com:8080/next",
true,
),
(
"https://example.com:8080/start",
"http://example.com:8080/next",
false,
),
(
"https://example.com:8080/start",
"https://other.example:8080/next",
false,
),
(
"https://example.com:8080/start",
"https://example.com:8081/next",
false,
),
] {
let previous = reqwest::Url::parse(previous).expect("previous URL should parse");
let next = reqwest::Url::parse(next).expect("next URL should parse");
let mut headers = HeaderMap::from_iter([
(AUTHORIZATION, HeaderValue::from_static("Bearer secret")),
(COOKIE, HeaderValue::from_static("session=secret")),
]);
remove_sensitive_headers(&mut headers, &previous, &next);
assert_eq!(
(
headers.contains_key(AUTHORIZATION),
headers.contains_key(COOKIE),
),
(retain_credentials, retain_credentials),
"credential handling for {previous} -> {next}"
);
}
}
@@ -0,0 +1,131 @@
use std::io;
use std::io::Read;
use std::io::Write;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use codex_http_client::ClientRouteClass;
use codex_http_client::HttpClientFactory;
use codex_http_client::OutboundProxyPolicy;
use codex_http_client::RouteAwareClientPool;
use http::StatusCode;
use tracing_subscriber::Layer;
use tracing_subscriber::layer::SubscriberExt;
#[tokio::test]
async fn disabled_pool_logging_does_not_expose_request_or_response_data() {
let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("HTTP listener should bind");
let address = listener
.local_addr()
.expect("HTTP listener should have an address");
let server_thread = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().expect("HTTP listener should accept");
stream
.set_read_timeout(Some(Duration::from_secs(2)))
.expect("HTTP stream should get a read timeout");
let mut buffer = [0_u8; 4096];
let _size = stream.read(&mut buffer).expect("HTTP request should read");
stream
.write_all(
b"HTTP/1.1 200 OK\r\nx-sensitive-response: response-secret-value\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok",
)
.expect("HTTP response should write");
});
let endpoint = format!(
"http://auth-user:password-secret-value@{address}/token?client_secret=query-secret-value"
);
let pool = RouteAwareClientPool::with_chatgpt_cloudflare_cookies_without_request_logging(
HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
ClientRouteClass::Api,
);
let buffer = Arc::new(Mutex::new(Vec::new()));
let subscriber = tracing_subscriber::registry().with(
tracing_subscriber::fmt::layer()
.with_ansi(false)
.with_writer(TestLogWriter {
buffer: Arc::clone(&buffer),
})
.with_filter(
tracing_subscriber::filter::Targets::new()
.with_target("codex_http_client", tracing::Level::TRACE),
),
);
let _guard = tracing::subscriber::set_default(subscriber);
tracing::debug!(target: "codex_http_client", "log capture sentinel");
let response = pool
.post(&endpoint)
.header("x-sensitive-request", "request-header-secret-value")
.body("request-body-secret-value")
.send()
.await
.expect("route-aware request should succeed");
assert_eq!(response.status(), StatusCode::OK);
server_thread.join().expect("server thread should finish");
let unresponsive_listener =
std::net::TcpListener::bind("127.0.0.1:0").expect("unresponsive listener should bind");
let unresponsive_address = unresponsive_listener
.local_addr()
.expect("unresponsive listener should have an address");
let unresponsive_endpoint = format!(
"http://auth-user:failure-password-secret-value@{unresponsive_address}/token?client_secret=failure-query-secret-value"
);
let error = pool
.post(&unresponsive_endpoint)
.timeout(Duration::from_millis(100))
.send()
.await
.expect_err("request to an unresponsive listener should time out");
assert!(error.is_timeout());
let logs = String::from_utf8(buffer.lock().expect("log buffer lock").clone())
.expect("logs should be UTF-8");
assert!(logs.contains("log capture sentinel"));
for secret in [
"password-secret-value",
"query-secret-value",
"request-header-secret-value",
"request-body-secret-value",
"response-secret-value",
"failure-password-secret-value",
"failure-query-secret-value",
] {
assert!(!logs.contains(secret), "logs exposed {secret}:\n{logs}");
}
}
#[derive(Clone)]
struct TestLogWriter {
buffer: Arc<Mutex<Vec<u8>>>,
}
struct TestLogSink {
buffer: Arc<Mutex<Vec<u8>>>,
}
impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for TestLogWriter {
type Writer = TestLogSink;
fn make_writer(&'a self) -> Self::Writer {
TestLogSink {
buffer: Arc::clone(&self.buffer),
}
}
}
impl Write for TestLogSink {
fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
let mut log_buffer = self
.buffer
.lock()
.map_err(|_| io::Error::other("log buffer lock was poisoned"))?;
log_buffer.extend(buffer);
Ok(buffer.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}