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

codex-rs/http-client/src/outbound_proxy_tests.rs

@@ -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 {
@@ -169,20 +240,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}"))]),
     };
@@ -206,7 +264,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!(
@@ -215,6 +276,92 @@ 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.contains("redirect-secret"));
+}
+
 #[test]
 fn parses_pac_proxy_tokens() {
     assert_eq!(

codex-rs/http-client/src/route_aware_redirect.rs

@@ -94,3 +94,7 @@ pub(super) fn remove_sensitive_headers(
         headers.remove("cookie2");
     }
 }
+
+#[cfg(test)]
+#[path = "route_aware_redirect_tests.rs"]
+mod tests;

codex-rs/http-client/src/route_aware_redirect_tests.rs

@@ -0,0 +1,64 @@
+use http::HeaderValue;
+use http::header::CONTENT_LENGTH;
+use http::header::CONTENT_TYPE;
+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")),
+        )
+    );
+}

codex-rs/http-client/tests/route_aware_client_pool.rs

@@ -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(())
+    }
+}