openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-29 19:08:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add a standalone codex-exec-server executable (#57589)

Browse Source 
## What changed

Extract executor startup into `codex-exec-server-cli`, shared by `codex exec-server` and the new standalone binary. Preserve executor arguments, authentication, configuration overrides, and sandbox helper dispatch across both entry points.

Build the standalone binary with `cargo build -p codex-exec-server-cli --bin codex-exec-server` and run it with `codex-exec-server --listen stdio`.

## Testing

Add standalone CLI parsing and help coverage, plus an integration test for version reporting, stdio initialization, filesystem reads, process execution, and sandbox helper re-execution when supported.

GitOrigin-RevId: ab3a81c6b91b6f45e4d72d8add2e01af8372e97f
This commit is contained in:
Dmitriy Kovalenko
2026-10-27 23:13:53 +00:00
committed by copyberry
parent 37f199dab1
commit a9ea49e78b
25 changed files with 1090 additions and 579 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/exec-server-cli/BUILD.bazel
+11
View File
@@ -0,0 +1,11 @@
load("//:defs.bzl", "codex_rust_crate")
codex_rust_crate(
name = "exec-server-cli",
binaries_with_build_commit = ["codex-exec-server"],
crate_name = "codex_exec_server_cli",
extra_binaries = ["//codex-rs/bwrap:bwrap"],
test_data_extra = glob(["src/**/snapshots/**"]),
# The standalone integration test launches native OS sandboxes itself.
test_tags = ["no-sandbox"],
)
codex-rs/exec-server-cli/Cargo.toml
+56
View File
@@ -0,0 +1,56 @@
[package]
name = "codex-exec-server-cli"
version.workspace = true
edition.workspace = true
license.workspace = true
[[bin]]
name = "codex-exec-server"
path = "src/main.rs"
[lib]
doctest = false
[lints]
workspace = true
[dependencies]
anyhow = { workspace = true }
clap = { workspace = true, features = ["derive", "env"] }
codex-api = { workspace = true }
codex-app-server = { workspace = true }
codex-arg0 = { workspace = true }
codex-aws-auth = { workspace = true }
codex-build-info = { workspace = true }
codex-cloud-config = { workspace = true }
codex-config = { workspace = true }
codex-core = { workspace = true }
codex-exec-server = { workspace = true }
codex-http-client = { workspace = true }
codex-login = { workspace = true }
codex-model-provider = { workspace = true }
codex-sandboxing = { workspace = true }
codex-utils-absolute-path = { workspace = true }
codex-utils-cli = { workspace = true }
codex-websocket-auth = { workspace = true }
http = { workspace = true }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt-multi-thread", "signal", "sync", "time"] }
toml = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true, features = ["env-filter"] }
url = { workspace = true }
[target.'cfg(all(target_os = "linux", target_env = "musl", any(target_arch = "x86_64", target_arch = "aarch64")))'.dependencies]
tikv-jemallocator = { workspace = true }
[dev-dependencies]
assert_matches = { workspace = true }
codex-protocol = { workspace = true }
codex-utils-cargo-bin = { workspace = true }
codex-utils-process = { workspace = true }
insta = { workspace = true }
pretty_assertions = { workspace = true }
tempfile = { workspace = true }
tokio = { workspace = true, features = ["io-util", "process"] }
codex-rs/exec-server-cli/src/command.rs
+391
View File
@@ -0,0 +1,391 @@
//! Command-line startup shared by `codex exec-server` and `codex-exec-server`.
//!
//! Transport, configuration, authentication, and shutdown are kept together.
use std::sync::Arc;
use clap::Parser;
use codex_arg0::Arg0DispatchPaths;
use codex_core::config::Config;
use codex_exec_server::ExecServerRuntimeOptions;
use codex_http_client::HttpClientFactory;
use codex_http_client::OutboundProxyPolicy;
use codex_utils_cli::CliConfigOverrides;
use codex_websocket_auth::WebsocketAuthArgs;
use crate::config::load_exec_server_config;
use crate::config::load_exec_server_remote_auth_provider;
use crate::exec_server_auth;
use crate::exec_server_credential::ExecutorCredentialPipe;
use crate::exec_server_telemetry;
/// [EXPERIMENTAL] Run the standalone exec-server service.
#[derive(Debug, Parser)]
pub struct ExecServerCommand {
#[command(subcommand)]
command: Option<ExecServerSubcommand>,
/// Error out when config.toml contains fields that are not recognized by this version of Codex.
#[arg(
id = "exec_server_strict_config",
long = "strict-config",
default_value_t = false,
global = true
)]
pub strict_config: bool,
/// Linux PID namespace: isolate (default) or inherit. Inherit allows signals to
/// other same-UID processes; enable only when provisioning a dedicated environment.
#[arg(long, value_name = "MODE", default_value = "isolate", global = true)]
linux_sandbox_pid_namespace: codex_sandboxing::LinuxSandboxPidNamespace,
/// Allow permitted private IP destinations to use the configured upstream proxy.
/// If no valid upstream proxy applies to the request protocol, connect directly.
/// Loopback stays local. This flag does not require upstream routing.
#[arg(
long,
env = "CODEX_EXEC_SERVER_PROXY_PRIVATE_IPS_VIA_UPSTREAM",
global = true
)]
proxy_private_ips_via_upstream: bool,
/// Maximum number of requests to process concurrently on each connection.
#[arg(
long = "concurrent-requests",
value_name = "COUNT",
default_value = "1"
)]
request_dispatch_mode: codex_exec_server::RequestDispatchMode,
/// Transport endpoint URL. Supported values: `ws://IP:PORT` (default), `stdio`, `stdio://`.
#[arg(
long = "listen",
value_name = "URL",
conflicts_with = "exec_server_remote"
)]
listen: Option<String>,
#[command(flatten)]
websocket_auth: WebsocketAuthArgs,
/// Register this exec-server as a remote environment using the given base URL.
#[arg(
long = "remote",
id = "exec_server_remote",
value_name = "URL",
requires = "environment_id",
global = true
)]
remote: Option<String>,
/// Transport used for the remote executor connection.
#[arg(
long = "remote-transport",
value_enum,
default_value_t = ExecServerRemoteTransport::Noise,
requires = "exec_server_remote",
requires_if("direct", "aws_sigv4"),
global = true
)]
remote_transport: ExecServerRemoteTransport,
/// Environment id to attach to when registering remotely.
#[arg(long = "environment-id", value_name = "ID", global = true)]
environment_id: Option<String>,
/// Human-readable environment name.
#[arg(long = "name", value_name = "NAME", global = true)]
name: Option<String>,
/// Use Agent Identity auth from CODEX_ACCESS_TOKEN for remote registration.
#[arg(
long = "use-agent-identity-auth",
requires = "exec_server_remote",
conflicts_with = "aws_sigv4",
global = true
)]
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
)]
executor_credential_stdin: bool,
/// Sign Direct registration and WebSocket handshake requests with AWS SigV4.
#[arg(long = "aws-sigv4", requires = "exec_server_remote", global = true)]
aws_sigv4: bool,
/// AWS profile used for SigV4 authentication.
#[arg(
long = "aws-profile",
value_name = "PROFILE",
requires = "aws_sigv4",
global = true
)]
aws_profile: Option<String>,
/// AWS signing region. Uses the SDK region chain when omitted.
#[arg(
long = "aws-region",
value_name = "REGION",
requires = "aws_sigv4",
global = true
)]
aws_region: Option<String>,
/// AWS signing service.
#[arg(
long = "aws-service",
value_name = "SERVICE",
default_value = "execute-api",
requires = "aws_sigv4",
global = true
)]
aws_service: String,
/// Exit when the parent-owned standard-input pipe closes.
#[arg(
long = "exit-on-stdin-close",
env = codex_exec_server::CODEX_EXEC_SERVER_EXIT_ON_STDIN_CLOSE_ENV_VAR,
requires_if("true", "exec_server_remote"),
global = true
)]
exit_on_stdin_close: bool,
}
impl ExecServerCommand {
/// Starts the executor with the given runtime paths and configuration overrides.
pub async fn run(
mut self,
arg0_paths: &Arg0DispatchPaths,
root_config_overrides: &CliConfigOverrides,
) -> anyhow::Result<()> {
let strict_config = self.strict_config;
self.validate_remote_transport()?;
let enable_cloud_config = self.should_load_cloud_config();
let websocket_auth = self.websocket_auth.try_into_settings()?;
if websocket_auth.config.is_some() && (self.remote.is_some() || self.command.is_some()) {
anyhow::bail!("WebSocket listener auth cannot be used with --remote or forward");
}
let codex_self_exe = arg0_paths
.codex_self_exe
.clone()
.ok_or_else(|| anyhow::anyhow!("Codex executable path is not configured"))?;
let mut runtime_paths = ExecServerRuntimeOptions::new(
codex_self_exe,
arg0_paths.codex_linux_sandbox_exe.clone(),
)?
.with_linux_sandbox_pid_namespace(self.linux_sandbox_pid_namespace)
.with_proxy_private_ips_via_upstream(self.proxy_private_ips_via_upstream);
// Config loading below owns error handling (non-strict stdio can proceed
// without config). Never retain other startup flags in the executor.
if let Ok(overrides) = root_config_overrides.parse_overrides() {
let cli_flags = codex_config::build_cli_overrides_layer(&overrides);
runtime_paths.prefer_mxc = cli_flags
.get("features")
.and_then(|features| features.get("prefer_mxc"))
.and_then(toml::Value::as_bool);
}
if let Some(base_url) = self.remote.take() {
let environment_id = self.environment_id.take().ok_or_else(|| {
anyhow::anyhow!("--environment-id is required when --remote is set")
})?;
let config =
load_exec_server_config(root_config_overrides, strict_config, enable_cloud_config)
.await?;
let direct_transport = self.remote_transport == ExecServerRemoteTransport::Direct;
// 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,
region: self.aws_region,
service: self.aws_service,
},
config.http_client_factory(),
)
.await?
} else {
load_exec_server_remote_auth_provider(
&config,
&base_url,
self.use_agent_identity_auth,
)
.await?
};
let mut remote_config = codex_exec_server::RemoteEnvironmentConfig::new_with_transport(
base_url,
environment_id,
if direct_transport {
codex_exec_server::RemoteEnvironmentTransport::Direct
} else {
codex_exec_server::RemoteEnvironmentTransport::Noise
},
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 (shutdown_sender, shutdown_receiver) = tokio::sync::oneshot::channel();
#[cfg(target_os = "macos")]
let runtime_paths = runtime_paths.with_allowed_symlinked_codex_home(
codex_config::allowed_symlinked_codex_home(
&config.config_layer_stack,
&config.codex_home,
),
);
let result = exec_server_telemetry::run_until_shutdown(
async move {
let shutdown = async move {
let _ = shutdown_receiver.await;
};
match self.command {
Some(ExecServerSubcommand::Forward { connect }) => {
codex_exec_server::run_remote_environment_forward_until_shutdown(
remote_config,
connect,
shutdown,
)
.await
}
None => {
codex_exec_server::run_remote_environment_until_shutdown(
remote_config,
runtime_paths,
shutdown,
)
.await
}
}
.map_err(anyhow::Error::new)
},
parent_lifetime,
exec_server_telemetry::ShutdownBehavior::Graceful(shutdown_sender),
)
.await;
if let Some(provider) = credential_provider {
provider.check_pipe_failure()?;
}
result
} else {
let config_result =
load_exec_server_config(root_config_overrides, strict_config, enable_cloud_config)
.await;
let config = if strict_config {
Some(config_result?)
} else {
config_result.ok()
};
let (_otel, telemetry) = exec_server_telemetry::init(config.as_ref());
#[cfg(target_os = "macos")]
let runtime_paths = runtime_paths.with_allowed_symlinked_codex_home(
config.as_ref().and_then(|config| {
codex_config::allowed_symlinked_codex_home(
&config.config_layer_stack,
&config.codex_home,
)
}),
);
let http_client_factory = config
.as_ref()
.map(Config::http_client_factory)
.unwrap_or_else(|| HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault));
let listen_url = self
.listen
.unwrap_or_else(|| codex_exec_server::DEFAULT_LISTEN_URL.to_string());
let run = exec_server_telemetry::run_until_shutdown(
codex_exec_server::run_main_with_telemetry(
&listen_url,
runtime_paths,
telemetry,
http_client_factory,
self.request_dispatch_mode,
websocket_auth,
),
exec_server_telemetry::ParentLifetime::Independent,
exec_server_telemetry::ShutdownBehavior::Immediate,
);
run.await.map_err(anyhow::Error::from_boxed)
}
}
fn should_load_cloud_config(&self) -> bool {
self.remote.is_some()
&& self.command.is_none()
&& !self.executor_credential_stdin
&& !self.aws_sigv4
&& !self.use_agent_identity_auth
}
fn validate_remote_transport(&self) -> anyhow::Result<()> {
match (self.remote_transport, self.aws_sigv4) {
(ExecServerRemoteTransport::Noise, true) => {
anyhow::bail!("--aws-sigv4 requires --remote-transport direct");
}
(ExecServerRemoteTransport::Direct, false) => {
anyhow::bail!("--remote-transport direct requires --aws-sigv4");
}
(ExecServerRemoteTransport::Noise, false)
| (ExecServerRemoteTransport::Direct, true) => {}
}
if self.remote_transport == ExecServerRemoteTransport::Direct
&& matches!(
self.command.as_ref(),
Some(ExecServerSubcommand::Forward { .. })
)
{
anyhow::bail!("direct exec-server transport does not support forwarding");
}
Ok(())
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, clap::ValueEnum)]
enum ExecServerRemoteTransport {
#[default]
Noise,
Direct,
}
#[derive(Debug, clap::Subcommand)]
enum ExecServerSubcommand {
/// Register an existing WebSocket exec-server as a remote environment.
Forward {
/// Destination exec-server WebSocket URL.
#[arg(long, value_name = "URL", requires = "exec_server_remote")]
connect: String,
},
}
#[cfg(test)]
#[path = "exec_server_args_tests.rs"]
mod tests;
codex-rs/exec-server-cli/src/config.rs
+168
View File
@@ -0,0 +1,168 @@
//! Loads executor startup configuration and remote authentication without changing CLI policy.
use codex_app_server::in_process::EmbeddedNetworkPolicy;
use codex_cloud_config::chatgpt_config_bundle_loader_for_storage;
use codex_config::LoaderOverrides;
use codex_core::config::ConfigBuilder;
use codex_core::config::ConfigLoadOptions;
use codex_core::config::bootstrap_auth_config;
use codex_core::config::find_codex_home;
use codex_core::config::load_config_toml_with_layer_stack;
use codex_login::AuthManager;
use codex_login::CodexAuth;
use codex_login::read_codex_access_token_from_env;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_cli::CliConfigOverrides;
use std::sync::Arc;
pub(super) async fn load_exec_server_remote_auth_provider(
config: &codex_core::config::Config,
base_url: &str,
use_agent_identity_auth: bool,
) -> anyhow::Result<codex_api::SharedAuthProvider> {
if use_agent_identity_auth {
read_codex_access_token_from_env().ok_or_else(|| {
anyhow::anyhow!("CODEX_ACCESS_TOKEN is required when --use-agent-identity-auth is set")
})?;
let auth = AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false)
.await?
.auth()
.await
.ok_or_else(|| anyhow::anyhow!("Agent Identity authentication is unavailable"))?;
if !matches!(auth, CodexAuth::AgentIdentity(_)) {
anyhow::bail!(
"CODEX_ACCESS_TOKEN did not provide permitted Agent Identity authentication"
);
}
return Ok(codex_model_provider::auth_provider_from_auth(&auth));
}
let (auth_manager, auth) = load_exec_server_remote_auth(
config,
"remote exec-server registration requires ChatGPT authentication or API key authentication; run `codex login` or set CODEX_API_KEY",
)
.await?;
if !is_supported_exec_server_remote_auth(&auth) {
anyhow::bail!(
"remote exec-server registration requires ChatGPT authentication or API key authentication; Agent Identity auth requires --use-agent-identity-auth"
);
}
if auth.is_api_key_auth() {
validate_api_key_remote_host(base_url)?;
}
if auth.is_chatgpt_auth() || auth_manager.is_workload_identity_selected() {
Ok(codex_model_provider::auth_provider_from_auth_manager(
auth_manager,
&auth,
))
} else {
Ok(codex_model_provider::auth_provider_from_auth(&auth))
}
}
pub(super) fn is_supported_exec_server_remote_auth(auth: &CodexAuth) -> bool {
auth.is_chatgpt_auth() || auth.is_api_key_auth()
}
pub(super) fn validate_api_key_remote_host(base_url: &str) -> anyhow::Result<()> {
let url = url::Url::parse(base_url)
.map_err(|err| anyhow::anyhow!("invalid remote exec-server registration URL: {err}"))?;
let host = url.host().ok_or_else(|| {
anyhow::anyhow!("remote exec-server registration URL must include a host")
})?;
let is_loopback = match &host {
url::Host::Domain(host) => host.eq_ignore_ascii_case("localhost"),
url::Host::Ipv4(ip) => ip.is_loopback(),
url::Host::Ipv6(ip) => ip.is_loopback(),
};
let is_openai_host = match &host {
url::Host::Domain(host) => ["openai.com", "openai.org"].into_iter().any(|domain| {
host.eq_ignore_ascii_case(domain)
|| host.to_ascii_lowercase().ends_with(&format!(".{domain}"))
}),
_ => false,
};
let is_allowed = match url.scheme() {
"https" => is_loopback || is_openai_host,
"http" => is_loopback,
_ => false,
};
if !is_allowed {
anyhow::bail!(
"remote exec-server API-key authentication is restricted to HTTPS openai.com and openai.org hosts and subdomains or loopback hosts"
);
}
Ok(())
}
pub(super) async fn load_exec_server_config(
root_config_overrides: &CliConfigOverrides,
strict_config: bool,
enable_cloud_config: bool,
) -> anyhow::Result<codex_core::config::Config> {
let cli_kv_overrides = root_config_overrides
.parse_overrides()
.map_err(anyhow::Error::msg)?;
let bootstrap_cli_overrides = cli_kv_overrides.clone();
let mut builder = ConfigBuilder::default()
.cli_overrides(cli_kv_overrides)
.strict_config(strict_config);
if enable_cloud_config {
let codex_home = find_codex_home()?;
let loader_overrides = LoaderOverrides::default();
let network_policy = EmbeddedNetworkPolicy::load(&loader_overrides).await;
let bootstrap_cwd = AbsolutePathBuf::current_dir()?;
let bootstrap_config = load_config_toml_with_layer_stack(
&codex_home,
Some(&bootstrap_cwd),
bootstrap_cli_overrides,
ConfigLoadOptions {
loader_overrides,
strict_config,
cloud_config_bundle: Default::default(),
},
)
.await?;
let bootstrap_auth_config = bootstrap_auth_config(&codex_home, &bootstrap_config)?;
let cloud_config_bundle = chatgpt_config_bundle_loader_for_storage(
network_policy.bind_bootstrap_auth(bootstrap_auth_config),
/*enable_codex_api_key_env*/ true,
)
.await?;
builder = builder.cloud_config_bundle(cloud_config_bundle);
}
let mut config = builder.build().await?;
EmbeddedNetworkPolicy::default().activate(&mut config);
Ok(config)
}
async fn load_exec_server_remote_auth(
config: &codex_core::config::Config,
missing_auth_error: &'static str,
) -> anyhow::Result<(Arc<AuthManager>, codex_login::CodexAuth)> {
let auth_manager =
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ true).await?;
let auth = match auth_manager.auth().await {
Some(auth) => auth,
None => {
auth_manager.reload().await;
auth_manager
.auth()
.await
.ok_or_else(|| anyhow::anyhow!(missing_auth_error))?
}
};
Ok((auth_manager, auth))
}
#[cfg(test)]
#[path = "config_tests.rs"]
mod tests;
codex-rs/exec-server-cli/src/config_tests.rs
+61
View File
@@ -0,0 +1,61 @@
#![cfg(test)]
use super::*;
use pretty_assertions::assert_eq;
#[test]
fn exec_server_remote_auth_accepts_api_key_auth() {
let auth = CodexAuth::from_api_key("sk-test");
assert!(is_supported_exec_server_remote_auth(&auth));
}
#[test]
fn exec_server_remote_api_key_auth_accepts_https_openai_domains() {
for base_url in [
"https://openai.com/api",
"https://service.openai.com/api",
"https://openai.org/api",
"https://service.openai.org/api",
] {
assert!(validate_api_key_remote_host(base_url).is_ok());
}
}
#[test]
fn exec_server_remote_api_key_auth_accepts_http_loopback() {
for base_url in [
"http://localhost:8098/api",
"http://127.0.0.1:8098/api",
"http://[::1]:8098/api",
] {
assert!(validate_api_key_remote_host(base_url).is_ok());
}
}
#[test]
fn exec_server_remote_api_key_auth_rejects_http_openai_domain() {
for base_url in [
"http://service.openai.com/api",
"http://service.openai.org/api",
] {
let error = validate_api_key_remote_host(base_url)
.expect_err("reject plaintext OpenAI destination");
assert_eq!(
error.to_string(),
"remote exec-server API-key authentication is restricted to HTTPS openai.com and openai.org hosts and subdomains or loopback hosts"
);
}
}
#[test]
fn exec_server_remote_api_key_auth_rejects_suffix_spoof() {
let error = validate_api_key_remote_host("https://service.openai.org.evil.example/api")
.expect_err("reject suffix spoof");
assert_eq!(
error.to_string(),
"remote exec-server API-key authentication is restricted to HTTPS openai.com and openai.org hosts and subdomains or loopback hosts"
);
}
codex-rs/exec-server-cli/src/exec_server_args_tests.rs
+339
View File
@@ -0,0 +1,339 @@
#![cfg(test)]
use assert_matches::assert_matches;
use clap::error::ErrorKind;
use pretty_assertions::assert_eq;
use super::*;
use crate::ExecServerCli;
fn exec_server_from_args(args: &[&str]) -> ExecServerCommand {
let cli = ExecServerCli::try_parse_from(
["codex-exec-server"]
.into_iter()
.chain(args.iter().copied()),
)
.expect("parse executor arguments");
cli.command
}
#[test]
fn exec_server_defaults_preserve_local_and_noise_modes() {
for args in [
vec![],
vec!["--listen", "stdio"],
vec![
"--remote",
"https://registry.example.com",
"--environment-id",
"env-1",
],
vec![
"--remote",
"https://registry.example.com",
"--environment-id",
"env-1",
"--remote-transport",
"noise",
"--use-agent-identity-auth",
],
] {
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");
}
}
#[test]
fn remote_forwarder_skips_cloud_config() {
let command = exec_server_from_args(&[
"forward",
"--connect",
"ws://localhost:8765",
"--remote",
"https://registry.example.com",
"--environment-id",
"env-1",
]);
assert!(!command.should_load_cloud_config());
}
#[test]
fn exec_server_parses_direct_aws_options() {
for (options, expected) in [
(vec![], (None, None, "execute-api")),
(
vec![
"--aws-profile",
"development",
"--aws-region",
"us-west-2",
"--aws-service",
"bedrock-mantle",
],
(Some("development"), Some("us-west-2"), "bedrock-mantle"),
),
] {
let mut args = vec![
"--remote",
"https://registry.example.com",
"--environment-id",
"env-1",
"--remote-transport",
"direct",
"--aws-sigv4",
];
args.extend(options);
let command = exec_server_from_args(&args);
assert_eq!(
command.remote.as_deref(),
Some("https://registry.example.com")
);
assert_eq!(command.environment_id.as_deref(), Some("env-1"));
assert_eq!(command.remote_transport, ExecServerRemoteTransport::Direct);
assert!(!command.should_load_cloud_config());
assert!(command.aws_sigv4);
assert_eq!(
(
command.aws_profile.as_deref(),
command.aws_region.as_deref(),
command.aws_service.as_str()
),
expected,
);
command
.validate_remote_transport()
.expect("valid Direct mode");
}
}
#[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 [
(
vec!["--remote-transport", "direct"],
ErrorKind::MissingRequiredArgument,
),
(
vec!["--aws-profile", "development"],
ErrorKind::MissingRequiredArgument,
),
(
vec!["--aws-region", "us-west-2"],
ErrorKind::MissingRequiredArgument,
),
(
vec!["--aws-service", "execute-api"],
ErrorKind::MissingRequiredArgument,
),
(
vec![
"--remote-transport",
"direct",
"--aws-sigv4",
"--use-agent-identity-auth",
],
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,
),
] {
let error = ExecServerCli::try_parse_from(
[
"codex-exec-server",
"--remote",
"https://registry.example.com",
"--environment-id",
"env-1",
]
.into_iter()
.chain(options),
)
.expect_err("reject invalid remote arguments");
assert_eq!(error.kind(), expected);
}
}
#[test]
fn exec_server_transport_and_aws_options_require_registration_arguments() {
for options in [
vec!["--remote-transport", "noise"],
vec!["--remote-transport", "direct", "--aws-sigv4"],
vec!["--aws-sigv4"],
vec!["--executor-credential-stdin"],
vec![
"--remote",
"https://registry.example.com",
"--remote-transport",
"direct",
"--aws-sigv4",
],
] {
let error = ExecServerCli::try_parse_from(["codex-exec-server"].into_iter().chain(options))
.expect_err("require remote URL and environment ID");
assert_eq!(error.kind(), ErrorKind::MissingRequiredArgument);
}
}
#[tokio::test]
async fn exec_server_sigv4_does_not_enable_aws_auth_for_noise() {
for options in [vec![], vec!["--remote-transport", "noise"]] {
let mut args = vec![
"--remote",
"https://registry.example.com",
"--environment-id",
"env-1",
"--aws-sigv4",
];
args.extend(options);
let command = exec_server_from_args(&args);
// Unset runtime paths prove validation runs before startup or credential loading.
let error = command
.run(
&Arg0DispatchPaths::default(),
&CliConfigOverrides::default(),
)
.await
.expect_err("Noise auth is unchanged");
assert_eq!(
error.to_string(),
"--aws-sigv4 requires --remote-transport direct"
);
}
}
#[tokio::test]
async fn exec_server_direct_forwarding_remains_rejected() {
let command = exec_server_from_args(&[
"forward",
"--connect",
"ws://127.0.0.1:8765",
"--remote",
"https://registry.example.com",
"--environment-id",
"env-1",
"--remote-transport",
"direct",
"--aws-sigv4",
"--aws-profile",
"development",
"--aws-region",
"us-west-2",
"--aws-service",
"bedrock-mantle",
]);
assert_eq!(
(
command.remote_transport,
command.aws_sigv4,
command.aws_profile.as_deref(),
command.aws_region.as_deref(),
command.aws_service.as_str()
),
(
ExecServerRemoteTransport::Direct,
true,
Some("development"),
Some("us-west-2"),
"bedrock-mantle"
),
);
let error = command
.run(
&Arg0DispatchPaths::default(),
&CliConfigOverrides::default(),
)
.await
.expect_err("Direct forwarding is unsupported before startup");
assert_eq!(
error.to_string(),
"direct exec-server transport does not support forwarding"
);
}
#[test]
fn exec_server_forward_parses_shared_remote_options() {
let cli = ExecServerCli::try_parse_from([
"codex-exec-server",
"forward",
"--connect",
"ws://127.0.0.1:8765",
"--remote",
"https://example.openai.com",
"--environment-id",
"env-1",
"--name",
"forwarded",
"--strict-config",
"--use-agent-identity-auth",
])
.expect("parse forward");
assert_matches!(cli.command, ExecServerCommand {
command: Some(ExecServerSubcommand::Forward { connect }),
remote: Some(remote),
environment_id: Some(environment_id),
name: Some(name),
strict_config: true,
use_agent_identity_auth: true,
..
} if connect == "ws://127.0.0.1:8765" && remote == "https://example.openai.com" && environment_id == "env-1" && name == "forwarded");
}
#[test]
fn standalone_help_documents_executor_options() {
use clap::CommandFactory;
let help = ExecServerCli::command()
.term_width(80)
.mut_arg("exit_on_stdin_close", |arg| arg.hide_env_values(true))
.mut_arg("proxy_private_ips_via_upstream", |arg| {
arg.hide_env_values(true)
})
.render_long_help()
.to_string()
.lines()
.map(str::trim_end)
.collect::<Vec<_>>()
.join("\n");
insta::assert_snapshot!(help);
}
codex-rs/exec-server-cli/src/exec_server_auth.rs
+80
View File
@@ -0,0 +1,80 @@
//! CLI-only glue between Codex authentication and generic AWS signing.
//!
//! Keeping this adapter here leaves `codex-api` and `codex-aws-auth` independent.
use std::sync::Arc;
use codex_api::AuthError;
use codex_api::AuthProvider;
use codex_api::SharedAuthProvider;
use codex_aws_auth::AwsAuthConfig;
use codex_aws_auth::AwsAuthContext;
use codex_aws_auth::AwsAuthError;
use codex_aws_auth::AwsRequestToSign;
use codex_http_client::Request;
use codex_http_client::RequestBody;
use codex_http_client::RequestCompression;
use http::HeaderMap;
/// Creates a SigV4 provider, preferring an explicit profile over the default credential chain.
pub(super) async fn aws_sigv4_auth_provider(
mut config: AwsAuthConfig,
http_client_factory: codex_http_client::HttpClientFactory,
) -> Result<SharedAuthProvider, AwsAuthError> {
config.profile = config
.profile
.map(|profile| profile.trim().to_string())
.filter(|profile| !profile.is_empty());
config.region = config
.region
.map(|region| region.trim().to_string())
.filter(|region| !region.is_empty());
let context = if config.profile.is_some() {
AwsAuthContext::load_profile(config, http_client_factory).await
} else {
AwsAuthContext::load(config, http_client_factory).await
}?;
Ok(Arc::new(AwsSigV4AuthProvider { context }))
}
#[derive(Debug)]
struct AwsSigV4AuthProvider {
context: AwsAuthContext,
}
impl AuthProvider for AwsSigV4AuthProvider {
fn add_auth_headers(&self, _headers: &mut HeaderMap) -> Result<(), AuthError> {
Ok(())
}
fn apply_auth(&self, mut request: Request) -> codex_api::AuthProviderFuture<'_> {
Box::pin(async move {
let prepared = request.prepare_body_for_send().map_err(AuthError::Build)?;
let signed = self
.context
.sign(AwsRequestToSign {
method: request.method.clone(),
url: request.url.clone(),
headers: prepared.headers.clone(),
body: prepared.body_bytes(),
})
.await
.map_err(|error| {
if error.is_retryable() {
AuthError::Transient(error.to_string())
} else {
AuthError::Build(error.to_string())
}
})?;
request.url = signed.url;
request.headers = signed.headers;
request.body = prepared.body.map(RequestBody::Raw);
request.compression = RequestCompression::None;
Ok(request)
})
}
}
#[cfg(test)]
#[path = "exec_server_auth_tests.rs"]
mod tests;
codex-rs/exec-server-cli/src/exec_server_auth_tests.rs
+107
View File
@@ -0,0 +1,107 @@
#![cfg(test)]
use std::time::Duration;
use codex_aws_auth::AwsAccessKeys;
use http::HeaderValue;
use http::Method;
use pretty_assertions::assert_eq;
use super::*;
async fn test_provider() -> AwsSigV4AuthProvider {
let context = AwsAuthContext::load_with_access_keys(
AwsAuthConfig {
profile: None,
region: Some("us-east-1".to_string()),
service: "execute-api".to_string(),
},
AwsAccessKeys {
access_key_id: "test-access-key".to_string(),
secret_access_key: "test-secret-key".to_string(),
session_token: Some("test-session-token".to_string()),
},
codex_http_client::HttpClientFactory::new(
codex_http_client::OutboundProxyPolicy::ReqwestDefault,
),
)
.await
.expect("load fixture signing context");
AwsSigV4AuthProvider { context }
}
#[tokio::test]
async fn signs_requests_without_changing_payload_or_metadata() {
let provider = test_provider().await;
let url = "https://executor.example.com/connect?environment_id=environment-1";
for mut request in [
Request::new(Method::GET, url.to_string()),
Request::new(Method::POST, url.to_string())
.with_json(&serde_json::json!({"transport": "direct_jsonrpc_v1"})),
Request::new(Method::POST, url.to_string())
.with_json(&serde_json::json!({"transport": "direct_jsonrpc_v1"}))
.with_compression(RequestCompression::Zstd),
] {
request
.headers
.insert("session_id", HeaderValue::from_static("session-1"));
request
.headers
.insert("x-custom-header", HeaderValue::from_static("preserved"));
request.timeout = Some(Duration::from_secs(3));
let method = request.method.clone();
let expected = request
.prepare_body_for_send()
.expect("prepare fixture body");
let signed = provider
.apply_auth(request)
.await
.expect("sign fixture request");
assert_eq!(signed.method, method);
assert_eq!(signed.url, url);
assert_eq!(signed.timeout, Some(Duration::from_secs(3)));
assert_eq!(signed.body, expected.body.clone().map(RequestBody::Raw));
assert_eq!(signed.compression, RequestCompression::None);
for (name, value) in &expected.headers {
assert_eq!(signed.headers.get(name), Some(value));
}
assert_eq!(signed.headers["x-amz-security-token"], "test-session-token");
assert!(signed.headers.contains_key("x-amz-date"));
let authorization = signed.headers[http::header::AUTHORIZATION]
.to_str()
.unwrap();
assert!(authorization.starts_with("AWS4-HMAC-SHA256 "));
assert!(authorization.contains("/us-east-1/execute-api/aws4_request"));
assert_eq!(signed.prepare_body_for_send().unwrap().body, expected.body);
}
}
#[tokio::test]
async fn invalid_signing_request_is_a_permanent_auth_error() {
let provider = test_provider().await;
let error = provider
.apply_auth(Request::new(Method::GET, "not a URL".to_string()))
.await
.expect_err("invalid URL should fail signing");
assert!(matches!(error, AuthError::Build(_)));
}
#[tokio::test]
async fn invalid_signing_configuration_is_rejected() {
let config = AwsAuthConfig {
profile: None,
region: Some("us-east-1".to_string()),
service: " ".to_string(),
};
let error = aws_sigv4_auth_provider(
config,
codex_http_client::HttpClientFactory::new(
codex_http_client::OutboundProxyPolicy::ReqwestDefault,
),
)
.await
.err()
.expect("empty service should fail configuration");
assert!(matches!(error, AwsAuthError::EmptyService));
}
codex-rs/exec-server-cli/src/exec_server_credential.rs
+268
View File
@@ -0,0 +1,268 @@
//! 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::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<ExecutorCredentialProvider>,
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: std::io::Read, F: FnOnce() -> 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<CredentialState>,
}
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 {
fn add_auth_headers(&self, headers: &mut HeaderMap) -> Result<(), AuthError> {
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);
headers.insert(AUTHORIZATION, authorization);
Ok(())
}
fn refresh_after_unauthorized(
&self,
rejected_headers: HeaderMap,
) -> codex_api::AuthRefreshFuture<'_> {
Box::pin(async move {
// The parent owns refresh; only retry if it already supplied a valid replacement.
self.apply_auth_headers(HeaderMap::new())
.await
.is_ok_and(|headers| {
headers.get(AUTHORIZATION) != rejected_headers.get(AUTHORIZATION)
})
})
}
}
enum CredentialState {
Pending,
Active { token: HeaderValue, expires_at: u64 },
Failed(CredentialError),
Closed,
}
impl CredentialState {
fn header(&self) -> Result<HeaderValue, CredentialError> {
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<CredentialState>,
) -> 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<CredentialState, CredentialError> {
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<u64, CredentialError> {
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;
codex-rs/exec-server-cli/src/exec_server_credential_tests.rs
+215
View File
@@ -0,0 +1,215 @@
#![cfg(test)]
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<u8> {
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<CredentialState>) {
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.apply_auth_headers(HeaderMap::new()).await.is_err());
assert!(provider.to_auth_headers().is_err());
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
.apply_auth_headers(HeaderMap::new())
.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"));
assert!(!provider.refresh_after_unauthorized(headers.clone()).await);
writer
.write_all(&frame("env-a", "rotated-scoped-token", expires))
.expect("rotate");
next_update(&mut updates).await;
// A request sent before the parent update can retry with the replacement.
assert!(provider.refresh_after_unauthorized(headers).await);
let headers = provider
.apply_auth_headers(HeaderMap::new())
.await
.expect("headers");
assert_eq!(headers.len(), 1);
assert_eq!(headers[AUTHORIZATION], "Bearer rotated-scoped-token");
assert!(headers[AUTHORIZATION].is_sensitive());
assert!(!provider.refresh_after_unauthorized(headers.clone()).await);
drop(writer);
pipe.parent_closed.await.expect("reader must monitor EOF");
assert!(provider.apply_auth_headers(HeaderMap::new()).await.is_err());
assert!(!provider.refresh_after_unauthorized(headers).await);
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
.apply_auth_headers(HeaderMap::new())
.await
.expect_err("reject auth");
assert_eq!(pipe.provider.check_pipe_failure(), Err(error));
assert!(
!pipe
.provider
.refresh_after_unauthorized(HeaderMap::new())
.await
);
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
.apply_auth_headers(HeaderMap::new())
.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.apply_auth_headers(HeaderMap::new()).await.is_err());
assert!(!provider.refresh_after_unauthorized(HeaderMap::new()).await);
updates.send_replace(CredentialState::Active {
token: HeaderValue::from_static("new-scoped-token"),
expires_at: expired + 60,
});
let headers = provider
.apply_auth_headers(HeaderMap::new())
.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"));
}
}
codex-rs/exec-server-cli/src/exec_server_telemetry.rs
+193
View File
@@ -0,0 +1,193 @@
//! Executor telemetry and shutdown coordination shared by both entry points.
use std::future::Future;
use tracing_subscriber::EnvFilter;
use tracing_subscriber::prelude::*;
const DEFAULT_ANALYTICS_ENABLED: bool = false;
const DEFAULT_LOG_FILTER: &str = "error,opentelemetry_sdk=off,opentelemetry_otlp=off";
const OTEL_SERVICE_NAME: &str = "codex-exec-server";
pub(crate) enum ParentLifetime {
Independent,
StdinPipe,
MonitoredStdin(tokio::sync::oneshot::Receiver<()>),
}
pub(crate) enum ShutdownBehavior {
Immediate,
Graceful(tokio::sync::oneshot::Sender<()>),
}
pub(crate) fn init(
config: Option<&codex_core::config::Config>,
) -> (impl Send + Sync, codex_exec_server::ExecServerTelemetry) {
let fmt_layer = tracing_subscriber::fmt::layer()
.with_writer(std::io::stderr)
.with_filter(stderr_env_filter());
let otel = match config {
Some(config) => codex_core::otel_init::build_provider(
config,
env!("CARGO_PKG_VERSION"),
Some(OTEL_SERVICE_NAME),
DEFAULT_ANALYTICS_ENABLED,
)
.unwrap_or_else(|error| {
eprintln!("Could not create otel exporter: {error}");
None
}),
None => None,
};
let provider = otel.as_ref();
codex_core::otel_init::record_process_start(provider, OTEL_SERVICE_NAME);
let otel_logger_layer = provider.and_then(|otel| otel.logger_layer());
let otel_tracing_layer = provider.and_then(|otel| otel.tracing_layer());
let telemetry = provider
.and_then(|otel| otel.metrics())
.cloned()
.map(codex_exec_server::ExecServerTelemetry::new)
.unwrap_or_default();
let _ = tracing_subscriber::registry()
.with(fmt_layer)
.with(otel_tracing_layer)
.with(otel_logger_layer)
.try_init();
tracing::callsite::rebuild_interest_cache();
(otel, telemetry)
}
pub(crate) async fn run_until_shutdown<F, E>(
run: F,
parent_lifetime: ParentLifetime,
shutdown_behavior: ShutdownBehavior,
) -> Result<(), E>
where
F: Future<Output = Result<(), E>>,
{
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 || {
if let Err(error) =
std::io::copy(&mut std::io::stdin().lock(), &mut std::io::sink())
{
tracing::warn!(%error, "Could not read exec-server parent lifetime pipe");
}
let _ = sender.send(());
});
Some(receiver)
}
};
let parent_disconnected = async {
match parent_disconnected {
Some(receiver) => {
let _ = receiver.await;
}
None => std::future::pending().await,
}
};
let shutdown_signal = match shutdown_signal() {
Ok(signal) => Some(signal),
Err(error) => {
eprintln!("Could not listen for exec-server shutdown signal: {error}");
None
}
};
let shutdown_signal = async {
match shutdown_signal {
Some(signal) => wait_for_shutdown_signal(signal).await,
None => std::future::pending().await,
}
};
run_until_shutdown_with_signals(run, parent_disconnected, shutdown_signal, shutdown_behavior)
.await
}
async fn run_until_shutdown_with_signals<F, E, P, S>(
run: F,
parent_disconnected: P,
shutdown_signal: S,
shutdown_behavior: ShutdownBehavior,
) -> Result<(), E>
where
F: Future<Output = Result<(), E>>,
P: Future<Output = ()>,
S: Future<Output = std::io::Result<()>>,
{
tokio::pin!(run, parent_disconnected, shutdown_signal);
let mut signal_enabled = true;
loop {
tokio::select! {
result = &mut run => return result,
_ = &mut parent_disconnected => {
tracing::info!("Stopping exec-server after its parent closed stdin");
break;
}
signal = &mut shutdown_signal, if signal_enabled => {
match signal {
Ok(()) => break,
Err(error) => {
eprintln!("Could not listen for exec-server shutdown signal: {error}");
signal_enabled = false;
}
}
}
}
}
match shutdown_behavior {
ShutdownBehavior::Immediate => Ok(()),
ShutdownBehavior::Graceful(sender) => {
let _ = sender.send(());
run.await
}
}
}
#[cfg(unix)]
struct ShutdownSignal {
terminate: tokio::signal::unix::Signal,
}
#[cfg(unix)]
fn shutdown_signal() -> std::io::Result<ShutdownSignal> {
Ok(ShutdownSignal {
terminate: tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?,
})
}
#[cfg(unix)]
async fn wait_for_shutdown_signal(mut shutdown_signal: ShutdownSignal) -> std::io::Result<()> {
tokio::select! {
result = tokio::signal::ctrl_c() => result,
_ = shutdown_signal.terminate.recv() => Ok(()),
}
}
#[cfg(not(unix))]
struct ShutdownSignal;
#[cfg(not(unix))]
fn shutdown_signal() -> std::io::Result<ShutdownSignal> {
Ok(ShutdownSignal)
}
#[cfg(not(unix))]
async fn wait_for_shutdown_signal(_: ShutdownSignal) -> std::io::Result<()> {
tokio::signal::ctrl_c().await
}
fn stderr_env_filter() -> EnvFilter {
EnvFilter::try_from_default_env()
.or_else(|_| EnvFilter::try_new(DEFAULT_LOG_FILTER))
.unwrap_or_else(|_| EnvFilter::new("error"))
}
#[cfg(test)]
#[path = "exec_server_telemetry_tests.rs"]
mod tests;
codex-rs/exec-server-cli/src/exec_server_telemetry_tests.rs
+33
View File
@@ -0,0 +1,33 @@
#![cfg(test)]
use std::time::Duration;
use super::ShutdownBehavior;
use super::run_until_shutdown_with_signals;
#[tokio::test]
async fn parent_disconnect_still_stops_executor_after_signal_listener_error() {
let (parent_sender, parent_receiver) = tokio::sync::oneshot::channel();
let (shutdown_sender, shutdown_receiver) = tokio::sync::oneshot::channel();
let run = async move {
let _ = shutdown_receiver.await;
Ok::<(), std::io::Error>(())
};
let task = tokio::spawn(run_until_shutdown_with_signals(
run,
async move {
let _ = parent_receiver.await;
},
std::future::ready(Err(std::io::Error::other("signal listener failed"))),
ShutdownBehavior::Graceful(shutdown_sender),
));
tokio::task::yield_now().await;
parent_sender.send(()).expect("parent lifetime receiver");
tokio::time::timeout(Duration::from_secs(1), task)
.await
.expect("parent disconnect should stop the executor")
.expect("shutdown task should not panic")
.expect("executor should shut down successfully");
}
codex-rs/exec-server-cli/src/lib.rs
+35
View File
@@ -0,0 +1,35 @@
//! Startup shared by the full Codex CLI and the standalone executor.
//! Internal sandbox and filesystem re-exec modes always dispatch before argument parsing.
use clap::CommandFactory;
use clap::Parser;
use codex_utils_cli::CliConfigOverrides;
use codex_utils_cli::collect_config_overrides;
mod command;
mod config;
mod exec_server_auth;
mod exec_server_credential;
mod exec_server_telemetry;
pub use command::ExecServerCommand;
#[derive(Debug, Parser)]
#[command(name = "codex-exec-server", bin_name = "codex-exec-server", version)]
struct ExecServerCli {
#[command(flatten)]
config_overrides: CliConfigOverrides,
#[command(flatten)]
command: ExecServerCommand,
}
/// Starts the executor after dispatching internal helper modes and preparing runtime paths.
pub fn run_standalone() -> anyhow::Result<()> {
codex_arg0::arg0_dispatch_or_else(|paths| async move {
let args: Vec<_> = std::env::args_os().collect();
let mut cli = ExecServerCli::parse_from(&args);
cli.config_overrides = collect_config_overrides(ExecServerCli::command(), args)
.unwrap_or_else(|error| error.exit());
cli.command.run(&paths, &cli.config_overrides).await
})
}
codex-rs/exec-server-cli/src/main.rs
+14
View File
@@ -0,0 +1,14 @@
//! Standalone executor entry point; startup and sandbox dispatch are shared with Codex.
#[cfg(all(
target_os = "linux",
target_env = "musl",
any(target_arch = "x86_64", target_arch = "aarch64")
))]
#[global_allocator]
static ALLOCATOR: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
fn main() -> anyhow::Result<()> {
codex_build_info::initialize!();
codex_exec_server_cli::run_standalone()
}
codex-rs/exec-server-cli/src/snapshots/codex_exec_server_cli__command__tests__standalone_help_documents_executor_options.snap
+124
View File
@@ -0,0 +1,124 @@
---
source: exec-server-cli/src/exec_server_args_tests.rs
assertion_line: 318
expression: help
---
[EXPERIMENTAL] Run the standalone exec-server service
Usage: codex-exec-server [OPTIONS] [COMMAND]
Commands:
forward Register an existing WebSocket exec-server as a remote environment
help Print this message or the help of the given subcommand(s)
Options:
-c, --config <key=value>
Override a configuration value that would otherwise be loaded from
`~/.codex/config.toml`. Use a dotted path (`foo.bar.baz`) to override
nested values. The `value` portion is parsed as TOML. If it fails to
parse as TOML, the raw string is used as a literal.
Examples: - `-c model="o3"` - `-c
'sandbox_permissions=["disk-full-read-access"]'` - `-c
shell_environment_policy.inherit=all`
--strict-config
Error out when config.toml contains fields that are not recognized by
this version of Codex
--linux-sandbox-pid-namespace <MODE>
Linux PID namespace: isolate (default) or inherit. Inherit allows
signals to other same-UID processes; enable only when provisioning a
dedicated environment
[default: isolate]
--proxy-private-ips-via-upstream
Allow permitted private IP destinations to use the configured upstream
proxy. If no valid upstream proxy applies to the request protocol,
connect directly. Loopback stays local. This flag does not require
upstream routing
[env: CODEX_EXEC_SERVER_PROXY_PRIVATE_IPS_VIA_UPSTREAM]
--concurrent-requests <COUNT>
Maximum number of requests to process concurrently on each connection
[default: 1]
--listen <URL>
Transport endpoint URL. Supported values: `ws://IP:PORT` (default),
`stdio`, `stdio://`
--ws-auth <MODE>
Authentication mode for incoming WebSocket connections
[possible values: capability-token, signed-bearer-token]
--ws-token-file <PATH>
Absolute path to the capability-token file
--ws-token-sha256 <HEX>
Hex-encoded SHA-256 digest of the capability token
--ws-shared-secret-file <PATH>
Absolute path to the shared secret file for signed JWT bearer tokens
--ws-issuer <ISSUER>
Expected issuer for signed JWT bearer tokens
--ws-audience <AUDIENCE>
Expected audience for signed JWT bearer tokens
--ws-max-clock-skew-seconds <SECONDS>
Maximum clock skew when validating signed JWT bearer tokens
--remote <URL>
Register this exec-server as a remote environment using the given base
URL
--remote-transport <REMOTE_TRANSPORT>
Transport used for the remote executor connection
[default: noise]
[possible values: noise, direct]
--environment-id <ID>
Environment id to attach to when registering remotely
--name <NAME>
Human-readable environment name
--use-agent-identity-auth
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
--aws-profile <PROFILE>
AWS profile used for SigV4 authentication
--aws-region <REGION>
AWS signing region. Uses the SDK region chain when omitted
--aws-service <SERVICE>
AWS signing service
[default: execute-api]
--exit-on-stdin-close
Exit when the parent-owned standard-input pipe closes
[env: CODEX_EXEC_SERVER_EXIT_ON_STDIN_CLOSE]
-h, --help
Print help (see a summary with '-h')
-V, --version
Print version
codex-rs/exec-server-cli/tests/standalone.rs
+150
View File
@@ -0,0 +1,150 @@
//! Exercise the shipped executor through its public stdio interface, including child re-exec.
use anyhow::Context;
use codex_exec_server::FileSystemSandboxContext;
#[cfg(unix)]
use codex_protocol::models::PermissionProfile;
#[cfg(unix)]
use codex_protocol::permissions::FileSystemAccessMode;
#[cfg(unix)]
use codex_protocol::permissions::FileSystemSandboxEntry;
#[cfg(unix)]
use codex_protocol::permissions::NetworkSandboxPolicy;
#[cfg(unix)]
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_cargo_bin::cargo_bin;
use codex_utils_process::background_command;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use std::process::Stdio;
use tokio::io::AsyncBufReadExt;
use tokio::io::AsyncWriteExt;
use tokio::io::BufReader;
#[tokio::test]
async fn standalone_initializes_reports_version_and_runs_helpers() -> anyhow::Result<()> {
let binary = cargo_bin("codex-exec-server")?;
let directory = tempfile::tempdir()?;
let cwd = url::Url::from_directory_path(directory.path())
.map_err(|()| anyhow::anyhow!("file URI"))?;
let version = tokio::process::Command::from(background_command(&binary))
.arg("--version")
.env("CODEX_HOME", directory.path())
.output()
.await?;
assert!(version.status.success(), "{version:?}");
assert_eq!(
String::from_utf8(version.stdout)?,
format!("codex-exec-server {}\n", env!("CARGO_PKG_VERSION"))
);
// The unsandboxed case also runs on Windows. Unix additionally verifies that
// sandboxed children and filesystem helpers can re-execute the standalone binary.
#[cfg(not(unix))]
let sandboxes: Vec<Option<FileSystemSandboxContext>> = vec![None];
#[cfg(unix)]
let sandboxes = {
let mut sandboxes = vec![None];
if codex_sandboxing::system_bwrap_warning(&PermissionProfile::read_only(), directory.path())
.is_none()
{
// Restrict reads as well, so fs/readFile must dispatch the filesystem helper.
let hidden = directory.path().join("hidden.txt");
std::fs::write(&hidden, "private")?;
let mut policy = PermissionProfile::read_only().file_system_sandbox_policy();
policy.entries.push(FileSystemSandboxEntry::new(
AbsolutePathBuf::try_from(hidden)?.into(),
FileSystemAccessMode::Deny,
));
let permissions = PermissionProfile::from_runtime_permissions(
&policy,
NetworkSandboxPolicy::Restricted,
);
sandboxes.push(Some(FileSystemSandboxContext::from_permission_profile(
permissions,
cwd.as_str().parse()?,
)));
} else {
eprintln!("sandbox case unavailable: system bubblewrap cannot run");
}
sandboxes
};
for sandbox in sandboxes {
let mut child = tokio::process::Command::from(background_command(&binary))
.args(["--listen", "stdio"])
.current_dir(directory.path())
.env("CODEX_HOME", directory.path())
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::inherit())
.kill_on_drop(true)
.spawn()?;
let mut stdin = child.stdin.take().context("executor stdin")?;
let mut stdout = BufReader::new(child.stdout.take().context("executor stdout")?).lines();
let request =
json!({"id": 1, "method": "initialize", "params": {"clientName": "standalone-test"}});
stdin.write_all(format!("{request}\n").as_bytes()).await?;
let response: Value =
serde_json::from_str(&stdout.next_line().await?.context("initialize response")?)?;
assert_eq!(response["id"], 1);
assert!(response["result"]["sessionId"].is_string(), "{response}");
let environment = response["result"]["environmentInfo"].clone();
assert_eq!(environment["executorVersion"], env!("CARGO_PKG_VERSION"));
let initialized = json!({"method": "initialized", "params": {}});
let info = json!({"id": 2, "method": "environment/info", "params": {}});
stdin
.write_all(format!("{initialized}\n{info}\n").as_bytes())
.await?;
let response: Value =
serde_json::from_str(&stdout.next_line().await?.context("environment response")?)?;
assert_eq!(response, json!({"id": 2, "result": environment}));
let file = directory.path().join("readable.txt");
std::fs::write(&file, "ok")?;
let read = json!({"id": 3, "method": "fs/readFile", "params": {
"path": url::Url::from_file_path(&file).unwrap(), "sandbox": sandbox,
}});
stdin.write_all(format!("{read}\n").as_bytes()).await?;
let response: Value =
serde_json::from_str(&stdout.next_line().await?.context("read response")?)?;
assert_eq!(response, json!({"id": 3, "result": {"dataBase64": "b2s="}}));
#[cfg(windows)]
let argv = vec!["cmd.exe", "/C", "exit 0"];
#[cfg(not(windows))]
let argv = vec![
"/bin/sh",
"-c",
"if touch forbidden 2>/dev/null; then exit 7; fi",
];
#[cfg(not(windows))]
let expected_exit = if sandbox.is_some() { 0 } else { 7 };
#[cfg(windows)]
let expected_exit = 0;
let start = json!({"id": 4, "method": "process/start", "params": {
"processId": "standalone-process", "cwd": cwd, "sandbox": sandbox,
"argv": argv, "env": {"PATH": std::env::var("PATH")?}, "tty": false, "pipeStdin": false, "arg0": null,
}});
stdin.write_all(format!("{start}\n").as_bytes()).await?;
let mut started = false;
let mut exit_code = None;
let mut notifications = Vec::new();
while !started || exit_code.is_none() {
let response: Value =
serde_json::from_str(&stdout.next_line().await?.context("process response")?)?;
if response["id"] == 4 {
assert!(response.get("error").is_none(), "{response}");
started = true;
}
if response["method"] == "process/exited" {
exit_code = response["params"]["exitCode"].as_i64();
}
notifications.push(response);
}
assert_eq!(exit_code, Some(expected_exit), "{notifications:#?}");
drop(stdin);
assert!(child.wait().await?.success());
}
Ok(())
}