Harden Windows sandbox helper cleanup (#40808)

## Why

Windows filesystem operations can complete while their helper process remains
alive, and elevated sandbox processes must not survive a disconnected control
pipe.

## What changed

- Read newline-delimited filesystem helper responses without imposing an
  operation deadline, then bound the time spent reaping the helper.
- Drain helper stderr concurrently, retain a bounded diagnostic on failure, and
  kill helpers that do not exit after responding.
- Terminate the elevated runner's job or process when its input pipe closes or
  fails.

## Testing

Added coverage for delayed responses, noisy stderr, stuck helpers, duplicated
file handles, and termination of elevated descendants after a control-pipe
disconnect.

GitOrigin-RevId: c0d3d5d76e13318de17638df4db71fcfb9a7397f

codex-rs/exec-server/src/fs_sandbox.rs

@@ -1,4 +1,6 @@
 use std::collections::HashMap;
+#[cfg(any(windows, test))]
+use std::time::Duration;
 
 use codex_exec_server_protocol::JSONRPCErrorError;
 use codex_protocol::models::PermissionProfile;
@@ -18,6 +20,10 @@ use codex_sandboxing::SandboxablePreference;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use codex_utils_absolute_path::canonicalize_preserving_symlinks;
 use codex_utils_path_uri::PathUri;
+#[cfg(any(windows, test))]
+use tokio::io::AsyncBufReadExt;
+#[cfg(any(windows, test))]
+use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWriteExt;
 use tokio::process::Command;
 
@@ -32,6 +38,10 @@ use crate::rpc::internal_error;
 use crate::rpc::invalid_request;
 
 const FS_HELPER_ENV_ALLOWLIST: &[&str] = &["PATH", "TMPDIR", "TMP", "TEMP"];
+#[cfg(any(windows, test))]
+const FS_HELPER_EXIT_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 2);
+#[cfg(any(windows, test))]
+const MAX_FS_HELPER_STDERR_BYTES: u64 = 4096;
 #[cfg(debug_assertions)]
 const FS_HELPER_BAZEL_BWRAP_ENV_ALLOWLIST: &[&str] = &[
     "CARGO_BIN_EXE_bwrap",
@@ -313,18 +323,110 @@ async fn run_command(
         .stdin
         .take()
         .ok_or_else(|| internal_error("failed to open fs sandbox helper stdin".to_string()))?;
-    stdin.write_all(&request_json).await.map_err(io_error)?;
-    stdin.shutdown().await.map_err(io_error)?;
-    drop(stdin);
 
-    let output = wait_for_helper_output(child).await?;
-    let response: FsHelperResponse = serde_json::from_slice(&output.stdout).map_err(json_error)?;
+    #[cfg(windows)]
+    let mut request_json = request_json;
+    #[cfg(windows)]
+    request_json.push(b'\n');
+    stdin.write_all(&request_json).await.map_err(io_error)?;
+
+    #[cfg(windows)]
+    let response = {
+        stdin.flush().await.map_err(io_error)?;
+        let stdout = child
+            .stdout
+            .take()
+            .ok_or_else(|| internal_error("failed to open fs sandbox helper stdout".to_string()))?;
+        let stderr = drain_helper_stderr(&mut child);
+        let response = read_helper_response(stdout).await;
+        drop(stdin);
+        reap_helper_after_response(child, stderr).await?;
+        response?
+    };
+
+    #[cfg(not(windows))]
+    let response = {
+        stdin.shutdown().await.map_err(io_error)?;
+        drop(stdin);
+        wait_for_helper_output(child).await?.stdout
+    };
+
+    let response = serde_json::from_slice(&response).map_err(json_error)?;
     match response {
         FsHelperResponse::Ok(payload) => Ok(payload),
         FsHelperResponse::Error(error) => Err(error),
     }
 }
 
+#[cfg(any(windows, test))]
+pub(crate) async fn read_helper_response(
+    stdout: impl tokio::io::AsyncRead + Unpin,
+) -> Result<Vec<u8>, JSONRPCErrorError> {
+    let mut response = Vec::new();
+    let bytes_read = tokio::io::BufReader::new(stdout)
+        .read_until(b'\n', &mut response)
+        .await
+        .map_err(io_error)?;
+    if bytes_read == 0 {
+        return Err(internal_error(
+            "fs sandbox helper closed stdout without responding".to_string(),
+        ));
+    }
+    Ok(response)
+}
+
+#[cfg(any(windows, test))]
+pub(crate) fn drain_helper_stderr(
+    child: &mut tokio::process::Child,
+) -> tokio::task::JoinHandle<Result<Vec<u8>, std::io::Error>> {
+    let stderr_pipe = child.stderr.take();
+    tokio::spawn(async move {
+        let mut stderr = Vec::new();
+        if let Some(mut stderr_pipe) = stderr_pipe {
+            (&mut stderr_pipe)
+                .take(MAX_FS_HELPER_STDERR_BYTES)
+                .read_to_end(&mut stderr)
+                .await?;
+            tokio::io::copy(&mut stderr_pipe, &mut tokio::io::sink()).await?;
+        }
+        Ok::<_, std::io::Error>(stderr)
+    })
+}
+
+#[cfg(any(windows, test))]
+pub(crate) async fn reap_helper_after_response(
+    mut child: tokio::process::Child,
+    stderr: tokio::task::JoinHandle<Result<Vec<u8>, std::io::Error>>,
+) -> Result<(), JSONRPCErrorError> {
+    let (status, stderr) = match tokio::time::timeout(FS_HELPER_EXIT_TIMEOUT, async {
+        tokio::try_join!(child.wait(), async {
+            stderr.await.map_err(std::io::Error::other)?
+        })
+    })
+    .await
+    {
+        Ok(result) => result.map_err(io_error)?,
+        Err(_) => {
+            tokio::time::timeout(FS_HELPER_EXIT_TIMEOUT, child.kill())
+                .await
+                .map_err(|_| {
+                    internal_error("fs sandbox helper did not stop after its response".to_string())
+                })?
+                .map_err(io_error)?;
+            return Ok(());
+        }
+    };
+    if status.success() {
+        return Ok(());
+    }
+
+    Err(internal_error(format!(
+        "fs sandbox helper failed with status {status}: {stderr}",
+        stderr = String::from_utf8_lossy(&stderr).trim()
+    )))
+}
+
+#[cfg(not(windows))]
 pub(crate) async fn wait_for_helper_output(
     child: tokio::process::Child,
 ) -> Result<std::process::Output, JSONRPCErrorError> {
@@ -392,6 +494,10 @@ fn json_error(err: serde_json::Error) -> JSONRPCErrorError {
     ))
 }
 
