Route environment registry requests through the shared HTTP client (#35034)

## Why

Noise environment registry requests need to follow the exec server's effective
outbound proxy policy without exposing registry URLs or response headers in HTTP
diagnostics.

## What changed

- Build the registry client from the supplied `HttpClientFactory` and use a
  route-aware API client with redirects and request logging disabled.
- Defer construction of the Noise connection provider until the outbound HTTP
  policy is available.
- Map route-aware request failures into registry errors while retaining timeout
  detection across response body reads.

## Testing

Add coverage for system-proxy routing, sensitive registry metadata redaction,
stalled response-body timeouts, and prepared Noise configuration validation.

GitOrigin-RevId: d312dfe037f72732085bf38109af44df76ed0b53
This commit is contained in:
Celia Chen
2026-07-23 23:17:56 +00:00
committed by copyberry
parent 41775559ca
commit d45055ae58
7 changed files with 385 additions and 40 deletions

View File

@@ -563,7 +563,7 @@ pub enum ExecServerError {
Server { code: i64, message: String },
#[error("environment registry request failed ({status}{code_suffix}): {message}", code_suffix = .code.as_ref().map(|code| format!(", {code}")).unwrap_or_default())]
EnvironmentRegistryHttp {
status: reqwest::StatusCode,
status: http::StatusCode,
code: Option<String>,
message: String,
},
@@ -572,7 +572,7 @@ pub enum ExecServerError {
#[error("environment registry authentication error: {0}")]
EnvironmentRegistryAuth(String),
#[error("environment registry request failed: {0}")]
EnvironmentRegistryRequest(#[from] reqwest::Error),
EnvironmentRegistryRequest(#[from] codex_http_client::RouteAwareRequestError),
#[error("exec-server connection attempt failed: {0}")]
ConnectionAttempt(#[source] Arc<ExecServerError>),
}

View File

@@ -597,12 +597,12 @@ fn is_retryable_registry_error(error: &ExecServerError) -> bool {
error,
ExecServerError::EnvironmentRegistryHttp { status, code, .. }
if status.is_server_error()
|| *status == reqwest::StatusCode::REQUEST_TIMEOUT
|| *status == reqwest::StatusCode::TOO_MANY_REQUESTS
|| *status == http::StatusCode::REQUEST_TIMEOUT
|| *status == http::StatusCode::TOO_MANY_REQUESTS
// TODO: Replace this coarse retry with an explicit registry/presence
// recovery FSM so `environment_offline` is retried only while the
// executor is expected to reconnect.
|| (*status == reqwest::StatusCode::CONFLICT
|| (*status == http::StatusCode::CONFLICT
&& code.as_deref() == Some("environment_offline"))
)
}

View File

@@ -226,6 +226,7 @@ impl EnvironmentManager {
local_runtime_paths: Option<ExecServerRuntimePaths>,
http_client_factory: HttpClientFactory,
) -> Result<Self, ExecServerError> {
let connect_provider = config.into_connect_provider(http_client_factory.clone())?;
let manager = Self {
default_environment: Some(REMOTE_ENVIRONMENT_ID.to_string()),
environments: RwLock::new(HashMap::new()),
@@ -233,10 +234,7 @@ impl EnvironmentManager {
local_runtime_paths,
http_client_factory,
};
manager.upsert_noise_environment(
REMOTE_ENVIRONMENT_ID.to_string(),
config.connect_provider(),
)?;
manager.upsert_noise_environment(REMOTE_ENVIRONMENT_ID.to_string(), connect_provider)?;
Ok(manager)
}
@@ -584,13 +582,13 @@ fn noise_environment_config_from_values(
}
};
let config = NoiseRendezvousEnvironmentConfig::new(
NoiseRendezvousEnvironmentConfig::new(
registry_url,
environment_id,
auth_token,
chatgpt_account_id,
)?;
Ok(Some(config))
)
.map(Some)
}
fn optional_environment_value(name: &str) -> Option<String> {
@@ -939,10 +937,14 @@ mod tests {
let manager = EnvironmentManager::from_noise_environment_config(
config,
/*local_runtime_paths*/ None,
legacy_http_client_factory(),
HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy),
)
.expect("build environment manager");
assert_eq!(
manager.http_client_factory().outbound_proxy_policy(),
OutboundProxyPolicy::RespectSystemProxy
);
assert_eq!(
manager.default_environment_id(),
Some(REMOTE_ENVIRONMENT_ID)

View File

@@ -12,6 +12,7 @@ use crate::LOCAL_ENVIRONMENT_ID;
use crate::REMOTE_ENVIRONMENT_ID;
use crate::environment_provider::EnvironmentDefault;
use crate::environment_provider::EnvironmentProviderSnapshot;
use crate::remote::NoiseRendezvousEnvironmentConfig;
#[test]
fn prepared_remote_environment_is_detected_without_starting_a_connection() {
@@ -35,6 +36,70 @@ fn prepared_remote_environment_is_detected_without_starting_a_connection() {
);
}
#[test]
fn prepared_noise_environment_is_detected_before_http_policy_is_resolved() {
let config = NoiseRendezvousEnvironmentConfig::new(
"https://registry-user:registry-password@registry.example/api?access_token=query-secret#fragment-secret"
.to_string(),
"environment-requested".to_string(),
"registry-token".to_string(),
Some("workspace-123".to_string()),
)
.expect("Noise environment configuration");
let prepared = PreparedEnvironmentManager {
source: PreparedEnvironmentSource::Noise(config),
};
assert!(prepared.default_environment_is_remote());
let debug = format!("{prepared:?}");
assert!(debug.contains("<redacted>"));
assert!(!debug.contains("registry-token"));
assert!(!debug.contains("workspace-123"));
assert!(!debug.contains("registry-user"));
assert!(!debug.contains("registry-password"));
assert!(!debug.contains("registry.example"));
assert!(!debug.contains("query-secret"));
assert!(!debug.contains("fragment-secret"));
}
#[test]
fn prepared_noise_environment_rejects_invalid_configuration() {
let invalid_configs = [
("", "environment-requested", "registry-token", None),
("https://registry.example", "", "registry-token", None),
(
"https://registry.example",
"environment-requested",
"",
None,
),
(
"https://registry.example",
"environment-requested",
"registry\ntoken",
None,
),
(
"https://registry.example",
"environment-requested",
"registry-token",
Some("workspace\n123"),
),
];
for (registry_url, environment_id, auth_token, chatgpt_account_id) in invalid_configs {
let result = NoiseRendezvousEnvironmentConfig::new(
registry_url.to_string(),
environment_id.to_string(),
auth_token.to_string(),
chatgpt_account_id.map(str::to_string),
);
assert!(result.is_err());
}
}
#[test]
fn prepared_local_and_disabled_environments_are_not_remote() {
let local = PreparedEnvironmentManager {

View File

@@ -4,7 +4,10 @@ use std::time::Instant;
use codex_api::AuthProvider;
use codex_api::SharedAuthProvider;
use codex_http_client::ClientRouteClass;
use codex_http_client::HttpClientFactory;
use codex_http_client::HttpResponse;
use codex_http_client::RouteAwareClientPool;
use futures::FutureExt;
use http::HeaderMap;
use http::HeaderName;
@@ -49,7 +52,7 @@ const NOISE_RELAY_SECURITY_PROFILE: &str = "noise_hybrid_ik_v1";
struct EnvironmentRegistryClient {
base_url: String,
auth_provider: SharedAuthProvider,
http: reqwest::Client,
http: RouteAwareClientPool,
connect_timeout: Duration,
telemetry: ExecServerTelemetry,
}
@@ -66,21 +69,28 @@ impl std::fmt::Debug for EnvironmentRegistryClient {
impl EnvironmentRegistryClient {
#[cfg(test)]
fn new(base_url: String, auth_provider: SharedAuthProvider) -> Result<Self, ExecServerError> {
Self::new_with_telemetry(base_url, auth_provider, ExecServerTelemetry::default())
Self::new_with_telemetry(
base_url,
auth_provider,
ExecServerTelemetry::default(),
HttpClientFactory::new(codex_http_client::OutboundProxyPolicy::ReqwestDefault),
)
}
fn new_with_telemetry(
base_url: String,
auth_provider: SharedAuthProvider,
telemetry: ExecServerTelemetry,
http_client_factory: HttpClientFactory,
) -> Result<Self, ExecServerError> {
let base_url = normalize_base_url(base_url)?;
Ok(Self {
base_url,
auth_provider,
http: reqwest::Client::builder()
.redirect(reqwest::redirect::Policy::none())
.build()?,
http: RouteAwareClientPool::new_without_redirects_or_request_logging(
http_client_factory,
ClientRouteClass::Api,
),
connect_timeout: DEFAULT_REMOTE_EXEC_SERVER_CONNECT_TIMEOUT,
telemetry,
})
@@ -206,15 +216,15 @@ impl EnvironmentRegistryClient {
})
}
async fn parse_json_response<R>(
&self,
response: reqwest::Response,
) -> Result<R, ExecServerError>
async fn parse_json_response<R>(&self, response: HttpResponse) -> Result<R, ExecServerError>
where
R: for<'de> Deserialize<'de>,
{
if response.status().is_success() {
return response.json::<R>().await.map_err(ExecServerError::from);
return response
.json::<R>()
.await
.map_err(|error| ExecServerError::EnvironmentRegistryRequest(error.into()));
}
let status = response.status();
@@ -276,7 +286,8 @@ impl HarnessKeyValidator for RegistryHarnessKeyValidator {
}
let response = response
.json::<EnvironmentRegistryHarnessKeyValidationResponse>()
.await?;
.await
.map_err(|error| ExecServerError::EnvironmentRegistryRequest(error.into()))?;
if !response.valid {
return Err(ExecServerError::Protocol(
"environment registry rejected Noise relay harness key".to_string(),
@@ -288,17 +299,22 @@ impl HarnessKeyValidator for RegistryHarnessKeyValidator {
/// Noise connection configuration for a Codex harness.
///
/// The provider holds the authenticated registry client so every reconnect
/// receives fresh URL and harness-key authorization material.
/// Configuration stays inert until the effective outbound HTTP policy is known.
/// Its connection provider then holds the authenticated registry client so every
/// reconnect receives fresh URL and harness-key authorization material.
#[derive(Clone)]
pub(crate) struct NoiseRendezvousEnvironmentConfig {
provider: Arc<dyn NoiseRendezvousConnectProvider>,
base_url: String,
environment_id: String,
auth_provider: SharedAuthProvider,
}
impl std::fmt::Debug for NoiseRendezvousEnvironmentConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("NoiseRendezvousEnvironmentConfig")
.field("provider", &"<redacted>")
.field("base_url", &"<redacted>")
.field("environment_id", &self.environment_id)
.field("auth_provider", &"<redacted>")
.finish()
}
}
@@ -310,23 +326,30 @@ impl NoiseRendezvousEnvironmentConfig {
bearer_token: String,
chatgpt_account_id: Option<String>,
) -> Result<Self, ExecServerError> {
let base_url = normalize_base_url(base_url)?;
let environment_id = normalize_environment_id(environment_id)?;
let auth_provider = static_bearer_auth_provider(bearer_token, chatgpt_account_id)?;
let client = EnvironmentRegistryClient::new_with_telemetry(
base_url,
auth_provider,
ExecServerTelemetry::default(),
)?;
Ok(Self {
provider: Arc::new(EnvironmentRegistryNoiseConnectProvider {
client,
environment_id,
}),
base_url,
environment_id,
auth_provider,
})
}
pub(crate) fn connect_provider(&self) -> Arc<dyn NoiseRendezvousConnectProvider> {
Arc::clone(&self.provider)
pub(crate) fn into_connect_provider(
self,
http_client_factory: HttpClientFactory,
) -> Result<Arc<dyn NoiseRendezvousConnectProvider>, ExecServerError> {
let client = EnvironmentRegistryClient::new_with_telemetry(
self.base_url,
self.auth_provider,
ExecServerTelemetry::default(),
http_client_factory,
)?;
Ok(Arc::new(EnvironmentRegistryNoiseConnectProvider {
client,
environment_id: self.environment_id,
}))
}
}
@@ -475,6 +498,7 @@ pub async fn run_remote_environment(
config.base_url.clone(),
config.auth_provider.clone(),
config.telemetry.clone(),
config.http_client_factory.clone(),
)?;
let processor = ConnectionProcessor::new_with_telemetry(
runtime_paths,
@@ -685,6 +709,8 @@ mod tests {
use opentelemetry::trace::TracerProvider as _;
use opentelemetry_sdk::trace::SdkTracerProvider;
use pretty_assertions::assert_eq;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpListener;
use tracing::Instrument;
use tracing_subscriber::prelude::*;
use wiremock::Mock;
@@ -805,7 +831,10 @@ mod tests {
.expect("noise configuration");
let bundle = config
.connect_provider()
.into_connect_provider(HttpClientFactory::new(
codex_http_client::OutboundProxyPolicy::ReqwestDefault,
))
.expect("Noise connect provider")
.connect_bundle(harness_public_key)
.await
.expect("Noise connect bundle");
@@ -850,6 +879,52 @@ mod tests {
));
}
#[tokio::test]
async fn connect_environment_times_out_when_registry_response_body_stalls() {
let listener = TcpListener::bind("127.0.0.1:0")
.await
.expect("registry listener should bind");
let registry_url = format!(
"http://{}",
listener
.local_addr()
.expect("registry listener should have an address")
);
tokio::spawn(async move {
let (mut stream, _) = listener
.accept()
.await
.expect("registry request should connect");
stream
.write_all(
b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 256\r\n\r\n{",
)
.await
.expect("registry response headers should write");
sleep(Duration::from_secs(1)).await;
});
let mut client =
EnvironmentRegistryClient::new(registry_url, static_registry_auth_provider())
.expect("client");
client.connect_timeout = Duration::from_millis(50);
let harness_public_key = NoiseChannelIdentity::generate()
.expect("identity")
.public_key();
let error = match client
.connect_environment("environment-requested", harness_public_key)
.await
{
Ok(_) => panic!("stalled connect response body should time out"),
Err(error) => error,
};
assert!(matches!(
error,
ExecServerError::EnvironmentRegistryRequest(error) if error.is_timeout()
));
}
#[tokio::test]
async fn register_environment_does_not_follow_redirects_with_auth_headers() {
let server = MockServer::start().await;

View File

@@ -1,4 +1,6 @@
use std::io::Write;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use anyhow::Result;
@@ -6,6 +8,7 @@ use codex_api::AuthProvider;
use codex_api::SharedAuthProvider;
use codex_http_client::HttpClientFactory;
use codex_http_client::OutboundProxyPolicy;
use codex_http_client::cache_system_proxy_route_for_test;
use http::HeaderMap;
use http::HeaderValue;
use tokio::io::AsyncReadExt;
@@ -13,6 +16,8 @@ use tokio::io::AsyncWriteExt;
use tokio::net::TcpListener;
use tokio::time::timeout;
use tokio_tungstenite::accept_async;
use tracing_subscriber::Layer;
use tracing_subscriber::layer::SubscriberExt;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
@@ -20,6 +25,7 @@ use wiremock::matchers::body_partial_json;
use wiremock::matchers::header;
use wiremock::matchers::method;
use wiremock::matchers::path;
use wiremock::matchers::query_param;
use super::*;
@@ -41,6 +47,123 @@ fn static_registry_auth_provider() -> SharedAuthProvider {
Arc::new(StaticRegistryAuthProvider)
}
#[tokio::test(flavor = "current_thread")]
async fn registry_requests_do_not_log_sensitive_urls_or_response_headers() -> Result<()> {
let log_buffer = Arc::new(Mutex::new(Vec::new()));
let writer_buffer = Arc::clone(&log_buffer);
let subscriber = tracing_subscriber::registry().with(
tracing_subscriber::fmt::layer()
.with_ansi(false)
.with_writer(move || RegistryLogWriter(Arc::clone(&writer_buffer)))
.with_filter(
tracing_subscriber::filter::Targets::new()
.with_target("codex_http_client", tracing::Level::TRACE)
.with_target("codex_exec_server", tracing::Level::TRACE),
),
);
let _guard = tracing::subscriber::set_default(subscriber);
tracing::debug!(target: "codex_exec_server", "registry log capture sentinel");
let server = MockServer::start().await;
let harness_public_key = NoiseChannelIdentity::generate()?.public_key();
let executor_public_key = NoiseChannelIdentity::generate()?.public_key();
for (operation, response, cookie_secret, location_secret) in [
(
"register",
serde_json::json!({
"environment_id": "environment-requested",
"url": "wss://rendezvous.test/environment",
"security_profile": NOISE_RELAY_SECURITY_PROFILE,
"executor_registration_id": "registration-1",
}),
"register-cookie-secret",
"register-location-secret",
),
(
"connect",
serde_json::json!({
"environment_id": "environment-requested",
"url": "wss://rendezvous.test/harness",
"security_profile": NOISE_RELAY_SECURITY_PROFILE,
"executor_registration_id": "registration-1",
"executor_public_key": executor_public_key.clone(),
"harness_key_authorization": HARNESS_KEY_AUTHORIZATION,
}),
"connect-cookie-secret",
"connect-location-secret",
),
(
"validate",
serde_json::json!({ "valid": true }),
"validate-cookie-secret",
"validate-location-secret",
),
] {
Mock::given(method("POST"))
.and(path("/registry-path-secret"))
.and(query_param(
"registry_token",
format!(
"registry-query-secret/cloud/environment/environment-requested/{operation}"
),
))
.respond_with(
ResponseTemplate::new(200)
.insert_header("set-cookie", format!("session={cookie_secret}"))
.insert_header(
"location",
format!("https://registry.example/private?token={location_secret}"),
)
.set_body_json(response),
)
.expect(1)
.mount(&server)
.await;
}
let registry_url =
server
.uri()
.replacen("http://", "http://registry-user:registry-password@", 1);
let registry_url =
format!("{registry_url}/registry-path-secret?registry_token=registry-query-secret");
let client = EnvironmentRegistryClient::new(registry_url, static_registry_auth_provider())?;
client
.register_environment("environment-requested", &executor_public_key)
.await?;
client
.connect_environment("environment-requested", harness_public_key.clone())
.await?;
RegistryHarnessKeyValidator {
client,
environment_id: "environment-requested".to_string(),
executor_registration_id: "registration-1".to_string(),
}
.validate_harness_key(&harness_public_key, HARNESS_KEY_AUTHORIZATION)
.await?;
let logs = String::from_utf8(log_buffer.lock().expect("log buffer lock").clone())?;
assert!(logs.contains("registry log capture sentinel"));
for secret in [
"registry-user",
"registry-password",
"registry-path-secret",
"registry-query-secret",
"registry-token",
HARNESS_KEY_AUTHORIZATION,
"register-cookie-secret",
"register-location-secret",
"connect-cookie-secret",
"connect-location-secret",
"validate-cookie-secret",
"validate-location-secret",
] {
assert!(!logs.contains(secret), "logs exposed {secret}:\n{logs}");
}
Ok(())
}
#[tokio::test]
async fn reconnect_reuses_registration_until_url_is_rejected() -> Result<()> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
@@ -95,6 +218,56 @@ async fn reconnect_reuses_registration_until_url_is_rejected() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn noise_connect_provider_uses_supplied_system_proxy_policy() -> Result<()> {
let proxy = MockServer::start().await;
let registry_url = "http://registry-policy-proxy.test";
let request_url = format!("{registry_url}/cloud/environment/environment-requested/connect");
cache_system_proxy_route_for_test(&request_url, proxy.uri());
let harness_public_key = NoiseChannelIdentity::generate()?.public_key();
let executor_public_key = NoiseChannelIdentity::generate()?.public_key();
Mock::given(method("POST"))
.and(path("/cloud/environment/environment-requested/connect"))
.and(header("authorization", "Bearer registry-token"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"environment_id": "environment-requested",
"url": "wss://rendezvous.test/cloud-agent/default/ws/environment/environment-requested",
"security_profile": NOISE_RELAY_SECURITY_PROFILE,
"executor_registration_id": "registration-1",
"executor_public_key": executor_public_key.clone(),
"harness_key_authorization": HARNESS_KEY_AUTHORIZATION,
})))
.expect(1)
.mount(&proxy)
.await;
let provider = NoiseRendezvousEnvironmentConfig::new(
registry_url.to_string(),
"environment-requested".to_string(),
"registry-token".to_string(),
/*chatgpt_account_id*/ None,
)?
.into_connect_provider(HttpClientFactory::new(
OutboundProxyPolicy::RespectSystemProxy,
))?;
let bundle = timeout(
Duration::from_secs(5),
provider.connect_bundle(harness_public_key),
)
.await??;
let requests = proxy
.received_requests()
.await
.expect("proxy request recording should be enabled");
assert_eq!(requests.len(), 1);
assert_eq!(requests[0].url.as_str(), request_url);
assert_eq!(bundle.executor_public_key, executor_public_key);
Ok(())
}
#[tokio::test]
async fn validate_harness_key_requires_explicit_valid_response() {
let server = MockServer::start().await;
@@ -164,3 +337,19 @@ async fn validate_harness_key_does_not_expose_error_body() {
if message == "environment registry harness key validation failed"
));
}
struct RegistryLogWriter(Arc<Mutex<Vec<u8>>>);
impl Write for RegistryLogWriter {
fn write(&mut self, buffer: &[u8]) -> std::io::Result<usize> {
self.0
.lock()
.expect("log buffer lock")
.extend_from_slice(buffer);
Ok(buffer.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}

View File

@@ -241,6 +241,20 @@ impl RouteAwareClientPool {
)
}
/// Creates a no-redirect pool without request URL or response-header diagnostics.
pub fn new_without_redirects_or_request_logging(
http_client_factory: HttpClientFactory,
route_class: ClientRouteClass,
) -> Self {
Self::with_builder(
http_client_factory,
route_class,
HttpClientBuilder::new()
.without_redirects()
.without_request_logging(),
)
}
/// Creates a pool whose clients limit only connection establishment.
///
/// The timeout applies to every client built for a resolved route, including redirect hops.