From f3741880f6dbc706252f4c5bb93f061dba18950d Mon Sep 17 00:00:00 2001 From: andrewgu-oai Date: Wed, 26 Aug 2026 23:31:16 +0000 Subject: [PATCH] Harden managed proxy listener handoff (#40999) ## Why Linux managed networking routed proxy connections through filesystem-backed Unix sockets, requiring an extra readable sandbox path and stale socket cleanup. ## What changed - Transfer loopback TCP listeners from the sandbox namespace to host bridge processes over anonymous Unix socket pairs. - Validate transferred descriptors, keep bootstrap channels private to each sandbox launch, and close them before running untrusted commands. - Tie proxy bridge lifetime to the sandbox so cancellation closes active upstream connections. ## Testing - Cover descriptor validation and interrupted handoffs. - Verify concurrent sandboxes keep proxy endpoints isolated and do not inherit privileged sockets. - Exercise managed networking end to end through `app-server`, confirming proxied access succeeds while direct host access remains blocked. GitOrigin-RevId: 6561e09c83b927c604951bf2aa00ff532d179da5 --- codex-rs/Cargo.lock | 2 + .../app-server/tests/suite/v2/turn_start.rs | 221 ++++++++++++ codex-rs/linux-sandbox/Cargo.toml | 2 + codex-rs/linux-sandbox/src/linux_run_main.rs | 25 +- codex-rs/linux-sandbox/src/proxy_lifecycle.rs | 258 +++++++------- .../src/proxy_lifecycle_tests.rs | 191 ++++++----- codex-rs/linux-sandbox/src/proxy_routing.rs | 323 ++++++------------ .../tests/suite/managed_proxy.rs | 279 +++++++++++++++ 8 files changed, 839 insertions(+), 462 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index e0a49eb1ab..d796679002 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3514,11 +3514,13 @@ dependencies = [ "landlock", "libc", "pretty_assertions", + "rustix 1.1.4", "seccompiler", "serde", "serde_json", "sha2 0.10.9", "tempfile", + "test-case", "tokio", "url", ] diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index 4d705dc70b..76cf9d4148 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -1662,6 +1662,227 @@ allow_local_binding = false Ok(()) } +#[cfg(target_os = "linux")] +#[tokio::test] +async fn turn_start_enforces_managed_network_sandbox() -> Result<()> { + skip_if_remote!( + Ok(()), + "network probe uses a host-local test executable and loopback server" + ); + + let server = responses::start_mock_server().await; + let codex_home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()) + .enable_feature(Feature::NetworkProxy) + .enable_feature(Feature::UnifiedExec) + .write(codex_home.path())?; + std::fs::write( + codex_home.path().join("requirements.toml"), + r#" +default_permissions = "managed-network" + +[allowed_permission_profiles] +managed-network = true +":read-only" = true + +[permissions.managed-network] +extends = ":read-only" + +[permissions.managed-network.network] +enabled = true +allow_local_binding = false +allow_upstream_proxy = false + +[permissions.managed-network.network.domains] +"127.0.0.2" = "allow" +"#, + )?; + + let mut app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + let ThreadStartResponse { thread, .. } = app_server + .start_thread(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + + // A different loopback address avoids confusing the host's target port with + // a coincidentally equal 127.0.0.1 listener port inside the sandbox namespace. + let listener = tokio::net::TcpListener::bind("127.0.0.2:0").await?; + let target = listener.local_addr()?; + let router = axum::Router::new().route( + "/sandbox", + axum::routing::get(|| async { "sandbox proxy response" }), + ); + // Keep the real origin listening throughout both the proxied and direct + // requests, and clean it up even if an assertion fails. + let _origin = tokio_util::task::AbortOnDropHandle::new(tokio::spawn(async move { + axum::serve(listener, router).await + })); + let test_executable = std::env::current_exe()?; + let test_executable = test_executable + .to_str() + .context("integration test executable path must be UTF-8")?; + // Libtest selectors omit the integration crate name. + let (_, test_module) = module_path!() + .split_once("::") + .context("network sandbox test must be in an integration test module")?; + let child_test = format!("{test_module}::managed_network_sandbox_probe"); + let target_env = format!("CODEX_TEST_MANAGED_NETWORK_TARGET={target}"); + let command = shlex::try_join([ + "env", + &target_env, + test_executable, + "--exact", + &child_test, + "--ignored", + "--nocapture", + "--test-threads=1", + ])?; + let call_id = "managed-network-probe"; + let probe_timeout = std::time::Duration::from_secs(/*secs*/ 30); + let _command_response = responses::mount_sse_once( + &server, + responses::sse(vec![ + responses::ev_response_created("resp-network-command"), + responses::ev_exec_command_call_with_args( + call_id, + &json!({ + "cmd": command, + "shell": "/bin/sh", + "login": false, + "yield_time_ms": probe_timeout.as_millis(), + }), + ), + responses::ev_completed("resp-network-command"), + ]), + ) + .await; + let final_response = responses::mount_sse_once( + &server, + create_final_assistant_message_sse_response("Network probe complete")?, + ) + .await; + + let _: TurnStartResponse = app_server + .request(|request_id| ClientRequest::TurnStart { + request_id, + params: TurnStartParams { + thread_id: thread.id, + input: vec![V2UserInput::Text { + text: "Verify managed sandbox networking".to_string(), + text_elements: Vec::new(), + }], + permissions: Some("managed-network".to_string()), + ..Default::default() + }, + }) + .await?; + + let expected_output = "proxied request succeeded; direct access blocked\n"; + timeout(probe_timeout, async { + loop { + let completed: ItemCompletedNotification = + app_server.read_notification("item/completed").await?; + if let ThreadItem::CommandExecution { + id, + status, + exit_code, + .. + } = completed.item + && id == call_id + { + assert_eq!( + (status, exit_code), + (CommandExecutionStatus::Completed, Some(0)) + ); + break; + } + } + anyhow::Ok(()) + }) + .await??; + let completed: TurnCompletedNotification = timeout( + probe_timeout, + app_server.read_notification("turn/completed"), + ) + .await??; + assert_eq!(completed.turn.status, TurnStatus::Completed); + // Early stdout can reach the tool-result buffer before the separate event + // subscriber attaches, so verify it in the model continuation instead. + // The marker also proves the child ran: libtest succeeds if no selector matches. + let output = final_response + .single_request() + .function_call_output_text(call_id) + .context("model continuation should contain the sandboxed command output")?; + assert!( + output.contains(expected_output), + "unexpected tool output: {output}" + ); + + Ok(()) +} + +/// Runs as the model-controlled command inside app-server's managed network sandbox. +#[cfg(target_os = "linux")] +#[test] +#[ignore = "invoked through app-server's managed network sandbox"] +fn managed_network_sandbox_probe() -> Result<()> { + use std::io::ErrorKind; + use std::io::Read; + use std::io::Write; + use std::net::SocketAddr; + use std::net::TcpStream; + use std::time::Duration; + + let target: SocketAddr = std::env::var("CODEX_TEST_MANAGED_NETWORK_TARGET")?.parse()?; + let proxy = url::Url::parse(&std::env::var("HTTP_PROXY")?)?; + let proxy_address = SocketAddr::new( + proxy.host_str().context("missing proxy host")?.parse()?, + proxy + .port_or_known_default() + .context("missing proxy port")?, + ); + let proxy_timeout = Duration::from_secs(/*secs*/ 5); + let mut connection = TcpStream::connect_timeout(&proxy_address, proxy_timeout)?; + connection.set_read_timeout(Some(proxy_timeout))?; + connection.set_write_timeout(Some(proxy_timeout))?; + // HTTP/1.0 keeps this small fixture response close-delimited, without chunked encoding. + write!( + connection, + "GET http://{target}/sandbox HTTP/1.0\r\nHost: {target}\r\nConnection: close\r\n\r\n" + )?; + let mut response = String::new(); + connection + .take(/*limit*/ 4096) + .read_to_string(&mut response)?; + let (headers, body) = response + .split_once("\r\n\r\n") + .context("incomplete proxy HTTP response")?; + let status = headers + .split_whitespace() + .nth(/*n*/ 1) + .context("missing HTTP response status")?; + assert_eq!((status, body), ("200", "sandbox proxy response")); + + let direct_error = TcpStream::connect_timeout(&target, Duration::from_secs(/*secs*/ 2)) + .expect_err("sandbox allowed direct host access"); + assert!( + matches!( + direct_error.kind(), + ErrorKind::ConnectionRefused + | ErrorKind::NetworkUnreachable + | ErrorKind::PermissionDenied + ), + "unexpected direct connection error: {direct_error}" + ); + println!("proxied request succeeded; direct access blocked"); + Ok(()) +} + #[tokio::test] async fn turn_start_rejects_unknown_environment_before_starting_turn() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; diff --git a/codex-rs/linux-sandbox/Cargo.toml b/codex-rs/linux-sandbox/Cargo.toml index 0f7c68a152..7655189c7e 100644 --- a/codex-rs/linux-sandbox/Cargo.toml +++ b/codex-rs/linux-sandbox/Cargo.toml @@ -27,6 +27,7 @@ codex-utils-absolute-path = { workspace = true } globset = { workspace = true } landlock = { workspace = true } libc = { workspace = true } +rustix = { workspace = true } seccompiler = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } @@ -37,6 +38,7 @@ url = { workspace = true } codex-core = { workspace = true } pretty_assertions = { workspace = true } tempfile = { workspace = true } +test-case = "3.3.1" tokio = { workspace = true, features = [ "io-std", "macros", diff --git a/codex-rs/linux-sandbox/src/linux_run_main.rs b/codex-rs/linux-sandbox/src/linux_run_main.rs index b50f19732b..43dfbb99fb 100644 --- a/codex-rs/linux-sandbox/src/linux_run_main.rs +++ b/codex-rs/linux-sandbox/src/linux_run_main.rs @@ -179,7 +179,7 @@ pub fn run_main() -> ! { ensure_inner_stage_mode_is_valid(apply_seccomp_then_exec, use_legacy_landlock); let EffectivePermissions { permission_profile, - mut file_system_sandbox_policy, + file_system_sandbox_policy, network_sandbox_policy, } = resolve_permission_profile(permission_profile).unwrap_or_else(|err| panic!("{err}")); ensure_legacy_landlock_mode_supports_policy( @@ -291,16 +291,12 @@ pub fn run_main() -> ! { // Outer stage: bubblewrap first, then re-enter this binary in the // sandboxed environment to apply seccomp. This path never falls back // to legacy Landlock on failure. - let proxy_route_spec = if allow_network_for_proxy { - let (proxy_route_spec, socket_dir) = prepare_host_proxy_route_spec() + let (proxy_route_spec, proxy_controls) = if allow_network_for_proxy { + let (proxy_route_spec, controls) = prepare_host_proxy_route_spec() .unwrap_or_else(|err| panic!("failed to prepare host proxy routing bridge: {err}")); - file_system_sandbox_policy = file_system_sandbox_policy.with_additional_readable_roots( - &sandbox_policy_cwd, - std::slice::from_ref(&socket_dir), - ); - Some(proxy_route_spec) + (Some(proxy_route_spec), controls) } else { - None + (None, Vec::new()) }; let inner = build_inner_seccomp_command(InnerSeccompCommandArgs { sandbox_policy_cwd: &sandbox_policy_cwd, @@ -314,10 +310,10 @@ pub fn run_main() -> ! { &sandbox_policy_cwd, command_cwd.as_deref(), &file_system_sandbox_policy, - network_sandbox_policy, + bwrap_network_mode(network_sandbox_policy, allow_network_for_proxy), inner, + proxy_controls, !no_proc, - allow_network_for_proxy, ); } @@ -398,12 +394,11 @@ fn run_bwrap_with_proc_fallback( sandbox_policy_cwd: &Path, command_cwd: Option<&Path>, file_system_sandbox_policy: &FileSystemSandboxPolicy, - network_sandbox_policy: NetworkSandboxPolicy, + network_mode: BwrapNetworkMode, inner: Vec, + proxy_controls: Vec, mount_proc: bool, - allow_network_for_proxy: bool, ) -> ! { - let network_mode = bwrap_network_mode(network_sandbox_policy, allow_network_for_proxy); let mut mount_proc = mount_proc; let command_cwd = command_cwd.unwrap_or(sandbox_policy_cwd); @@ -429,6 +424,7 @@ fn run_bwrap_with_proc_fallback( options, ) .unwrap_or_else(|err| exit_with_bwrap_build_error(err)); + bwrap_args.preserved_files.extend(proxy_controls); apply_inner_command_argv0(&mut bwrap_args.args); run_or_exec_bwrap(bwrap_args); } @@ -597,6 +593,7 @@ fn run_bwrap_in_child_with_synthetic_mount_cleanup(bwrap_args: crate::bwrap::Bwr exec_bwrap(args, preserved_files); } + drop(preserved_files); close_child_exec_start_read(exec_start_pipe[0]); let protected_create_monitor = ProtectedCreateMonitor::start(&protected_create_targets); let signal_forwarders = install_bwrap_signal_forwarders(pid); diff --git a/codex-rs/linux-sandbox/src/proxy_lifecycle.rs b/codex-rs/linux-sandbox/src/proxy_lifecycle.rs index 3d30d777ff..1ee0148120 100644 --- a/codex-rs/linux-sandbox/src/proxy_lifecycle.rs +++ b/codex-rs/linux-sandbox/src/proxy_lifecycle.rs @@ -1,121 +1,116 @@ +//! Manage privileged bridge processes and transfer their loopback TCP listeners. +//! +//! Received descriptors are CLOEXEC and owned even when validation rejects them. +//! Shape validation does not authenticate the sender: the caller must retain exclusive +//! control of the anonymous channel and close bootstrap descriptors before untrusted execution. + +use rustix::io::retry_on_intr; +use rustix::net::RecvAncillaryBuffer; +use rustix::net::RecvAncillaryMessage; +use rustix::net::RecvFlags; +use rustix::net::ReturnFlags; +use rustix::net::SendAncillaryBuffer; +use rustix::net::SendAncillaryMessage; +use rustix::net::SendFlags; +use rustix::net::SocketType; +use rustix::net::ipproto; +use rustix::net::sockopt; use std::io; -use std::path::Path; -use std::path::PathBuf; -use std::time::Duration; +use std::io::IoSlice; +use std::io::IoSliceMut; +use std::mem::MaybeUninit; +use std::net::Ipv4Addr; +use std::net::SocketAddr; +use std::net::TcpListener; +use std::os::fd::AsFd; +use std::os::fd::AsRawFd; +use std::os::fd::FromRawFd; +use std::os::fd::OwnedFd; +use std::os::unix::net::UnixStream; -pub(crate) const PROXY_SOCKET_DIR_PREFIX: &str = "codex-linux-sandbox-proxy-"; +const LISTENER_MESSAGE: u8 = 1; -pub(crate) fn cleanup_stale_proxy_socket_dirs_in(temp_dir: &Path) -> io::Result<()> { - for entry in std::fs::read_dir(temp_dir)? { - let entry = match entry { - Ok(entry) => entry, - Err(_) => continue, - }; - let file_type = match entry.file_type() { - Ok(file_type) => file_type, - Err(_) => continue, - }; - if !file_type.is_dir() { - continue; - } - - let file_name = entry.file_name(); - let file_name = file_name.to_string_lossy(); - let Some(owner_pid) = parse_proxy_socket_dir_owner_pid(file_name.as_ref()) else { - continue; - }; - if is_pid_alive(owner_pid) { - continue; - } - - let _ = cleanup_proxy_socket_dir(entry.path().as_path()); +pub(crate) fn send_listener(channel: &UnixStream, listener: &TcpListener) -> io::Result<()> { + let descriptors = [listener.as_fd()]; + let mut space = [MaybeUninit::uninit(); rustix::cmsg_space!(ScmRights(1))]; + let mut control = SendAncillaryBuffer::new(&mut space); + if !control.push(SendAncillaryMessage::ScmRights(&descriptors)) { + return Err(io::Error::other("missing proxy listener control header")); + } + // rustix surfaces EINTR on both its libc and raw-syscall backends. + // Retry interrupted sends so a handled signal does not abort the handoff. + let sent = retry_on_intr(|| { + rustix::net::sendmsg( + channel, + &[IoSlice::new(&[LISTENER_MESSAGE])], + &mut control, + SendFlags::NOSIGNAL, + ) + })?; + if sent != 1 { + return Err(io::Error::other("failed to transfer proxy listener")); } - Ok(()) } -fn parse_proxy_socket_dir_owner_pid(file_name: &str) -> Option { - let suffix = file_name.strip_prefix(PROXY_SOCKET_DIR_PREFIX)?; - let (pid_raw, _) = suffix.split_once('-')?; - pid_raw.parse::().ok().filter(|pid| *pid != 0) -} +pub(crate) fn receive_listener(channel: &UnixStream) -> io::Result { + let mut byte = [0_u8]; + let mut space = [MaybeUninit::uninit(); rustix::cmsg_space!(ScmRights(1))]; + let mut control = RecvAncillaryBuffer::new(&mut space); + // rustix surfaces EINTR on both its libc and raw-syscall backends. + // Retry interrupted receives so a handled signal does not abort the handoff. + let message = retry_on_intr(|| { + rustix::net::recvmsg( + channel, + &mut [IoSliceMut::new(&mut byte)], + &mut control, + RecvFlags::CMSG_CLOEXEC, + ) + })?; -fn is_pid_alive(pid: u32) -> bool { - let Ok(pid) = libc::pid_t::try_from(pid) else { - return false; - }; - is_pid_alive_raw(pid) -} - -fn is_pid_alive_raw(pid: libc::pid_t) -> bool { - is_pid_alive_in_proc_root(pid, Path::new("/proc")) -} - -fn is_pid_alive_in_proc_root(pid: libc::pid_t, proc_root: &Path) -> bool { - let status = unsafe { libc::kill(pid, 0) }; - if status != 0 { - let err = io::Error::last_os_error(); - return !matches!(err.raw_os_error(), Some(libc::ESRCH)); - } - - match std::fs::read_to_string(proc_root.join(format!("{pid}/stat"))) { - Ok(status) => !matches!( - status - .rsplit_once(") ") - .and_then(|(_, fields)| fields.chars().next()), - Some('Z') - ), - Err(err) if err.kind() == io::ErrorKind::NotFound => !proc_root.is_dir(), - Err(_) => true, - } -} - -pub(crate) fn spawn_proxy_socket_dir_cleanup_worker( - socket_dir: PathBuf, - host_bridge_pids: Vec, -) -> io::Result<()> { - let pid = unsafe { libc::fork() }; - if pid < 0 { - return Err(io::Error::last_os_error()); - } - - if pid == 0 { - if detach_bridge_stdio().is_err() { - unsafe { libc::_exit(1) }; - } - - loop { - if host_bridge_pids - .iter() - .copied() - .all(|bridge_pid| !is_pid_alive_raw(bridge_pid)) - { - break; - } - std::thread::sleep(Duration::from_millis(100)); - } - - let _ = cleanup_proxy_socket_dir(socket_dir.as_path()); - unsafe { libc::_exit(0) }; - } - - Ok(()) -} - -fn cleanup_proxy_socket_dir(socket_dir: &Path) -> io::Result<()> { - for _ in 0..20 { - match std::fs::remove_dir_all(socket_dir) { - Ok(()) => return Ok(()), - Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(()), - Err(_) => std::thread::sleep(Duration::from_millis(100)), + // Alignment headroom can fit extra descriptors. Own all of them before checking + // the count; rustix also closes any descriptors left undrained on error paths. + let mut descriptors = Vec::new(); + let mut unexpected_message = false; + for ancillary in control.drain() { + match ancillary { + RecvAncillaryMessage::ScmRights(received) => descriptors.extend(received), + _ => unexpected_message = true, } } - - match std::fs::remove_dir_all(socket_dir) { - Ok(()) => Ok(()), - Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(()), - Err(err) => Err(err), + if message.bytes != 1 + || byte != [LISTENER_MESSAGE] + || message + .flags + .intersects(ReturnFlags::CTRUNC | ReturnFlags::TRUNC) + || unexpected_message + || descriptors.len() != 1 + { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "invalid proxy listener handoff", + )); } + + let listener = TcpListener::from(descriptors.remove(/*index*/ 0)); + if sockopt::socket_type(&listener)? != SocketType::STREAM + || sockopt::socket_protocol(&listener)? != Some(ipproto::TCP) + || !sockopt::socket_acceptconn(&listener)? + { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "proxy handoff requires a TCP listener", + )); + } + if !matches!(listener.local_addr()?, SocketAddr::V4(address) + if *address.ip() == Ipv4Addr::LOCALHOST && address.port() != 0) + { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "proxy listener must be bound to IPv4 loopback", + )); + } + Ok(listener) } pub(crate) fn harden_bridge_process(expected_parent_pid: libc::pid_t) -> io::Result<()> { @@ -197,49 +192,26 @@ fn redirect_bridge_output(source_fd: libc::c_int) -> io::Result<()> { Ok(()) } -pub(crate) fn create_ready_pipe() -> io::Result<(libc::c_int, libc::c_int)> { - let mut pipe_fds = [0; 2]; - let res = unsafe { libc::pipe2(pipe_fds.as_mut_ptr(), libc::O_CLOEXEC) }; - if res != 0 { - return Err(io::Error::last_os_error()); - } - - let read_fd = match move_fd_above_stdio(pipe_fds[0]) { - Ok(fd) => fd, - Err(err) => { - let _ = close_fd(pipe_fds[0]); - let _ = close_fd(pipe_fds[1]); - return Err(err); - } - }; - let write_fd = match move_fd_above_stdio(pipe_fds[1]) { - Ok(fd) => fd, - Err(err) => { - let _ = close_fd(read_fd); - let _ = close_fd(pipe_fds[1]); - return Err(err); - } - }; - - Ok((read_fd, write_fd)) -} - -fn move_fd_above_stdio(fd: libc::c_int) -> io::Result { - if fd > libc::STDERR_FILENO { +pub(crate) fn move_fd_above_stdio(fd: OwnedFd) -> io::Result { + if fd.as_raw_fd() > libc::STDERR_FILENO { return Ok(fd); } - let relocated_fd = unsafe { libc::fcntl(fd, libc::F_DUPFD_CLOEXEC, libc::STDERR_FILENO + 1) }; + // SAFETY: F_DUPFD_CLOEXEC takes an integer lower bound, not a pointer; + // `fd` owns the live descriptor throughout the call. + let relocated_fd = unsafe { + libc::fcntl( + fd.as_raw_fd(), + libc::F_DUPFD_CLOEXEC, + libc::STDERR_FILENO + 1, + ) + }; if relocated_fd < 0 { return Err(io::Error::last_os_error()); } - if let Err(err) = close_fd(fd) { - let _ = close_fd(relocated_fd); - return Err(err); - } - - Ok(relocated_fd) + // SAFETY: F_DUPFD_CLOEXEC returned a new descriptor; the original owner drops here. + Ok(unsafe { OwnedFd::from_raw_fd(relocated_fd) }) } pub(crate) fn close_fd(fd: libc::c_int) -> io::Result<()> { diff --git a/codex-rs/linux-sandbox/src/proxy_lifecycle_tests.rs b/codex-rs/linux-sandbox/src/proxy_lifecycle_tests.rs index a18e26fdae..3726a3b0db 100644 --- a/codex-rs/linux-sandbox/src/proxy_lifecycle_tests.rs +++ b/codex-rs/linux-sandbox/src/proxy_lifecycle_tests.rs @@ -1,91 +1,124 @@ -use super::PROXY_SOCKET_DIR_PREFIX; -use super::cleanup_proxy_socket_dir; -use super::cleanup_stale_proxy_socket_dirs_in; -use super::is_pid_alive_in_proc_root; -use super::is_pid_alive_raw; -use super::parse_proxy_socket_dir_owner_pid; +//! Verify listener ownership, handoff validation, and descriptor cleanup on rejection. + +use super::LISTENER_MESSAGE; +use super::receive_listener; +use super::send_listener; use pretty_assertions::assert_eq; -use std::process::Command; +use rustix::io::FdFlags; +use rustix::io::fcntl_getfd; +use rustix::net::SendAncillaryBuffer; +use rustix::net::SendAncillaryMessage; +use rustix::net::SendFlags; +use std::io; +use std::io::IoSlice; +use std::io::Read; +use std::io::Write; +use std::mem::MaybeUninit; +use std::net::Ipv4Addr; +use std::net::TcpListener; +use std::net::TcpStream; +use std::os::fd::AsFd; +use std::os::fd::BorrowedFd; +use std::os::unix::net::UnixStream; use std::time::Duration; -use std::time::Instant; +use test_case::test_case; #[test] -fn cleanup_proxy_socket_dir_removes_bridge_artifacts() { - let root = tempfile::tempdir().expect("tempdir should create"); - let socket_dir = root.path().join("codex-linux-sandbox-proxy-test"); - std::fs::create_dir(&socket_dir).expect("socket dir should create"); - let marker = socket_dir.join("bridge.sock"); - std::fs::write(&marker, b"test").expect("marker should write"); +fn transferred_listener_survives_sender_close_and_is_cloexec() -> io::Result<()> { + let (sender, receiver) = UnixStream::pair()?; + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0))?; + let address = listener.local_addr()?; + send_listener(&sender, &listener)?; + drop(listener); + drop(sender); - cleanup_proxy_socket_dir(socket_dir.as_path()).expect("cleanup should succeed"); - - assert_eq!(socket_dir.exists(), false); -} - -#[test] -fn parse_proxy_socket_dir_owner_pid_reads_owner_pid() { + let listener = receive_listener(&receiver)?; + let flags = fcntl_getfd(&listener)?; assert_eq!( - parse_proxy_socket_dir_owner_pid("codex-linux-sandbox-proxy-1234-0"), - Some(1234) + (listener.local_addr()?, flags & FdFlags::CLOEXEC), + (address, FdFlags::CLOEXEC) ); - assert_eq!( - parse_proxy_socket_dir_owner_pid("codex-linux-sandbox-proxy-1234-1000-0"), - Some(1234) - ); - assert_eq!( - parse_proxy_socket_dir_owner_pid("codex-linux-sandbox-proxy-x"), - None - ); - assert_eq!(parse_proxy_socket_dir_owner_pid("not-a-proxy-dir"), None); + let mut client = TcpStream::connect(address)?; + client.write_all(b"proxy")?; + let (mut accepted, _) = listener.accept()?; + let mut bytes = [0; 5]; + accepted.read_exact(&mut bytes)?; + assert_eq!(bytes, *b"proxy"); + Ok(()) } -#[test] -fn cleanup_stale_proxy_socket_dirs_removes_dead_pid_directories() { - let root = tempfile::tempdir().expect("tempdir should create"); - let dead_dir = root - .path() - .join(format!("{PROXY_SOCKET_DIR_PREFIX}{}-0", u32::MAX)); - std::fs::create_dir(&dead_dir).expect("dead dir should create"); - - let alive_dir = root - .path() - .join(format!("{PROXY_SOCKET_DIR_PREFIX}{}-1", std::process::id())); - std::fs::create_dir(&alive_dir).expect("alive dir should create"); - - let unrelated_dir = root.path().join("unrelated-proxy-dir"); - std::fs::create_dir(&unrelated_dir).expect("unrelated dir should create"); - - cleanup_stale_proxy_socket_dirs_in(root.path()).expect("stale cleanup should succeed"); - - assert_eq!(dead_dir.exists(), false); - assert_eq!(alive_dir.exists(), true); - assert_eq!(unrelated_dir.exists(), true); -} - -#[test] -fn missing_procfs_keeps_live_process_alive() { - let root = tempfile::tempdir().expect("tempdir should create"); - let missing_proc_root = root.path().join("missing-proc"); - let pid = libc::pid_t::try_from(std::process::id()).expect("current pid should fit"); - - assert_eq!(is_pid_alive_in_proc_root(pid, &missing_proc_root), true); -} - -#[test] -fn zombie_process_is_not_alive() { - let mut child = Command::new("sh") - .arg("-c") - .arg("exit 0") - .spawn() - .expect("child should spawn"); - let pid = libc::pid_t::try_from(child.id()).expect("child pid should fit"); - let deadline = Instant::now() + Duration::from_secs(5); - - while is_pid_alive_raw(pid) && Instant::now() < deadline { - std::thread::sleep(Duration::from_millis(10)); +#[test_case(1; "non_listener")] +#[test_case(2; "extra_descriptors")] +// Eight descriptors exceed the receive buffer even with alignment headroom. +#[test_case(8; "truncated_control_message")] +fn malformed_handoff_closes_all_received_descriptors(count: usize) -> io::Result<()> { + let (sender, receiver) = UnixStream::pair()?; + let mut peers = Vec::new(); + let mut passed = Vec::new(); + for _ in 0..count { + let (peer, descriptor) = UnixStream::pair()?; + peer.set_read_timeout(Some(Duration::from_secs(/*secs*/ 1)))?; + peers.push(peer); + passed.push(descriptor); } - let zombie_was_recognized = !is_pid_alive_raw(pid); - child.wait().expect("child should be reaped"); + let descriptors = passed.iter().map(AsFd::as_fd).collect::>(); + send_descriptors(&sender, LISTENER_MESSAGE, &descriptors)?; + drop(passed); + assert!(receive_listener(&receiver).is_err()); + for mut peer in peers { + assert_eq!(peer.read(&mut [0])?, 0); + } + Ok(()) +} - assert_eq!(zombie_was_recognized, true); +#[test] +fn rejects_missing_descriptors_and_invalid_payloads() -> io::Result<()> { + let (mut sender, receiver) = UnixStream::pair()?; + sender.write_all(&[LISTENER_MESSAGE])?; + assert!(receive_listener(&receiver).is_err()); + + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0))?; + send_descriptors(&sender, /*payload*/ 0, &[listener.as_fd()])?; + assert!(receive_listener(&receiver).is_err()); + Ok(()) +} + +#[test_case(2; "extra_descriptors")] +#[test_case(8; "truncated_control_message")] +fn rejects_extra_descriptors_even_when_first_is_a_listener(count: usize) -> io::Result<()> { + let (sender, receiver) = UnixStream::pair()?; + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0))?; + send_descriptors(&sender, LISTENER_MESSAGE, &vec![listener.as_fd(); count])?; + assert!(receive_listener(&receiver).is_err()); + Ok(()) +} + +#[test] +fn rejects_listener_bound_to_all_interfaces() -> io::Result<()> { + let (sender, receiver) = UnixStream::pair()?; + let listener = TcpListener::bind((Ipv4Addr::UNSPECIFIED, 0))?; + send_listener(&sender, &listener)?; + assert!(receive_listener(&receiver).is_err()); + Ok(()) +} + +fn send_descriptors( + channel: &UnixStream, + payload: u8, + descriptors: &[BorrowedFd<'_>], +) -> io::Result<()> { + let message = SendAncillaryMessage::ScmRights(descriptors); + let mut space = vec![MaybeUninit::uninit(); message.size()]; + let mut control = SendAncillaryBuffer::new(&mut space); + assert!(control.push(message)); + let sent = rustix::io::retry_on_intr(|| { + rustix::net::sendmsg( + channel, + &[IoSlice::new(&[payload])], + &mut control, + SendFlags::NOSIGNAL, + ) + })?; + assert_eq!(sent, 1); + Ok(()) } diff --git a/codex-rs/linux-sandbox/src/proxy_routing.rs b/codex-rs/linux-sandbox/src/proxy_routing.rs index 241c5b410d..a008299da5 100644 --- a/codex-rs/linux-sandbox/src/proxy_routing.rs +++ b/codex-rs/linux-sandbox/src/proxy_routing.rs @@ -1,19 +1,15 @@ -use crate::proxy_lifecycle::PROXY_SOCKET_DIR_PREFIX; -use crate::proxy_lifecycle::cleanup_stale_proxy_socket_dirs_in; use crate::proxy_lifecycle::close_fd; -use crate::proxy_lifecycle::create_ready_pipe; use crate::proxy_lifecycle::harden_bridge_process; -use crate::proxy_lifecycle::spawn_proxy_socket_dir_cleanup_worker; +use crate::proxy_lifecycle::move_fd_above_stdio; +use crate::proxy_lifecycle::receive_listener; +use crate::proxy_lifecycle::send_listener; use codex_network_proxy::PROXY_ATTRIBUTION_TOKEN_ENV_KEY; use codex_network_proxy::write_attribution_frame; -use codex_utils_absolute_path::AbsolutePathBuf; use serde::Deserialize; use serde::Serialize; use std::collections::BTreeMap; use std::collections::HashMap; -use std::fs::DirBuilder; use std::fs::File; -use std::fs::Permissions; use std::io; use std::io::Read; use std::io::Write; @@ -23,14 +19,11 @@ use std::net::Shutdown; use std::net::SocketAddr; use std::net::TcpListener; use std::net::TcpStream; +use std::os::fd::AsRawFd; use std::os::fd::FromRawFd; -use std::os::unix::ffi::OsStrExt; -use std::os::unix::fs::DirBuilderExt; -use std::os::unix::fs::PermissionsExt; -use std::os::unix::net::UnixListener; +use std::os::fd::OwnedFd; use std::os::unix::net::UnixStream; -use std::path::Path; -use std::path::PathBuf; +use std::time::Duration; use url::Url; const PROXY_ENV_KEYS: &[&str] = &[ @@ -54,8 +47,7 @@ const PROXY_ENV_KEYS: &[&str] = &[ const HOST_BRIDGE_READY: u8 = 1; const LOOPBACK_INTERFACE_NAME: &[u8] = b"lo"; -// Linux sockaddr_un.sun_path allows 108 bytes, including the trailing NUL. -const UNIX_SOCKET_PATH_MAX_BYTES: usize = 107; +const HANDOFF_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 30); #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub(crate) struct ProxyRouteSpec { @@ -65,7 +57,7 @@ pub(crate) struct ProxyRouteSpec { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] struct ProxyRouteEntry { env_key: String, - uds_path: PathBuf, + control_fd: libc::c_int, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -80,7 +72,7 @@ struct ProxyRoutePlan { has_proxy_config: bool, } -pub(crate) fn prepare_host_proxy_route_spec() -> io::Result<(String, AbsolutePathBuf)> { +pub(crate) fn prepare_host_proxy_route_spec() -> io::Result<(String, Vec)> { let (attribution_token, plan) = extract_attribution_token_and_plan(std::env::vars().collect()); // SAFETY: the sandbox helper is single-threaded here, before it forks bridge workers or // executes the user command. @@ -97,48 +89,35 @@ pub(crate) fn prepare_host_proxy_route_spec() -> io::Result<(String, AbsolutePat return Err(io::Error::new(io::ErrorKind::InvalidInput, message)); } - let socket_parent_dir = proxy_socket_parent_dir(); - let _ = cleanup_stale_proxy_socket_dirs_in(socket_parent_dir.as_path()); - - let socket_dir = create_proxy_socket_dir()?; - let readable_socket_dir = AbsolutePathBuf::relative_to_current_dir(&socket_dir)?; - let mut socket_by_endpoint: BTreeMap = BTreeMap::new(); - let mut next_index = 0usize; + let mut control_by_endpoint = BTreeMap::new(); for route in &plan.routes { - if socket_by_endpoint.contains_key(&route.endpoint) { + if control_by_endpoint.contains_key(&route.endpoint) { continue; } - let socket_path = socket_dir.join(format!("proxy-route-{next_index}.sock")); - next_index += 1; - socket_by_endpoint.insert(route.endpoint, socket_path); - } - - let mut host_bridge_pids = Vec::with_capacity(socket_by_endpoint.len()); - for (endpoint, socket_path) in &socket_by_endpoint { - host_bridge_pids.push(spawn_host_bridge( - *endpoint, - socket_path, + let control = spawn_host_bridge( + route.endpoint, attribution_token.as_deref(), - )?); + &mut control_by_endpoint, + )?; + control_by_endpoint.insert(route.endpoint, control); } - spawn_proxy_socket_dir_cleanup_worker(socket_dir, host_bridge_pids)?; let mut routes = Vec::with_capacity(plan.routes.len()); for route in plan.routes { - let Some(uds_path) = socket_by_endpoint.get(&route.endpoint) else { + let Some(control) = control_by_endpoint.get(&route.endpoint) else { return Err(io::Error::other(format!( - "missing UDS path for endpoint {}", + "missing bootstrap channel for endpoint {}", route.endpoint ))); }; routes.push(ProxyRouteEntry { env_key: route.env_key, - uds_path: uds_path.clone(), + control_fd: control.as_raw_fd(), }); } let spec = serde_json::to_string(&ProxyRouteSpec { routes }).map_err(io::Error::other)?; - Ok((spec, readable_socket_dir)) + Ok((spec, control_by_endpoint.into_values().collect())) } fn extract_attribution_token_and_plan( @@ -159,21 +138,50 @@ pub(crate) fn activate_proxy_routes_in_netns(serialized_spec: &str) -> io::Resul )); } - let mut local_port_by_uds_path: BTreeMap = BTreeMap::new(); + let mut local_port_by_control = BTreeMap::new(); for route in &spec.routes { - if local_port_by_uds_path.contains_key(&route.uds_path) { + if local_port_by_control.contains_key(&route.control_fd) { continue; } - let local_port = spawn_local_bridge(route.uds_path.as_path())?; - local_port_by_uds_path.insert(route.uds_path.clone(), local_port); + if route.control_fd <= libc::STDERR_FILENO { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "proxy bootstrap channel must not use standard descriptors", + )); + } + // SAFETY: F_GETFD takes only a descriptor, with no pointer arguments; + // an invalid inherited descriptor returns EBADF. + let flags = unsafe { libc::fcntl(route.control_fd, libc::F_GETFD) }; + if flags < 0 { + return Err(io::Error::last_os_error()); + } + // SAFETY: the outer helper left this descriptor open across exec, which + // discarded its Rust owner. F_GETFD above confirmed it is still valid. + // The map ensures each descriptor is claimed once, even for proxy aliases. + let mut control = unsafe { UnixStream::from_raw_fd(route.control_fd) }; + // SAFETY: F_SETFD takes integer flags, not a pointer; `control` owns the + // live descriptor throughout the call. + if unsafe { libc::fcntl(control.as_raw_fd(), libc::F_SETFD, flags | libc::FD_CLOEXEC) } < 0 + { + return Err(io::Error::last_os_error()); + } + control.set_read_timeout(Some(HANDOFF_TIMEOUT))?; + let listener = bind_local_loopback_listener()?; + let local_port = listener.local_addr()?.port(); + send_listener(&control, &listener)?; + let mut ready = [0_u8; 1]; + control.read_exact(&mut ready)?; + if ready != [HOST_BRIDGE_READY] { + return Err(io::Error::other("host bridge did not accept its listener")); + } + // Loop scope closes both descriptors, leaving only the host's listener. + // Neither can reach the namespace reaper or the untrusted command. + local_port_by_control.insert(route.control_fd, local_port); } for route in spec.routes { - let Some(local_port) = local_port_by_uds_path.get(&route.uds_path) else { - return Err(io::Error::other(format!( - "missing local bridge port for UDS path {}", - route.uds_path.display() - ))); + let Some(local_port) = local_port_by_control.get(&route.control_fd) else { + return Err(io::Error::other("missing proxy listener port")); }; let original_value = std::env::var(&route.env_key).map_err(|_| { io::Error::new( @@ -306,129 +314,62 @@ fn rewrite_proxy_env_value(proxy_url: &str, local_port: u16) -> Option { Some(rewritten) } -fn create_proxy_socket_dir() -> io::Result { - let temp_dir = proxy_socket_parent_dir(); - let pid = std::process::id(); - let uid = unsafe { libc::geteuid() }; - for attempt in 0..128 { - let candidate = temp_dir.join(format!("{PROXY_SOCKET_DIR_PREFIX}{pid}-{uid}-{attempt}")); - // The bridge UDS paths live under a shared temp root, so the per-run - // directory should not be traversable by other processes. - let mut dir_builder = DirBuilder::new(); - dir_builder.mode(0o700); - match dir_builder.create(&candidate) { - Ok(()) => return Ok(candidate), - Err(err) if err.kind() == io::ErrorKind::AlreadyExists => continue, - Err(err) => return Err(err), - } - } - Err(io::Error::new( - io::ErrorKind::AlreadyExists, - format!( - "failed to allocate proxy routing temp dir under {}", - temp_dir.display() - ), - )) -} - -fn proxy_socket_parent_dir() -> PathBuf { - if let Some(codex_home) = std::env::var_os("CODEX_HOME") { - let candidate = PathBuf::from(codex_home).join("tmp"); - if proxy_socket_paths_fit(candidate.as_path()) - && ensure_private_proxy_socket_parent_dir(candidate.as_path()).is_ok() - { - return candidate; - } - } - let temp_dir = std::env::temp_dir(); - if proxy_socket_paths_fit(temp_dir.as_path()) { - temp_dir - } else { - PathBuf::from("/tmp") - } -} - -fn proxy_socket_paths_fit(parent: &Path) -> bool { - let socket_path = parent - .join(format!( - "{PROXY_SOCKET_DIR_PREFIX}{}-{}-127", - u32::MAX, - libc::uid_t::MAX - )) - .join(format!("proxy-route-{}.sock", usize::MAX)); - socket_path.as_os_str().as_bytes().len() <= UNIX_SOCKET_PATH_MAX_BYTES -} - -fn ensure_private_proxy_socket_parent_dir(path: &Path) -> io::Result<()> { - let mut dir_builder = DirBuilder::new(); - dir_builder.recursive(true); - dir_builder.mode(0o700); - match dir_builder.create(path) { - Ok(()) => {} - Err(err) if err.kind() == io::ErrorKind::AlreadyExists => {} - Err(err) => return Err(err), - } - std::fs::set_permissions(path, Permissions::from_mode(0o700)) -} - fn spawn_host_bridge( endpoint: SocketAddr, - uds_path: &Path, attribution_token: Option<&str>, -) -> io::Result { - let (read_fd, write_fd) = create_ready_pipe()?; + inherited_controls: &mut BTreeMap, +) -> io::Result { + let (host_control, control) = UnixStream::pair()?; + let host_control = UnixStream::from(move_fd_above_stdio(host_control.into())?); + let mut control = UnixStream::from(move_fd_above_stdio(control.into())?); + control.set_read_timeout(Some(HANDOFF_TIMEOUT))?; let parent_pid = unsafe { libc::getpid() }; let pid = unsafe { libc::fork() }; if pid < 0 { - let err = io::Error::last_os_error(); - close_fd(read_fd)?; - close_fd(write_fd)?; - return Err(err); + return Err(io::Error::last_os_error()); } if pid == 0 { - if close_fd(read_fd).is_err() { - unsafe { libc::_exit(1) }; - } - let result = run_host_bridge(endpoint, uds_path, write_fd, attribution_token, parent_pid); + drop(control); + // A worker must not retain another route's inner bootstrap endpoint. + // Drop this child's inherited owners; the parent's copies are unaffected by fork. + inherited_controls.clear(); + let result = run_host_bridge(endpoint, host_control, attribution_token, parent_pid); if result.is_err() { unsafe { libc::_exit(1) }; } unsafe { libc::_exit(0) }; } - close_fd(write_fd)?; + drop(host_control); let mut ready = [0_u8; 1]; - let mut read_file = unsafe { File::from_raw_fd(read_fd) }; - read_file.read_exact(&mut ready)?; - if ready[0] != HOST_BRIDGE_READY { + control.read_exact(&mut ready)?; + if ready != [HOST_BRIDGE_READY] { return Err(io::Error::other( "host bridge did not acknowledge readiness", )); } - Ok(pid) + Ok(File::from(OwnedFd::from(control))) } fn run_host_bridge( endpoint: SocketAddr, - uds_path: &Path, - ready_fd: libc::c_int, + mut control: UnixStream, attribution_token: Option<&str>, parent_pid: libc::pid_t, ) -> io::Result<()> { harden_bridge_process(parent_pid)?; - if uds_path.exists() { - std::fs::remove_file(uds_path)?; - } - let listener = UnixListener::bind(uds_path)?; - - let mut ready_file = unsafe { File::from_raw_fd(ready_fd) }; - ready_file.write_all(&[HOST_BRIDGE_READY])?; - drop(ready_file); + control.write_all(&[HOST_BRIDGE_READY])?; + // The outer helper must launch bubblewrap before this listener can arrive. + // The socket keeps its isolated network namespace even though we accept in + // the host process; newly opened upstream sockets use our host namespace. + let listener = receive_listener(&control)?; + control.write_all(&[HOST_BRIDGE_READY])?; + drop(control); let attribution_token = attribution_token.map(str::to_owned); loop { - let (unix_stream, _) = listener.accept()?; + let (client_stream, _) = listener.accept()?; let attribution_token = attribution_token.clone(); std::thread::spawn(move || { let mut tcp_stream = match TcpStream::connect(endpoint) { @@ -442,63 +383,7 @@ fn run_host_bridge( // application bytes if this bridge cannot prove the exec attribution first. return; } - let _ = proxy_bidirectional(tcp_stream, unix_stream); - }); - } -} - -fn spawn_local_bridge(uds_path: &Path) -> io::Result { - let (read_fd, write_fd) = create_ready_pipe()?; - let parent_pid = unsafe { libc::getpid() }; - let pid = unsafe { libc::fork() }; - if pid < 0 { - let err = io::Error::last_os_error(); - close_fd(read_fd)?; - close_fd(write_fd)?; - return Err(err); - } - - if pid == 0 { - if close_fd(read_fd).is_err() { - unsafe { libc::_exit(1) }; - } - let result = run_local_bridge(uds_path, write_fd, parent_pid); - if result.is_err() { - unsafe { libc::_exit(1) }; - } - unsafe { libc::_exit(0) }; - } - - close_fd(write_fd)?; - let mut port_bytes = [0_u8; 2]; - let mut read_file = unsafe { File::from_raw_fd(read_fd) }; - read_file.read_exact(&mut port_bytes)?; - Ok(u16::from_be_bytes(port_bytes)) -} - -fn run_local_bridge( - uds_path: &Path, - ready_fd: libc::c_int, - parent_pid: libc::pid_t, -) -> io::Result<()> { - harden_bridge_process(parent_pid)?; - let listener = bind_local_loopback_listener()?; - let port = listener.local_addr()?.port(); - - let mut ready_file = unsafe { File::from_raw_fd(ready_fd) }; - ready_file.write_all(&port.to_be_bytes())?; - drop(ready_file); - - let uds_path = uds_path.to_path_buf(); - loop { - let (tcp_stream, _) = listener.accept()?; - let socket_path = uds_path.clone(); - std::thread::spawn(move || { - let unix_stream = match UnixStream::connect(socket_path) { - Ok(stream) => stream, - Err(_) => return, - }; - let _ = proxy_bidirectional(tcp_stream, unix_stream); + let _ = proxy_bidirectional(tcp_stream, client_stream); }); } } @@ -583,21 +468,21 @@ fn ensure_loopback_interface_up() -> io::Result<()> { close_fd(fd) } -fn proxy_bidirectional(mut tcp_stream: TcpStream, mut unix_stream: UnixStream) -> io::Result<()> { - let mut tcp_reader = tcp_stream.try_clone()?; - let mut unix_writer = unix_stream.try_clone()?; - let tcp_to_unix = std::thread::spawn(move || { - let result = std::io::copy(&mut tcp_reader, &mut unix_writer); - let _ = unix_writer.shutdown(Shutdown::Write); +fn proxy_bidirectional(mut upstream: TcpStream, mut client: TcpStream) -> io::Result<()> { + let mut upstream_reader = upstream.try_clone()?; + let mut client_writer = client.try_clone()?; + let upstream_to_client = std::thread::spawn(move || { + let result = std::io::copy(&mut upstream_reader, &mut client_writer); + let _ = client_writer.shutdown(Shutdown::Write); result }); - let unix_to_tcp = std::io::copy(&mut unix_stream, &mut tcp_stream); - let _ = tcp_stream.shutdown(Shutdown::Write); - let tcp_to_unix = tcp_to_unix + let client_to_upstream = std::io::copy(&mut client, &mut upstream); + let _ = upstream.shutdown(Shutdown::Write); + let upstream_to_client = upstream_to_client .join() .map_err(|_| io::Error::other("bridge thread panicked"))?; - tcp_to_unix?; - unix_to_tcp?; + upstream_to_client?; + client_to_upstream?; Ok(()) } @@ -611,12 +496,10 @@ mod tests { use super::is_proxy_env_key; use super::parse_loopback_proxy_endpoint; use super::plan_proxy_routes; - use super::proxy_socket_paths_fit; use super::rewrite_proxy_env_value; use pretty_assertions::assert_eq; use std::collections::HashMap; use std::net::SocketAddr; - use std::path::PathBuf; #[test] fn recognizes_proxy_env_keys_case_insensitively() { @@ -717,31 +600,19 @@ mod tests { assert_eq!(default_proxy_port("socks5h"), 1080); } - #[test] - fn proxy_socket_paths_enforce_linux_path_limit() { - assert_eq!( - proxy_socket_paths_fit(PathBuf::from("/tmp").as_path()), - true - ); - assert_eq!( - proxy_socket_paths_fit(PathBuf::from(format!("/tmp/{}", "a".repeat(96))).as_path()), - false - ); - } - #[test] fn proxy_route_spec_serialization_omits_proxy_urls() { let spec = ProxyRouteSpec { routes: vec![ProxyRouteEntry { env_key: "HTTP_PROXY".to_string(), - uds_path: PathBuf::from("/tmp/proxy-route-0.sock"), + control_fd: 3, }], }; let serialized = serde_json::to_string(&spec).expect("proxy route spec should serialize"); assert_eq!( serialized, - r#"{"routes":[{"env_key":"HTTP_PROXY","uds_path":"/tmp/proxy-route-0.sock"}]}"# + r#"{"routes":[{"env_key":"HTTP_PROXY","control_fd":3}]}"# ); } } diff --git a/codex-rs/linux-sandbox/tests/suite/managed_proxy.rs b/codex-rs/linux-sandbox/tests/suite/managed_proxy.rs index 2b03c9d020..0493245d98 100644 --- a/codex-rs/linux-sandbox/tests/suite/managed_proxy.rs +++ b/codex-rs/linux-sandbox/tests/suite/managed_proxy.rs @@ -1,7 +1,13 @@ +//! Linux managed-proxy integration tests exercise the real sandbox with isolated environments. +//! Handoff must preserve routing and attribution without leaking privileged sockets; cancellation +//! must close active routes while filesystem and network restrictions remain enforced. + #![cfg(target_os = "linux")] #![allow(clippy::unwrap_used)] use codex_core::exec_env::create_env; +use codex_network_proxy::PROXY_ATTRIBUTION_TOKEN_ENV_KEY; +use codex_network_proxy::write_attribution_frame; use codex_protocol::config_types::ShellEnvironmentPolicy; use codex_protocol::models::PermissionProfile; use codex_protocol::permissions::FileSystemAccessMode; @@ -16,17 +22,24 @@ use std::collections::HashMap; use std::io::Read; use std::io::Write; use std::net::Ipv4Addr; +use std::net::Shutdown; use std::net::TcpListener; +use std::net::TcpStream; use std::os::unix::fs::MetadataExt; use std::os::unix::fs::PermissionsExt; +use std::path::Path; use std::process::Output; use std::process::Stdio; use std::time::Duration; +use std::time::Instant; use tempfile::NamedTempFile; use tokio::process::Command; +use url::Url; const BWRAP_UNAVAILABLE_ERR: &str = "bubblewrap is unavailable: no system bwrap was found"; const NETWORK_TIMEOUT_MS: u64 = 4_000; +const OPERATION_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 8); +const SANDBOX_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 60); const MANAGED_PROXY_PERMISSION_ERR_SNIPPETS: &[&str] = &[ "loopback: Failed RTM_NEWADDR", "loopback: Failed RTM_NEWLINK", @@ -691,3 +704,269 @@ async fn managed_proxy_mode_denies_af_unix_socket_but_allows_socketpair() { String::from_utf8_lossy(&output.stderr) ); } + +/// Runs as the model-controlled command, after the trusted listener handoff. +#[test] +#[ignore = "invoked inside the managed proxy sandbox"] +fn handoff_client() { + let Ok(label) = std::env::var("CODEX_TEST_HANDOFF_LABEL") else { + return; + }; + assert!(std::env::var_os(PROXY_ATTRIBUTION_TOKEN_ENV_KEY).is_none()); + let inherited_sockets = std::fs::read_dir("/proc/self/fd") + .expect("enumerate inherited descriptors") + .filter_map(Result::ok) + .filter_map(|entry| { + let target = std::fs::read_link(entry.path()).ok()?; + target + .to_string_lossy() + .starts_with("socket:") + .then(|| (entry.file_name(), target)) + }) + .collect::>(); + assert_eq!(inherited_sockets, Vec::new(), "privileged socket leaked"); + + let shared = std::env::var("CODEX_TEST_HANDOFF_SHARED").expect("shared directory"); + let shared = Path::new(&shared); + let peer = std::env::var("CODEX_TEST_HANDOFF_PEER").expect("peer label"); + std::fs::write(shared.join(&label), b"ready").expect("announce sandbox readiness"); + let deadline = Instant::now() + OPERATION_TIMEOUT; + while !shared.join(&peer).exists() { + assert!(Instant::now() < deadline, "peer sandbox did not start"); + std::thread::sleep(Duration::from_millis(/*millis*/ 10)); + } + + let http = + Url::parse(&std::env::var("HTTP_PROXY").expect("HTTP proxy")).expect("parse HTTP proxy"); + let https = + Url::parse(&std::env::var("HTTPS_PROXY").expect("HTTPS proxy")).expect("parse HTTPS proxy"); + let other = + Url::parse(&std::env::var("ALL_PROXY").expect("other proxy")).expect("parse other proxy"); + for (route, proxy) in [("http", http), ("https", https), ("all", other)] { + let endpoint = ( + proxy.host_str().expect("proxy host"), + proxy.port().expect("proxy port"), + ); + let mut stream = TcpStream::connect(endpoint).expect("connect through handoff listener"); + stream + .set_read_timeout(Some(OPERATION_TIMEOUT)) + .expect("read timeout"); + stream + .set_write_timeout(Some(OPERATION_TIMEOUT)) + .expect("write timeout"); + let expected = format!("{label}/{route}"); + stream + .write_all(expected.as_bytes()) + .expect("send route marker"); + stream.shutdown(Shutdown::Write).expect("finish request"); + let mut response = String::new(); + stream + .read_to_string(&mut response) + .expect("read routed reply"); + assert_eq!(response, expected); + } + println!("handoff routes verified for {label}"); +} + +/// Both filesystem policies must preserve route identity across concurrent launches. +#[tokio::test] +async fn handoff_isolates_concurrent_endpoints_and_closes_privileged_descriptors() { + if let Some(reason) = managed_proxy_skip_reason().await { + eprintln!("skipping managed proxy test: {reason}"); + return; + } + + let shared = tempfile::tempdir_in("/tmp").expect("shared writable test directory"); + let test_executable = std::env::current_exe().expect("integration test executable"); + let test_executable = test_executable.to_str().expect("UTF-8 executable path"); + // Libtest selectors include the module path but omit the integration crate name. + let (_, test_module) = module_path!() + .split_once("::") + .expect("handoff tests live in an integration test module"); + let client_test = format!("{test_module}::handoff_client"); + let mut commands = Vec::new(); + let mut servers = Vec::new(); + for (label, peer, profile) in [ + ("first", "second", PermissionProfile::workspace_write()), + ("second", "first", PermissionProfile::Disabled), + ] { + let mut env = create_env_from_core_vars(); + strip_proxy_env(&mut env); + env.insert( + PROXY_ATTRIBUTION_TOKEN_ENV_KEY.to_string(), + format!("handoff-{label}"), + ); + env.insert("CODEX_TEST_HANDOFF_LABEL".to_string(), label.to_string()); + env.insert("CODEX_TEST_HANDOFF_PEER".to_string(), peer.to_string()); + env.insert( + "CODEX_TEST_HANDOFF_SHARED".to_string(), + shared.path().to_string_lossy().into_owned(), + ); + + for (routes, keys) in [ + (&["http", "https"][..], &["HTTP_PROXY", "HTTPS_PROXY"][..]), + (&["all"][..], &["ALL_PROXY"][..]), + ] { + let listener = + TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).expect("local proxy listener"); + let endpoint = listener.local_addr().expect("proxy endpoint"); + for key in keys { + env.insert((*key).to_string(), format!("http://{endpoint}")); + } + listener + .set_nonblocking(/*nonblocking*/ true) + .expect("bounded accept"); + servers.push(std::thread::spawn(move || { + for route in routes { + let deadline = Instant::now() + OPERATION_TIMEOUT; + let (mut stream, _) = loop { + match listener.accept() { + Ok(connection) => break connection, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + assert!(Instant::now() < deadline, "proxy request timed out"); + std::thread::sleep(Duration::from_millis(/*millis*/ 10)); + } + Err(error) => panic!("accept proxy request: {error}"), + } + }; + stream + .set_read_timeout(Some(OPERATION_TIMEOUT)) + .expect("read timeout"); + stream + .set_write_timeout(Some(OPERATION_TIMEOUT)) + .expect("write timeout"); + let marker = format!("{label}/{route}"); + let mut expected = Vec::new(); + write_attribution_frame(&mut expected, &format!("handoff-{label}")) + .expect("expected attribution frame"); + expected.extend_from_slice(marker.as_bytes()); + let mut request = Vec::new(); + stream + .read_to_end(&mut request) + .expect("read attributed marker"); + assert_eq!(request, expected); + stream + .write_all(marker.as_bytes()) + .expect("reply with route marker"); + } + })); + } + + let mut command = linux_sandbox_command( + &[ + test_executable, + "--exact", + &client_test, + "--ignored", + "--nocapture", + "--test-threads=1", + ], + &profile, + /*allow_network_for_proxy*/ true, + env, + ); + command.kill_on_drop(true); + commands.push(command); + } + + let mut second = commands.pop().expect("second sandbox command"); + let mut first = commands.pop().expect("first sandbox command"); + let outputs = tokio::time::timeout(SANDBOX_TIMEOUT, async { + tokio::join!(first.output(), second.output()) + }) + .await + .expect("concurrent sandboxes should finish and release output"); + for (label, output) in [("first", outputs.0), ("second", outputs.1)] { + let output = output.expect("run sandbox"); + assert_eq!( + output.status.success(), + true, + "{label} sandbox failed; stdout={}; stderr={}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + String::from_utf8_lossy(&output.stdout) + .contains(&format!("handoff routes verified for {label}")) + ); + } + for server in servers { + server.join().expect("proxy responder should succeed"); + } +} + +#[tokio::test] +async fn cancelling_sandbox_closes_active_proxy_connection() { + if let Some(reason) = managed_proxy_skip_reason().await { + eprintln!("skipping managed proxy test: {reason}"); + return; + } + + for profile in [ + PermissionProfile::workspace_write(), + PermissionProfile::Disabled, + ] { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).expect("local proxy listener"); + listener + .set_nonblocking(/*nonblocking*/ true) + .expect("bounded accept"); + let mut env = create_env_from_core_vars(); + strip_proxy_env(&mut env); + env.remove(PROXY_ATTRIBUTION_TOKEN_ENV_KEY); + env.insert( + "HTTP_PROXY".to_string(), + format!("http://{}", listener.local_addr().expect("proxy endpoint")), + ); + let mut command = linux_sandbox_command( + &[ + "bash", + "-c", + concat!( + "set -eu; proxy=${HTTP_PROXY#*://}; ", + "exec 3<>/dev/tcp/${proxy%:*}/${proxy##*:}; ", + "printf ready >&3; IFS= read -r response <&3", + ), + ], + &profile, + /*allow_network_for_proxy*/ true, + env, + ); + command.kill_on_drop(true); + let mut child = command.spawn().expect("launch sandbox"); + let (mut upstream, _) = tokio::time::timeout(OPERATION_TIMEOUT, async { + loop { + match listener.accept() { + Ok(connection) => break connection, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + tokio::time::sleep(Duration::from_millis(/*millis*/ 10)).await; + } + Err(error) => panic!("accept proxy connection: {error}"), + } + } + }) + .await + .expect("sandbox should establish its proxy connection"); + upstream + .set_read_timeout(Some(OPERATION_TIMEOUT)) + .expect("bounded read"); + let mut ready = [0; 5]; + upstream + .read_exact(&mut ready) + .expect("read client readiness"); + assert_eq!(ready, *b"ready"); + + child.start_kill().expect("cancel sandbox"); + let output = tokio::time::timeout(OPERATION_TIMEOUT, child.wait_with_output()) + .await + .expect("cancelled sandbox should release output") + .expect("wait for cancelled sandbox"); + assert_eq!(output.status.success(), false); + assert_eq!( + upstream + .read(&mut [0]) + .expect("host bridge should close its active connection"), + 0, + "cancelling the sandbox must stop its host bridge" + ); + } +}