From 16e578dd1ddaf4f5db9cd1f3dbfea0958fd74339 Mon Sep 17 00:00:00 2001 From: viyatb-oai Date: Thu, 15 Oct 2026 22:21:45 +0000 Subject: [PATCH] Add rotating stdin credentials for remote exec-server (#54434) ## What changed - Add `--executor-credential-stdin` to read newline-delimited JSON credentials bound to the configured environment. Require an initial credential within 30 seconds and accept subsequent rotations. - Reject expired credentials on each request, and stop the executor on invalid input or stdin closure. Limit credential frames to 16 KiB. - Use scoped bearer authentication for registry requests without account credential fallback or WebSocket authentication headers. Skip workload identity loading and configured telemetry exporters in this mode. ## Testing Add unit and integration coverage for rotation, expiry, malformed frames, environment mismatches, stdin closure, registry-only authentication, credential redaction, and telemetry exporter isolation. GitOrigin-RevId: 30a94478165395144a765450f5ec6b3697acdd55 --- codex-rs/cli/src/exec_server_args_tests.rs | 36 +++ codex-rs/cli/src/exec_server_command.rs | 51 ++- codex-rs/cli/src/exec_server_credential.rs | 261 ++++++++++++++++ .../cli/src/exec_server_credential_tests.rs | 190 ++++++++++++ codex-rs/cli/src/exec_server_telemetry.rs | 2 + codex-rs/cli/src/main.rs | 1 + ..._server_help_documents_remote_options.snap | 4 + codex-rs/cli/tests/exec_server.rs | 291 ++++++++++++++++++ 8 files changed, 826 insertions(+), 10 deletions(-) create mode 100644 codex-rs/cli/src/exec_server_credential.rs create mode 100644 codex-rs/cli/src/exec_server_credential_tests.rs diff --git a/codex-rs/cli/src/exec_server_args_tests.rs b/codex-rs/cli/src/exec_server_args_tests.rs index 1724d221e9..79178c9865 100644 --- a/codex-rs/cli/src/exec_server_args_tests.rs +++ b/codex-rs/cli/src/exec_server_args_tests.rs @@ -65,6 +65,7 @@ fn exec_server_defaults_preserve_local_and_noise_modes() { let command = exec_server_from_args(&args); assert_eq!(command.remote_transport, ExecServerRemoteTransport::Noise); assert!(!command.aws_sigv4); + assert!(!command.executor_credential_stdin); command .validate_remote_transport() .expect("valid existing mode"); @@ -120,6 +121,27 @@ fn exec_server_parses_direct_aws_options() { } } +#[test] +fn exec_server_parses_executor_credential_options() { + let args = [ + "--remote", + "https://registry.example.com", + "--environment-id", + "env-1", + ]; + for options in [ + vec!["--executor-credential-stdin"], + vec!["--executor-credential-stdin", "--exit-on-stdin-close"], + ] { + let options: Vec<_> = args.iter().copied().chain(options).collect(); + let command = exec_server_from_args(&options); + assert!(command.executor_credential_stdin); + command + .validate_remote_transport() + .expect("valid scoped mode"); + } +} + #[test] fn exec_server_rejects_missing_or_conflicting_remote_options() { for (options, expected) in [ @@ -148,6 +170,19 @@ fn exec_server_rejects_missing_or_conflicting_remote_options() { ], ErrorKind::ArgumentConflict, ), + ( + vec!["--executor-credential-stdin", "--use-agent-identity-auth"], + ErrorKind::ArgumentConflict, + ), + ( + vec![ + "--executor-credential-stdin", + "--remote-transport", + "direct", + "--aws-sigv4", + ], + ErrorKind::ArgumentConflict, + ), ( vec!["--remote-transport", "unsupported"], ErrorKind::InvalidValue, @@ -176,6 +211,7 @@ fn exec_server_transport_and_aws_options_require_registration_arguments() { vec!["--remote-transport", "noise"], vec!["--remote-transport", "direct", "--aws-sigv4"], vec!["--aws-sigv4"], + vec!["--executor-credential-stdin"], vec![ "--remote", "https://registry.example.com", diff --git a/codex-rs/cli/src/exec_server_command.rs b/codex-rs/cli/src/exec_server_command.rs index 568ceaf98c..703fadc10c 100644 --- a/codex-rs/cli/src/exec_server_command.rs +++ b/codex-rs/cli/src/exec_server_command.rs @@ -26,6 +26,7 @@ use codex_utils_cli::CliConfigOverrides; use codex_websocket_auth::WebsocketAuthArgs; use crate::exec_server_auth; +use crate::exec_server_credential::ExecutorCredentialPipe; use crate::exec_server_telemetry; /// [EXPERIMENTAL] Run the standalone exec-server service. @@ -115,6 +116,15 @@ pub(super) struct ExecServerCommand { )] pub(super) use_agent_identity_auth: bool, + /// Read rotating executor credentials from stdin; exit when the pipe closes. + #[arg( + long = "executor-credential-stdin", + requires = "exec_server_remote", + conflicts_with_all = ["use_agent_identity_auth", "aws_sigv4"], + global = true + )] + pub(super) executor_credential_stdin: bool, + /// Sign Direct registration and WebSocket handshake requests with AWS SigV4. #[arg(long = "aws-sigv4", requires = "exec_server_remote", global = true)] pub(super) aws_sigv4: bool, @@ -196,12 +206,31 @@ impl ExecServerCommand { let config = load_exec_server_config( root_config_overrides, strict_config, - /*enable_workload_identity*/ true, + /*enable_workload_identity*/ !self.executor_credential_stdin, ) .await?; let direct_transport = self.remote_transport == ExecServerRemoteTransport::Direct; - let (_otel, telemetry) = exec_server_telemetry::init(Some(&config)); - let auth_provider = if self.aws_sigv4 { + // Configured exporters may contain account-wide credentials or client keys. + // The scoped executor must not initialize those exporters. + let telemetry_config = (!self.executor_credential_stdin).then_some(&config); + let (_otel, telemetry) = exec_server_telemetry::init(telemetry_config); + let mut credential_provider = None; + let mut parent_lifetime = if self.exit_on_stdin_close { + exec_server_telemetry::ParentLifetime::StdinPipe + } else { + exec_server_telemetry::ParentLifetime::Independent + }; + let auth_provider: codex_api::SharedAuthProvider = if self.executor_credential_stdin { + let ExecutorCredentialPipe { + provider, + parent_closed, + } = ExecutorCredentialPipe::from_stdin(environment_id.trim().to_string()); + provider.wait_for_initial().await?; + parent_lifetime = + exec_server_telemetry::ParentLifetime::MonitoredStdin(parent_closed); + credential_provider = Some(Arc::clone(&provider)); + provider + } else if self.aws_sigv4 { exec_server_auth::aws_sigv4_auth_provider( codex_aws_auth::AwsAuthConfig { profile: self.aws_profile, @@ -230,16 +259,14 @@ impl ExecServerCommand { auth_provider, config.http_client_factory(), )?; + if self.executor_credential_stdin { + remote_config = remote_config.with_executor_credential_auth()?; + } if let Some(name) = self.name { remote_config.name = name; } remote_config.request_dispatch_mode = self.request_dispatch_mode; let remote_config = remote_config.with_telemetry(telemetry); - let parent_lifetime = if self.exit_on_stdin_close { - exec_server_telemetry::ParentLifetime::StdinPipe - } else { - exec_server_telemetry::ParentLifetime::Independent - }; let (shutdown_sender, shutdown_receiver) = tokio::sync::oneshot::channel(); #[cfg(target_os = "macos")] let runtime_paths = runtime_paths.with_allowed_symlinked_codex_home( @@ -248,7 +275,7 @@ impl ExecServerCommand { &config.codex_home, ), ); - exec_server_telemetry::run_until_shutdown( + let result = exec_server_telemetry::run_until_shutdown( async move { let shutdown = async move { let _ = shutdown_receiver.await; @@ -276,7 +303,11 @@ impl ExecServerCommand { parent_lifetime, exec_server_telemetry::ShutdownBehavior::Graceful(shutdown_sender), ) - .await + .await; + if let Some(provider) = credential_provider { + provider.check_pipe_failure()?; + } + result } else { let config_result = load_exec_server_config( root_config_overrides, diff --git a/codex-rs/cli/src/exec_server_credential.rs b/codex-rs/cli/src/exec_server_credential.rs new file mode 100644 index 0000000000..72d034e9a5 --- /dev/null +++ b/codex-rs/cli/src/exec_server_credential.rs @@ -0,0 +1,261 @@ +//! Reads scoped registry credentials from a trusted parent's stdin. +//! +//! The parent sends newline-delimited JSON frames with `version: 1`, +//! `environment_id`, `token`, and `expires_at` (Unix seconds). A frame, including +//! its newline, must be at most 16 KiB; a valid first frame is required within +//! 30 seconds. Each valid frame replaces the current credential. Frames for another +//! environment or otherwise invalid input clear it and stop the executor, as does +//! EOF. Requests reject expired credentials, so the parent must renew before expiry. +//! Only registry requests carry the credential, as an Authorization bearer. + +use std::fmt; +use std::io::BufRead; +use std::sync::Arc; +use std::time::Duration; +use std::time::SystemTime; +use std::time::UNIX_EPOCH; + +use codex_api::AuthError; +use codex_api::AuthHeadersFuture; +use codex_api::AuthProvider; +use http::HeaderMap; +use http::HeaderValue; +use http::header::AUTHORIZATION; +use serde::Deserialize; +use tokio::sync::oneshot; +use tokio::sync::watch; + +const MAX_FRAME_BYTES: usize = 16 * 1024; +const INITIAL_CREDENTIAL_TIMEOUT: Duration = Duration::from_secs(30); + +pub(super) struct ExecutorCredentialPipe { + pub(super) provider: Arc, + pub(super) parent_closed: oneshot::Receiver<()>, +} + +impl ExecutorCredentialPipe { + pub(super) fn from_stdin(environment_id: String) -> Self { + Self::spawn_with_reader(environment_id, std::io::stdin) + } + + fn spawn_with_reader R + Send + 'static>( + environment_id: String, + open_reader: F, + ) -> Self { + let (updates, current) = watch::channel(CredentialState::Pending); + let (closed, parent_closed) = oneshot::channel(); + std::thread::spawn(move || { + let mut reader = std::io::BufReader::new(open_reader()); + let result = read_frames(&mut reader, &environment_id, &updates); + updates.send_replace(match result { + Ok(()) => CredentialState::Closed, + Err(error) => CredentialState::Failed(error), + }); + let _ = closed.send(()); + }); + Self { + provider: Arc::new(ExecutorCredentialProvider { current }), + parent_closed, + } + } +} + +pub(super) struct ExecutorCredentialProvider { + current: watch::Receiver, +} + +impl fmt::Debug for ExecutorCredentialProvider { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ExecutorCredentialProvider") + .finish_non_exhaustive() + } +} + +impl ExecutorCredentialProvider { + pub(super) async fn wait_for_initial(&self) -> Result<(), CredentialError> { + self.wait_for_initial_with_timeout(INITIAL_CREDENTIAL_TIMEOUT) + .await + } + + async fn wait_for_initial_with_timeout( + &self, + timeout: Duration, + ) -> Result<(), CredentialError> { + let mut current = self.current.clone(); + tokio::time::timeout(timeout, async move { + loop { + match &*current.borrow_and_update() { + CredentialState::Pending => {} + state => return state.header().map(|_| ()), + } + current + .changed() + .await + .map_err(|_| CredentialError::Closed)?; + } + }) + .await + .map_err(|_| CredentialError::InitialTimeout)? + } + + pub(super) fn check_pipe_failure(&self) -> Result<(), CredentialError> { + if let CredentialState::Failed(error) = &*self.current.borrow() { + return Err(*error); + } + Ok(()) + } +} + +impl AuthProvider for ExecutorCredentialProvider { + // Only the fallible API can prevent an unauthenticated registry request. + fn add_auth_headers(&self, _headers: &mut HeaderMap) {} + + fn resolve_auth_headers(&self) -> AuthHeadersFuture<'_> { + Box::pin(async { + let token = self + .current + .borrow() + .header() + .map_err(|error| AuthError::Build(error.to_string()))?; + let token = token + .to_str() + .map_err(|_| AuthError::Build(CredentialError::InvalidFrame.to_string()))?; + let mut authorization = HeaderValue::from_str(&format!("Bearer {token}")) + .map_err(|_| AuthError::Build(CredentialError::InvalidFrame.to_string()))?; + authorization.set_sensitive(true); + let mut headers = HeaderMap::new(); + headers.insert(AUTHORIZATION, authorization); + Ok(headers) + }) + } +} + +enum CredentialState { + Pending, + Active { token: HeaderValue, expires_at: u64 }, + Failed(CredentialError), + Closed, +} + +impl CredentialState { + fn header(&self) -> Result { + match self { + Self::Active { token, expires_at } if *expires_at > unix_timestamp()? => { + Ok(token.clone()) + } + Self::Active { .. } => Err(CredentialError::Expired), + Self::Failed(error) => Err(*error), + Self::Pending => Err(CredentialError::Missing), + Self::Closed => Err(CredentialError::Closed), + } + } +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct CredentialFrame { + version: u8, + environment_id: String, + token: String, + expires_at: u64, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum CredentialError { + InvalidFrame, + FrameTooLarge, + WrongEnvironment, + Expired, + Missing, + Closed, + Read, + Clock, + InitialTimeout, +} + +impl fmt::Display for CredentialError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(match self { + Self::InvalidFrame => "invalid exec-server executor credential frame", + Self::FrameTooLarge => "exec-server executor credential frame exceeds the size limit", + Self::WrongEnvironment => { + "exec-server executor credential is for a different environment" + } + Self::Expired => "exec-server executor credential is expired", + Self::Missing => "exec-server executor credential is unavailable", + Self::Closed => "exec-server executor credential pipe closed", + Self::Read => "could not read exec-server executor credential pipe", + Self::Clock => "could not check exec-server executor credential expiration", + Self::InitialTimeout => { + "timed out waiting for the initial exec-server executor credential" + } + }) + } +} + +impl std::error::Error for CredentialError {} + +fn read_frames( + reader: &mut impl BufRead, + environment_id: &str, + updates: &watch::Sender, +) -> Result<(), CredentialError> { + let mut frame = Vec::new(); + loop { + let chunk = reader.fill_buf().map_err(|_| CredentialError::Read)?; + if chunk.is_empty() { + return if frame.is_empty() { + Ok(()) + } else { + Err(CredentialError::InvalidFrame) + }; + } + let newline = chunk.iter().position(|byte| *byte == b'\n'); + let consumed = newline.map_or(chunk.len(), |index| index + 1); + if frame.len() + consumed > MAX_FRAME_BYTES { + return Err(CredentialError::FrameTooLarge); + } + frame.extend_from_slice(&chunk[..consumed]); + reader.consume(consumed); + if newline.is_some() { + let parsed = parse_frame(&frame, environment_id)?; + updates.send_replace(parsed); + frame.clear(); + } + } +} + +fn parse_frame(frame: &[u8], environment_id: &str) -> Result { + let frame: CredentialFrame = + serde_json::from_slice(frame).map_err(|_| CredentialError::InvalidFrame)?; + if frame.version != 1 + || frame.token.is_empty() + || !frame.token.bytes().all(|byte| byte.is_ascii_graphic()) + { + return Err(CredentialError::InvalidFrame); + } + if environment_id.is_empty() || frame.environment_id != environment_id { + return Err(CredentialError::WrongEnvironment); + } + if frame.expires_at <= unix_timestamp()? { + return Err(CredentialError::Expired); + } + let mut token = + HeaderValue::from_str(&frame.token).map_err(|_| CredentialError::InvalidFrame)?; + token.set_sensitive(true); + Ok(CredentialState::Active { + token, + expires_at: frame.expires_at, + }) +} + +fn unix_timestamp() -> Result { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|elapsed| elapsed.as_secs()) + .map_err(|_| CredentialError::Clock) +} + +#[cfg(test)] +#[path = "exec_server_credential_tests.rs"] +mod tests; diff --git a/codex-rs/cli/src/exec_server_credential_tests.rs b/codex-rs/cli/src/exec_server_credential_tests.rs new file mode 100644 index 0000000000..8d5e7956e6 --- /dev/null +++ b/codex-rs/cli/src/exec_server_credential_tests.rs @@ -0,0 +1,190 @@ +use std::io::PipeWriter; +use std::io::Write; + +use pretty_assertions::assert_eq; + +use super::*; + +fn pipe() -> (PipeWriter, ExecutorCredentialPipe) { + let (reader, writer) = std::io::pipe().expect("create credential pipe"); + let pipe = ExecutorCredentialPipe::spawn_with_reader("env-a".to_string(), move || reader); + (writer, pipe) +} + +fn frame(environment_id: &str, token: &str, expires_at: u64) -> Vec { + let mut value = serde_json::to_vec(&serde_json::json!({ + "version": 1, + "environment_id": environment_id, + "token": token, + "expires_at": expires_at, + })) + .expect("serialize frame"); + value.push(b'\n'); + value +} + +async fn next_update(updates: &mut watch::Receiver) { + tokio::time::timeout(Duration::from_secs(2), updates.changed()) + .await + .expect("credential reader should publish") + .expect("reader should remain open"); +} + +#[tokio::test] +async fn waits_for_initial_rotation_replaces_token_and_eof_clears_auth() { + let (mut writer, pipe) = pipe(); + let provider = pipe.provider; + let mut updates = provider.current.clone(); + let expires = unix_timestamp().expect("clock") + 60; + assert_eq!( + provider + .wait_for_initial_with_timeout(Duration::from_millis(5)) + .await, + Err(CredentialError::InitialTimeout) + ); + assert!(provider.resolve_auth_headers().await.is_err()); + assert!(provider.to_auth_headers().is_empty()); + writer + .write_all(&frame("env-a", "first-scoped-token", expires)) + .expect("write"); + provider.wait_for_initial().await.expect("initial token"); + next_update(&mut updates).await; + let headers = provider.resolve_auth_headers().await.expect("headers"); + assert_eq!(headers.len(), 1); + assert_eq!(headers[AUTHORIZATION], "Bearer first-scoped-token"); + assert!(headers[AUTHORIZATION].is_sensitive()); + assert!(!format!("{provider:?} {headers:?}").contains("first-scoped-token")); + + writer + .write_all(&frame("env-a", "rotated-scoped-token", expires)) + .expect("rotate"); + next_update(&mut updates).await; + let headers = provider.resolve_auth_headers().await.expect("headers"); + assert_eq!(headers.len(), 1); + assert_eq!(headers[AUTHORIZATION], "Bearer rotated-scoped-token"); + assert!(headers[AUTHORIZATION].is_sensitive()); + + drop(writer); + pipe.parent_closed.await.expect("reader must monitor EOF"); + assert!(provider.resolve_auth_headers().await.is_err()); + assert_eq!(provider.check_pipe_failure(), Ok(())); +} + +#[tokio::test] +async fn malformed_rotations_latch_failure_and_clear_existing_credential() { + let expires = unix_timestamp().expect("clock") + 60; + let mut oversized = vec![b'x'; MAX_FRAME_BYTES]; + oversized.push(b'\n'); + for (bad_frame, error) in [ + ( + frame("env-b", "do-not-log-cross-env", expires), + CredentialError::WrongEnvironment, + ), + ( + frame("env-a", "do-not-log-expired", /*expires_at*/ 1), + CredentialError::Expired, + ), + ( + frame("env-a", "invalid\ntoken", expires), + CredentialError::InvalidFrame, + ), + ( + b"invalid do-not-log-malformed\n".to_vec(), + CredentialError::InvalidFrame, + ), + (b"\n".to_vec(), CredentialError::InvalidFrame), + (oversized, CredentialError::FrameTooLarge), + ] { + let (mut writer, pipe) = pipe(); + writer + .write_all(&frame("env-a", "initial-scoped-token", expires)) + .expect("initial frame"); + pipe.provider + .wait_for_initial() + .await + .expect("initial credential"); + writer.write_all(&bad_frame).expect("bad frame"); + pipe.parent_closed + .await + .expect("bad frame terminates without EOF"); + let auth_error = pipe + .provider + .resolve_auth_headers() + .await + .expect_err("reject auth"); + assert_eq!(pipe.provider.check_pipe_failure(), Err(error)); + assert!(!auth_error.to_string().contains("do-not-log")); + assert!(!auth_error.to_string().contains("initial-scoped-token")); + let _ = writer.write_all(&frame("env-a", "later-scoped-token", expires)); + assert_eq!(pipe.provider.check_pipe_failure(), Err(error)); + } +} + +#[tokio::test] +async fn eof_before_a_complete_frame_never_authenticates() { + let expires = unix_timestamp().expect("clock") + 60; + for (bytes, expected) in [ + (Vec::new(), CredentialError::Closed), + ( + frame("env-a", "scoped-token", expires) + .split_last() + .expect("newline") + .1 + .to_vec(), + CredentialError::InvalidFrame, + ), + ] { + let (mut writer, pipe) = pipe(); + if !bytes.is_empty() { + writer.write_all(&bytes).expect("partial frame"); + } + drop(writer); + pipe.parent_closed.await.expect("reader monitors EOF"); + assert_eq!(pipe.provider.wait_for_initial().await, Err(expected)); + assert!(pipe.provider.resolve_auth_headers().await.is_err()); + } +} + +#[tokio::test] +async fn expiry_is_checked_again_on_every_request() { + let expired = unix_timestamp().expect("clock"); + let (updates, current) = watch::channel(CredentialState::Active { + token: HeaderValue::from_static("unused-expired-token"), + expires_at: expired, + }); + let provider = ExecutorCredentialProvider { current }; + assert_eq!( + provider.wait_for_initial().await, + Err(CredentialError::Expired) + ); + assert!(provider.resolve_auth_headers().await.is_err()); + updates.send_replace(CredentialState::Active { + token: HeaderValue::from_static("new-scoped-token"), + expires_at: expired + 60, + }); + let headers = provider + .resolve_auth_headers() + .await + .expect("rotation restores auth"); + assert_eq!(headers[AUTHORIZATION], "Bearer new-scoped-token"); +} + +#[test] +fn parser_rejects_unsupported_schema_without_exposing_input() { + let expires = unix_timestamp().expect("clock") + 60; + for invalid in [ + serde_json::json!({"version": 2, "environment_id": "env-a", "token": "secret", "expires_at": expires}), + serde_json::json!({"version": 1, "environment_id": "env-a", "token": "secret", "expires_at": expires, "unexpected": true}), + serde_json::json!({"version": 1, "environment_id": "env-a", "token": "secret", "expires_at": -1}), + serde_json::json!({"version": 1, "environment_id": "env-a", "token": " ", "expires_at": expires}), + serde_json::json!({"version": 1, "environment_id": "env-a", "token": "secret token", "expires_at": expires}), + serde_json::json!({"version": 1, "environment_id": "env-a", "token": "secret-☃", "expires_at": expires}), + ] { + let serialized = serde_json::to_vec(&invalid).expect("serialize test input"); + let error = parse_frame(&serialized, "env-a") + .err() + .expect("invalid schema"); + assert_eq!(error, CredentialError::InvalidFrame); + assert!(!error.to_string().contains("secret")); + } +} diff --git a/codex-rs/cli/src/exec_server_telemetry.rs b/codex-rs/cli/src/exec_server_telemetry.rs index 5e0e76c504..c20895b513 100644 --- a/codex-rs/cli/src/exec_server_telemetry.rs +++ b/codex-rs/cli/src/exec_server_telemetry.rs @@ -10,6 +10,7 @@ const OTEL_SERVICE_NAME: &str = "codex-exec-server"; pub(crate) enum ParentLifetime { Independent, StdinPipe, + MonitoredStdin(tokio::sync::oneshot::Receiver<()>), } pub(crate) enum ShutdownBehavior { @@ -65,6 +66,7 @@ where { let parent_disconnected = match parent_lifetime { ParentLifetime::Independent => None, + ParentLifetime::MonitoredStdin(receiver) => Some(receiver), ParentLifetime::StdinPipe => { let (sender, receiver) = tokio::sync::oneshot::channel(); std::thread::spawn(move || { diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index c1537bfa00..616c3f0f51 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -66,6 +66,7 @@ mod exec_args_tests; mod exec_server_args_tests; mod exec_server_auth; mod exec_server_command; +mod exec_server_credential; mod exec_server_telemetry; mod marketplace_cmd; mod mcp_cmd; diff --git a/codex-rs/cli/src/snapshots/codex__exec_server_args_tests__exec_server_help_documents_remote_options.snap b/codex-rs/cli/src/snapshots/codex__exec_server_args_tests__exec_server_help_documents_remote_options.snap index 2010b1e0c5..706c032121 100644 --- a/codex-rs/cli/src/snapshots/codex__exec_server_args_tests__exec_server_help_documents_remote_options.snap +++ b/codex-rs/cli/src/snapshots/codex__exec_server_args_tests__exec_server_help_documents_remote_options.snap @@ -99,6 +99,10 @@ Options: Use Agent Identity auth from CODEX_ACCESS_TOKEN for remote registration + --executor-credential-stdin + Read rotating executor credentials from stdin; exit when the pipe + closes + --aws-sigv4 Sign Direct registration and WebSocket handshake requests with AWS SigV4 diff --git a/codex-rs/cli/tests/exec_server.rs b/codex-rs/cli/tests/exec_server.rs index cc100b7f6c..b0b8e8da69 100644 --- a/codex-rs/cli/tests/exec_server.rs +++ b/codex-rs/cli/tests/exec_server.rs @@ -47,6 +47,9 @@ use tokio::io::BufReader; use tokio::net::TcpListener; use tokio_tungstenite::WebSocketStream; use tokio_tungstenite::accept_async; +use tokio_tungstenite::accept_hdr_async; +use tokio_tungstenite::tungstenite::handshake::server::Request; +use tokio_tungstenite::tungstenite::handshake::server::Response; use wiremock::Mock; use wiremock::MockServer; use wiremock::ResponseTemplate; @@ -449,6 +452,294 @@ metrics_exporter = {{ otlp-http = {{ endpoint = "{collector_url}/v1/metrics", pr Ok(()) } +#[tokio::test] +async fn remote_exec_server_scoped_credentials_authenticate_only_register_and_validate() +-> Result<()> { + const ENVIRONMENT: &str = "environment-scoped-credentials"; + const CREDENTIAL: &str = "synthetic-scoped-credential-never-log-1234"; + const ROTATED_CREDENTIAL: &str = "rotated-scoped-credential"; + const TIMEOUT: Duration = Duration::from_secs(30); + + let listener = TcpListener::bind("127.0.0.1:0").await?; + let rendezvous = format!("ws://{}", listener.local_addr()?); + let registry = MockServer::start().await; + let (registration_tx, mut registration_rx) = tokio::sync::mpsc::unbounded_channel(); + // A confirmed pre-write auth outage must retry with the same scoped credential. + Mock::given(method("POST")) + .and(path(format!("/cloud/environment/{ENVIRONMENT}/register"))) + .respond_with(ResponseTemplate::new(502).set_body_json(serde_json::json!({ + "detail": {"code": "authentication_service_unavailable", "message": CREDENTIAL}, + }))) + .with_priority(1) + .up_to_n_times(1) + .expect(1) + .mount(®istry) + .await; + let registration_response = ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "environment_id": ENVIRONMENT, + "url": format!("{rendezvous}/relay?role=environment"), + "security_profile": "noise_hybrid_ik_v1", + "executor_registration_id": "registration-scoped", + })); + Mock::given(method("POST")) + .and(path(format!("/cloud/environment/{ENVIRONMENT}/register"))) + .respond_with(move |request: &wiremock::Request| { + let credential = request + .headers + .get("authorization") + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.strip_prefix("Bearer ")) + .unwrap_or_default() + .to_string(); + let _ = registration_tx.send(credential); + registration_response.clone() + }) + .mount(®istry) + .await; + Mock::given(method("POST")) + .and(path(format!("/cloud/environment/{ENVIRONMENT}/validate"))) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"valid": true}))) + .mount(®istry) + .await; + + let collector = MockServer::start().await; + let home = TempDir::new()?; + std::fs::write( + home.path().join("config.toml"), + format!( + r#" +[analytics] +enabled = true +[otel] +metrics_exporter = {{ otlp-http = {{ endpoint = "{}/v1/metrics", protocol = "json", headers = {{ Authorization = "general-telemetry-token" }} }} }} +"#, + collector.uri() + ), + )?; + std::fs::write( + home.path().join("auth.json"), + r#"{"OPENAI_API_KEY":"general-file-token"}"#, + )?; + let mut command = tokio::process::Command::new(codex_utils_cargo_bin::cargo_bin("codex")?); + command + .env("CODEX_HOME", home.path()) + .env_remove(codex_exec_server::CODEX_EXEC_SERVER_EXIT_ON_STDIN_CLOSE_ENV_VAR) + .env("CODEX_API_KEY", "general-api-key") + .env("OPENAI_API_KEY", "general-openai-key") + .env("CODEX_ACCESS_TOKEN", "general-agent-identity") + .env( + "OPENAI_WORKLOAD_IDENTITY_CONTEXT", + "invalid-general-workload-identity", + ) + .env("NO_PROXY", "127.0.0.1,localhost") + .env("no_proxy", "127.0.0.1,localhost") + .args([ + "exec-server", + "--remote", + ®istry.uri(), + "--environment-id", + ENVIRONMENT, + "--executor-credential-stdin", + ]) + .stdin(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + let mut child = command.spawn()?; + let mut stdin = child + .stdin + .take() + .context("credential stdin was not piped")?; + send_json_line(&mut stdin, &serde_json::json!({ + "version": 1, "environment_id": ENVIRONMENT, "token": CREDENTIAL, + "expires_at": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_secs() + 120, + })).await?; + + let (socket, _) = tokio::time::timeout(TIMEOUT, listener.accept()) + .await + .context("executor did not reach rendezvous")??; + let environment_websocket = tokio::time::timeout( + TIMEOUT, + accept_hdr_async(socket, |request: &Request, response: Response| { + for name in ["authorization", "chatgpt-account-id"] { + assert!( + !request.headers().contains_key(name), + "registry authentication must not be sent in the WebSocket handshake" + ); + } + Ok(response) + }), + ) + .await + .context("executor WebSocket handshake did not complete")??; + let executor_public_key = registered_parent_lifetime_executor_public_key(®istry).await?; + let args = NoiseRendezvousConnectArgs { + bundle: NoiseRendezvousConnectBundle { + websocket_url: format!("{rendezvous}/relay?role=harness"), + environment_id: ENVIRONMENT.to_string(), + executor_registration_id: "registration-scoped".to_string(), + executor_public_key, + harness_key_authorization: "synthetic-harness-authorization".to_string(), + }, + harness_identity: NoiseChannelIdentity::generate()?, + client_name: "scoped-credential-test".to_string(), + connect_timeout: TIMEOUT, + initialize_timeout: TIMEOUT, + resume_session_id: None, + http_client_factory: HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + }; + let connect = + tokio::spawn(async move { ExecServerClient::connect_noise_rendezvous(args).await }); + let harness_websocket = accept_parent_lifetime_websocket(&listener, TIMEOUT).await?; + let relay = tokio::spawn(proxy_parent_lifetime_relay( + environment_websocket, + harness_websocket, + )); + let client = tokio::time::timeout(TIMEOUT, connect) + .await + .context("harness timed out")???; + let before_rotation = client.environment_info().await?; + send_json_line(&mut stdin, &serde_json::json!({ + "version": 1, "environment_id": ENVIRONMENT, "token": ROTATED_CREDENTIAL, + "expires_at": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_secs() + 120, + })).await?; + assert_eq!( + &client.force_environment_info().await?, + before_rotation.as_ref() + ); + let requests = registry + .received_requests() + .await + .context("registry requests")?; + assert_eq!( + requests + .iter() + .map(|request| request.url.path()) + .collect::>(), + [ + format!("/cloud/environment/{ENVIRONMENT}/register"), + format!("/cloud/environment/{ENVIRONMENT}/register"), + format!("/cloud/environment/{ENVIRONMENT}/validate"), + ] + ); + for request in requests { + assert_eq!( + request.headers["authorization"], + format!("Bearer {CREDENTIAL}") + ); + assert_eq!(request.headers.get_all("authorization").iter().count(), 1); + for name in [ + "chatgpt-account-id", + "x-openai-biscuit", + "biscuit", + "x-openai-access-token", + ] { + assert!( + !request.headers.contains_key(name), + "unexpected registry credential header {name}" + ); + } + } + + assert_eq!(registration_rx.try_recv()?, CREDENTIAL); + drop(client); + relay.abort(); + let _ = relay.await; + + // Reject reconnects to trigger re-registration while the parent update is processed. + tokio::time::timeout(TIMEOUT, async { + loop { + let (socket, _) = listener.accept().await?; + let rejected = http::Response::builder() + .status(http::StatusCode::UNAUTHORIZED) + .body(Some("expired registration".to_string()))?; + let result = accept_hdr_async(socket, |_: &Request, _: Response| Err(rejected)).await; + assert!(result.is_err()); + + let credential = registration_rx + .recv() + .await + .context("missing re-registration")?; + if credential == ROTATED_CREDENTIAL { + break Ok::<(), anyhow::Error>(()); + } + assert_eq!(credential, CREDENTIAL); + } + }) + .await + .context("executor did not re-register with the rotated parent credential")??; + + drop(stdin); + let output = tokio::time::timeout(TIMEOUT, child.wait_with_output()) + .await + .context("stdin EOF did not stop executor")??; + let stderr = String::from_utf8_lossy(&output.stderr); + anyhow::ensure!(output.status.success(), "scoped executor failed: {stderr}"); + assert!(!stderr.contains(CREDENTIAL)); + assert!(!stderr.contains(ROTATED_CREDENTIAL)); + assert!( + collector + .received_requests() + .await + .context("telemetry requests")? + .is_empty() + ); + Ok(()) +} + +#[tokio::test] +async fn remote_exec_server_scoped_credentials_never_fall_back_for_another_environment() +-> Result<()> { + let registry = MockServer::start().await; + let home = TempDir::new()?; + std::fs::write( + home.path().join("auth.json"), + r#"{"OPENAI_API_KEY":"general-file-token"}"#, + )?; + let mut command = tokio::process::Command::new(codex_utils_cargo_bin::cargo_bin("codex")?); + let mut child = command + .env("CODEX_HOME", home.path()) + .env("CODEX_API_KEY", "general-api-key") + .env("CODEX_ACCESS_TOKEN", "general-agent-identity") + .env("NO_PROXY", "127.0.0.1,localhost") + .env("no_proxy", "127.0.0.1,localhost") + .args([ + "exec-server", + "--remote", + ®istry.uri(), + "--environment-id", + "configured-environment", + "--executor-credential-stdin", + ]) + .stdin(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .spawn()?; + let mut stdin = child + .stdin + .take() + .context("credential stdin was not piped")?; + send_json_line(&mut stdin, &serde_json::json!({ + "version": 1, "environment_id": "another-environment", "token": "never-print-this-scoped-token", + "expires_at": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_secs() + 120, + })).await?; + let output = tokio::time::timeout(Duration::from_secs(30), child.wait_with_output()) + .await + .context("invalid scoped frame did not stop executor")??; + assert!(!output.status.success()); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("different environment")); + assert!(!stderr.contains("never-print-this-scoped-token")); + assert!( + registry + .received_requests() + .await + .context("registry requests")? + .is_empty() + ); + drop(stdin); + Ok(()) +} + async fn accept_parent_lifetime_websocket( listener: &TcpListener, timeout: Duration,