mirror of
https://github.com/openai/codex.git
synced 2026-10-20 17:38:45 +00:00
Replay exec-server shell snapshots through unnamed files (#48078)
## Why Large shell snapshots need a replay path that does not add their state to the child process environment. ## What changed - Keep captured state cached in memory and give each Bash or Zsh launch an independent, read-only file descriptor to source. Unlink the file before writing shell state, create it under the sandbox-protected daemon directory, and close the carrier descriptor before restoring functions. - Probe descriptor-path access inside the capture sandbox. Retain environment replay for `sh` and when descriptor paths are unavailable; use normal shell startup if preparing the replay file fails. - Group options and aliases during capture so sourcing preserves their parsing behavior. ## Testing Add coverage for concurrent replays with 480 KiB payloads, pipe and TTY execution, stdin preservation, descriptor cleanup, sandbox protection, blocked descriptor-path fallback, and alias/option parsing equivalence. GitOrigin-RevId: 8103d5c37ce4d1885628a4efba3551549f12bea8
This commit is contained in:
@@ -46,6 +46,8 @@ mod sandboxed_file_system;
|
||||
mod server;
|
||||
#[cfg(unix)]
|
||||
mod shell_snapshot;
|
||||
#[cfg(unix)]
|
||||
mod shell_snapshot_file;
|
||||
mod telemetry;
|
||||
mod trace_context;
|
||||
mod websocket_pong_watchdog;
|
||||
|
||||
@@ -379,7 +379,8 @@ impl LocalProcess {
|
||||
#[cfg(unix)]
|
||||
let mut prepared = prepared;
|
||||
#[cfg(unix)]
|
||||
self.inner
|
||||
let snapshot_file = self
|
||||
.inner
|
||||
.shell_snapshots
|
||||
.prepare(
|
||||
¶ms,
|
||||
@@ -413,6 +414,13 @@ impl LocalProcess {
|
||||
);
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
let inherited_fds = snapshot_file
|
||||
.iter()
|
||||
.map(std::os::fd::AsRawFd::as_raw_fd)
|
||||
.collect::<Vec<_>>();
|
||||
#[cfg(not(unix))]
|
||||
let inherited_fds = Vec::new();
|
||||
let spawned_result = codex_sandboxing::spawn_process(codex_sandboxing::SpawnRequest {
|
||||
command: &prepared.command,
|
||||
cwd: prepared.cwd.as_path(),
|
||||
@@ -422,9 +430,11 @@ impl LocalProcess {
|
||||
windows_sandbox: prepared.windows_sandbox_spawn_request(),
|
||||
tty: params.tty,
|
||||
stdin_open: params.tty || params.pipe_stdin,
|
||||
inherited_fds: codex_utils_pty::ChildFds::Inherited(&[]),
|
||||
inherited_fds: codex_utils_pty::ChildFds::Attached(&inherited_fds),
|
||||
})
|
||||
.await;
|
||||
#[cfg(unix)]
|
||||
drop(snapshot_file);
|
||||
let spawned = match spawned_result {
|
||||
Ok(spawned) => spawned,
|
||||
Err(err) => {
|
||||
@@ -827,6 +837,7 @@ impl ExecBackend for LocalProcess {
|
||||
CapturePurpose::Prewarm,
|
||||
)
|
||||
.await
|
||||
.map(|_| ())
|
||||
.map_err(map_handler_error)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
//! Cache captured shell state in executor memory. Replay through a per-launch
|
||||
//! unnamed reader when the capture sandbox permits it; otherwise retain env replay.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::collections::VecDeque;
|
||||
use std::fs::File;
|
||||
use std::os::fd::AsRawFd;
|
||||
use std::process::Stdio;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
@@ -13,7 +18,7 @@ use codex_shell_command::shell_detect::ShellType;
|
||||
use codex_shell_command::shell_snapshot::CapturedSnapshot;
|
||||
use codex_shell_command::shell_snapshot::SnapshotCaptureOptions;
|
||||
use codex_shell_command::shell_snapshot::SnapshotStartup;
|
||||
use codex_shell_command::shell_snapshot::snapshot_capture_script;
|
||||
use codex_shell_command::shell_snapshot::snapshot_source_capture_script;
|
||||
use codex_utils_path_uri::PathUri;
|
||||
use tokio::io::AsyncReadExt;
|
||||
use tokio::process::Command;
|
||||
@@ -58,6 +63,7 @@ struct CachedShellSnapshot {
|
||||
|
||||
struct ShellSnapshot {
|
||||
state: String,
|
||||
file_source: bool,
|
||||
environment: HashMap<String, String>,
|
||||
}
|
||||
|
||||
@@ -85,9 +91,9 @@ impl ShellSnapshotCache {
|
||||
prepared: &mut PreparedExecRequest,
|
||||
telemetry: &ExecServerTelemetry,
|
||||
purpose: CapturePurpose,
|
||||
) -> Result<(), JSONRPCErrorError> {
|
||||
) -> Result<Option<File>, JSONRPCErrorError> {
|
||||
let Some(request) = params.shell_snapshot.as_ref() else {
|
||||
return Ok(());
|
||||
return Ok(None);
|
||||
};
|
||||
if request.scope_id.is_empty() || request.scope_id.len() > MAX_SNAPSHOT_SCOPE_BYTES {
|
||||
return Err(invalid_params(format!(
|
||||
@@ -100,7 +106,7 @@ impl ShellSnapshotCache {
|
||||
|| params.argv[1] != "-lc"
|
||||
|| !prepared.command.ends_with(¶ms.argv)
|
||||
{
|
||||
return Ok(());
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
let shell_type = match request.shell.name.as_str() {
|
||||
@@ -199,12 +205,34 @@ impl ShellSnapshotCache {
|
||||
}
|
||||
};
|
||||
let Ok(snapshot) = snapshot else {
|
||||
return Ok(());
|
||||
return Ok(None);
|
||||
};
|
||||
if purpose == CapturePurpose::Prewarm {
|
||||
return Ok(());
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
// POSIX sh cannot portably close arbitrary descriptors above 9. Keep its
|
||||
// existing replay until the launcher supports remapping child descriptors.
|
||||
let reader = if snapshot.file_source {
|
||||
let state = snapshot.state.clone();
|
||||
match tokio::task::spawn_blocking(move || {
|
||||
crate::shell_snapshot_file::materialize(shell_type, &state)
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(Ok(reader)) => Some(reader),
|
||||
error => {
|
||||
tracing::warn!(
|
||||
?error,
|
||||
"cannot prepare shell snapshot transport; using normal startup"
|
||||
);
|
||||
return Ok(None);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
let request_overrides = params
|
||||
.env
|
||||
.iter()
|
||||
@@ -226,24 +254,29 @@ impl ShellSnapshotCache {
|
||||
.env
|
||||
.retain(|name, _| !shell_environment::is_non_inheritable_env_var(name));
|
||||
|
||||
let mut state = snapshot.state.as_str();
|
||||
let mut state_variables = Vec::new();
|
||||
while !state.is_empty() {
|
||||
let mut end = state.len().min(MAX_SNAPSHOT_ENV_VALUE_BYTES);
|
||||
while !state.is_char_boundary(end) {
|
||||
end -= 1;
|
||||
let restore = if let Some(reader) = &reader {
|
||||
format!(". /dev/fd/{}", reader.as_raw_fd())
|
||||
} else {
|
||||
let mut state = snapshot.state.as_str();
|
||||
let mut state_variables = Vec::new();
|
||||
while !state.is_empty() {
|
||||
let mut end = state.len().min(MAX_SNAPSHOT_ENV_VALUE_BYTES);
|
||||
while !state.is_char_boundary(end) {
|
||||
end -= 1;
|
||||
}
|
||||
let (chunk, remaining) = state.split_at(end);
|
||||
let name = format!("__CODEX_SHELL_SNAPSHOT_STATE_{}", state_variables.len());
|
||||
prepared.env.insert(name.clone(), chunk.to_string());
|
||||
state_variables.push(name);
|
||||
state = remaining;
|
||||
}
|
||||
let (chunk, remaining) = state.split_at(end);
|
||||
let name = format!("__CODEX_SHELL_SNAPSHOT_STATE_{}", state_variables.len());
|
||||
prepared.env.insert(name.clone(), chunk.to_string());
|
||||
state_variables.push(name);
|
||||
state = remaining;
|
||||
}
|
||||
let state_expansion = state_variables
|
||||
.iter()
|
||||
.map(|name| format!("${{{name}}}"))
|
||||
.collect::<String>();
|
||||
let state_variables = state_variables.join(" ");
|
||||
let state_expansion = state_variables
|
||||
.iter()
|
||||
.map(|name| format!("${{{name}}}"))
|
||||
.collect::<String>();
|
||||
let state_variables = state_variables.join(" ");
|
||||
format!("eval \"unset {state_variables}\n{state_expansion}\"")
|
||||
};
|
||||
let shell_start = prepared.command.len() - params.argv.len();
|
||||
// Automatic startup files run before the restoration script and could
|
||||
// reintroduce environment variables that the snapshot already filtered.
|
||||
@@ -255,11 +288,11 @@ impl ShellSnapshotCache {
|
||||
};
|
||||
prepared.command[shell_start + 1] = shell_flag.to_string();
|
||||
prepared.command[shell_start + 2] = format!(
|
||||
"{startup}if ! eval \"unset {state_variables}\n{state_expansion}\" >/dev/null; then printf 'failed to restore shell snapshot\\n' >&2; fi\n{}",
|
||||
"{startup}if ! {restore} >/dev/null; then printf 'failed to restore shell snapshot\\n' >&2; fi\n{}",
|
||||
params.argv[2]
|
||||
);
|
||||
|
||||
Ok(())
|
||||
Ok(reader)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -273,7 +306,7 @@ async fn capture_snapshot(
|
||||
prepared: &PreparedExecRequest,
|
||||
shell_type: ShellType,
|
||||
) -> CaptureResult {
|
||||
let script = snapshot_capture_script(
|
||||
let mut script = snapshot_source_capture_script(
|
||||
shell_type,
|
||||
SnapshotCaptureOptions {
|
||||
startup: SnapshotStartup::Interactive,
|
||||
@@ -287,6 +320,27 @@ async fn capture_snapshot(
|
||||
invalid_params("unsupported shell snapshot script".to_string()),
|
||||
)
|
||||
})?;
|
||||
// Probe under the exact capture sandbox, including its /proc fallback and
|
||||
// read policy. Keep cached env replay when descriptor paths cannot be opened.
|
||||
let probe = if shell_type == ShellType::Sh {
|
||||
None
|
||||
} else {
|
||||
tokio::task::spawn_blocking(move || crate::shell_snapshot_file::materialize(shell_type, ""))
|
||||
.await
|
||||
.ok()
|
||||
.and_then(Result::ok)
|
||||
};
|
||||
const SOURCE_MARKER: &str = "CODEX_SNAPSHOT_SOURCE";
|
||||
script = if let Some(probe) = &probe {
|
||||
let fd = probe.as_raw_fd();
|
||||
// Automatic startup has already run: do not call exec/unset here,
|
||||
// which may be user functions. Capture exit closes this harmless probe.
|
||||
format!(
|
||||
"if case '' in '') ;; esac 2>/dev/null </dev/fd/{fd}; then printf '\\0%s\\0%s\\0' '{SOURCE_MARKER}' 1; else printf '\\0%s\\0%s\\0' '{SOURCE_MARKER}' 0; fi\n{script}"
|
||||
)
|
||||
} else {
|
||||
format!("printf '\\0%s\\0%s\\0' '{SOURCE_MARKER}' 0\n{script}")
|
||||
};
|
||||
let shell_start = prepared.command.len() - params.argv.len();
|
||||
let mut argv = prepared.command.clone();
|
||||
argv[shell_start + 2] = script;
|
||||
@@ -310,12 +364,26 @@ async fn capture_snapshot(
|
||||
if let Some(arg0) = &prepared.arg0 {
|
||||
command.arg0(arg0);
|
||||
}
|
||||
if let Some(probe) = &probe {
|
||||
let fd = probe.as_raw_fd();
|
||||
// SAFETY: the caller owns the CLOEXEC reader through spawn. This only
|
||||
// changes the child's descriptor table, without allocating after fork.
|
||||
unsafe {
|
||||
command.pre_exec(move || {
|
||||
if libc::fcntl(fd, libc::F_SETFD, 0) == -1 {
|
||||
return Err(std::io::Error::last_os_error());
|
||||
}
|
||||
Ok(())
|
||||
});
|
||||
}
|
||||
}
|
||||
let mut child = command.spawn().map_err(|err| {
|
||||
(
|
||||
"spawn_failed",
|
||||
internal_error(format!("cannot capture shell snapshot: {err}")),
|
||||
)
|
||||
})?;
|
||||
drop(probe);
|
||||
let stdout = child.stdout.take().ok_or_else(|| {
|
||||
(
|
||||
"missing_output",
|
||||
@@ -365,7 +433,26 @@ async fn capture_snapshot(
|
||||
)
|
||||
})??;
|
||||
|
||||
parse_snapshot(shell_type, &output, params.env_policy.as_ref())
|
||||
let marker = format!("\0{SOURCE_MARKER}\0");
|
||||
let marker_end = output
|
||||
.windows(marker.len())
|
||||
.position(|part| part == marker.as_bytes())
|
||||
.map(|index| index + marker.len())
|
||||
.ok_or_else(|| {
|
||||
(
|
||||
"invalid_capture",
|
||||
internal_error("missing snapshot source probe".to_string()),
|
||||
)
|
||||
})?;
|
||||
let (flag, captured) = output[marker_end..].split_at_checked(2).ok_or_else(|| {
|
||||
(
|
||||
"invalid_capture",
|
||||
internal_error("incomplete snapshot source probe".to_string()),
|
||||
)
|
||||
})?;
|
||||
let mut snapshot = parse_snapshot(shell_type, captured, params.env_policy.as_ref())?;
|
||||
snapshot.file_source = flag == b"1\0";
|
||||
Ok(snapshot)
|
||||
}
|
||||
|
||||
fn parse_snapshot(
|
||||
@@ -411,7 +498,11 @@ fn parse_snapshot(
|
||||
environment.remove("OLDPWD");
|
||||
environment.retain(|name, _| !shell_environment::is_non_inheritable_env_var(name));
|
||||
|
||||
Ok(ShellSnapshot { state, environment })
|
||||
Ok(ShellSnapshot {
|
||||
state,
|
||||
file_source: false,
|
||||
environment,
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
//! Per-launch snapshot transport, created under the sandbox-protected daemon root.
|
||||
//! No named file ever contains shell state, and the cache owns bytes, not descriptors.
|
||||
//! The caller keeps the reader alive through spawn.
|
||||
|
||||
use codex_shell_command::shell_detect::ShellType;
|
||||
use std::fs::File;
|
||||
use std::io;
|
||||
use std::io::Write;
|
||||
use std::os::fd::AsRawFd;
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
|
||||
pub(super) fn materialize(shell: ShellType, state: &str) -> io::Result<File> {
|
||||
// Mode 0700 alone cannot stop a same-user sandboxed command from replacing
|
||||
// the pathname or retaining a writable handle before unlink. Reuse the root
|
||||
// that every filesystem-restricted sandbox hides, independent of CODEX_HOME.
|
||||
let root = codex_uds::prepare_shared_daemon_socket_directory()?;
|
||||
let directory = tempfile::tempdir_in(root)?;
|
||||
let writer = tempfile::NamedTempFile::new_in(directory.path())?;
|
||||
// Open independently, rather than dup: writing must not advance the reader.
|
||||
let reader = File::open(writer.path())?;
|
||||
let reader = File::from(rustix::io::fcntl_dupfd_cloexec(&reader, /*min*/ 10)?);
|
||||
let (mut writer, path) = writer.into_parts();
|
||||
path.close()?;
|
||||
directory.close()?;
|
||||
if reader.metadata()?.nlink() != 0 {
|
||||
return Err(io::Error::other("shell snapshot transport is still linked"));
|
||||
}
|
||||
|
||||
let fd = reader.as_raw_fd();
|
||||
// Source has opened its own reader before this runs. Close the carrier before
|
||||
// restoring functions that may shadow exec/unset, without borrowing stdin.
|
||||
let close = match shell {
|
||||
ShellType::Bash => format!("\\exec {fd}<&-\n"),
|
||||
ShellType::Zsh => format!(
|
||||
"__codex_snapshot_fd={fd}\nexec {{__codex_snapshot_fd}}<&-\nunset __codex_snapshot_fd\n"
|
||||
),
|
||||
ShellType::Sh | ShellType::PowerShell | ShellType::Cmd => {
|
||||
return Err(io::Error::other("unsupported snapshot shell"));
|
||||
}
|
||||
};
|
||||
writer.write_all(close.as_bytes())?;
|
||||
writer.write_all(state.as_bytes())?;
|
||||
Ok(reader)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "shell_snapshot_file_tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,74 @@
|
||||
use super::materialize;
|
||||
use codex_shell_command::shell_detect::ShellType;
|
||||
use pretty_assertions::assert_eq;
|
||||
use std::collections::HashMap;
|
||||
use std::io::Write;
|
||||
use std::os::fd::AsRawFd;
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
use std::path::Path;
|
||||
use test_case::test_case;
|
||||
|
||||
#[test_case("/bin/bash", ShellType::Bash; "bash")]
|
||||
#[cfg_attr(target_os = "macos", test_case("/bin/zsh", ShellType::Zsh; "zsh"))]
|
||||
#[tokio::test]
|
||||
async fn unnamed_snapshot_preserves_stdin_and_closes_read_only_carrier(
|
||||
shell: &str,
|
||||
shell_type: ShellType,
|
||||
) -> anyhow::Result<()> {
|
||||
let state = "helper() { printf 'restored:%s' \"$1\"; }\nexec() { exit 42; }\n";
|
||||
let mut reader = materialize(shell_type, state)?;
|
||||
let fd = reader.as_raw_fd();
|
||||
#[cfg(target_os = "linux")]
|
||||
assert!(
|
||||
std::fs::read_link(format!("/proc/self/fd/{fd}"))?
|
||||
.starts_with(codex_uds::shared_daemon_socket_directory()?),
|
||||
"the transport must be protected even before it is unlinked"
|
||||
);
|
||||
let flags = unsafe { libc::fcntl(fd, libc::F_GETFD) };
|
||||
assert_eq!(
|
||||
(
|
||||
reader.metadata()?.nlink(),
|
||||
flags & libc::FD_CLOEXEC,
|
||||
reader.write(b"x").unwrap_err().raw_os_error()
|
||||
),
|
||||
(0, libc::FD_CLOEXEC, Some(libc::EBADF)),
|
||||
);
|
||||
let closed_check = format!("if [ -e /dev/fd/{fd} ]; then exit 43; fi\n");
|
||||
let args = vec![
|
||||
"-c".to_string(),
|
||||
format!(". /dev/fd/{fd}\n{closed_check}IFS= read -r line; helper \"$line\"; exit 7"),
|
||||
];
|
||||
let env = HashMap::from([("PATH".to_string(), "/usr/bin:/bin".to_string())]);
|
||||
let spawned =
|
||||
codex_utils_pty::spawn_pipe_process(shell, &args, Path::new("/"), &env, &None, &[fd])
|
||||
.await?;
|
||||
assert_eq!(unsafe { libc::fcntl(fd, libc::F_GETFD) }, flags);
|
||||
drop(reader);
|
||||
let codex_utils_pty::SpawnedProcess {
|
||||
session,
|
||||
mut stdout_rx,
|
||||
mut stderr_rx,
|
||||
exit_rx,
|
||||
} = spawned;
|
||||
session.writer_sender().send(b"input\n".to_vec()).await?;
|
||||
let result = tokio::time::timeout(std::time::Duration::from_secs(10), async {
|
||||
let mut output = Vec::new();
|
||||
while let Some(chunk) = stdout_rx.recv().await {
|
||||
output.extend(chunk);
|
||||
}
|
||||
let mut errors = Vec::new();
|
||||
while let Some(chunk) = stderr_rx.recv().await {
|
||||
errors.extend(chunk);
|
||||
}
|
||||
Ok::<_, anyhow::Error>((String::from_utf8(output)?, errors, exit_rx.await?))
|
||||
})
|
||||
.await;
|
||||
session.terminate();
|
||||
let (output, errors, status) = result??;
|
||||
assert!(
|
||||
output.ends_with("restored:input"),
|
||||
"stdout={output:?}, stderr={errors:?}, status={status}"
|
||||
);
|
||||
assert_eq!((errors, status), (Vec::new(), 7));
|
||||
Ok(())
|
||||
}
|
||||
@@ -123,12 +123,20 @@ async fn snapshot_failure_retries_are_bounded_and_single_flight(
|
||||
cache.prepare(¶ms, &mut prepared, &telemetry, purpose),
|
||||
cache.prepare(¶ms, &mut concurrent, &telemetry, CapturePurpose::Execution),
|
||||
);
|
||||
if prewarming {
|
||||
let first = if prewarming {
|
||||
first.expect_err("prewarm must report failure without caching it");
|
||||
None
|
||||
} else {
|
||||
first.expect("capture failure must preserve command fallback");
|
||||
first.expect("capture failure must preserve command fallback")
|
||||
};
|
||||
let second = second.expect("waiting command must complete even when prewarm fails");
|
||||
for (prepared, reader) in [(&mut prepared, first), (&mut concurrent, second)] {
|
||||
if let Some(reader) = reader {
|
||||
let fd = std::os::fd::AsRawFd::as_raw_fd(&reader);
|
||||
prepared.command[2] =
|
||||
prepared.command[2].replace(&format!("/dev/fd/{fd}"), "/dev/fd/SNAPSHOT");
|
||||
}
|
||||
}
|
||||
second.expect("waiting command must complete even when prewarm fails");
|
||||
assert_eq!(
|
||||
(&prepared.command, &prepared.env),
|
||||
(&concurrent.command, &concurrent.env)
|
||||
|
||||
Reference in New Issue
Block a user