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
This commit is contained in:
viyatb-oai
2026-10-15 22:32:24 +00:00
committed by copyberry
parent 98fd26206d
commit 16e578dd1d
8 changed files with 826 additions and 10 deletions
@@ -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",
+41 -10
View File
@@ -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,
+261
View File
@@ -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<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 {
// 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<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;
@@ -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<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.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"));
}
}
@@ -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 || {
+1
View File
@@ -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;
@@ -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