+#[cfg(test)]
+#[path = "fs_sandbox_windows_tests.rs"]
+mod windows_tests;
+
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;

codex-rs/exec-server/src/fs_sandbox_windows_tests.rs

@@ -0,0 +1,261 @@
+#[cfg(windows)]
+use std::collections::HashMap;
+#[cfg(windows)]
+use std::path::Path;
+use std::process::Stdio;
+use std::time::Duration;
+
+#[cfg(windows)]
+use codex_protocol::config_types::WindowsSandboxLevel;
+#[cfg(windows)]
+use codex_protocol::models::PermissionProfile;
+#[cfg(windows)]
+use codex_sandboxing::SandboxExecRequest;
+#[cfg(windows)]
+use codex_sandboxing::SandboxType;
+#[cfg(windows)]
+use codex_utils_path_uri::PathUri;
+use pretty_assertions::assert_eq;
+#[cfg(windows)]
+use tokio::io::AsyncReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::process::Command;
+
+use super::drain_helper_stderr;
+use super::read_helper_response;
+use super::reap_helper_after_response;
+#[cfg(windows)]
+use super::run_command;
+#[cfg(windows)]
+use crate::fs_helper::FsHelperPayload;
+#[cfg(windows)]
+use crate::protocol::FsReadFileResponse;
+#[cfg(windows)]
+use crate::protocol::FsWriteFileResponse;
+
+#[tokio::test(start_paused = true)]
+async fn filesystem_operation_is_not_limited_by_helper_response_deadline() {
+    let (reader, mut writer) = tokio::io::duplex(/*max_buf_size*/ 256);
+    let response = tokio::spawn(async move { read_helper_response(reader).await });
+
+    tokio::task::yield_now().await;
+    tokio::time::advance(Duration::from_secs(/*secs*/ 31)).await;
+    tokio::task::yield_now().await;
+    assert!(
+        !response.is_finished(),
+        "a filesystem operation must not time out before the helper responds"
+    );
+
+    writer
+        .write_all(b"completed after the old deadline\n")
+        .await
+        .expect("helper response");
+
+    assert_eq!(
+        response
+            .await
+            .expect("response task")
+            .expect("unbounded filesystem operation"),
+        b"completed after the old deadline\n"
+    );
+}
+
+#[tokio::test]
+async fn noisy_failing_helper_preserves_exit_status_and_bounded_stderr() {
+    #[cfg(unix)]
+    let mut command = {
+        let mut command = Command::new("sh");
+        command.arg("-c").arg(
+            "printf 'expected helper diagnostic' >&2; i=0; while [ \"$i\" -lt 1024 ]; do printf '%0128d' 0 >&2; i=$((i + 1)); done; exit 7",
+        );
+        command
+    };
+    #[cfg(windows)]
+    let mut command = {
+        let system_root = std::env::var("SystemRoot").expect("Windows system root");
+        let powershell = Path::new(&system_root)
+            .join("System32")
+            .join("WindowsPowerShell")
+            .join("v1.0")
+            .join("powershell.exe");
+        let mut command = Command::new(powershell);
+        command
+            .arg("-NoProfile")
+            .arg("-Command")
+            .arg("[Console]::Error.Write('expected helper diagnostic' + ('x' * 131072)); exit 7");
+        command
+    };
+    command.stdout(Stdio::null());
+    command.stderr(Stdio::piped());
+    command.kill_on_drop(true);
+    let mut child = command.spawn().expect("noisy helper process");
+    let stderr = drain_helper_stderr(&mut child);
+
+    let error = tokio::time::timeout(
+        Duration::from_secs(/*secs*/ 8),
+        reap_helper_after_response(child, stderr),
+    )
+    .await
+    .expect("helper stderr must be drained during bounded cleanup")
+    .expect_err("nonzero helper exit should fail after its stderr pipe fills");
+
+    assert!(error.message.contains('7'), "{}", error.message);
+    assert!(
+        error.message.contains("expected helper diagnostic"),
+        "{}",
+        error.message
+    );
+    assert!(error.message.len() < 4400, "{}", error.message.len());
+}
+
+#[tokio::test]
+async fn helper_stderr_is_drained_before_the_response() {
+    #[cfg(unix)]
+    let mut command = {
+        let mut command = Command::new("sh");
+        command.arg("-c").arg(
+            "printf 'expected pre-response diagnostic' >&2; i=0; while [ \"$i\" -lt 1024 ]; do printf '%0128d' 0 >&2; i=$((i + 1)); done; printf 'completed after noisy stderr\\n'",
+        );
+        command
+    };
+    #[cfg(windows)]
+    let mut command = {
+        let system_root = std::env::var("SystemRoot").expect("Windows system root");
+        let powershell = Path::new(&system_root)
+            .join("System32")
+            .join("WindowsPowerShell")
+            .join("v1.0")
+            .join("powershell.exe");
+        let mut command = Command::new(powershell);
+        command.arg("-NoProfile").arg("-Command").arg(
+            "[Console]::Error.Write('expected pre-response diagnostic' + ('x' * 131072)); [Console]::Out.WriteLine('completed after noisy stderr')",
+        );
+        command
+    };
+    command.stdout(Stdio::piped());
+    command.stderr(Stdio::piped());
+    command.kill_on_drop(true);
+    let mut child = command.spawn().expect("noisy helper process");
+    let stdout = child.stdout.take().expect("helper stdout");
+    let stderr = drain_helper_stderr(&mut child);
+
+    let response = tokio::time::timeout(
+        Duration::from_secs(/*secs*/ 2),
+        read_helper_response(stdout),
+    )
+    .await
+    .expect("helper stderr must be drained before awaiting its response")
+    .expect("helper response");
+
+    assert_eq!(response.trim_ascii_end(), b"completed after noisy stderr");
+    reap_helper_after_response(child, stderr)
+        .await
+        .expect("noisy helper should be cleaned up after its response");
+}
+
+#[cfg(windows)]
+#[tokio::test]
+async fn completed_windows_image_read_does_not_wait_for_a_stuck_helper() {
+    let directory = tempfile::tempdir().expect("temporary directory");
+    let path = directory.path().join("image.png");
+    let operations = [
+        (
+            r#"[IO.File]::WriteAllText($env:CODEX_FS_HELPER_TEST_PATH, 'image')
+[Console]::Out.WriteLine('{"status":"ok","payload":{"operation":"fs/writeFile","response":{}}}')"#,
+            FsHelperPayload::WriteFile(FsWriteFileResponse {}),
+        ),
+        (
+            r#"$data = [Convert]::ToBase64String([IO.File]::ReadAllBytes($env:CODEX_FS_HELPER_TEST_PATH))
+[Console]::Out.WriteLine('{"status":"ok","payload":{"operation":"fs/readFile","response":{"dataBase64":"' + $data + '"}}}')"#,
+            FsHelperPayload::ReadFile(FsReadFileResponse {
+                data_base64: "aW1hZ2U=".to_string(),
+            }),
+        ),
+    ];
+
+    for (operation, expected) in operations {
+        let script = format!(
+            "[Console]::In.ReadLine() | Out-Null\n{operation}\n[Console]::Out.Flush()\n[Threading.Thread]::Sleep(30000)"
+        );
+        let command = powershell_command(&script, &path).expect("PowerShell helper command");
+        let result = tokio::time::timeout(
+            Duration::from_secs(/*secs*/ 8),
+            run_command(command, b"{}".to_vec()),
+        )
+        .await
+        .expect("the completed operation must not wait for helper termination")
+        .expect("helper response");
+
+        assert_eq!(result, expected);
+    }
+    assert_eq!(std::fs::read(&path).expect("created file"), b"image");
+}
+
+#[cfg(windows)]
+#[tokio::test]
+async fn duplicated_windows_file_handle_survives_bounded_helper_cleanup() {
+    let directory = tempfile::tempdir().expect("temporary directory");
+    let path = directory.path().join("image.png");
+    std::fs::write(&path, b"image").expect("image file");
+    let command = powershell_command(
+        r#"[Console]::In.ReadLine() | Out-Null
+$file = [IO.File]::OpenRead($env:CODEX_FS_HELPER_TEST_PATH)
+$handle = $file.SafeFileHandle.DangerousGetHandle().ToInt64()
+[Console]::Out.WriteLine('{"status":"ok","payload":{"operation":"fs/open","response":{"processId":' + $PID + ',"fileHandle":' + $handle + '}}}')
+[Console]::Out.Flush()
+[Threading.Thread]::Sleep(30000)"#,
+        &path,
+    )
+    .expect("PowerShell helper command");
+
+    let mut file = tokio::time::timeout(
+        Duration::from_secs(/*secs*/ 8),
+        crate::sandboxed_file_open::open(
+            command,
+            PathUri::from_host_native_path(&path).expect("image path URI"),
+        ),
+    )
+    .await
+    .expect("the opened file must not wait for helper termination")
+    .expect("duplicated file handle");
+    let mut data = Vec::new();
+    file.read_to_end(&mut data).await.expect("image contents");
+
+    assert_eq!(data, b"image");
+}
+
+#[cfg(windows)]
+fn powershell_command(script: &str, path: &Path) -> anyhow::Result<SandboxExecRequest> {
+    let system_root = std::env::var("SystemRoot")?;
+    let powershell = Path::new(&system_root)
+        .join("System32")
+        .join("WindowsPowerShell")
+        .join("v1.0")
+        .join("powershell.exe");
+    let cwd = PathUri::from_host_native_path(std::env::current_dir()?)?;
+
+    Ok(SandboxExecRequest {
+        command: vec![
+            powershell.to_string_lossy().into_owned(),
+            "-NoProfile".to_string(),
+            "-Command".to_string(),
+            script.to_string(),
+        ],
+        cwd: cwd.clone(),
+        sandbox_policy_cwd: cwd,
+        env: HashMap::from([
+            ("SystemRoot".to_string(), system_root),
+            (
+                "CODEX_FS_HELPER_TEST_PATH".to_string(),
+                path.to_string_lossy().into_owned(),
+            ),
+        ]),
+        network: None,
+        network_environment_id: None,
+        sandbox: SandboxType::None,
+        windows_sandbox_level: WindowsSandboxLevel::Disabled,
+        windows_sandbox_private_desktop: false,
+        permission_profile: PermissionProfile::Disabled,
+        arg0: None,
+    })
+}

