fix(sandbox): accept local connections natively on loopback-confined sockets

crates/openshell-cli/Cargo.toml

@@ -88,7 +88,7 @@ tracing-subscriber = { workspace = true }
 workspace = true
 
 [target.'cfg(unix)'.dependencies]
-nix = { workspace = true }
+nix = { workspace = true, features = ["poll"] }
 
 [dev-dependencies]
 # Tests import the example profiles from providers/ the way an operator

crates/openshell-cli/src/run.rs

@@ -1868,7 +1868,7 @@ enum PipedStdin {
 /// never waits on a thread parked in `read(2)`. The thread exits at EOF, on a
 /// read error, or when the receiver is dropped.
 fn spawn_piped_stdin_reader(
-    mut reader: impl Read + Send + 'static,
+    mut reader: impl PipedInput,
 ) -> tokio::sync::mpsc::Receiver<std::io::Result<Vec<u8>>> {
     let (tx, rx) = tokio::sync::mpsc::channel::<std::io::Result<Vec<u8>>>(64);
     std::thread::spawn(move || {
@@ -1877,6 +1877,16 @@ fn spawn_piped_stdin_reader(
             match reader.read(&mut buf) {
                 Ok(0) => return,
                 Err(error) if error.kind() == ErrorKind::Interrupted => {}
+                // Processes that inherit the same stdin share its open file
+                // description, so another process may have made it
+                // nonblocking. Wait for input instead of failing.
+                #[cfg(unix)]
+                Err(error) if error.kind() == ErrorKind::WouldBlock => {
+                    if let Err(error) = wait_until_readable(&reader) {
+                        let _ = tx.blocking_send(Err(error));
+                        return;
+                    }
+                }
                 Err(error) => {
                     let _ = tx.blocking_send(Err(error));
                     return;
@@ -1892,6 +1902,30 @@ fn spawn_piped_stdin_reader(
     rx
 }
 
+/// Input the piped-stdin reader accepts. On Unix it must expose a descriptor
+/// so a nonblocking stream can be waited on.
+#[cfg(unix)]
+trait PipedInput: Read + std::os::fd::AsFd + Send + 'static {}
+#[cfg(unix)]
+impl<T: Read + std::os::fd::AsFd + Send + 'static> PipedInput for T {}
+#[cfg(not(unix))]
+trait PipedInput: Read + Send + 'static {}
+#[cfg(not(unix))]
+impl<T: Read + Send + 'static> PipedInput for T {}
+
+/// Block until `reader` has input or reaches end of file.
+#[cfg(unix)]
+fn wait_until_readable(reader: &impl std::os::fd::AsFd) -> std::io::Result<()> {
+    use nix::poll::{PollFd, PollFlags, PollTimeout, poll};
+    let mut fds = [PollFd::new(reader.as_fd(), PollFlags::POLLIN)];
+    loop {
+        match poll(&mut fds, PollTimeout::NONE) {
+            Err(nix::errno::Errno::EINTR) => {}
+            result => return result.map(drop).map_err(std::io::Error::from),
+        }
+    }
+}
+
 /// Collect piped stdin until EOF or until `grace` elapses, whichever comes
 /// first. Input beyond `limit` bytes is rejected with the upload hint.
 async fn collect_piped_stdin(
@@ -8585,6 +8619,42 @@ mod tests {
         );
     }
 
+    #[cfg(unix)]
+    #[test]
+    fn piped_stdin_left_nonblocking_by_another_process_still_streams() {
+        // Processes that inherit the same stdin share one open file
+        // description, so any of them can make it nonblocking for all. A
+        // read with no input yet then fails with EAGAIN instead of waiting.
+        use std::os::fd::AsRawFd as _;
+        let (reader, mut writer) = std::io::pipe().expect("pipe");
+        nix::fcntl::fcntl(
+            reader.as_raw_fd(),
+            nix::fcntl::FcntlArg::F_SETFL(nix::fcntl::OFlag::O_NONBLOCK),
+        )
+        .expect("make the shared pipe nonblocking");
+        let runtime = exec_stdin_runtime();
+        let collected = runtime.block_on(super::collect_piped_stdin(
+            super::spawn_piped_stdin_reader(reader),
+            Duration::from_millis(100),
+            super::MAX_EXEC_STDIN_BYTES,
+        ));
+        let super::PipedStdin::Open { prefix, mut rest } = collected.expect("collect") else {
+            panic!("an open pipe must start the command before EOF");
+        };
+        assert!(prefix.is_empty());
+        writer.write_all(b"late").unwrap();
+        drop(writer);
+        let next = runtime
+            .block_on(rest.recv())
+            .expect("late chunk")
+            .expect("read");
+        assert_eq!(next, b"late");
+        assert!(
+            runtime.block_on(rest.recv()).is_none(),
+            "EOF closes the channel"
+        );
+    }
+
     #[test]
     fn piped_stdin_over_the_limit_is_rejected_with_the_upload_hint() {
         let (reader, mut writer) = std::io::pipe().expect("pipe");

crates/openshell-isolation-interface/Cargo.toml

@@ -22,7 +22,8 @@ tokio = { workspace = true }
 libc = "0.2"
 
 [target.'cfg(target_os = "linux")'.dependencies]
-rustix = { workspace = true, features = ["fs", "process"] }
+rustix = { workspace = true, features = ["fs", "net", "process"] }
+socket2 = { workspace = true, features = ["all"] }
 
 [dev-dependencies]
 tokio = { workspace = true }

crates/openshell-isolation-interface/src/linux/child_seccomp.rs

@@ -29,6 +29,7 @@ const SECCOMP_DATA_ARGS_OFFSET: u32 = 16;
 const X32_SYSCALL_BIT: u32 = 0x4000_0000;
 
 const CLOSE_RANGE_UNSHARE_FLAG: u32 = 1 << 1;
+const CLOSE_RANGE_CLOEXEC_FLAG: u32 = 1 << 2;
 const F_SETOWN_COMMAND: u32 = 8;
 const F_SETSIG_COMMAND: u32 = 10;
 const F_SETOWN_EX_COMMAND: u32 = 15;
@@ -53,6 +54,7 @@ impl ChildHardeningProgram {
     /// The caller must invoke this from the post-fork child after all
     /// sandbox-wide TSYNC work and the launcher's `NEW_LISTENER` filter.
     pub fn install(&mut self) -> io::Result<()> {
+        mark_inherited_descriptors_close_on_exec()?;
         set_no_new_privileges()?;
         let len = u16::try_from(self.instructions.len()).map_err(|_| {
             io::Error::new(
@@ -88,13 +90,43 @@ impl ChildHardeningProgram {
     }
 }
 
+/// Mark every descriptor above stdio close-on-exec in the post-fork child.
+///
+/// Workloads receive INET sockets only through broker injection, which binds
+/// them to loopback first. A descriptor the sandbox process inherited from its
+/// container runtime, or opened without `O_CLOEXEC`, must never cross `exec`
+/// as an unconfined socket. The command's stdio is already installed on 0-2
+/// when `pre_exec` hooks run. This is a single async-signal-safe syscall.
+///
+/// # Errors
+///
+/// Returns the kernel error; kernels without `CLOSE_RANGE_CLOEXEC` (before
+/// Linux 5.11) fail closed.
+pub fn mark_inherited_descriptors_close_on_exec() -> io::Result<()> {
+    // SAFETY: close_range takes scalar arguments and only sets FD_CLOEXEC.
+    let result = unsafe {
+        libc::syscall(
+            libc::SYS_close_range,
+            3_u32,
+            u32::MAX,
+            CLOSE_RANGE_CLOEXEC_FLAG,
+        )
+    };
+    if result < 0 {
+        Err(io::Error::last_os_error())
+    } else {
+        Ok(())
+    }
+}
+
 /// Build the same-UID workload self-protection program before `fork`.
 ///
 /// `sandbox_tgid` is the sandbox PID as visible from its workload namespace.
 /// The filter blocks thread-targeting operations that name the trusted sandbox
 /// leader and blocks process-directed operations with the same target. The
-/// ordinary workload listener additionally mediates `kill`, `tkill`, and
-/// `rt_sigqueueinfo`: Linux accepts nonleader TIDs for these operations, so a
+/// ordinary workload listener additionally mediates `kill`, `tkill`,
+/// `rt_sigqueueinfo`, and `SIGCONT` sent with `tgkill` or
+/// `rt_tgsigqueueinfo`: Linux accepts nonleader TIDs for these operations, so a
 /// static TGID comparison alone cannot protect future sandbox worker threads.
 pub fn prepare(sandbox_tgid: u32) -> io::Result<ChildHardeningProgram> {
     if sandbox_tgid == 0 {
@@ -352,6 +384,48 @@ fn set_no_new_privileges() -> io::Result<()> {
 mod tests {
     use super::*;
 
+    #[test]
+    fn inherited_sockets_are_marked_close_on_exec_but_stdio_is_not() {
+        // The sweep changes every descriptor in the calling process, so run
+        // it in a fresh copy of this test binary rather than the harness.
+        const CHILD_MARKER: &str = "OPENSHELL_CLOEXEC_SWEEP_CHILD";
+        if std::env::var_os(CHILD_MARKER).is_some() {
+            let socket = rustix::net::socket(
+                rustix::net::AddressFamily::INET,
+                rustix::net::SocketType::STREAM,
+                None,
+            )
+            .expect("inheritable socket");
+            assert!(
+                !rustix::io::fcntl_getfd(&socket)
+                    .unwrap()
+                    .contains(rustix::io::FdFlags::CLOEXEC)
+            );
+            mark_inherited_descriptors_close_on_exec().expect("sweep descriptors");
+            assert!(
+                rustix::io::fcntl_getfd(&socket)
+                    .unwrap()
+                    .contains(rustix::io::FdFlags::CLOEXEC)
+            );
+            assert!(
+                !rustix::io::fcntl_getfd(io::stderr())
+                    .unwrap()
+                    .contains(rustix::io::FdFlags::CLOEXEC)
+            );
+            return;
+        }
+        let status = std::process::Command::new(std::env::current_exe().unwrap())
+            .args([
+                "--exact",
+                "linux::child_seccomp::tests::inherited_sockets_are_marked_close_on_exec_but_stdio_is_not",
+                "--nocapture",
+            ])
+            .env(CHILD_MARKER, "1")
+            .status()
+            .expect("run isolated sweep test");
+        assert!(status.success(), "isolated sweep test failed");
+    }
+
     #[test]
     fn rejects_zero_sandbox_tgid() {
         assert_eq!(

crates/openshell-isolation-interface/src/linux/mod.rs

@@ -11,6 +11,7 @@ pub mod landlock;
 pub mod proc_fd;
 pub mod process_signal;
 pub mod seccomp_notify;
+pub mod socket_confinement;
 pub mod socket_registry;
 pub mod task_memory;
 pub mod workload_launcher;

crates/openshell-isolation-interface/src/linux/process_signal.rs

@@ -12,6 +12,7 @@
 
 use std::io;
 use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
+use std::sync::atomic::{AtomicBool, Ordering};
 
 use crate::linux::seccomp_notify::{Notification, NotificationListener};
 use crate::linux::task_memory;
@@ -27,6 +28,7 @@ pub fn mediate_process_signal(
     listener: &NotificationListener,
     notification: Notification,
     sandbox_tgid: u32,
+    workload_frozen: &AtomicBool,
 ) -> io::Result<()> {
     listener.validate_id(notification.id)?;
     let target = scalar_int(notification.args[0]);
@@ -38,6 +40,7 @@ pub fn mediate_process_signal(
             libc::EINVAL
         }));
     }
+    refuse_resume_while_frozen(signal, workload_frozen)?;
     let target = u32::try_from(target).map_err(|_| io::Error::from_raw_os_error(libc::ESRCH))?;
     let retained = retain_signal_target(target, sandbox_tgid)?;
     // SAFETY: all-zero siginfo consists of valid integer/pointer fields. A
@@ -64,6 +67,7 @@ pub fn mediate_process_signal(
         _ => return Err(io::Error::from_raw_os_error(libc::ENOSYS)),
     };
     listener.validate_id(notification.id)?;
+    refuse_resume_while_frozen(signal, workload_frozen)?;
     // SAFETY: retained owns a live pidfd; info is null or a complete trusted
     // copy. The kernel targets that process object, never a reused numeric PID.
     let result = unsafe {
@@ -81,37 +85,62 @@ pub fn mediate_process_signal(
     listener.respond_value(notification.id, 0)
 }
 
-/// Continue a positive-target `tkill` only when the target thread belongs to
-/// an untrusted workload process rather than the sandbox runtime itself.
+/// Continue a thread-directed signal (`tkill`, `tgkill`,
+/// `rt_tgsigqueueinfo`) aimed at an untrusted workload thread.
 ///
 /// Continuing preserves Linux's thread-directed signal semantics, including
-/// the cancellation signal used by musl. A target that exits between the
-/// ownership check and continuation can only be reused inside the same PID
-/// namespace; the static child filter still rejects the sandbox leader.
+/// the cancellation signal used by musl. The static child filter rejects the
+/// sandbox leader as a `tgkill`/`rt_tgsigqueueinfo` group, and the kernel
+/// rejects a thread outside the named group, so those two are notified only
+/// for `SIGCONT`. A `tkill` names a bare thread, so its group is resolved here;
+/// a target reused between this check and continuation stays inside the same
+/// PID namespace.
 pub fn mediate_thread_signal(
     listener: &NotificationListener,
     notification: Notification,
     sandbox_tgid: u32,
+    workload_frozen: &AtomicBool,
 ) -> io::Result<()> {
     listener.validate_id(notification.id)?;
-    let target = scalar_int(notification.args[0]);
-    let signal = scalar_int(notification.args[1]);
-    if target <= 0 || !(0..=64).contains(&signal) {
-        return Err(io::Error::from_raw_os_error(if target <= 0 {
-            libc::EPERM
-        } else {
-            libc::EINVAL
-        }));
-    }
-    let target = u32::try_from(target).map_err(|_| io::Error::from_raw_os_error(libc::ESRCH))?;
-    let target_group = thread_group_id(target)?;
-    if target_group == sandbox_tgid || target_group == 0 {
-        return Err(io::Error::from_raw_os_error(libc::EPERM));
+    let signal = match i64::from(notification.syscall) {
+        libc::SYS_tkill => {
+            let target = scalar_int(notification.args[0]);
+            let signal = scalar_int(notification.args[1]);
+            if target <= 0 {
+                return Err(io::Error::from_raw_os_error(libc::EPERM));
+            }
+            let target =
+                u32::try_from(target).map_err(|_| io::Error::from_raw_os_error(libc::ESRCH))?;
+            let target_group = thread_group_id(target)?;
+            if target_group == sandbox_tgid || target_group == 0 {
+                return Err(io::Error::from_raw_os_error(libc::EPERM));
+            }
+            signal
+        }
+        libc::SYS_tgkill | libc::SYS_rt_tgsigqueueinfo => scalar_int(notification.args[2]),
+        _ => return Err(io::Error::from_raw_os_error(libc::ENOSYS)),
+    };
+    if !(0..=64).contains(&signal) {
+        return Err(io::Error::from_raw_os_error(libc::EINVAL));
     }
     listener.validate_id(notification.id)?;
+    refuse_resume_while_frozen(signal, workload_frozen)?;
     listener.respond_continue(notification.id)
 }
 
+/// While the boundary has stopped the workload for supervisor recovery, a
+/// workload process that was not yet stopped must not resume the others.
+///
+/// The flag is read immediately before delivery. The freezer does not wait
+/// for in-flight notifications, so a signal already past this check when the
+/// freeze begins can still be delivered.
+fn refuse_resume_while_frozen(signal: i32, workload_frozen: &AtomicBool) -> io::Result<()> {
+    if signal == libc::SIGCONT && workload_frozen.load(Ordering::Acquire) {
+        return Err(io::Error::from_raw_os_error(libc::EPERM));
+    }
+    Ok(())
+}
+
 fn scalar_int(value: u64) -> i32 {
     let bytes = value.to_ne_bytes();
     #[cfg(target_endian = "little")]
@@ -176,8 +205,13 @@ mod tests {
             .recv_timeout(std::time::Duration::from_secs(5))
             .unwrap();
         let notification = listener.receive().unwrap();
-        let error =
-            mediate_process_signal(&listener, notification, std::process::id()).unwrap_err();
+        let error = mediate_process_signal(
+            &listener,
+            notification,
+            std::process::id(),
+            &AtomicBool::new(false),
+        )
+        .unwrap_err();
         assert_eq!(error.raw_os_error(), Some(libc::EPERM));
         listener
             .respond_errno(notification.id, libc::EPERM)