mirror of
https://github.com/openai/codex.git
synced 2026-11-12 22:42:36 +00:00
Re-resolve system proxy routes across redirects (#34479)
## Why Route-aware clients select a direct or proxy route from the complete request URL. Letting `reqwest` follow redirects internally reuses the client chosen for the original URL, so the redirect destination cannot receive its own PAC or system proxy decision. ## What changed - Follow redirects through `RouteAwareClientPool` when respecting the system proxy, resolving and caching a route for every hop. - Preserve standard redirect method and body behavior, enforce one timeout across the request chain, and cap redirects at 10 hops. - Strip origin-sensitive credentials and stale proxy authorization when the destination or selected route changes, and apply safe `Referer` handling. - Log one final request outcome instead of intermediate redirect responses. ## Testing Added an integration test covering a redirect from a direct route to a proxy route, credential handling, and final-outcome logging. GitOrigin-RevId: e5cb2cc962e99d33df61ef649465e956b92279bb
This commit is contained in:
1 parent
c04452a240
commit
841e47b8fb
7 files changed
+476
-24
No files matched your search
@@ -118,6 +118,18 @@ impl HttpClient {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
pub(crate) fn log_error_summary(&self, method: &Method, url: &str, error: &reqwest::Error) {
|
||||||
|
if self.request_logging == RequestLogging::Enabled {
|
||||||
|
tracing::debug!(
|
||||||
|
method = %method,
|
||||||
|
url = %url,
|
||||||
|
status = error.status().map(|status| status.as_u16()),
|
||||||
|
is_timeout = error.is_timeout(),
|
||||||
|
is_connect = error.is_connect(),
|
||||||
|
"Request failed"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ mod error;
|
|||||||
mod outbound_proxy;
|
mod outbound_proxy;
|
||||||
mod request;
|
mod request;
|
||||||
mod route_aware_client_pool;
|
mod route_aware_client_pool;
|
||||||
|
mod route_aware_redirect;
|
||||||
mod transport;
|
mod transport;
|
||||||
|
|
||||||
pub use crate::chatgpt_cloudflare_cookies::with_chatgpt_cloudflare_cookie_store;
|
pub use crate::chatgpt_cloudflare_cookies::with_chatgpt_cloudflare_cookie_store;
|
||||||
|
|||||||
@@ -819,6 +819,10 @@ fn proxy_env_value(env: &dyn EnvSource, upper: &str) -> Option<String> {
|
|||||||
.filter(|value| !value.is_empty())
|
.filter(|value| !value.is_empty())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
#[path = "route_aware_redirect_integration_tests.rs"]
|
||||||
|
mod redirect_integration_tests;
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
#[path = "outbound_proxy_tests.rs"]
|
#[path = "outbound_proxy_tests.rs"]
|
||||||
mod tests;
|
mod tests;
|
||||||
@@ -12,6 +12,7 @@ use http::HeaderValue;
|
|||||||
use http::Method;
|
use http::Method;
|
||||||
use http::StatusCode;
|
use http::StatusCode;
|
||||||
use http::header::CONTENT_TYPE;
|
use http::header::CONTENT_TYPE;
|
||||||
|
use http::header::PROXY_AUTHORIZATION;
|
||||||
use reqwest::IntoUrl;
|
use reqwest::IntoUrl;
|
||||||
use serde::Serialize;
|
use serde::Serialize;
|
||||||
|
|
||||||
@@ -21,6 +22,12 @@ use crate::HttpClient;
|
|||||||
use crate::HttpClientFactory;
|
use crate::HttpClientFactory;
|
||||||
use crate::OutboundProxyPolicy;
|
use crate::OutboundProxyPolicy;
|
||||||
use crate::OutboundProxyRoute;
|
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;
|
use crate::with_chatgpt_cloudflare_cookie_store;
|
||||||
|
|
||||||
const MAX_CACHED_ROUTES: usize = 16;
|
const MAX_CACHED_ROUTES: usize = 16;
|
||||||
@@ -28,7 +35,8 @@ const MAX_CACHED_ROUTES: usize = 16;
|
|||||||
/// Reuses transport clients by resolved route while selecting a route for every request URL.
|
/// 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
|
/// 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)]
|
#[derive(Clone)]
|
||||||
pub struct RouteAwareClientPool {
|
pub struct RouteAwareClientPool {
|
||||||
http_client_factory: HttpClientFactory,
|
http_client_factory: HttpClientFactory,
|
||||||
@@ -73,18 +81,28 @@ pub enum RouteAwareRequestError {
|
|||||||
Route(#[from] RouteAwareClientPoolError),
|
Route(#[from] RouteAwareClientPoolError),
|
||||||
#[error("failed to build route-aware request: {0}")]
|
#[error("failed to build route-aware request: {0}")]
|
||||||
Build(String),
|
Build(String),
|
||||||
|
#[error("redirect target uses unsupported URL scheme: {0}")]
|
||||||
|
UnsupportedRedirectScheme(String),
|
||||||
|
#[error("too many redirects")]
|
||||||
|
TooManyRedirects,
|
||||||
|
#[error("route-aware request timed out")]
|
||||||
|
Timeout,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl RouteAwareRequestError {
|
impl RouteAwareRequestError {
|
||||||
pub fn status(&self) -> Option<StatusCode> {
|
pub fn status(&self) -> Option<StatusCode> {
|
||||||
match self {
|
match self {
|
||||||
Self::Request(error) => error.status(),
|
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 {
|
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 {
|
pub fn is_connect(&self) -> bool {
|
||||||
@@ -308,28 +326,137 @@ impl RouteAwareClientPool {
|
|||||||
&self,
|
&self,
|
||||||
request: reqwest::Request,
|
request: reqwest::Request,
|
||||||
) -> Result<reqwest::Response, RouteAwareRequestError> {
|
) -> Result<reqwest::Response, RouteAwareRequestError> {
|
||||||
let client = self.client_for_url(request.url().as_str()).await?;
|
|
||||||
Ok(client.execute(request).await?)
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn client_for_url(
|
|
||||||
&self,
|
|
||||||
request_url: &str,
|
|
||||||
) -> Result<HttpClient, RouteAwareClientPoolError> {
|
|
||||||
let http_client_factory = self.http_client_factory.clone();
|
let http_client_factory = self.http_client_factory.clone();
|
||||||
self.client_for_url_with_resolver(request_url, move |request_url| async move {
|
self.send_with_resolver(request, move |request_url| {
|
||||||
http_client_factory
|
let http_client_factory = http_client_factory.clone();
|
||||||
.resolve_proxy_route_async(request_url)
|
async move {
|
||||||
.await
|
http_client_factory
|
||||||
|
.resolve_proxy_route_async(request_url)
|
||||||
|
.await
|
||||||
|
}
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn send_with_resolver<F, Fut>(
|
||||||
|
&self,
|
||||||
|
mut request: reqwest::Request,
|
||||||
|
resolve_route: F,
|
||||||
|
) -> Result<reqwest::Response, RouteAwareRequestError>
|
||||||
|
where
|
||||||
|
F: Fn(String) -> Fut,
|
||||||
|
Fut: Future<Output = io::Result<OutboundProxyRoute>>,
|
||||||
|
{
|
||||||
|
let request_method = request.method().clone();
|
||||||
|
let request_url = request.url().to_string();
|
||||||
|
let follows_redirects_manually = self.http_client_factory.outbound_proxy_policy()
|
||||||
|
== OutboundProxyPolicy::RespectSystemProxy;
|
||||||
|
let timeout_deadline = request
|
||||||
|
.timeout()
|
||||||
|
.copied()
|
||||||
|
.map(|timeout| tokio::time::Instant::now() + timeout);
|
||||||
|
let mut redirects = 0;
|
||||||
|
let mut previous_route = None;
|
||||||
|
loop {
|
||||||
|
let current_url = request.url().clone();
|
||||||
|
let (current_route, client) = match timeout_deadline {
|
||||||
|
Some(timeout_deadline) => tokio::time::timeout_at(
|
||||||
|
timeout_deadline,
|
||||||
|
self.client_for_url_with_resolver(current_url.as_str(), &resolve_route),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.map_err(|_| RouteAwareRequestError::Timeout)??,
|
||||||
|
None => {
|
||||||
|
self.client_for_url_with_resolver(current_url.as_str(), &resolve_route)
|
||||||
|
.await?
|
||||||
|
}
|
||||||
|
};
|
||||||
|
if previous_route
|
||||||
|
.as_ref()
|
||||||
|
.is_some_and(|previous_route| previous_route != ¤t_route)
|
||||||
|
{
|
||||||
|
request.headers_mut().remove(PROXY_AUTHORIZATION);
|
||||||
|
}
|
||||||
|
previous_route = Some(current_route);
|
||||||
|
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 execute_request = async {
|
||||||
|
if follows_redirects_manually {
|
||||||
|
client.execute_without_request_logging(request).await
|
||||||
|
} else {
|
||||||
|
client.execute(request).await
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let response = match match timeout_deadline {
|
||||||
|
Some(timeout_deadline) => {
|
||||||
|
tokio::time::timeout_at(timeout_deadline, execute_request)
|
||||||
|
.await
|
||||||
|
.map_err(|_| RouteAwareRequestError::Timeout)?
|
||||||
|
}
|
||||||
|
None => execute_request.await,
|
||||||
|
} {
|
||||||
|
Ok(response) => response,
|
||||||
|
Err(error) => {
|
||||||
|
if follows_redirects_manually {
|
||||||
|
client.log_error_summary(&request_method, &request_url, &error);
|
||||||
|
}
|
||||||
|
return Err(error.into());
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let status = response.status();
|
||||||
|
if !is_redirect(status) {
|
||||||
|
if follows_redirects_manually {
|
||||||
|
client.log_response(&request_method, &request_url, &response);
|
||||||
|
}
|
||||||
|
return Ok(response);
|
||||||
|
}
|
||||||
|
let Some(next_url) = redirect_url(&response) else {
|
||||||
|
if follows_redirects_manually {
|
||||||
|
client.log_response(&request_method, &request_url, &response);
|
||||||
|
}
|
||||||
|
return Ok(response);
|
||||||
|
};
|
||||||
|
let Some(mut next_request) =
|
||||||
|
redirect_request(status, method, headers, version, timeout, replay, next_url)
|
||||||
|
else {
|
||||||
|
if follows_redirects_manually {
|
||||||
|
client.log_response(&request_method, &request_url, &response);
|
||||||
|
}
|
||||||
|
return Ok(response);
|
||||||
|
};
|
||||||
|
let next_request_url = next_request.url().clone();
|
||||||
|
if !matches!(next_request_url.scheme(), "http" | "https") {
|
||||||
|
return Err(RouteAwareRequestError::UnsupportedRedirectScheme(
|
||||||
|
next_request_url.scheme().to_string(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
if redirects >= MAX_REDIRECTS {
|
||||||
|
return Err(RouteAwareRequestError::TooManyRedirects);
|
||||||
|
}
|
||||||
|
remove_sensitive_headers(next_request.headers_mut(), ¤t_url, &next_request_url);
|
||||||
|
insert_referer(next_request.headers_mut(), ¤t_url, &next_request_url);
|
||||||
|
request = next_request;
|
||||||
|
redirects += 1;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
async fn client_for_url_with_resolver<F, Fut>(
|
async fn client_for_url_with_resolver<F, Fut>(
|
||||||
&self,
|
&self,
|
||||||
request_url: &str,
|
request_url: &str,
|
||||||
resolve_route: F,
|
resolve_route: F,
|
||||||
) -> Result<HttpClient, RouteAwareClientPoolError>
|
) -> Result<(OutboundProxyRoute, HttpClient), RouteAwareClientPoolError>
|
||||||
where
|
where
|
||||||
F: FnOnce(String) -> Fut,
|
F: FnOnce(String) -> Fut,
|
||||||
Fut: Future<Output = io::Result<OutboundProxyRoute>>,
|
Fut: Future<Output = io::Result<OutboundProxyRoute>>,
|
||||||
@@ -342,7 +469,7 @@ impl RouteAwareClientPool {
|
|||||||
Err(error) => panic!("route-aware client cache lock should not be poisoned: {error}"),
|
Err(error) => panic!("route-aware client cache lock should not be poisoned: {error}"),
|
||||||
};
|
};
|
||||||
if let Some(client) = clients.get(&route) {
|
if let Some(client) = clients.get(&route) {
|
||||||
return Ok(client.clone());
|
return Ok((route, client.clone()));
|
||||||
}
|
}
|
||||||
drop(clients);
|
drop(clients);
|
||||||
|
|
||||||
@@ -365,15 +492,15 @@ impl RouteAwareClientPool {
|
|||||||
Err(error) => panic!("route-aware client cache lock should not be poisoned: {error}"),
|
Err(error) => panic!("route-aware client cache lock should not be poisoned: {error}"),
|
||||||
};
|
};
|
||||||
if let Some(existing_client) = clients.get(&route) {
|
if let Some(existing_client) = clients.get(&route) {
|
||||||
return Ok(existing_client.clone());
|
return Ok((route, existing_client.clone()));
|
||||||
}
|
}
|
||||||
if clients.len() >= MAX_CACHED_ROUTES
|
if clients.len() >= MAX_CACHED_ROUTES
|
||||||
&& let Some(route_to_evict) = clients.keys().next().cloned()
|
&& let Some(route_to_evict) = clients.keys().next().cloned()
|
||||||
{
|
{
|
||||||
clients.remove(&route_to_evict);
|
clients.remove(&route_to_evict);
|
||||||
}
|
}
|
||||||
clients.insert(route, client.clone());
|
clients.insert(route.clone(), client.clone());
|
||||||
Ok(client)
|
Ok((route, client))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -249,8 +249,10 @@ async fn resolve_with(
|
|||||||
request_url: &str,
|
request_url: &str,
|
||||||
) -> Result<HttpClient, RouteAwareClientPoolError> {
|
) -> Result<HttpClient, RouteAwareClientPoolError> {
|
||||||
let resolver = resolver.clone();
|
let resolver = resolver.clone();
|
||||||
pool.client_for_url_with_resolver(request_url, move |request_url| async move {
|
let (_, client) = pool
|
||||||
resolver.resolve(request_url).await
|
.client_for_url_with_resolver(request_url, move |request_url| async move {
|
||||||
})
|
resolver.resolve(request_url).await
|
||||||
.await
|
})
|
||||||
|
.await?;
|
||||||
|
Ok(client)
|
||||||
}
|
}
|
||||||
@@ -0,0 +1,138 @@
|
|||||||
|
//! Redirect semantics for requests whose proxy route is selected from the complete URL.
|
||||||
|
//!
|
||||||
|
//! Reqwest normally follows redirects inside one `Client::execute` call. That is safe for
|
||||||
|
//! `ReqwestDefault`, where reqwest owns proxy selection, but not for `RespectSystemProxy`: Codex
|
||||||
|
//! resolves PAC and operating-system proxy settings before building a client pinned to the
|
||||||
|
//! resulting direct or concrete-proxy route. An internal reqwest redirect would reuse that client
|
||||||
|
//! for the new URL without giving [`crate::RouteAwareClientPool`] an opportunity to resolve the
|
||||||
|
//! redirect destination.
|
||||||
|
//!
|
||||||
|
//! The pool therefore disables reqwest redirects for `RespectSystemProxy`, observes each redirect
|
||||||
|
//! response, and uses these helpers to construct the next request before resolving its route. The
|
||||||
|
//! helpers preserve the relevant reqwest method/body behavior while applying stricter
|
||||||
|
//! origin-sensitive credential and `Referer` rules at the boundary between requests.
|
||||||
|
|
||||||
|
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,
|
||||||
|
) {
|
||||||
|
if !same_origin(previous, next) {
|
||||||
|
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,
|
||||||
|
) {
|
||||||
|
headers.remove(REFERER);
|
||||||
|
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 !same_origin(previous, next) {
|
||||||
|
referer.set_path("/");
|
||||||
|
referer.set_query(None);
|
||||||
|
}
|
||||||
|
if let Ok(value) = referer.as_str().parse() {
|
||||||
|
headers.insert(REFERER, value);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn same_origin(previous: &reqwest::Url, next: &reqwest::Url) -> bool {
|
||||||
|
previous.scheme() == next.scheme()
|
||||||
|
&& previous.host_str() == next.host_str()
|
||||||
|
&& previous.port_or_known_default() == next.port_or_known_default()
|
||||||
|
}
|
||||||
@@ -0,0 +1,168 @@
|
|||||||
|
use std::io;
|
||||||
|
use std::io::Read;
|
||||||
|
use std::io::Write;
|
||||||
|
use std::sync::Arc;
|
||||||
|
use std::sync::Mutex;
|
||||||
|
use std::time::Duration;
|
||||||
|
use std::time::Instant;
|
||||||
|
|
||||||
|
use http::header::AUTHORIZATION;
|
||||||
|
use http::header::COOKIE;
|
||||||
|
use http::header::PROXY_AUTHORIZATION;
|
||||||
|
use pretty_assertions::assert_eq;
|
||||||
|
use tracing_subscriber::Layer;
|
||||||
|
use tracing_subscriber::layer::SubscriberExt;
|
||||||
|
|
||||||
|
use super::*;
|
||||||
|
use crate::RouteAwareClientPool;
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn route_aware_pool_re_resolves_redirects_and_logs_only_final_outcome() {
|
||||||
|
let (proxy_addr, proxy_thread) =
|
||||||
|
spawn_response("HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok");
|
||||||
|
let (redirect_addr, redirect_thread) = spawn_response(
|
||||||
|
"HTTP/1.1 302 Found\r\nLocation: /final?token=redirect-target-secret\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
|
||||||
|
);
|
||||||
|
let initial_url = format!("http://{redirect_addr}/start");
|
||||||
|
let redirected_url = format!("http://{redirect_addr}/final?token=redirect-target-secret");
|
||||||
|
cache_system_proxy_decision(&initial_url, SystemProxyDecision::Direct);
|
||||||
|
cache_system_proxy_decision(
|
||||||
|
&redirected_url,
|
||||||
|
SystemProxyDecision::Proxy {
|
||||||
|
url: format!("http://{proxy_addr}"),
|
||||||
|
},
|
||||||
|
);
|
||||||
|
let pool = 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);
|
||||||
|
|
||||||
|
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("redirected request should use both selected routes");
|
||||||
|
redirect_thread
|
||||||
|
.join()
|
||||||
|
.expect("redirect listener should finish");
|
||||||
|
let redirected_request = proxy_thread.join().expect("proxy listener should finish");
|
||||||
|
assert_eq!(
|
||||||
|
redirected_request.lines().next(),
|
||||||
|
Some(format!("GET {redirected_url} HTTP/1.1").as_str())
|
||||||
|
);
|
||||||
|
assert!(has_header(&redirected_request, "authorization"));
|
||||||
|
assert!(has_header(&redirected_request, "cookie"));
|
||||||
|
assert!(!has_header(&redirected_request, "proxy-authorization"));
|
||||||
|
|
||||||
|
let logs = String::from_utf8(log_buffer.lock().expect("log buffer lock").clone())
|
||||||
|
.expect("logs should be UTF-8");
|
||||||
|
assert!(logs.contains(&initial_url));
|
||||||
|
assert_eq!(logs.matches("Request completed").count(), 1);
|
||||||
|
assert!(!logs.contains("redirect-target-secret"));
|
||||||
|
}
|
||||||
|
|
||||||
|
fn spawn_response(
|
||||||
|
response: &'static str,
|
||||||
|
) -> (std::net::SocketAddr, std::thread::JoinHandle<String>) {
|
||||||
|
let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).expect("bind listener");
|
||||||
|
let address = listener.local_addr().expect("HTTP listener address");
|
||||||
|
listener.set_nonblocking(true).expect("set 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, "request timed out");
|
||||||
|
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("read timeout");
|
||||||
|
let request = read_http_headers(&mut stream);
|
||||||
|
stream
|
||||||
|
.write_all(response.as_bytes())
|
||||||
|
.expect("write response");
|
||||||
|
request
|
||||||
|
});
|
||||||
|
(address, thread)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn read_http_headers(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 headers should read");
|
||||||
|
if bytes_read == 0 {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
buffer.extend_from_slice(&chunk[..bytes_read]);
|
||||||
|
if buffer.windows(4).any(|window| window == b"\r\n\r\n") {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
String::from_utf8_lossy(&buffer).into_owned()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn has_header(request: &str, expected_name: &str) -> bool {
|
||||||
|
request.lines().any(|line| {
|
||||||
|
line.split_once(':')
|
||||||
|
.is_some_and(|(name, _)| name.eq_ignore_ascii_case(expected_name))
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
#[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(())
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in new issue
Block a user