codex-rs/exec-server/src/sandboxed_file_open.rs

@@ -7,8 +7,15 @@ use crate::fs_helper::FsHelperOpenResponse;
 use crate::fs_helper::FsHelperPayload;
 use crate::fs_helper::FsHelperRequest;
 use crate::fs_helper::FsHelperResponse;
+#[cfg(windows)]
+use crate::fs_sandbox::drain_helper_stderr;
 use crate::fs_sandbox::io_error;
+#[cfg(windows)]
+use crate::fs_sandbox::read_helper_response;
+#[cfg(windows)]
+use crate::fs_sandbox::reap_helper_after_response;
 use crate::fs_sandbox::spawn_command;
+#[cfg(unix)]
 use crate::fs_sandbox::wait_for_helper_output;
 use crate::protocol::FsReadFileParams;
 use crate::rpc::internal_error;
@@ -69,7 +76,6 @@ async fn open_platform(
     command: SandboxExecRequest,
     mut request: Vec<u8>,
 ) -> Result<tokio::fs::File, JSONRPCErrorError> {
-    use tokio::io::AsyncBufReadExt;
     use tokio::io::AsyncWriteExt;
 
     let mut child = spawn_command(command, std::process::Stdio::piped())?;
@@ -83,19 +89,17 @@ async fn open_platform(
         .ok_or_else(|| internal_error("missing fs sandbox helper stdout".to_string()))?;
     request.push(b'\n');
     stdin.write_all(&request).await.map_err(io_error)?;
+    stdin.flush().await.map_err(io_error)?;
+    let stderr = drain_helper_stderr(&mut child);
 
     let result = async {
-        let mut response = Vec::new();
-        tokio::io::BufReader::new(stdout)
-            .read_until(b'\n', &mut response)
-            .await
-            .map_err(io_error)?;
+        let response = read_helper_response(stdout).await?;
         let response = open_response(&response)?;
         duplicate_file_handle(response.process_id, response.file_handle).map_err(io_error)
     }
     .await;
     drop(stdin);
-    wait_for_helper_output(child).await?;
+    reap_helper_after_response(child, stderr).await?;
     result.map(tokio::fs::File::from_std)
 }
 

codex-rs/windows-sandbox-rs/src/bin/command_runner/win.rs

@@ -10,6 +10,9 @@
 #![allow(unsafe_op_in_unsafe_fn)]
 
 mod cwd_junction;
+#[cfg(test)]
+#[path = "win/input_loop_tests.rs"]
+mod input_loop_tests;
 
 use anyhow::Context;
 use anyhow::Result;
@@ -442,8 +445,10 @@ fn spawn_input_loop(
         loop {
             let msg = match read_frame(&mut reader) {
                 Ok(Some(v)) => v,
-                Ok(None) => break,
-                Err(_) => break,
+                Ok(None) | Err(_) => {
+                    terminate_job_or_process(&job, process, log_dir.as_deref());
+                    break;
+                }
             };
             match msg.message {
                 Message::Stdin { payload } => {

codex-rs/windows-sandbox-rs/src/bin/command_runner/win/input_loop_tests.rs

@@ -0,0 +1,85 @@
+use std::os::windows::io::AsRawHandle;
+use std::os::windows::io::FromRawHandle;
+use std::os::windows::io::OwnedHandle;
+use std::process::Stdio;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::time::Duration;
+
+use anyhow::Context;
+use codex_utils_pty::JobObject;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::BufReader;
+use tokio::process::Command;
+use windows_sys::Win32::Foundation::WAIT_OBJECT_0;
+use windows_sys::Win32::System::Threading::OpenProcess;
+use windows_sys::Win32::System::Threading::PROCESS_SYNCHRONIZE;
+use windows_sys::Win32::System::Threading::WaitForSingleObject;
+
+use super::spawn_input_loop;
+
+#[tokio::test]
+async fn disconnected_control_pipe_terminates_elevated_helper_and_descendants() -> anyhow::Result<()>
+{
+    let mut command = Command::new("python");
+    command
+        .args([
+            "-u",
+            "-c",
+            "import subprocess,sys; child=subprocess.Popen([sys.executable,'-c','import time; time.sleep(60)']); print(child.pid,flush=True); child.wait()",
+        ])
+        .stdin(Stdio::null())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::null());
+
+    let job = Arc::new(JobObject::create_without_breakaway()?);
+    let mut helper = job.spawn_contained(&mut command)?;
+    let stdout = helper.stdout.take().context("helper stdout")?;
+    let mut stdout = BufReader::new(stdout);
+    let mut descendant_pid = String::new();
+    tokio::time::timeout(
+        Duration::from_secs(/*secs*/ 10),
+        stdout.read_line(&mut descendant_pid),
+    )
+    .await??;
+    let descendant_pid = descendant_pid.trim().parse()?;
+    let descendant = unsafe {
+        OpenProcess(
+            PROCESS_SYNCHRONIZE,
+            /*bInheritHandle*/ 0,
+            descendant_pid,
+        )
+    };
+    anyhow::ensure!(descendant != 0, "failed to open sandbox descendant");
+    let descendant = unsafe { OwnedHandle::from_raw_handle(descendant as _) };
+    let process = helper.raw_handle().context("helper process handle")? as _;
+
+    let input_loop = spawn_input_loop(
+        tempfile::tempfile()?,
+        /*stdin_handle*/ None,
+        Arc::new(Mutex::new(None)),
+        Arc::clone(&job),
+        process,
+        /*log_dir*/ None,
+    );
+    input_loop
+        .join()
+        .expect("elevated runner input loop should stop after pipe disconnect");
+
+    let status = tokio::time::timeout(Duration::from_secs(/*secs*/ 2), helper.wait())
+        .await
+        .context("elevated helper survived control-pipe disconnect")??;
+    anyhow::ensure!(!status.success(), "elevated helper unexpectedly succeeded");
+    let descendant_status = unsafe {
+        WaitForSingleObject(
+            descendant.as_raw_handle() as _,
+            /*dwMilliseconds*/ 2_000,
+        )
+    };
+    anyhow::ensure!(
+        descendant_status == WAIT_OBJECT_0,
+        "elevated descendant survived control-pipe disconnect"
+    );
+
+    Ok(())
+}