From 0003732489a514748e61a0f76a652ac992a5f4f5 Mon Sep 17 00:00:00 2001 From: viyatb-oai Date: Thu, 15 Oct 2026 16:40:58 +0000 Subject: [PATCH] Add scoped executor credential auth for remote exec-server (#54346) ## What changed - Add `RemoteEnvironmentConfig::with_executor_credential_auth()` for Noise transport, requiring exactly one `Authorization: Bearer ` header without accompanying account credentials. - Require HTTPS registry URLs except on loopback, and bypass system proxies for loopback HTTP requests. - Mark credentials as sensitive, suppress registry diagnostics that could echo them, and reject registration responses containing the credential while preserving recognized retry error codes. - Reject scoped executor credentials for Direct registration and WebSocket connections. Keep legacy authentication as the default. ## Testing Add tests for invalid credential headers, HTTPS and loopback restrictions, proxy bypass, credential echoes in registry responses, and rejection before Direct transport sends requests. GitOrigin-RevId: a427d648a0520d2e187dd40ca63268903a949f94 --- codex-rs/exec-server/src/remote.rs | 565 +++++++++++++++++- codex-rs/exec-server/src/remote/direct.rs | 13 + .../exec-server/src/remote/direct_tests.rs | 52 ++ 3 files changed, 600 insertions(+), 30 deletions(-) diff --git a/codex-rs/exec-server/src/remote.rs b/codex-rs/exec-server/src/remote.rs index d432853ed4..aeb4467000 100644 --- a/codex-rs/exec-server/src/remote.rs +++ b/codex-rs/exec-server/src/remote.rs @@ -13,6 +13,7 @@ use http::HeaderMap; use http::HeaderName; use http::HeaderValue; use http::StatusCode; +use http::header::AUTHORIZATION; use serde::Deserialize; use tokio::time::sleep; use tokio::time::timeout_at; @@ -72,11 +73,19 @@ pub enum RemoteEnvironmentTransport { Direct, } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum RegistryAuthMode { + Legacy, + ExecutorCredential, +} + #[derive(Clone)] struct EnvironmentRegistryClient { base_url: String, auth_provider: SharedAuthProvider, + auth_mode: RegistryAuthMode, http: RouteAwareClientPool, + http_direct: RouteAwareClientPool, connect_timeout: Duration, telemetry: ExecServerTelemetry, } @@ -111,10 +120,16 @@ impl EnvironmentRegistryClient { Ok(Self { base_url, auth_provider, + auth_mode: RegistryAuthMode::Legacy, http: RouteAwareClientPool::new_without_redirects_or_request_logging( - http_client_factory, + http_client_factory.clone(), ClientRouteClass::Api, ), + http_direct: RouteAwareClientPool::new_without_redirects_or_request_logging( + http_client_factory, + ClientRouteClass::Api, + ) + .with_legacy_direct_proxy_and_custom_ca_fallback(), connect_timeout: DEFAULT_REMOTE_EXEC_SERVER_CONNECT_TIMEOUT, telemetry, }) @@ -161,15 +176,18 @@ impl EnvironmentRegistryClient { security_profile: NOISE_RELAY_SECURITY_PROFILE.to_string(), executor_public_key: executor_public_key.clone(), }; - let response = timeout_at(deadline, async { - self.http + let (response, executor_credential) = timeout_at(deadline, async { + let (auth_headers, executor_credential) = self.resolve_auth().await?; + let response = self + .http_for_auth() .post(url) - .headers(self.resolve_auth_headers().await?) + .headers(auth_headers) .headers(current_trace_context_headers()) .json(&body) .send() .await - .map_err(ExecServerError::EnvironmentRegistryRequest) + .map_err(ExecServerError::EnvironmentRegistryRequest)?; + Ok((response, executor_credential)) }) .await .unwrap_or_else(|_| { @@ -179,22 +197,30 @@ impl EnvironmentRegistryClient { })?; let status = response.status(); // Read diagnostics within the same attempt budget, preserving a known error status. - let response: EnvironmentRegistryRegistrationResponse = - timeout_at(deadline, self.parse_json_response(response)) - .await - .unwrap_or_else(|_| { - Err(match status { - StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => { - environment_registry_auth_error(status, "response body timed out") - } - status if !status.is_success() => { - environment_registry_http_error(status, "response body timed out") - } - _ => ExecServerError::EnvironmentRegistryRequest( - codex_http_client::RouteAwareRequestError::Timeout, - ), - }) - })?; + let response: EnvironmentRegistryRegistrationResponse = timeout_at(deadline, async { + if let Some(credential) = executor_credential + .as_ref() + .and_then(|value| value.to_str().ok()) + { + parse_executor_registration_response(response, credential).await + } else { + self.parse_json_response(response).await + } + }) + .await + .unwrap_or_else(|_| { + Err(match status { + StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => { + environment_registry_auth_error(status, "response body timed out") + } + status if !status.is_success() => { + environment_registry_http_error(status, "response body timed out") + } + _ => ExecServerError::EnvironmentRegistryRequest( + codex_http_client::RouteAwareRequestError::Timeout, + ), + }) + })?; if response.environment_id != environment_id { return Err(ExecServerError::Protocol( "environment registry returned a different environment id".to_string(), @@ -240,10 +266,11 @@ impl EnvironmentRegistryClient { &format!("/cloud/environment/{environment_id}/connect"), ); let body = EnvironmentRegistryConnectRequest { harness_public_key }; + let (auth_headers, _) = self.resolve_auth().await?; let response = self - .http + .http_for_auth() .post(url) - .headers(self.resolve_auth_headers().await?) + .headers(auth_headers) .headers(current_trace_context_headers()) .json(&body) .timeout(self.connect_timeout) @@ -279,15 +306,59 @@ impl EnvironmentRegistryClient { }) } - async fn resolve_auth_headers(&self) -> Result { - self.auth_provider + async fn resolve_auth(&self) -> Result<(HeaderMap, Option), ExecServerError> { + let mut headers = self + .auth_provider .resolve_auth_headers() .await .map_err(|error| { ExecServerError::EnvironmentRegistryAuth(format!( "failed to resolve environment registry authentication: {error}" )) - }) + })?; + let executor_credential = if self.auth_mode == RegistryAuthMode::ExecutorCredential { + // Scoped mode sends exactly one bearer and never accompanying account credentials. + let token = (headers.len() == 1) + .then(|| headers.get(AUTHORIZATION)) + .flatten() + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.strip_prefix("Bearer ")) + .filter(|token| { + !token.is_empty() + && token.len() <= 16 * 1024 + && token.bytes().all(|byte| byte.is_ascii_graphic()) + }) + .ok_or_else(|| { + ExecServerError::EnvironmentRegistryAuth( + "invalid executor credential authorization".to_string(), + ) + })?; + let mut credential = HeaderValue::from_str(token).map_err(|_| { + ExecServerError::EnvironmentRegistryAuth( + "invalid executor credential authorization".to_string(), + ) + })?; + credential.set_sensitive(true); + if let Some(authorization) = headers.get_mut(AUTHORIZATION) { + authorization.set_sensitive(true); + } + direct::require_tls_or_loopback(&self.base_url, "https")?; + Some(credential) + } else { + None + }; + Ok((headers, executor_credential)) + } + + fn http_for_auth(&self) -> &RouteAwareClientPool { + // A system proxy can otherwise receive a loopback HTTP credential in plaintext. + if self.auth_mode == RegistryAuthMode::ExecutorCredential + && url::Url::parse(&self.base_url).is_ok_and(|url| url.scheme() == "http") + { + &self.http_direct + } else { + &self.http + } } async fn parse_json_response(&self, response: HttpResponse) -> Result @@ -348,11 +419,12 @@ impl HarnessKeyValidator for RegistryHarnessKeyValidator { harness_public_key: harness_public_key.clone(), harness_key_authorization: authorization.to_string(), }; + let (auth_headers, executor_credential) = self.client.resolve_auth().await?; let response = self .client - .http + .http_for_auth() .post(url) - .headers(self.client.resolve_auth_headers().await?) + .headers(auth_headers) .headers(current_trace_context_headers()) .json(&body) .send() @@ -375,7 +447,15 @@ impl HarnessKeyValidator for RegistryHarnessKeyValidator { let response = response .json::() .await - .map_err(ExecServerError::EnvironmentRegistryRequest)?; + .map_err(|error| { + if executor_credential.is_some() { + ExecServerError::Protocol( + "invalid environment registry harness key validation response".to_string(), + ) + } else { + ExecServerError::EnvironmentRegistryRequest(error) + } + })?; if !response.valid { return Err(ExecServerError::Protocol( "environment registry rejected Noise relay harness key".to_string(), @@ -526,6 +606,11 @@ fn static_bearer_auth_provider( } /// Configuration for registering an exec-server for remote use. +/// +/// Scoped mode uses one `Authorization: Bearer ` header with +/// the Noise transport. It requires an HTTPS registry URL, except that HTTP is +/// permitted for loopback. Scoped executor credentials are not supported by +/// the Direct transport. #[derive(Clone)] pub struct RemoteEnvironmentConfig { pub base_url: String, @@ -534,6 +619,7 @@ pub struct RemoteEnvironmentConfig { pub request_dispatch_mode: RequestDispatchMode, transport: RemoteEnvironmentTransport, auth_provider: SharedAuthProvider, + auth_mode: RegistryAuthMode, telemetry: ExecServerTelemetry, http_client_factory: HttpClientFactory, } @@ -547,6 +633,7 @@ impl std::fmt::Debug for RemoteEnvironmentConfig { .field("request_dispatch_mode", &self.request_dispatch_mode) .field("transport", &self.transport) .field("auth_provider", &"") + .field("auth_mode", &self.auth_mode) .finish() } } @@ -584,11 +671,24 @@ impl RemoteEnvironmentConfig { request_dispatch_mode: RequestDispatchMode::Inline, transport, auth_provider, + auth_mode: RegistryAuthMode::Legacy, telemetry: ExecServerTelemetry::default(), http_client_factory, }) } + /// Requires one environment-bound bearer from the auth provider, without account fallback. + pub fn with_executor_credential_auth(mut self) -> Result { + if self.transport == RemoteEnvironmentTransport::Direct { + return Err(ExecServerError::EnvironmentRegistryConfig( + "direct transport does not support scoped executor credentials".to_string(), + )); + } + direct::require_tls_or_loopback(&self.base_url, "https")?; + self.auth_mode = RegistryAuthMode::ExecutorCredential; + Ok(self) + } + pub fn with_telemetry(mut self, telemetry: ExecServerTelemetry) -> Self { self.telemetry = telemetry; self @@ -687,12 +787,13 @@ where T: std::future::Future>, { ensure_rustls_crypto_provider(); - let client = EnvironmentRegistryClient::new_with_telemetry( + let mut client = EnvironmentRegistryClient::new_with_telemetry( config.base_url.clone(), config.auth_provider.clone(), config.telemetry.clone(), config.http_client_factory.clone(), )?; + client.auth_mode = config.auth_mode; let run = run_loop(config, client); tokio::pin!(run, shutdown); tokio::select! { @@ -892,6 +993,63 @@ fn endpoint_url(base_url: &str, path: &str) -> String { format!("{base_url}/{}", path.trim_start_matches('/')) } +async fn parse_executor_registration_response( + response: HttpResponse, + credential: &str, +) -> Result { + const SAFE_ERROR: &str = "invalid environment registry executor registration response"; + let status = response.status(); + if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) { + return Err(environment_registry_auth_error( + status, + "response body omitted", + )); + } + let body = response + .text() + .await + .map_err(|_| ExecServerError::Protocol(SAFE_ERROR.to_string()))?; + if !status.is_success() { + // Only these fixed codes affect retry behavior. Registry diagnostics can echo the credential. + let code = serde_json::from_str::(&body) + .ok() + .and_then(RegistryErrorBody::into_error) + .and_then(|error| error.code) + .filter(|code| { + matches!( + code.as_str(), + "registration_conflict" | "authentication_service_unavailable" + ) + }); + return Err(ExecServerError::EnvironmentRegistryHttp { + status, + code, + message: "environment registry executor registration failed".to_string(), + }); + } + let value: serde_json::Value = serde_json::from_str(&body) + .map_err(|_| ExecServerError::Protocol(SAFE_ERROR.to_string()))?; + if body.contains(credential) || json_contains_executor_credential(&value, credential) { + return Err(ExecServerError::Protocol(SAFE_ERROR.to_string())); + } + serde_json::from_value(value).map_err(|_| ExecServerError::Protocol(SAFE_ERROR.to_string())) +} + +fn json_contains_executor_credential(value: &serde_json::Value, credential: &str) -> bool { + match value { + serde_json::Value::String(value) => value.contains(credential), + serde_json::Value::Array(values) => values + .iter() + .any(|value| json_contains_executor_credential(value, credential)), + serde_json::Value::Object(fields) => fields.iter().any(|(key, value)| { + key.contains(credential) || json_contains_executor_credential(value, credential) + }), + serde_json::Value::Null | serde_json::Value::Bool(_) | serde_json::Value::Number(_) => { + false + } + } +} + fn environment_registry_auth_error(status: StatusCode, body: &str) -> ExecServerError { let message = registry_error_message(body).unwrap_or_else(|| "empty error body".to_string()); ExecServerError::EnvironmentRegistryAuth(format!( @@ -994,6 +1152,349 @@ mod tests { Arc::new(StaticRegistryAuthProvider) } + struct ScopedExecutorAuth; + + impl AuthProvider for ScopedExecutorAuth { + fn add_auth_headers(&self, headers: &mut HeaderMap) { + headers.insert( + AUTHORIZATION, + HeaderValue::from_static("Bearer synthetic-scoped-token"), + ); + } + } + + #[tokio::test] + async fn invalid_executor_credential_headers_are_never_sent() { + struct InvalidExecutorAuth(HeaderMap); + + impl AuthProvider for InvalidExecutorAuth { + fn add_auth_headers(&self, headers: &mut HeaderMap) { + headers.extend(self.0.clone()); + } + } + + let server = MockServer::start().await; + let public_key = NoiseChannelIdentity::generate() + .expect("identity") + .public_key(); + for credentials in [ + vec![], + vec![(AUTHORIZATION, HeaderValue::from_static("Bearer "))], + vec![( + AUTHORIZATION, + HeaderValue::from_static("Basic synthetic-secret"), + )], + vec![( + AUTHORIZATION, + HeaderValue::from_bytes(b"Bearer synthetic-\xff-secret").expect("header value"), + )], + vec![ + ( + AUTHORIZATION, + HeaderValue::from_static("Bearer synthetic-first-secret"), + ), + ( + AUTHORIZATION, + HeaderValue::from_static("Bearer synthetic-second-secret"), + ), + ], + vec![ + ( + AUTHORIZATION, + HeaderValue::from_static("Bearer synthetic-secret"), + ), + ( + HeaderName::from_static("chatgpt-account-id"), + HeaderValue::from_static("account-id"), + ), + ], + ] { + let mut headers = HeaderMap::new(); + for (name, value) in credentials { + headers.append(name, value); + } + let mut client = EnvironmentRegistryClient::new( + server.uri(), + Arc::new(InvalidExecutorAuth(headers)), + ) + .expect("registry client"); + client.auth_mode = RegistryAuthMode::ExecutorCredential; + let registration = client + .register_environment("environment-requested", &public_key) + .await + .expect_err("invalid credentials must be rejected"); + assert!(matches!( + registration, + ExecServerError::EnvironmentRegistryAuth(_) + )); + assert!(!format!("{registration:?} {registration}").contains("synthetic")); + + let validator = RegistryHarnessKeyValidator { + client, + environment_id: "environment-requested".to_string(), + executor_registration_id: "registration".to_string(), + }; + let validation = validator + .validate_harness_key(&public_key, "harness-authorization") + .await + .expect_err("invalid credentials must be rejected"); + assert!(matches!( + validation, + ExecServerError::EnvironmentRegistryAuth(_) + )); + assert!(!format!("{validation:?} {validation}").contains("synthetic")); + } + assert!( + server + .received_requests() + .await + .expect("requests") + .is_empty() + ); + } + + #[tokio::test] + async fn executor_credentials_require_https_or_loopback() { + for (url, allowed) in [ + ("https://registry.example", true), + ("http://localhost:8080", true), + ("http://127.0.0.1:8080", true), + ("http://[::1]:8080", true), + ("http://registry.example", false), + ("http://localhost.example", false), + ("not a url", false), + ] { + let mut client = + EnvironmentRegistryClient::new(url.to_string(), Arc::new(ScopedExecutorAuth)) + .expect("registry client"); + client.auth_mode = RegistryAuthMode::ExecutorCredential; + assert_eq!(client.resolve_auth().await.is_ok(), allowed, "{url}"); + } + + let legacy = EnvironmentRegistryClient::new( + "http://registry.example".to_string(), + static_registry_auth_provider(), + ) + .expect("registry client"); + let (headers, credential) = legacy.resolve_auth().await.expect("legacy auth"); + assert_eq!(headers[AUTHORIZATION], "Bearer registry-token"); + assert_eq!(headers["ChatGPT-Account-ID"], "workspace-123"); + assert!(credential.is_none()); + } + + #[tokio::test] + async fn insecure_registry_never_receives_executor_credentials() { + let proxy = MockServer::start().await; + let registry_url = "http://scoped-registry-proxy.invalid"; + for endpoint in ["register", "validate"] { + let request_url = + format!("{registry_url}/cloud/environment/environment-requested/{endpoint}"); + codex_http_client::cache_system_proxy_route_for_test(&request_url, proxy.uri()); + } + let mut client = EnvironmentRegistryClient::new_with_telemetry( + registry_url.to_string(), + Arc::new(ScopedExecutorAuth), + ExecServerTelemetry::default(), + HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy), + ) + .expect("registry client"); + client.auth_mode = RegistryAuthMode::ExecutorCredential; + let public_key = NoiseChannelIdentity::generate() + .expect("identity") + .public_key(); + let error = client + .register_environment("environment-requested", &public_key) + .await + .expect_err("registration must reject a non-loopback HTTP registry"); + assert!(matches!( + error, + ExecServerError::EnvironmentRegistryConfig(_) + )); + + let validator = RegistryHarnessKeyValidator { + client, + environment_id: "environment-requested".to_string(), + executor_registration_id: "registration".to_string(), + }; + let error = validator + .validate_harness_key(&public_key, "harness-authorization") + .await + .expect_err("validation must reject a non-loopback HTTP registry"); + assert!(matches!( + error, + ExecServerError::EnvironmentRegistryConfig(_) + )); + assert!( + proxy + .received_requests() + .await + .expect("proxy requests") + .is_empty() + ); + } + + #[tokio::test] + async fn scoped_loopback_registry_bypasses_a_configured_proxy() { + let registry = MockServer::start().await; + let proxy = MockServer::start().await; + let public_key = NoiseChannelIdentity::generate() + .expect("identity") + .public_key(); + for endpoint in ["register", "validate"] { + let request_url = format!( + "{}/cloud/environment/environment-requested/{endpoint}", + registry.uri() + ); + codex_http_client::cache_system_proxy_route_for_test(&request_url, proxy.uri()); + } + Mock::given(method("POST")) + .and(path("/cloud/environment/environment-requested/register")) + .and(header(AUTHORIZATION, "Bearer synthetic-scoped-token")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "environment_id": "environment-requested", + "url": "wss://rendezvous.test/", + "security_profile": NOISE_RELAY_SECURITY_PROFILE, + "executor_registration_id": "registration", + }))) + .expect(1) + .mount(®istry) + .await; + Mock::given(method("POST")) + .and(path("/cloud/environment/environment-requested/validate")) + .and(header(AUTHORIZATION, "Bearer synthetic-scoped-token")) + .respond_with( + ResponseTemplate::new(200).set_body_json(serde_json::json!({"valid": true})), + ) + .expect(1) + .mount(®istry) + .await; + let http_client_factory = HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy); + assert!(matches!( + http_client_factory.resolve_proxy_route(&format!( + "{}/cloud/environment/environment-requested/register", + registry.uri() + )), + codex_http_client::OutboundProxyRoute::Proxy { .. } + )); + let mut client = EnvironmentRegistryClient::new_with_telemetry( + registry.uri(), + Arc::new(ScopedExecutorAuth), + ExecServerTelemetry::default(), + http_client_factory, + ) + .expect("registry client"); + client.auth_mode = RegistryAuthMode::ExecutorCredential; + let (headers, credential) = client.resolve_auth().await.expect("scoped auth"); + assert_eq!(headers.len(), 1); + assert_eq!(headers[AUTHORIZATION], "Bearer synthetic-scoped-token"); + assert!(headers[AUTHORIZATION].is_sensitive()); + let credential = credential.expect("raw executor credential"); + assert_eq!(credential, "synthetic-scoped-token"); + assert!(credential.is_sensitive()); + client + .register_environment("environment-requested", &public_key) + .await + .expect("register directly"); + RegistryHarnessKeyValidator { + client, + environment_id: "environment-requested".to_string(), + executor_registration_id: "registration".to_string(), + } + .validate_harness_key(&public_key, "harness-authorization") + .await + .expect("validate directly"); + registry.verify().await; + assert!( + proxy + .received_requests() + .await + .expect("proxy requests") + .is_empty() + ); + } + + #[tokio::test] + async fn executor_credential_is_never_returned_in_registry_diagnostics() { + const TOKEN: &str = "synthetic-scoped-registry-secret"; + struct ExecutorAuth; + impl AuthProvider for ExecutorAuth { + fn add_auth_headers(&self, headers: &mut HeaderMap) { + let mut credential = + HeaderValue::from_str(&format!("Bearer {TOKEN}")).expect("bearer"); + credential.set_sensitive(true); + headers.insert(AUTHORIZATION, credential); + } + } + let encoded_echo = serde_json::json!({ + "environment_id": "environment-requested", + "url": "wss://rendezvous.test/", + "security_profile": NOISE_RELAY_SECURITY_PROFILE, + "executor_registration_id": TOKEN, + }) + .to_string() + .replace("synthetic", "\\u0073ynthetic"); + for (status, body) in [ + (401, format!(r#"{{"error":{{"message":"{TOKEN}"}}}}"#)), + ( + 502, + format!( + r#"{{"detail":{{"code":"authentication_service_unavailable","message":"{TOKEN}"}}}}"# + ), + ), + ( + 500, + format!(r#"{{"error":{{"code":"{TOKEN}","message":"{TOKEN}"}}}}"#), + ), + (200, format!(r#"{{"environment_id":"{TOKEN}"}}"#)), + (200, encoded_echo), + ] { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/cloud/environment/environment-requested/register")) + .and(header(AUTHORIZATION, format!("Bearer {TOKEN}"))) + .respond_with(ResponseTemplate::new(status).set_body_string(body.clone())) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/cloud/environment/environment-requested/validate")) + .and(header(AUTHORIZATION, format!("Bearer {TOKEN}"))) + .respond_with( + ResponseTemplate::new(status) + .set_body_string(format!(r#"{{"valid":"{TOKEN}"}}"#)), + ) + .mount(&server) + .await; + let mut client = EnvironmentRegistryClient::new(server.uri(), Arc::new(ExecutorAuth)) + .expect("registry client"); + client.auth_mode = RegistryAuthMode::ExecutorCredential; + let public_key = NoiseChannelIdentity::generate() + .expect("identity") + .public_key(); + let registration = client + .register_environment("environment-requested", &public_key) + .await + .expect_err("registry echoes cannot enter diagnostics"); + assert!(!format!("{registration:?} {registration}").contains(TOKEN)); + if status == 502 { + assert!(matches!( + ®istration, + ExecServerError::EnvironmentRegistryHttp { code: Some(code), .. } + if code == "authentication_service_unavailable" + )); + } + let validator = RegistryHarnessKeyValidator { + client, + environment_id: "environment-requested".to_string(), + executor_registration_id: "registration".to_string(), + }; + let validation = validator + .validate_harness_key(&public_key, "harness-authorization") + .await + .expect_err("invalid validation response"); + assert!(!format!("{validation:?} {validation}").contains(TOKEN)); + } + } + #[tokio::test(flavor = "current_thread")] async fn register_environment_posts_with_auth_provider_headers() { let provider = SdkTracerProvider::builder().build(); @@ -1333,6 +1834,10 @@ mod tests { .expect("config"); assert_eq!(config.transport, RemoteEnvironmentTransport::Direct); + assert!(matches!( + config.with_executor_credential_auth(), + Err(ExecServerError::EnvironmentRegistryConfig(_)) + )); } #[test] diff --git a/codex-rs/exec-server/src/remote/direct.rs b/codex-rs/exec-server/src/remote/direct.rs index a99454e177..99b0a2aa78 100644 --- a/codex-rs/exec-server/src/remote/direct.rs +++ b/codex-rs/exec-server/src/remote/direct.rs @@ -17,6 +17,7 @@ use tracing::info; use tracing::warn; use super::EnvironmentRegistryClient; +use super::RegistryAuthMode; use super::RemoteEnvironmentConfig; use crate::ExecServerError; use crate::client::is_retryable_recovery_error; @@ -80,6 +81,11 @@ impl EnvironmentRegistryClient { ) -> Result { let started_at = Instant::now(); let response = async { + if self.auth_mode == RegistryAuthMode::ExecutorCredential { + return Err(ExecServerError::EnvironmentRegistryConfig( + "direct transport does not support scoped executor credentials".to_string(), + )); + } let url = super::endpoint_url( &self.base_url, &format!("/cloud/environment/{environment_id}/direct/register"), @@ -154,6 +160,7 @@ pub(super) async fn run_direct_environment( match connect_direct( ®istration.url, config.auth_provider.as_ref(), + config.auth_mode, &config.http_client_factory, ) .await @@ -252,8 +259,14 @@ fn is_policy_recoverable( async fn connect_direct( url: &str, auth_provider: &dyn AuthProvider, + auth_mode: RegistryAuthMode, http_client_factory: &codex_http_client::HttpClientFactory, ) -> Result { + if auth_mode == RegistryAuthMode::ExecutorCredential { + return Err(ExecServerError::EnvironmentRegistryConfig( + "direct transport does not support scoped executor credentials".to_string(), + )); + } let mut request = url.into_client_request() .map_err(|source| ExecServerError::WebSocketConnect { diff --git a/codex-rs/exec-server/src/remote/direct_tests.rs b/codex-rs/exec-server/src/remote/direct_tests.rs index ae40fa5db6..9c49254824 100644 --- a/codex-rs/exec-server/src/remote/direct_tests.rs +++ b/codex-rs/exec-server/src/remote/direct_tests.rs @@ -175,6 +175,57 @@ impl AuthProvider for QueryAuthProvider { } } +struct ScopedExecutorAuth; + +impl AuthProvider for ScopedExecutorAuth { + fn add_auth_headers(&self, headers: &mut HeaderMap) { + headers.insert( + http::header::AUTHORIZATION, + HeaderValue::from_static("Bearer synthetic-scoped-token"), + ); + } +} + +#[tokio::test] +async fn direct_transport_rejects_scoped_credentials_before_sending() -> Result<()> { + let registry = MockServer::start().await; + let mut client = EnvironmentRegistryClient::new(registry.uri(), Arc::new(ScopedExecutorAuth))?; + client.auth_mode = RegistryAuthMode::ExecutorCredential; + let error = client + .register_direct_environment("environment-requested") + .await + .expect_err("scoped credentials must not be sent on direct registration"); + assert!(matches!( + error, + ExecServerError::EnvironmentRegistryConfig(_) + )); + assert!(registry.received_requests().await.unwrap().is_empty()); + + let listener = TcpListener::bind("127.0.0.1:0").await?; + let url = format!("ws://{}/connect", listener.local_addr()?); + let error = match connect_direct( + &url, + &ScopedExecutorAuth, + RegistryAuthMode::ExecutorCredential, + &HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + ) + .await + { + Err(error) => error, + Ok(_) => panic!("scoped credentials must not be sent to the WebSocket"), + }; + assert!(matches!( + error, + ExecServerError::EnvironmentRegistryConfig(_) + )); + assert!( + timeout(Duration::from_millis(50), listener.accept()) + .await + .is_err() + ); + Ok(()) +} + #[tokio::test] async fn direct_websocket_signs_only_host_and_preserves_handshake_headers() -> Result<()> { let mut request = "wss://executor.example.com/connect".into_client_request()?; @@ -213,6 +264,7 @@ async fn direct_websocket_uses_authenticated_url_query_parameters() -> Result<() let connection = connect_direct( &url, &QueryAuthProvider, + RegistryAuthMode::Legacy, &HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), ) .await?;