mirror of
https://github.com/openai/codex.git
synced 2026-10-19 17:28:46 +00:00
Expand route-aware proxy redirect coverage (#34483)
## What changed - Add end-to-end coverage that exact-URL system proxy decisions select the expected route, including when a redirect switches routes. - Verify redirects strip credential headers across origins while retaining them for the same origin and route. - Check that redirect chains emit only the final request outcome, omit sensitive redirect details from logs, and honor disabled request logging. - Share local HTTP listener helpers across the outbound proxy tests. GitOrigin-RevId: afd1640e1910442d71fa1dc77b20b4fb1a6f1b6a
This commit is contained in:
@@ -0,0 +1,217 @@
|
||||
use super::*;
|
||||
use pretty_assertions::assert_eq;
|
||||
use std::sync::Arc;
|
||||
use std::sync::Mutex;
|
||||
use std::time::Duration;
|
||||
use tracing_subscriber::Layer;
|
||||
use tracing_subscriber::layer::SubscriberExt;
|
||||
|
||||
#[tokio::test]
|
||||
async fn route_aware_pool_strips_credentials_on_cross_origin_redirect() {
|
||||
let (destination_addr, destination_thread) = spawn_proxy_listener();
|
||||
let destination_url = format!("http://{destination_addr}/final");
|
||||
let (redirect_addr, redirect_thread) = spawn_redirect_listener(&destination_url);
|
||||
let initial_url = format!("http://{redirect_addr}/start");
|
||||
cache_system_proxy_decision(&initial_url, SystemProxyDecision::Direct);
|
||||
cache_system_proxy_decision(&destination_url, SystemProxyDecision::Direct);
|
||||
let pool = crate::RouteAwareClientPool::new(
|
||||
HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy),
|
||||
ClientRouteClass::Api,
|
||||
);
|
||||
|
||||
let response = tokio::time::timeout(
|
||||
Duration::from_secs(2),
|
||||
pool.get(&initial_url)
|
||||
.header(AUTHORIZATION, "Bearer origin-secret")
|
||||
.header(COOKIE, "session=origin-secret")
|
||||
.header(PROXY_AUTHORIZATION, "Basic proxy-secret")
|
||||
.send(),
|
||||
)
|
||||
.await
|
||||
.expect("redirected request should finish")
|
||||
.expect("cross-origin redirect should succeed");
|
||||
let initial_request = only_request(redirect_thread, "redirect");
|
||||
let destination_request = only_request(destination_thread, "destination");
|
||||
|
||||
assert_eq!(response.url().as_str(), destination_url);
|
||||
assert_eq!(
|
||||
[
|
||||
credential_headers(&initial_request),
|
||||
credential_headers(&destination_request),
|
||||
],
|
||||
[(true, true, true), (false, false, false)]
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn route_aware_pool_retains_credentials_for_same_origin_and_route() {
|
||||
let (proxy_addr, proxy_thread) = spawn_http_listener(vec![
|
||||
"HTTP/1.1 302 Found\r\nLocation: /final\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
|
||||
.to_string(),
|
||||
"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok".to_string(),
|
||||
]);
|
||||
let initial_url = "http://same-route.test/start";
|
||||
let redirected_url = "http://same-route.test/final";
|
||||
let route = SystemProxyDecision::Proxy {
|
||||
url: format!("http://{proxy_addr}"),
|
||||
};
|
||||
cache_system_proxy_decision(initial_url, route.clone());
|
||||
cache_system_proxy_decision(redirected_url, route);
|
||||
let pool = crate::RouteAwareClientPool::new(
|
||||
HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy),
|
||||
ClientRouteClass::Api,
|
||||
);
|
||||
|
||||
let response = tokio::time::timeout(
|
||||
Duration::from_secs(2),
|
||||
pool.get(initial_url)
|
||||
.header(AUTHORIZATION, "Bearer origin-secret")
|
||||
.header(COOKIE, "session=origin-secret")
|
||||
.header(PROXY_AUTHORIZATION, "Basic proxy-secret")
|
||||
.send(),
|
||||
)
|
||||
.await
|
||||
.expect("redirected request should finish")
|
||||
.expect("same-route redirect should succeed");
|
||||
let requests = proxy_thread.join().expect("proxy thread should finish");
|
||||
|
||||
assert_eq!(response.url().as_str(), redirected_url);
|
||||
assert_eq!(
|
||||
requests
|
||||
.iter()
|
||||
.map(|request| credential_headers(request))
|
||||
.collect::<Vec<_>>(),
|
||||
vec![(true, true, true), (true, true, true)]
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn route_aware_pool_sanitizes_redirected_failure_logs() {
|
||||
let log_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(&log_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 (enabled_failure_addr, enabled_failure_thread) = spawn_failing_listener();
|
||||
let enabled_target_url =
|
||||
format!("http://{enabled_failure_addr}/final?token=enabled-target-secret");
|
||||
let (enabled_redirect_addr, enabled_redirect_thread) =
|
||||
spawn_redirect_listener(&enabled_target_url);
|
||||
let enabled_initial_url = format!("http://{enabled_redirect_addr}/start");
|
||||
cache_system_proxy_decision(&enabled_initial_url, SystemProxyDecision::Direct);
|
||||
cache_system_proxy_decision(&enabled_target_url, SystemProxyDecision::Direct);
|
||||
let enabled_pool = crate::RouteAwareClientPool::new(
|
||||
HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy),
|
||||
ClientRouteClass::Api,
|
||||
);
|
||||
|
||||
enabled_pool
|
||||
.get(&enabled_initial_url)
|
||||
.timeout(Duration::from_secs(2))
|
||||
.send()
|
||||
.await
|
||||
.expect_err("redirect target should fail");
|
||||
only_request(enabled_redirect_thread, "enabled redirect");
|
||||
enabled_failure_thread
|
||||
.join()
|
||||
.expect("enabled failure thread should finish");
|
||||
|
||||
let (disabled_failure_addr, disabled_failure_thread) = spawn_failing_listener();
|
||||
let disabled_target_url =
|
||||
format!("http://{disabled_failure_addr}/final?token=disabled-target-secret");
|
||||
let (disabled_redirect_addr, disabled_redirect_thread) =
|
||||
spawn_redirect_listener(&disabled_target_url);
|
||||
let disabled_initial_url = format!("http://{disabled_redirect_addr}/start");
|
||||
cache_system_proxy_decision(&disabled_initial_url, SystemProxyDecision::Direct);
|
||||
cache_system_proxy_decision(&disabled_target_url, SystemProxyDecision::Direct);
|
||||
let disabled_pool = crate::RouteAwareClientPool::new_without_request_logging(
|
||||
HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy),
|
||||
ClientRouteClass::Api,
|
||||
);
|
||||
|
||||
disabled_pool
|
||||
.get(&disabled_initial_url)
|
||||
.timeout(Duration::from_secs(2))
|
||||
.send()
|
||||
.await
|
||||
.expect_err("redirect target should fail");
|
||||
only_request(disabled_redirect_thread, "disabled redirect");
|
||||
disabled_failure_thread
|
||||
.join()
|
||||
.expect("disabled failure thread should finish");
|
||||
|
||||
let logs = String::from_utf8(log_buffer.lock().expect("log buffer lock").clone())
|
||||
.expect("logs should be UTF-8");
|
||||
assert!(logs.contains("log capture sentinel"));
|
||||
assert!(logs.contains(&enabled_initial_url));
|
||||
assert!(!logs.contains(&disabled_initial_url));
|
||||
assert_eq!(logs.matches("Request failed").count(), 1);
|
||||
assert!(logs.contains("is_timeout"));
|
||||
assert!(logs.contains("is_connect"));
|
||||
for secret in [
|
||||
"enabled-target-secret",
|
||||
"disabled-target-secret",
|
||||
enabled_target_url.as_str(),
|
||||
disabled_target_url.as_str(),
|
||||
"error=",
|
||||
] {
|
||||
assert!(!logs.contains(secret), "logs exposed {secret}:\n{logs}");
|
||||
}
|
||||
}
|
||||
|
||||
fn credential_headers(request: &str) -> (bool, bool, bool) {
|
||||
let names = request
|
||||
.lines()
|
||||
.filter_map(|line| {
|
||||
line.split_once(':')
|
||||
.map(|(name, _)| name.to_ascii_lowercase())
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
(
|
||||
names.iter().any(|name| name == "authorization"),
|
||||
names.iter().any(|name| name == "cookie"),
|
||||
names.iter().any(|name| name == "proxy-authorization"),
|
||||
)
|
||||
}
|
||||
|
||||
fn spawn_failing_listener() -> (std::net::SocketAddr, std::thread::JoinHandle<()>) {
|
||||
let listener =
|
||||
std::net::TcpListener::bind(("127.0.0.1", 0)).expect("failing listener should bind");
|
||||
let address = listener
|
||||
.local_addr()
|
||||
.expect("failing listener should have an address");
|
||||
listener
|
||||
.set_nonblocking(true)
|
||||
.expect("failing listener should become nonblocking");
|
||||
let thread = std::thread::spawn(move || {
|
||||
let deadline = Instant::now() + Duration::from_secs(2);
|
||||
let (mut stream, _) = loop {
|
||||
match listener.accept() {
|
||||
Ok(connection) => break connection,
|
||||
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
|
||||
assert!(
|
||||
Instant::now() < deadline,
|
||||
"failing listener should receive a request"
|
||||
);
|
||||
std::thread::sleep(Duration::from_millis(10));
|
||||
}
|
||||
Err(error) => panic!("failing listener should accept: {error}"),
|
||||
}
|
||||
};
|
||||
stream
|
||||
.set_read_timeout(Some(Duration::from_secs(2)))
|
||||
.expect("failing stream should get a read timeout");
|
||||
read_http_message(&mut stream);
|
||||
});
|
||||
(address, thread)
|
||||
}
|
||||
@@ -1,15 +1,121 @@
|
||||
//! Shared outbound proxy policy tests.
|
||||
|
||||
use super::*;
|
||||
use http::header::AUTHORIZATION;
|
||||
use http::header::COOKIE;
|
||||
use http::header::PROXY_AUTHORIZATION;
|
||||
use pretty_assertions::assert_eq;
|
||||
use std::io::Read;
|
||||
use std::io::Write;
|
||||
use std::sync::Arc;
|
||||
use std::sync::Mutex;
|
||||
use tracing_subscriber::Layer;
|
||||
use tracing_subscriber::layer::SubscriberExt;
|
||||
|
||||
#[path = "outbound_proxy_redirect_coverage_tests.rs"]
|
||||
mod redirect_coverage_tests;
|
||||
|
||||
struct MapEnv {
|
||||
values: HashMap<String, String>,
|
||||
}
|
||||
|
||||
fn spawn_proxy_listener() -> (std::net::SocketAddr, std::thread::JoinHandle<Vec<String>>) {
|
||||
spawn_http_listener(vec![
|
||||
"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok".to_string(),
|
||||
])
|
||||
}
|
||||
|
||||
fn spawn_redirect_listener(
|
||||
location: &str,
|
||||
) -> (std::net::SocketAddr, std::thread::JoinHandle<Vec<String>>) {
|
||||
spawn_http_listener(vec![format!(
|
||||
"HTTP/1.1 302 Found\r\nLocation: {location}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
|
||||
)])
|
||||
}
|
||||
|
||||
fn spawn_http_listener(
|
||||
responses: Vec<String>,
|
||||
) -> (std::net::SocketAddr, std::thread::JoinHandle<Vec<String>>) {
|
||||
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");
|
||||
listener
|
||||
.set_nonblocking(true)
|
||||
.expect("HTTP listener should become nonblocking");
|
||||
let thread = std::thread::spawn(move || {
|
||||
let mut requests = Vec::new();
|
||||
for response in responses {
|
||||
let deadline = Instant::now() + Duration::from_secs(2);
|
||||
let (mut stream, _) = loop {
|
||||
match listener.accept() {
|
||||
Ok(connection) => break connection,
|
||||
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
|
||||
assert!(
|
||||
Instant::now() < deadline,
|
||||
"HTTP listener should receive the next request"
|
||||
);
|
||||
std::thread::sleep(Duration::from_millis(10));
|
||||
}
|
||||
Err(error) => panic!("HTTP listener should accept: {error}"),
|
||||
}
|
||||
};
|
||||
stream
|
||||
.set_read_timeout(Some(Duration::from_secs(2)))
|
||||
.expect("HTTP stream should get a read timeout");
|
||||
requests.push(read_http_message(&mut stream));
|
||||
stream
|
||||
.write_all(response.as_bytes())
|
||||
.expect("HTTP listener should write response");
|
||||
}
|
||||
requests
|
||||
});
|
||||
(address, thread)
|
||||
}
|
||||
|
||||
fn only_request(thread: std::thread::JoinHandle<Vec<String>>, source: &str) -> String {
|
||||
let requests = thread
|
||||
.join()
|
||||
.unwrap_or_else(|_| panic!("{source} thread should finish"));
|
||||
let [request]: [String; 1] = requests.try_into().unwrap_or_else(|requests: Vec<String>| {
|
||||
panic!(
|
||||
"{source} should receive one request, got {}",
|
||||
requests.len()
|
||||
)
|
||||
});
|
||||
request
|
||||
}
|
||||
|
||||
fn read_http_message(stream: &mut impl Read) -> String {
|
||||
let mut buffer = Vec::new();
|
||||
let mut chunk = [0_u8; 1024];
|
||||
loop {
|
||||
let bytes_read = stream.read(&mut chunk).expect("HTTP message should read");
|
||||
if bytes_read == 0 {
|
||||
break;
|
||||
}
|
||||
buffer.extend_from_slice(&chunk[..bytes_read]);
|
||||
if let Some(header_end) = buffer.windows(4).position(|window| window == b"\r\n\r\n") {
|
||||
let body_start = header_end + 4;
|
||||
let headers = String::from_utf8_lossy(&buffer[..body_start]);
|
||||
let content_length = headers
|
||||
.lines()
|
||||
.filter_map(|line| line.split_once(':'))
|
||||
.find_map(|(name, value)| {
|
||||
name.eq_ignore_ascii_case("content-length")
|
||||
.then(|| value.trim().parse::<usize>().ok())
|
||||
.flatten()
|
||||
})
|
||||
.unwrap_or(0);
|
||||
if buffer.len() >= body_start + content_length {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
String::from_utf8_lossy(&buffer).into_owned()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn websocket_route_uses_http_equivalent_for_system_resolution() {
|
||||
let env = MapEnv {
|
||||
@@ -194,20 +300,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 +324,7 @@ 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 = only_request(proxy_thread, "proxy");
|
||||
|
||||
assert_eq!(response.status(), reqwest::StatusCode::OK);
|
||||
assert_eq!(
|
||||
@@ -240,6 +333,88 @@ 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 = only_request(proxy_thread, "proxy");
|
||||
|
||||
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_logs_only_the_final_redirect_outcome() {
|
||||
let (proxy_addr, proxy_thread) = spawn_proxy_listener();
|
||||
let redirected_url = "http://redirect-target.test/final?token=redirect-target-secret-value";
|
||||
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 log_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(&log_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 = tokio::time::timeout(Duration::from_secs(2), pool.get(&initial_url).send())
|
||||
.await
|
||||
.expect("redirected request should finish")
|
||||
.expect("redirected request should use selected routes");
|
||||
only_request(redirect_thread, "redirect");
|
||||
only_request(proxy_thread, "proxy");
|
||||
|
||||
assert_eq!(response.status(), reqwest::StatusCode::OK);
|
||||
let logs = String::from_utf8(log_buffer.lock().expect("log buffer lock").clone())
|
||||
.expect("logs should be UTF-8");
|
||||
assert!(logs.contains("log capture sentinel"));
|
||||
assert!(logs.contains(&initial_url));
|
||||
assert_eq!(logs.matches("Request completed").count(), 1);
|
||||
for secret in ["redirect-target-secret-value", redirected_url, "location"] {
|
||||
assert!(
|
||||
!logs
|
||||
.to_ascii_lowercase()
|
||||
.contains(&secret.to_ascii_lowercase())
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_pac_proxy_tokens() {
|
||||
assert_eq!(
|
||||
@@ -264,6 +439,40 @@ fn unavailable_system_proxy_decision_is_cached() {
|
||||
assert_eq!(cached_system_proxy_decision(request_url), Some(decision));
|
||||
}
|
||||
|
||||
#[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(())
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn system_proxy_resolution_is_single_flight() {
|
||||
let cache = Arc::new(Mutex::new(HashMap::new()));
|
||||
|
||||
Reference in New Issue
Block a user