From f13d2464384be5eb5f6522372220b506d72b401a Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 24 Oct 2026 11:20:40 -0600 Subject: [PATCH 1/6] REVERT ME: Work around ctfs refusing O_CLOEXEC opens Before illumos 18395[1], files in ctfs fail open(2) with `EINVAL` when `O_CLOEXEC` is set, and Rust's `std::fs` always sets it. So instead we open the contract files with `rustix` and set `FD_CLOEXEC` after. To be reverted once we're up to date with [2]. [1] https://www.illumos.org/issues/18395 [2] https://github.com/oxidecomputer/illumos-gate/commit/460f7917fa4f Co-Authored-By: Claude --- server/src/contract.rs | 32 +++++++++++++++++--------------- 1 file changed, 17 insertions(+), 15 deletions(-) diff --git a/server/src/contract.rs b/server/src/contract.rs index 0bade71c..4fe57326 100644 --- a/server/src/contract.rs +++ b/server/src/contract.rs @@ -14,11 +14,13 @@ //! login session, in OpenSSH's `openbsd-compat/port-solaris.c`. use std::ffi::{CString, c_char, c_int, c_uint}; -use std::fs::{File, OpenOptions}; +use std::fs::File; use std::io; use std::os::fd::AsRawFd as _; use libc::ctid_t; +use rustix::fs::{Mode, OFlags}; +use rustix::io::{Errno, FdFlags, fcntl_setfd}; use sush_common::jobs::JobId; @@ -61,11 +63,17 @@ fn check(what: &str, errno: c_int) -> io::Result<()> { } } -/// Open a ctfs file, naming it in any error. -fn open(options: &OpenOptions, path: String) -> io::Result { - options - .open(&path) - .map_err(|error| io::Error::new(error.kind(), format!("opening {path}: {error}"))) +/// Open a ctfs file, naming it in any error. Until illumos 18395, ctfs +/// refuses opens with `O_CLOEXEC`, which `std::fs` always sets, so we +/// open without it and set `FD_CLOEXEC` afterward. +fn open(flags: OFlags, path: String) -> io::Result { + let named = |error: Errno| { + let error = io::Error::from(error); + io::Error::new(error.kind(), format!("opening {path}: {error}")) + }; + let fd = rustix::fs::open(&path, flags, Mode::empty()).map_err(named)?; + fcntl_setfd(&fd, FdFlags::CLOEXEC).map_err(named)?; + Ok(File::from(fd)) } /// Run `spawn` with a job's contract template active, then abandon the @@ -92,10 +100,7 @@ impl Template { /// As in sshd, a hardware error kills only the process group of the /// process it struck. The contract is labeled by job ID. fn activate(job_id: &JobId) -> io::Result { - let file = open( - OpenOptions::new().read(true).write(true), - format!("{PROCESS}/template"), - )?; + let file = open(OFlags::RDWR, format!("{PROCESS}/template"))?; let fd = file.as_raw_fd(); let aux = CString::new(format!("sush job {job_id}")).map_err(io::Error::other)?; unsafe { @@ -118,7 +123,7 @@ impl Template { /// Abandon the contract created by the last `fork` on this thread, /// and return its ID. fn abandon_latest(self) -> io::Result { - let latest = open(OpenOptions::new().read(true), format!("{PROCESS}/latest"))?; + let latest = open(OFlags::RDONLY, format!("{PROCESS}/latest"))?; let ctid = unsafe { let mut status: ct_stathdl_t = std::ptr::null_mut(); check( @@ -129,10 +134,7 @@ impl Template { ct_status_free(status); ctid }; - let ctl = open( - OpenOptions::new().write(true), - format!("{PROCESS}/{ctid}/ctl"), - )?; + let ctl = open(OFlags::WRONLY, format!("{PROCESS}/{ctid}/ctl"))?; check("abandoning contract", unsafe { ct_ctl_abandon(ctl.as_raw_fd()) })?; From 2a6b60824b6a5ef1195a9351933d06e316971447 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 24 Oct 2026 14:51:09 -0600 Subject: [PATCH 2/6] Spawn jobs with posix_spawn instead of fork Forking sled-agent to start a job copies its large address space and stops all of its threads. On illumos, posix_spawn builds the child in the kernel instead, and does neither. posix_spawn runs none of our code in the child, so a small bash shim sets the job's limits before it execs the job. JobLimits::clamp now also lowers each limit to our own hard limit, which a job cannot exceed. JobChild replaces tokio's Child, and reaps the job when SIGCHLD arrives. Errors from libcontract and posix_spawn now go through one check_errno, which names the call that failed. Co-Authored-By: Claude --- Cargo.toml | 2 +- common/src/jobs.rs | 48 +++--- server/src/contract.rs | 45 +++--- server/src/error.rs | 18 +++ server/src/executor.rs | 251 +++++++++++------------------ server/src/io.rs | 39 +++-- server/src/job.rs | 51 +++--- server/src/lib.rs | 1 + server/src/spawn.rs | 315 +++++++++++++++++++++++++++++++++++++ tests/src/manager_tests.rs | 111 +++++++++++++ 10 files changed, 638 insertions(+), 243 deletions(-) create mode 100644 server/src/spawn.rs diff --git a/Cargo.toml b/Cargo.toml index a1c4f945..1963ac4d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -68,7 +68,7 @@ sprockets-tls-test-utils = { git = "https://github.com/oxidecomputer/sprockets.g ssh-key = { version = "0.6", features = ["ed25519", "p256", "serde"] } tempfile = "3" thiserror = "1" -tokio = { version = "1", features = ["macros", "process", "rt", "sync", "time"] } +tokio = { version = "1", features = ["macros", "process", "rt", "signal", "sync", "time"] } tokio-rustls = "0.26" tokio-stream = "0.1" tokio-fd = "0.3" diff --git a/common/src/jobs.rs b/common/src/jobs.rs index fb308bb1..a8587506 100644 --- a/common/src/jobs.rs +++ b/common/src/jobs.rs @@ -740,37 +740,24 @@ pub struct JobLimits { } impl JobLimits { - /// Clamp each limit to the default ceiling. Limits arrive as - /// unsigned request parameters, which may narrow the server's - /// limits but never widen them. - pub fn clamp(self) -> Self { + /// Clamp each limit to the default ceiling and to this process's + /// hard limit (since a job inherits these). Limits are unsigned + /// request parameters, so they may only narrow the server's limits. + pub fn clamp(self) -> Result { let ceiling = Self::default(); - Self { - max_cpu: self.max_cpu.min(ceiling.max_cpu), - max_mem: self.max_mem.min(ceiling.max_mem), - max_fsize: self.max_fsize.min(ceiling.max_fsize), - } - } - - pub fn apply(&self) -> Result<(), IoError> { - let Self { - max_cpu, - max_mem, - max_fsize, - } = *self; - self.set_limit(Resource::CPU, max_cpu)?; - self.set_limit(Resource::AS, max_mem)?; - self.set_limit(Resource::FSIZE, max_fsize)?; - Ok(()) - } - - fn set_limit(&self, resource: Resource, value: u64) -> Result<(), IoError> { - let (_soft, hard) = resource.get()?; - resource.set(value.min(hard), hard) + let hard = |resource: Resource| resource.get().map(|(_soft, hard)| hard); + Ok(Self { + max_cpu: self.max_cpu.min(ceiling.max_cpu).min(hard(Resource::CPU)?), + max_mem: self.max_mem.min(ceiling.max_mem).min(hard(Resource::AS)?), + max_fsize: self + .max_fsize + .min(ceiling.max_fsize) + .min(hard(Resource::FSIZE)?), + }) } } -/// The default limits are also the ceilings: see [`JobLimits::clamp`]. +/// The default limits are also the ceilings; see [`JobLimits::clamp`]. impl Default for JobLimits { fn default() -> Self { JobLimits { @@ -1045,12 +1032,15 @@ mod test { max_mem: GB, max_fsize: GB, }; - assert_eq!(narrowed.clone().clamp(), narrowed); + assert_eq!(narrowed.clone().clamp().unwrap(), narrowed); let widened = JobLimits { max_cpu: u64::MAX, max_mem: u64::MAX, max_fsize: u64::MAX, }; - assert_eq!(widened.clamp(), JobLimits::default()); + assert_eq!( + widened.clamp().unwrap(), + JobLimits::default().clamp().unwrap() + ); } } diff --git a/server/src/contract.rs b/server/src/contract.rs index 4fe57326..7d5537f4 100644 --- a/server/src/contract.rs +++ b/server/src/contract.rs @@ -24,6 +24,8 @@ use rustix::io::{Errno, FdFlags, fcntl_setfd}; use sush_common::jobs::JobId; +use crate::error::check_errno; + #[allow(non_camel_case_types)] type ct_stathdl_t = *mut libc::c_void; @@ -53,16 +55,6 @@ const CTD_COMMON: c_int = 0; const PROCESS: &str = "/system/contract/process"; -/// libcontract returns an error number rather than setting `errno`. -fn check(what: &str, errno: c_int) -> io::Result<()> { - if errno == 0 { - Ok(()) - } else { - let error = io::Error::from_raw_os_error(errno); - Err(io::Error::new(error.kind(), format!("{what}: {error}"))) - } -} - /// Open a ctfs file, naming it in any error. Until illumos 18395, ctfs /// refuses opens with `O_CLOEXEC`, which `std::fs` always sets, so we /// open without it and set `FD_CLOEXEC` afterward. @@ -77,10 +69,11 @@ fn open(flags: OFlags, path: String) -> io::Result { } /// Run `spawn` with a job's contract template active, then abandon the -/// contract its `fork` created. The template is per thread, so `spawn` -/// must fork on the calling thread; being synchronous, it cannot move -/// to another. If `spawn` fails after forking or abandoning fails, -/// the empty contract lingers until we exit. +/// new contract that holds the child. The template is per thread, so +/// `spawn` must create the child on the calling thread; being +/// synchronous, it cannot move to another. If `spawn` fails after +/// creating a child, or abandoning fails, the empty contract lingers +/// until we exit. pub fn spawn_in_contract( job_id: &JobId, spawn: impl FnOnce() -> io::Result, @@ -91,8 +84,8 @@ pub fn spawn_in_contract( Ok((child, ctid)) } -/// An active process contract template. While it lives, every `fork` -/// on the calling thread creates a new contract. +/// An active process contract template. While it lives, every child +/// the calling thread creates starts in a new contract. struct Template(File); impl Template { @@ -104,29 +97,29 @@ impl Template { let fd = file.as_raw_fd(); let aux = CString::new(format!("sush job {job_id}")).map_err(io::Error::other)?; unsafe { - check( + check_errno( "setting fatal events", ct_pr_tmpl_set_fatal(fd, CT_PR_EV_HWERR), )?; - check("setting critical events", ct_tmpl_set_critical(fd, 0))?; - check("setting informative events", ct_tmpl_set_informative(fd, 0))?; - check( + check_errno("setting critical events", ct_tmpl_set_critical(fd, 0))?; + check_errno("setting informative events", ct_tmpl_set_informative(fd, 0))?; + check_errno( "setting parameters", ct_pr_tmpl_set_param(fd, CT_PR_PGRPONLY), )?; - check("setting label", ct_pr_tmpl_set_svc_aux(fd, aux.as_ptr()))?; - check("activating template", ct_tmpl_activate(fd))?; + check_errno("setting label", ct_pr_tmpl_set_svc_aux(fd, aux.as_ptr()))?; + check_errno("activating template", ct_tmpl_activate(fd))?; } Ok(Self(file)) } - /// Abandon the contract created by the last `fork` on this thread, - /// and return its ID. + /// Abandon the newest contract this thread created, and return its + /// ID. fn abandon_latest(self) -> io::Result { let latest = open(OFlags::RDONLY, format!("{PROCESS}/latest"))?; let ctid = unsafe { let mut status: ct_stathdl_t = std::ptr::null_mut(); - check( + check_errno( "reading latest contract", ct_status_read(latest.as_raw_fd(), CTD_COMMON, &mut status), )?; @@ -135,7 +128,7 @@ impl Template { ctid }; let ctl = open(OFlags::WRONLY, format!("{PROCESS}/{ctid}/ctl"))?; - check("abandoning contract", unsafe { + check_errno("abandoning contract", unsafe { ct_ctl_abandon(ctl.as_raw_fd()) })?; Ok(ctid) diff --git a/server/src/error.rs b/server/src/error.rs index 6dc8d239..ba16f63e 100644 --- a/server/src/error.rs +++ b/server/src/error.rs @@ -4,6 +4,9 @@ //! Job server errors. +use std::ffi::c_int; +use std::fmt::Display; +use std::io; use std::path::{Path, PathBuf}; use dropshot::{ClientErrorStatusCode, HttpError}; @@ -14,6 +17,21 @@ use sush_common::interactive::InteractiveJobError; use sush_common::jobs::{ExecutionError, JobId, JobOutputHash, SessionId}; use sush_common::keys::{KeyError, KeyId}; +/// Prefix an I/O error with `what` we were doing. +pub fn io_context(what: impl Display, error: io::Error) -> io::Error { + io::Error::new(error.kind(), format!("{what}: {error}")) +} + +/// Check the result of a C function that returns an error number rather +/// than setting the global `errno`, as libcontract and posix_spawn(3C) do. +pub fn check_errno(what: impl Display, errno: c_int) -> io::Result<()> { + if errno == 0 { + Ok(()) + } else { + Err(io_context(what, io::Error::from_raw_os_error(errno))) + } +} + /// What went wrong processing a client job request. #[derive(Debug, Error)] pub enum JobError { diff --git a/server/src/executor.rs b/server/src/executor.rs index 87204aea..41d2f2f6 100644 --- a/server/src/executor.rs +++ b/server/src/executor.rs @@ -8,22 +8,18 @@ //! machine, but session agnostic. use std::collections::{BTreeMap, BTreeSet}; +use std::ffi::OsString; use std::io; -use std::mem::MaybeUninit; -use std::os::fd::AsRawFd as _; -use std::process::Stdio; -use std::ptr::null_mut; use std::sync::{Arc, RwLock}; use chrono::Utc; use futures::Stream; use pwd::Passwd; -use rustix::io::{Errno, close}; -use rustix::process::{Pid, Signal, ioctl_tiocsctty, kill_process_group, setsid}; -use slog::{Logger, debug, error, o, warn}; +#[cfg(target_os = "illumos")] +use slog::debug; +use slog::{Logger, error, o, warn}; use tokio::fs::{DirBuilder, OpenOptions}; use tokio::io::AsyncWriteExt as _; -use tokio::process::{Child, Command}; use tokio::spawn; use tokio::sync::{mpsc, watch}; use tokio_stream::wrappers::ReceiverStream; @@ -42,11 +38,25 @@ use crate::job::{Job, SocketSender}; use crate::locker::boundary::{Boundary, BoundaryStore}; use crate::messages::v0::{Event, JobEvent}; use crate::output::JobOutputDir; -use crate::pty::open_pty; +use crate::spawn::{JobChild, PosixSpawn, SpawnStdio}; pub const DEFAULT_PATH: &str = "/usr/sbin:/usr/bin:/sbin:/bin"; pub const DEFAULT_TERM: &str = "vt100"; +const BASH: &str = "/usr/bin/bash"; + +/// A bash script that sets a job's soft limits and then execs the job +/// in `$BASH`, the bash running the script. posix_spawn cannot set +/// resource limits on the child, so the child must set its own before +/// it execs the job. The script's arguments are the maximum CPU seconds, +/// address space (in KiB), file size (in KiB), and the job command. +/// +/// For compatibility with `ssh`, we set `SSH_CLIENT` so that bash reads +/// `/root/.bashrc`. We set it only for the exec'd job, so that bash reads +/// it only once. +const LIMITS_SHIM: &str = + r#"ulimit -S -t "$1" -v "$2" -f "$3" && SSH_CLIENT=sush exec "$BASH" -c "$4""#; + /// This queue size is an arbitrary choice. We expect to consume /// messages from the channel very rapidly, so should not /// experience backpressure; if we do, something is wrong. @@ -255,7 +265,14 @@ async fn job_spawn(launch: Launch) { cols, wait: _, } = params; - let mut limits = requested.clone().clamp(); + let mut limits = match requested.clone().clamp() { + Ok(limits) => limits, + Err(error) => { + let error = ProcessError::io("reading resource limits", error); + send_error(&log, &job_id, &events, error).await; + return; + } + }; if limits != requested { warn!( log, "clamped requested job limits"; @@ -364,141 +381,71 @@ async fn job_spawn(launch: Launch) { format!("writing job request file `{}`", job_path.display()) ); - // Set up the job command. - let mut cmd = Command::new("bash"); - cmd.kill_on_drop(true); - cmd.arg("-c").arg(&command); - - // Set up basic environment. - cmd.env_clear(); + // Set up the job's environment. + let mut env = Vec::new(); match path_isolation { - PathIsolation::Enable => { - cmd.env("PATH", DEFAULT_PATH); - } + PathIsolation::Enable => env.push(("PATH", OsString::from(DEFAULT_PATH))), PathIsolation::InsecureDisable => { if let Some(path) = std::env::var_os("PATH") { - cmd.env("PATH", path); + env.push(("PATH", path)); } } } - if let Some(pwd) = Passwd::current_user() { - cmd.current_dir(&pwd.dir) - .env("HOME", &pwd.dir) - .env("LOGNAME", &pwd.name) - .env("USER", &pwd.name); + let pwd = Passwd::current_user(); + if let Some(pwd) = &pwd { + env.push(("HOME", pwd.dir.clone().into())); + env.push(("LOGNAME", pwd.name.clone().into())); + env.push(("USER", pwd.name.clone().into())); } - cmd.env("SSH_CLIENT", "sush") // read bashrc - .env("SUSH_JOB_ID", job_id.to_string()) - .env("SUSH_COMMAND", &command) - .env("SUSH_JOB_OUTPUT_DIR", &job_dir); - - // Set process limits. - unsafe { - let limits = limits.clone(); - cmd.pre_exec(move || limits.apply()); - } - - // Always create a new process session and group. - // Job stop kills the whole group. - unsafe { - cmd.pre_exec(move || { - setsid()?; - Ok(()) - }); - } - - // Reset signal dispositions and the signal mask. - // macOS has no realtime signals. - #[cfg(target_os = "macos")] - let max_signal = 31; - #[cfg(not(target_os = "macos"))] - let max_signal = libc::SIGRTMAX(); - unsafe { - cmd.pre_exec(move || { - for signal in 1..=max_signal { - libc::signal(signal, libc::SIG_DFL); - } - let mut none = MaybeUninit::::uninit(); - libc::sigemptyset(none.as_mut_ptr()); - libc::sigprocmask(libc::SIG_SETMASK, none.as_ptr(), null_mut()); - Ok(()) - }); - } - - let job = if mode.is_interactive() { - // Create a pseudoterminal and wire the child up to it. - let (pty, writer, pts, pts_path) = - with_io_err!(open_pty(), "opening pseudoterminal".to_string()); - macro_rules! pts_clone { - () => { - with_io_err!(pts.try_clone(), "cloning pseudoterminal".to_string()) - }; - } - cmd.env("SUSH_TTY", &pts_path) - .stdin(pts_clone!()) - .stdout(pts_clone!()) - .stderr(pts_clone!()); - - unsafe { - let pty = pty.as_raw_fd(); - cmd.pre_exec(move || { - close(pty); // not needed in the child - ioctl_tiocsctty(&pts)?; // set controlling terminal - Ok(()) - }); - } + env.push(("SUSH_JOB_ID", job_id.to_string().into())); + env.push(("SUSH_COMMAND", command.clone().into())); + env.push(("SUSH_JOB_OUTPUT_DIR", job_dir.clone().into())); + + let args = [ + BASH.to_string(), + "-c".to_string(), + LIMITS_SHIM.to_string(), + "sush".to_string(), + limits.max_cpu.to_string(), + limits.max_mem.div_ceil(1024).to_string(), + limits.max_fsize.div_ceil(1024).to_string(), + command, + ]; - // Initialize the pseudoterminal. - cmd.env("TERM", term.unwrap_or_else(|| DEFAULT_TERM.to_string())); + let (mut io, stdio) = with_io_err!( + JobIo::open(mode, stop.child_token()), + "opening job I/O".to_string() + ); + if let SpawnStdio::Terminal { path, .. } = &stdio { + env.push(("SUSH_TTY", path.clone().into())); + env.push(( + "TERM", + term.unwrap_or_else(|| DEFAULT_TERM.to_string()).into(), + )); if let (Some(rows), Some(cols)) = (rows, cols) { with_io_err!( - pty.set_window_size(WindowSize { rows, cols }), + io.set_window_size(WindowSize { rows, cols }), "resizing pseudoterminal window".to_string() ); } + } - // Start the interactive job. - let child = with_io_err!( - spawn_job(&log, &job_id, &mut cmd), - "spawning interactive job".to_string() - ); - let io = JobIo::interactive(pty, writer, stop.child_token()); - Job::start( - log.new(o!("mode" => mode.as_str())), - limits, - child, - io, - stdout_file, - stderr_file, - mode, - stop, - ) - } else { - // For batch jobs, close stdin and send output to pipes. - cmd.stdin(Stdio::null()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()); - - // Start the batch job. - let mut child = with_io_err!( - spawn_job(&log, &job_id, &mut cmd), - "spawning batch job".to_string() - ); - let io = JobIo::batch( - child.stdout.take().expect("batch job should have stdout"), - child.stderr.take().expect("batch job should have stderr"), - ); - Job::start( - log.new(o!("mode" => mode.as_str())), - limits, - child, - io, - stdout_file, - stderr_file, - mode, - stop, - ) - }; + // Start the job in $HOME. + let home = pwd.as_ref().map(|pwd| pwd.dir.as_str()); + let child = with_io_err!( + spawn_job(&log, &job_id, home, stdio, &args, &env), + format!("spawning {} job", mode.as_str()) + ); + let job = Job::start( + log.new(o!("mode" => mode.as_str())), + limits, + child, + io, + stdout_file, + stderr_file, + mode, + stop, + ); // Send the attachment point. if tx_attachment.send(Some(job.attachment())).is_err() { @@ -536,18 +483,31 @@ async fn job_spawn(launch: Launch) { }); } -/// Spawn a job process, with a fresh contract under illumos. -fn spawn_job(log: &Logger, job_id: &JobId, cmd: &mut Command) -> io::Result { +/// Spawn a job's shell in `dir`, on illumos into a contract of its own. +fn spawn_job( + log: &Logger, + job_id: &JobId, + dir: Option<&str>, + stdio: SpawnStdio, + args: &[String], + env: &[(&str, OsString)], +) -> io::Result { + let mut posix_spawn = PosixSpawn::new()?; + if let Some(dir) = dir { + posix_spawn.chdir(dir)?; + } + posix_spawn.stdio(stdio)?; + let spawn = || posix_spawn.spawn(BASH, args, env); #[cfg(target_os = "illumos")] { - let (child, ctid) = spawn_in_contract(job_id, || cmd.spawn())?; + let (child, ctid) = spawn_in_contract(job_id, spawn)?; debug!(log, "spawned job"; "ctid" => ctid); Ok(child) } #[cfg(not(target_os = "illumos"))] { let _ = (log, job_id); - cmd.spawn() + spawn() } } @@ -569,27 +529,6 @@ async fn send_error( } } -/// A job's process group ID, which its `child` leads. Capture it -/// before the leader is reaped, after which `id` returns nothing. -pub fn job_pgid(child: &Child) -> Option { - Pid::from_raw(child.id()?.try_into().ok()?) -} - -/// Send a signal to a job's whole process group. -pub fn kill_job(log: &Logger, pgid: Option, signal: Signal) { - let Some(pgid) = pgid else { - debug!(log, "job has no process group"); - return; - }; - match kill_process_group(pgid, signal) { - Ok(()) => debug!(log, "signalled job processes"; "signal" => ?signal), - Err(Errno::SRCH) => debug!(log, "job processes are already dead"), - Err(error) => { - error!(log, "unable to signal job"; "signal" => ?signal, "error" => %error) - } - } -} - #[derive(Debug, Clone, Copy)] pub enum PathIsolation { Enable, diff --git a/server/src/io.rs b/server/src/io.rs index e98dcc1e..05e5e8df 100644 --- a/server/src/io.rs +++ b/server/src/io.rs @@ -10,7 +10,8 @@ //! to batch jobs is dropped; input to interactive jobs is relayed over //! a bounded channel, and dropped if the channel is full. -use std::io; +use std::io::{self, pipe}; +use std::os::fd::OwnedFd; use std::time::Duration; use bytes::{Bytes, BytesMut}; @@ -22,9 +23,11 @@ use tokio::{select, spawn}; use tokio_util::sync::CancellationToken; use sush_common::interactive::WindowSize; +use sush_common::jobs::JobMode; use sush_common::jobs::JobOutputStream::{self, *}; -use crate::pty::{PtyReader, PtyWriter}; +use crate::pty::{PtyReader, PtyWriter, open_pty}; +use crate::spawn::SpawnStdio; const INPUT_CHANNEL_CAPACITY: usize = 100; pub const BATCH_OUTPUT_BUFFER_SIZE: usize = 0x10000; @@ -60,6 +63,7 @@ impl StreamState { } } +/// The sush server's end of a job's I/O. pub enum JobIo { Interactive { pty: StreamState, @@ -72,6 +76,26 @@ pub enum JobIo { } impl JobIo { + /// Create both ends of a job's I/O: a pseudoterminal for an + /// interactive job, and pipes for any other job. + pub fn open(mode: JobMode, stop: CancellationToken) -> io::Result<(Self, SpawnStdio)> { + if mode.is_interactive() { + let (pty, writer, pts, path) = open_pty()?; + let stdio = SpawnStdio::Terminal { pts, path }; + Ok((Self::interactive(pty, writer, stop), stdio)) + } else { + let (stdout, stdout_writer) = pipe()?; + let (stderr, stderr_writer) = pipe()?; + let stdio = SpawnStdio::Pipes { + stdout: stdout_writer.into(), + stderr: stderr_writer.into(), + }; + let stdout = ChildStdout::from_std(OwnedFd::from(stdout).into())?; + let stderr = ChildStderr::from_std(OwnedFd::from(stderr).into())?; + Ok((Self::batch(stdout, stderr), stdio)) + } + } + pub fn interactive(pty: PtyReader, writer: PtyWriter, stop: CancellationToken) -> Self { let (tx_input, rx_input) = mpsc::channel::(INPUT_CHANNEL_CAPACITY); spawn(Self::relay_input(rx_input, writer, stop)); @@ -168,17 +192,8 @@ impl JobIo { } /// How long to keep reading output after the child dies. - /// - /// For a pty there is no portable EOF signal; a short window - /// after death is the best we can do (OpenSSH does this too). - /// Pipes deliver EOF once every writer exits, so this is only a - /// backstop against a descendant that escaped the process group - /// while holding the inherited pipe open. pub fn drain_timeout(&self) -> Duration { - match self { - Self::Interactive { .. } => Duration::from_millis(10), - Self::Batch { .. } => Duration::from_secs(5), - } + Duration::from_secs(1) } pub fn stop_recording(&mut self, stream: JobOutputStream) { diff --git a/server/src/job.rs b/server/src/job.rs index bf3437af..34055a51 100644 --- a/server/src/job.rs +++ b/server/src/job.rs @@ -33,7 +33,6 @@ use slog::{Logger, debug, error, info, warn}; use sush_common::hash::Hasher; use tokio::fs::File; use tokio::io::{AsyncReadExt as _, AsyncSeekExt as _, AsyncWriteExt as _}; -use tokio::process::Child; use tokio::sync::mpsc; use tokio::task::{JoinError, JoinHandle}; use tokio::time::{Duration, Instant, sleep, timeout}; @@ -49,9 +48,9 @@ use sush_common::jobs::{ }; use tokio_util::sync::CancellationToken; -use crate::executor::{job_pgid, kill_job}; use crate::io::JobIo; use crate::mux::WebSocketMux; +use crate::spawn::JobChild; pub type SocketStream = WebSocketStream; pub type SocketSender = mpsc::Sender<(SocketStream, Access)>; @@ -85,7 +84,7 @@ impl Job { pub fn start( log: Logger, limits: JobLimits, - child: Child, + child: JobChild, io: JobIo, stdout: File, stderr: File, @@ -116,7 +115,7 @@ impl Job { async fn job( log: Logger, JobLimits { max_fsize, .. }: JobLimits, - mut child: Child, + mut child: JobChild, mut io: JobIo, mut stdout_file: File, mut stderr_file: File, @@ -128,7 +127,6 @@ async fn job( let mut stderr_hasher = Hasher::new(); let mut clients = WebSocketMux::new(); let mut fatal = Option::::None; - let pgid = job_pgid(&child); let mut killed = false; let mut kills = 0; let mut dead = false; @@ -349,7 +347,7 @@ async fn job( // Stop job on cancellation signal: terminate first, then kill. _ = stop.cancelled(), if !killed => { - kill_job(&log, pgid, Signal::TERM); + kill_job(&log, &child, Signal::TERM); kill_timeout.as_mut().reset(Instant::now() + KILL_GRACE); killed = true; } @@ -358,7 +356,7 @@ async fn job( // period, even if its leader is already dead. _ = &mut kill_timeout, if killed && kills < KILL_ATTEMPTS => { warn!(log, "job survived SIGTERM grace period, upgrading to SIGKILL"); - kill_job(&log, pgid, Signal::KILL); + kill_job(&log, &child, Signal::KILL); kill_timeout.as_mut().reset(Instant::now() + KILL_GRACE); kills += 1; } @@ -427,11 +425,11 @@ async fn job( let exit_status = select! { status = child.wait() => status, _ = stop.cancelled(), if !killed => { - kill_job(&log, pgid, Signal::KILL); + kill_job(&log, &child, Signal::KILL); child.wait().await } _ = sleep(KILL_GRACE), if killed && !dead => { - kill_job(&log, pgid, Signal::KILL); + kill_job(&log, &child, Signal::KILL); child.wait().await } }; @@ -439,7 +437,7 @@ async fn job( Some(error) => Err(error), None => match exit_status { Ok(status) => process_exit(status), - Err(error) => Err(ProcessError::Interactive(error.to_string())), + Err(error) => Err(ProcessError::io("reaping job process", error)), }, }; info!(log, "job stopped"); @@ -487,6 +485,19 @@ async fn stream_output( .await; } +/// Send a signal to every process in a job. +fn kill_job(log: &Logger, child: &JobChild, signal: Signal) { + match child.kill(signal) { + Ok(()) => debug!(log, "signalled job processes"; "signal" => ?signal), + Err(error) if error.raw_os_error() == Some(libc::ESRCH) => { + debug!(log, "job processes are already dead") + } + Err(error) => { + error!(log, "unable to signal job"; "signal" => ?signal, "error" => %error) + } + } +} + fn process_exit(exit_status: ExitStatus) -> Result { if let Some(code) = exit_status.code() { Ok(code) @@ -528,10 +539,9 @@ mod test { use slog::{Drain as _, o}; use slog_term::{FullFormat, PlainSyncDecorator, TestStdoutWriter}; use tempfile::NamedTempFile; - use tokio::process::Command; use crate::io::JobIo; - use crate::pty::open_pty; + use crate::spawn::{PosixSpawn, SpawnStdio}; use super::*; @@ -544,17 +554,20 @@ mod test { let decorator = PlainSyncDecorator::new(TestStdoutWriter); let drain = FullFormat::new(decorator).build().fuse(); let log = Logger::root(drain, o!("test" => function_name!())); - let (pty, writer, pts, pts_path) = open_pty().unwrap(); + let stop = CancellationToken::new(); + let (io, stdio) = JobIo::open(JobMode::Interactive, stop.child_token()).unwrap(); + let SpawnStdio::Terminal { path: pts_path, .. } = &stdio else { + panic!("interactive job has no terminal"); + }; + let pts_path = pts_path.clone(); assert!(pts_path.starts_with("/dev/pts/")); let output_file = NamedTempFile::new().unwrap(); - let child = Command::new("tty") - .stdin(pts.try_clone().unwrap()) - .stdout(pts.try_clone().unwrap()) - .spawn() + let mut posix_spawn = PosixSpawn::new().unwrap(); + posix_spawn.stdio(stdio).unwrap(); + let child = posix_spawn + .spawn("/usr/bin/tty", &["tty"], &[] as &[(&str, &str)]) .unwrap(); - let stop = CancellationToken::new(); - let io = JobIo::interactive(pty, writer, stop.child_token()); let job = Job::start( log, JobLimits::default(), diff --git a/server/src/lib.rs b/server/src/lib.rs index 2d9988d3..20546ce3 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -31,6 +31,7 @@ pub mod proxy; pub mod pty; pub mod rack; pub mod server; +pub mod spawn; pub mod state; pub use error::JobError; diff --git a/server/src/spawn.rs b/server/src/spawn.rs new file mode 100644 index 00000000..3ef97c70 --- /dev/null +++ b/server/src/spawn.rs @@ -0,0 +1,315 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at https://mozilla.org/MPL/2.0/. + +//! Spawn and reap job processes without forking. +//! +//! On illumos, posix_spawn(3C) builds the child in the kernel, without +//! copying our address space or stopping our threads. Avoiding both +//! matters because we run inside `sled-agent`, a large, heavily +//! multithreaded program. We call posix_spawn ourselves because +//! [`std::process::Command`] falls back to fork on illumos whenever a +//! child needs a new session, a reset signal mask, or a controlling +//! terminal. + +#[cfg(target_vendor = "apple")] +use std::ffi::c_char; +use std::ffi::{CString, OsStr, c_int, c_short}; +use std::io; +use std::mem::MaybeUninit; +use std::os::fd::{AsRawFd as _, OwnedFd, RawFd}; +use std::os::unix::ffi::OsStrExt as _; +use std::os::unix::process::ExitStatusExt as _; +use std::path::{Path, PathBuf}; +use std::process::ExitStatus; +use std::ptr::null_mut; + +use libc::{ + EINTR, EIO, O_RDONLY, O_RDWR, POSIX_SPAWN_SETSIGDEF, POSIX_SPAWN_SETSIGMASK, SIGKILL, SIGSTOP, + WNOHANG, posix_spawn, posix_spawn_file_actions_adddup2, posix_spawn_file_actions_addopen, + posix_spawn_file_actions_destroy, posix_spawn_file_actions_init, posix_spawn_file_actions_t, + posix_spawnattr_destroy, posix_spawnattr_init, posix_spawnattr_setflags, + posix_spawnattr_setsigdefault, posix_spawnattr_setsigmask, posix_spawnattr_t, sigdelset, + sigemptyset, sigfillset, sigset_t, waitpid, +}; +#[cfg(not(target_vendor = "apple"))] +use libc::{POSIX_SPAWN_SETSID, posix_spawn_file_actions_addchdir_np}; +use rustix::process::{Pid, Signal, kill_process_group}; +use tokio::runtime::Handle; +use tokio::signal::unix::{Signal as SignalStream, SignalKind, signal}; + +use crate::error::{check_errno, io_context}; + +// The libc crate lacks these for macOS, which has had both since 10.15. +#[cfg(target_vendor = "apple")] +const POSIX_SPAWN_SETSID: c_short = 0x0400; +#[cfg(target_vendor = "apple")] +unsafe extern "C" { + fn posix_spawn_file_actions_addchdir_np( + actions: *mut posix_spawn_file_actions_t, + path: *const c_char, + ) -> c_int; +} + +fn c_string(bytes: impl Into>) -> io::Result { + CString::new(bytes).map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error)) +} + +/// Our handles to the child's side of its standard input and output. +/// [`PosixSpawn::stdio`] takes them, and [`PosixSpawn::spawn`] closes +/// them once the child is spawned. +pub enum SpawnStdio { + /// The child will open the pts at `path` as its standard input and + /// output. Until it does, we hold `pts` open; on illumos, the last + /// close of a pts pops its terminal modules, and the child would + /// then open a stream that is not a terminal. + Terminal { pts: OwnedFd, path: PathBuf }, + + /// The child will read from `/dev/null` and write to these pipes. + /// A pipe signals EOF only after every write end is closed, + /// so we must close these once the child has its copies. + Pipes { stdout: OwnedFd, stderr: OwnedFd }, +} + +/// Spawns job processes. Each child leads a new session, starts with +/// every signal at its default action, and blocks no signals. +pub struct PosixSpawn { + attr: Box>, + actions: Box>, + stdio: Option, +} + +impl PosixSpawn { + pub fn new() -> io::Result { + let mut attr = Box::new(MaybeUninit::uninit()); + check_errno("initializing spawn attributes", unsafe { + posix_spawnattr_init(attr.as_mut_ptr()) + })?; + + let mut actions = Box::new(MaybeUninit::uninit()); + if let Err(error) = check_errno("initializing spawn file actions", unsafe { + posix_spawn_file_actions_init(actions.as_mut_ptr()) + }) { + unsafe { posix_spawnattr_destroy(attr.as_mut_ptr()) }; + return Err(error); + } + let mut posix_spawn = Self { + attr, + actions, + stdio: None, + }; + + unsafe { + let attr = posix_spawn.attr.as_mut_ptr(); + let mut defaults = MaybeUninit::::uninit(); + sigfillset(defaults.as_mut_ptr()); + // SIGKILL and SIGSTOP can't be changed. + sigdelset(defaults.as_mut_ptr(), SIGKILL); + sigdelset(defaults.as_mut_ptr(), SIGSTOP); + check_errno( + "setting default signals", + posix_spawnattr_setsigdefault(attr, defaults.as_ptr()), + )?; + + let mut mask = MaybeUninit::::uninit(); + sigemptyset(mask.as_mut_ptr()); + check_errno( + "setting signal mask", + posix_spawnattr_setsigmask(attr, mask.as_ptr()), + )?; + + let flags = POSIX_SPAWN_SETSID as c_short + | POSIX_SPAWN_SETSIGDEF as c_short + | POSIX_SPAWN_SETSIGMASK as c_short; + check_errno("setting spawn flags", posix_spawnattr_setflags(attr, flags))?; + } + + Ok(posix_spawn) + } + + /// Run the child in `dir`. + pub fn chdir(&mut self, dir: impl AsRef) -> io::Result<()> { + let what = format!("changing to `{}`", dir.as_ref().display()); + let dir = c_string(dir.as_ref().as_os_str().as_bytes())?; + check_errno(what, unsafe { + posix_spawn_file_actions_addchdir_np(self.actions.as_mut_ptr(), dir.as_ptr()) + }) + } + + /// Open `path` as the child's `fd`. The child opens it after it + /// starts its new session, so if `path` is a terminal, it becomes + /// the child's controlling terminal. + fn open(&mut self, fd: RawFd, path: impl AsRef, flags: c_int) -> io::Result<()> { + let what = format!("opening `{}` as fd {fd}", path.as_ref().display()); + let path = c_string(path.as_ref().as_os_str().as_bytes())?; + check_errno(what, unsafe { + posix_spawn_file_actions_addopen(self.actions.as_mut_ptr(), fd, path.as_ptr(), flags, 0) + }) + } + + /// Duplicate our descriptor `from` onto the child's descriptor `to`. + fn dup2(&mut self, from: RawFd, to: RawFd) -> io::Result<()> { + check_errno(format!("duplicating fd {from} onto fd {to}"), unsafe { + posix_spawn_file_actions_adddup2(self.actions.as_mut_ptr(), from, to) + }) + } + + /// Give the child `stdio` as its standard input and output. + pub fn stdio(&mut self, stdio: SpawnStdio) -> io::Result<()> { + match &stdio { + SpawnStdio::Terminal { path, .. } => { + self.open(0, path, O_RDWR)?; + self.dup2(0, 1)?; + self.dup2(0, 2)?; + } + SpawnStdio::Pipes { stdout, stderr } => { + self.open(0, "/dev/null", O_RDONLY)?; + self.dup2(stdout.as_raw_fd(), 1)?; + self.dup2(stderr.as_raw_fd(), 2)?; + } + } + self.stdio = Some(stdio); + Ok(()) + } + + /// Start `program`, which must be an absolute path, because + /// posix_spawn does not search `PATH`. `args` begins with `argv[0]`, + /// and `env` is the child's entire environment. + pub fn spawn( + self, + program: impl AsRef, + args: &[impl AsRef], + env: &[(impl AsRef, impl AsRef)], + ) -> io::Result { + let path = program.as_ref(); + if !path.is_absolute() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!("`{}` is not an absolute path", path.display()), + )); + } + let program = c_string(path.as_os_str().as_bytes())?; + let args = args + .iter() + .map(|arg| c_string(arg.as_ref().as_bytes())) + .collect::>>()?; + let env = env + .iter() + .map(|(name, value)| { + let (name, value) = (name.as_ref().as_bytes(), value.as_ref().as_bytes()); + c_string([name, b"=", value].concat()) + }) + .collect::>>()?; + let mut argv: Vec<_> = args.iter().map(|arg| arg.as_ptr().cast_mut()).collect(); + argv.push(null_mut()); + let mut envp: Vec<_> = env.iter().map(|var| var.as_ptr().cast_mut()).collect(); + envp.push(null_mut()); + + let sigchld = signal(SignalKind::child()) + .map_err(|error| io_context("listening for SIGCHLD", error))?; + let mut pid = 0; + check_errno(format!("spawning `{}`", path.display()), unsafe { + posix_spawn( + &mut pid, + program.as_ptr(), + self.actions.as_ptr(), + self.attr.as_ptr(), + argv.as_ptr(), + envp.as_ptr(), + ) + })?; + let pid = Pid::from_raw(pid).ok_or_else(|| io::Error::other("spawned pid 0"))?; + Ok(JobChild::new(pid, sigchld)) + } +} + +impl Drop for PosixSpawn { + fn drop(&mut self) { + unsafe { + posix_spawn_file_actions_destroy(self.actions.as_mut_ptr()); + posix_spawnattr_destroy(self.attr.as_mut_ptr()); + } + } +} + +/// A spawned job process. Only [`JobChild::wait`] reaps it. +pub struct JobChild { + pid: Pid, + sigchld: SignalStream, + /// The exit status, or the error number `waitpid` failed with. + reaped: Option>, +} + +impl JobChild { + fn new(pid: Pid, sigchld: SignalStream) -> Self { + Self { + pid, + sigchld, + reaped: None, + } + } + + /// Send `signal` to every process in the job's process group, which + /// the child leads. + pub fn kill(&self, signal: Signal) -> io::Result<()> { + Ok(kill_process_group(self.pid, signal)?) + } + + /// Wait for the child to exit, and reap it. Cancelling the wait + /// never loses the exit status. + pub async fn wait(&mut self) -> io::Result { + let reaped = match self.reaped { + Some(reaped) => reaped, + None => *self.reaped.insert(reap(self.pid, &mut self.sigchld).await), + }; + reaped.map_err(io::Error::from_raw_os_error) + } +} + +/// Kill a job that [`JobChild::wait`] never reaped, and reap its child +/// in the background. An unreaped child still holds its pid, so the +/// kill cannot reach another process. If [`JobChild::wait`] failed to +/// reap the child, another process may hold its pid by now, so we leave +/// it alone. +impl Drop for JobChild { + fn drop(&mut self) { + if self.reaped.is_some() { + return; + } + let _ = self.kill(Signal::KILL); + if let Ok(handle) = Handle::try_current() { + let pid = self.pid; + handle.spawn(async move { + if let Ok(mut sigchld) = signal(SignalKind::child()) { + let _ = reap(pid, &mut sigchld).await; + } + }); + } + } +} + +/// Wait for `pid` to exit, and reap it after the last `.await`. Returns +/// the error number if `waitpid` fails. +async fn reap(pid: Pid, sigchld: &mut SignalStream) -> Result { + loop { + if let Some(status) = try_wait(pid)? { + return Ok(status); + } + sigchld.recv().await; + } +} + +/// Reap `pid` if it has exited. +fn try_wait(pid: Pid) -> Result, i32> { + let mut status = 0; + loop { + match unsafe { waitpid(pid.as_raw_nonzero().get(), &mut status, WNOHANG) } { + 0 => return Ok(None), + -1 => match io::Error::last_os_error().raw_os_error() { + Some(EINTR) => continue, + errno => return Err(errno.unwrap_or(EIO)), + }, + _ => return Ok(Some(ExitStatus::from_raw(status))), + } + } +} diff --git a/tests/src/manager_tests.rs b/tests/src/manager_tests.rs index 4ecdb0ee..b87675ed 100644 --- a/tests/src/manager_tests.rs +++ b/tests/src/manager_tests.rs @@ -340,6 +340,117 @@ async fn job_contract() { check_status_stopped(status, &job_id, Ok(0), None, None); } +/// A job's bash sees the job's limits as soft limits, and sees +/// `SSH_CLIENT` set. +#[named] +#[tokio::test] +async fn job_limits_and_environment() { + let log = test_logger(function_name!()); + let (mgr, mut root, _dir, _shutdown) = manager_and_test_root(log).await; + let authn = fake_identity(&mut root).await; + let SessionCeremony { + session_id, + client_nonce, + signer_nonce, + seq, + .. + } = session_ceremony(&mgr, 1); + let session = Session::new(session_id); + mgr.session_start(&authn, session_id, client_nonce, signer_nonce, seq) + .await + .unwrap(); + + let job_id = session.next_job_id(); + let command = r#"echo "$(ulimit -S -t) $(ulimit -S -v) $(ulimit -S -f) $SSH_CLIENT""#; + let job = root + .sign_job_request(job_id, session_id, command, false) + .await; + mgr.job_start( + &authn, + job.into_signed(), + JobStartParams { + wait: JobWait::Stop, + ..Default::default() + }, + ) + .await + .unwrap(); + let baseboard_id = mgr.own_baseboard(); + let stdout = mgr + .job_output(&authn, &job_id, baseboard_id, Stdout, None) + .await + .unwrap() + .into_bytes() + .await; + let JobLimits { + max_cpu, + max_mem, + max_fsize, + } = JobLimits::default().clamp().unwrap(); + assert_eq!( + String::from_utf8_lossy(&stdout), + format!( + "{max_cpu} {} {} sush\n", + max_mem.div_ceil(1024), + max_fsize.div_ceil(1024) + ) + ); +} + +/// An interactive job leads its own session, with its pseudoterminal +/// as its controlling terminal. +#[named] +#[tokio::test] +async fn interactive_job_session() { + let log = test_logger(function_name!()); + let (mgr, mut root, _dir, _shutdown) = manager_and_test_root(log).await; + let authn = fake_identity(&mut root).await; + let SessionCeremony { + session_id, + client_nonce, + signer_nonce, + seq, + .. + } = session_ceremony(&mgr, 1); + let session = Session::new(session_id); + mgr.session_start(&authn, session_id, client_nonce, signer_nonce, seq) + .await + .unwrap(); + + let job_id = session.next_job_id(); + let command = r#" + test $(ps -o sid= -p $$) -eq $$ || echo "not a session leader" + { : /dev/null || echo "no controlling terminal" + test "$(tty)" = "$SUSH_TTY" || echo "stdin is not the job's terminal""#; + let job = root + .sign_job_request_for( + job_id, + session_id, + command, + true, + test_baseboard_id().into(), + ) + .await; + mgr.job_start( + &authn, + job.into_signed(), + JobStartParams { + wait: JobWait::Stop, + ..Default::default() + }, + ) + .await + .unwrap(); + let baseboard_id = mgr.own_baseboard(); + let stdout = mgr + .job_output(&authn, &job_id, baseboard_id, Stdout, None) + .await + .unwrap() + .into_bytes() + .await; + assert!(stdout.is_empty(), "{}", String::from_utf8_lossy(&stdout)); +} + /// A job signed for anything but the active session is refused with /// an error status, not silently discarded. #[named] From 7c9bca5d58754c8bbcc998f01432c9368cfbcf8e Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 24 Oct 2026 14:51:09 -0600 Subject: [PATCH 3/6] Stop jobs by contract on illumos A stop used to signal the job's process group, so a descendant that started its own process group escaped it. Signalling the job's contract reaches every process in the job. Co-Authored-By: Claude --- server/src/contract.rs | 14 ++++++++ server/src/executor.rs | 2 +- server/src/spawn.rs | 25 +++++++++++-- tests/src/manager_tests.rs | 74 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 112 insertions(+), 3 deletions(-) diff --git a/server/src/contract.rs b/server/src/contract.rs index 7d5537f4..5bc3f224 100644 --- a/server/src/contract.rs +++ b/server/src/contract.rs @@ -21,6 +21,7 @@ use std::os::fd::AsRawFd as _; use libc::ctid_t; use rustix::fs::{Mode, OFlags}; use rustix::io::{Errno, FdFlags, fcntl_setfd}; +use rustix::process::Signal; use sush_common::jobs::JobId; @@ -44,6 +45,10 @@ unsafe extern "C" { fn ct_ctl_abandon(fd: c_int) -> c_int; } +unsafe extern "C" { + fn sigsend(idtype: libc::idtype_t, id: libc::id_t, sig: c_int) -> c_int; +} + /// Process had an uncorrectable hardware error. const CT_PR_EV_HWERR: c_uint = 0x20; @@ -68,6 +73,15 @@ fn open(flags: OFlags, path: String) -> io::Result { Ok(File::from(fd)) } +/// Send `signal` to every process in contract `ctid`. +pub fn signal_contract(ctid: ctid_t, signal: Signal) -> io::Result<()> { + if unsafe { sigsend(libc::P_CTID, ctid, signal.as_raw()) } == 0 { + Ok(()) + } else { + Err(io::Error::last_os_error()) + } +} + /// Run `spawn` with a job's contract template active, then abandon the /// new contract that holds the child. The template is per thread, so /// `spawn` must create the child on the calling thread; being diff --git a/server/src/executor.rs b/server/src/executor.rs index 41d2f2f6..371c770b 100644 --- a/server/src/executor.rs +++ b/server/src/executor.rs @@ -502,7 +502,7 @@ fn spawn_job( { let (child, ctid) = spawn_in_contract(job_id, spawn)?; debug!(log, "spawned job"; "ctid" => ctid); - Ok(child) + Ok(child.in_contract(ctid)) } #[cfg(not(target_os = "illumos"))] { diff --git a/server/src/spawn.rs b/server/src/spawn.rs index 3ef97c70..ff3b50db 100644 --- a/server/src/spawn.rs +++ b/server/src/spawn.rs @@ -24,6 +24,8 @@ use std::path::{Path, PathBuf}; use std::process::ExitStatus; use std::ptr::null_mut; +#[cfg(target_os = "illumos")] +use libc::ctid_t; use libc::{ EINTR, EIO, O_RDONLY, O_RDWR, POSIX_SPAWN_SETSIGDEF, POSIX_SPAWN_SETSIGMASK, SIGKILL, SIGSTOP, WNOHANG, posix_spawn, posix_spawn_file_actions_adddup2, posix_spawn_file_actions_addopen, @@ -38,6 +40,8 @@ use rustix::process::{Pid, Signal, kill_process_group}; use tokio::runtime::Handle; use tokio::signal::unix::{Signal as SignalStream, SignalKind, signal}; +#[cfg(target_os = "illumos")] +use crate::contract::signal_contract; use crate::error::{check_errno, io_context}; // The libc crate lacks these for macOS, which has had both since 10.15. @@ -235,6 +239,8 @@ impl Drop for PosixSpawn { /// A spawned job process. Only [`JobChild::wait`] reaps it. pub struct JobChild { pid: Pid, + #[cfg(target_os = "illumos")] + contract: Option, sigchld: SignalStream, /// The exit status, or the error number `waitpid` failed with. reaped: Option>, @@ -244,14 +250,29 @@ impl JobChild { fn new(pid: Pid, sigchld: SignalStream) -> Self { Self { pid, + #[cfg(target_os = "illumos")] + contract: None, sigchld, reaped: None, } } - /// Send `signal` to every process in the job's process group, which - /// the child leads. + /// Record the contract that holds the child and its descendants. + #[cfg(target_os = "illumos")] + pub fn in_contract(mut self, ctid: ctid_t) -> Self { + self.contract = Some(ctid); + self + } + + /// Send `signal` to every process in the job: its contract if we + /// know it, or else its process group, which the child leads. + /// Signalling the contract also reaches descendants that left the + /// process group. pub fn kill(&self, signal: Signal) -> io::Result<()> { + #[cfg(target_os = "illumos")] + if let Some(ctid) = self.contract { + return signal_contract(ctid, signal); + } Ok(kill_process_group(self.pid, signal)?) } diff --git a/tests/src/manager_tests.rs b/tests/src/manager_tests.rs index b87675ed..f0239e3b 100644 --- a/tests/src/manager_tests.rs +++ b/tests/src/manager_tests.rs @@ -340,6 +340,80 @@ async fn job_contract() { check_status_stopped(status, &job_id, Ok(0), None, None); } +/// Stopping a job kills every process in its contract, including one +/// in a process group of its own. +#[cfg(target_os = "illumos")] +#[named] +#[tokio::test] +async fn job_stop_reaches_contract() { + let log = test_logger(function_name!()); + let (mgr, mut root, _dir, _shutdown) = manager_and_test_root(log).await; + let authn = fake_identity(&mut root).await; + let SessionCeremony { + session_id, + client_nonce, + signer_nonce, + seq, + .. + } = session_ceremony(&mgr, 1); + let session = Session::new(session_id); + mgr.session_start(&authn, session_id, client_nonce, signer_nonce, seq) + .await + .unwrap(); + + // With job control on, bash runs `sleep` in its own process group. + let job_id = session.next_job_id(); + let command = "set -m; sleep 1000 & echo $!; wait"; + let job = root + .sign_job_request(job_id, session_id, command, false) + .await; + mgr.job_start( + &authn, + job.into_signed(), + JobStartParams { + wait: JobWait::Start, + ..Default::default() + }, + ) + .await + .unwrap(); + let baseboard_id = mgr.own_baseboard(); + let pid: i32 = timeout(Duration::from_secs(10), async { + loop { + let stdout = mgr + .job_output(&authn, &job_id, baseboard_id, Stdout, None) + .await + .unwrap() + .into_bytes() + .await; + if let Ok(pid) = String::from_utf8_lossy(&stdout).trim().parse() { + break pid; + } + sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("job never reported its sleep's pid"); + + mgr.job_stop( + &authn, + &job_id, + JobStopParams { + wait: JobWait::Stop, + ..Default::default() + }, + ) + .await + .unwrap(); + timeout(Duration::from_secs(10), async { + while unsafe { libc::kill(pid, 0) } == 0 { + sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("sleep outside the job's process group survived the stop"); +} + /// A job's bash sees the job's limits as soft limits, and sees /// `SSH_CLIENT` set. #[named] From bdc30df975775d9230d0ad539189a28fa2279cc4 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sun, 25 Oct 2026 11:07:58 -0600 Subject: [PATCH 4/6] Register each pty handle only for the direction it uses On illumos, after a pty's subsidiary closes, polling the master for both reading and writing can report nothing but a write error, even though an EOF is ready to read. The job's pty reader was previously registered for both directions, so it never saw the EOF, and the drain timeout expired instead. Co-authored-by: Claude --- server/src/pty.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/server/src/pty.rs b/server/src/pty.rs index e2f618dc..e8b8c19a 100644 --- a/server/src/pty.rs +++ b/server/src/pty.rs @@ -21,7 +21,7 @@ use rustix::pty::{OpenptFlags, grantpt, openpt, ptsname, unlockpt}; use rustix::termios::{tcgetwinsize, tcsetwinsize}; use tokio::io::unix::AsyncFd; -use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; +use tokio::io::{AsyncRead, AsyncWrite, Interest, ReadBuf}; use sush_common::interactive::WindowSize; @@ -55,9 +55,9 @@ pub fn open_pty() -> io::Result<(PtyReader, PtyWriter, OwnedFd, PathBuf)> { } } - let dup = AsyncFd::new(pty.try_clone()?)?; - let fd = AsyncFd::new(pty)?; - Ok((PtyReader(fd), PtyWriter(dup), pts, pts_path)) + let read_fd = AsyncFd::with_interest(pty.try_clone()?, Interest::READABLE)?; + let write_fd = AsyncFd::with_interest(pty, Interest::WRITABLE)?; + Ok((PtyReader(read_fd), PtyWriter(write_fd), pts, pts_path)) } #[derive(Debug)] From 81a86b1e21f441d3b5b127d23415e8c14cb17173 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Mon, 26 Oct 2026 08:17:13 -0600 Subject: [PATCH 5/6] Revert "REVERT ME: Work around ctfs refusing O_CLOEXEC opens" This reverts commit f13d2464384be5eb5f6522372220b506d72b401a. Co-Authored-By: Claude --- server/src/contract.rs | 32 +++++++++++++++----------------- 1 file changed, 15 insertions(+), 17 deletions(-) diff --git a/server/src/contract.rs b/server/src/contract.rs index 5bc3f224..48546f7c 100644 --- a/server/src/contract.rs +++ b/server/src/contract.rs @@ -14,13 +14,11 @@ //! login session, in OpenSSH's `openbsd-compat/port-solaris.c`. use std::ffi::{CString, c_char, c_int, c_uint}; -use std::fs::File; +use std::fs::{File, OpenOptions}; use std::io; use std::os::fd::AsRawFd as _; use libc::ctid_t; -use rustix::fs::{Mode, OFlags}; -use rustix::io::{Errno, FdFlags, fcntl_setfd}; use rustix::process::Signal; use sush_common::jobs::JobId; @@ -60,17 +58,11 @@ const CTD_COMMON: c_int = 0; const PROCESS: &str = "/system/contract/process"; -/// Open a ctfs file, naming it in any error. Until illumos 18395, ctfs -/// refuses opens with `O_CLOEXEC`, which `std::fs` always sets, so we -/// open without it and set `FD_CLOEXEC` afterward. -fn open(flags: OFlags, path: String) -> io::Result { - let named = |error: Errno| { - let error = io::Error::from(error); - io::Error::new(error.kind(), format!("opening {path}: {error}")) - }; - let fd = rustix::fs::open(&path, flags, Mode::empty()).map_err(named)?; - fcntl_setfd(&fd, FdFlags::CLOEXEC).map_err(named)?; - Ok(File::from(fd)) +/// Open a ctfs file, naming it in any error. +fn open(options: &OpenOptions, path: String) -> io::Result { + options + .open(&path) + .map_err(|error| io::Error::new(error.kind(), format!("opening {path}: {error}"))) } /// Send `signal` to every process in contract `ctid`. @@ -107,7 +99,10 @@ impl Template { /// As in sshd, a hardware error kills only the process group of the /// process it struck. The contract is labeled by job ID. fn activate(job_id: &JobId) -> io::Result { - let file = open(OFlags::RDWR, format!("{PROCESS}/template"))?; + let file = open( + OpenOptions::new().read(true).write(true), + format!("{PROCESS}/template"), + )?; let fd = file.as_raw_fd(); let aux = CString::new(format!("sush job {job_id}")).map_err(io::Error::other)?; unsafe { @@ -130,7 +125,7 @@ impl Template { /// Abandon the newest contract this thread created, and return its /// ID. fn abandon_latest(self) -> io::Result { - let latest = open(OFlags::RDONLY, format!("{PROCESS}/latest"))?; + let latest = open(OpenOptions::new().read(true), format!("{PROCESS}/latest"))?; let ctid = unsafe { let mut status: ct_stathdl_t = std::ptr::null_mut(); check_errno( @@ -141,7 +136,10 @@ impl Template { ct_status_free(status); ctid }; - let ctl = open(OFlags::WRONLY, format!("{PROCESS}/{ctid}/ctl"))?; + let ctl = open( + OpenOptions::new().write(true), + format!("{PROCESS}/{ctid}/ctl"), + )?; check_errno("abandoning contract", unsafe { ct_ctl_abandon(ctl.as_raw_fd()) })?; From 18c0280276c348fb7fa136814fac487ea2cef724 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Mon, 26 Oct 2026 08:17:23 -0600 Subject: [PATCH 6/6] Probe: run Helios CI on helios-3.0 and print its OS build Scratch commit to learn whether the helios-3.0 image includes illumos 18395, so the ctfs workaround above can go. Co-authored-by: Claude --- .github/buildomat/jobs/ci-helios.sh | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/buildomat/jobs/ci-helios.sh b/.github/buildomat/jobs/ci-helios.sh index 4ca9c21d..b8129912 100644 --- a/.github/buildomat/jobs/ci-helios.sh +++ b/.github/buildomat/jobs/ci-helios.sh @@ -1,7 +1,7 @@ #!/bin/bash #: name = "Build Helios client / CI" #: variety = "basic" -#: target = "helios-2.0" +#: target = "helios-3.0" #: rust_toolchain = true #: output_rules = ["=/work/sush"] #: access_repos = ["oxidecomputer/permission-slip"] @@ -15,6 +15,7 @@ set -o errexit set -o pipefail set -o xtrace +uname -v cargo --version rustc --version