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 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 0bade71c..48546f7c 100644 --- a/server/src/contract.rs +++ b/server/src/contract.rs @@ -19,9 +19,12 @@ use std::io; use std::os::fd::AsRawFd as _; use libc::ctid_t; +use rustix::process::Signal; use sush_common::jobs::JobId; +use crate::error::check_errno; + #[allow(non_camel_case_types)] type ct_stathdl_t = *mut libc::c_void; @@ -40,6 +43,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; @@ -51,16 +58,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. fn open(options: &OpenOptions, path: String) -> io::Result { options @@ -68,11 +65,21 @@ fn open(options: &OpenOptions, path: String) -> io::Result { .map_err(|error| io::Error::new(error.kind(), format!("opening {path}: {error}"))) } +/// 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 -/// 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, @@ -83,8 +90,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 { @@ -99,29 +106,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(OpenOptions::new().read(true), 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), )?; @@ -133,7 +140,7 @@ impl Template { OpenOptions::new().write(true), 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..371c770b 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) + Ok(child.in_contract(ctid)) } #[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/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)] diff --git a/server/src/spawn.rs b/server/src/spawn.rs new file mode 100644 index 00000000..ff3b50db --- /dev/null +++ b/server/src/spawn.rs @@ -0,0 +1,336 @@ +// 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; + +#[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, + 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}; + +#[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. +#[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, + #[cfg(target_os = "illumos")] + contract: Option, + 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, + #[cfg(target_os = "illumos")] + contract: None, + sigchld, + reaped: None, + } + } + + /// 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)?) + } + + /// 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..f0239e3b 100644 --- a/tests/src/manager_tests.rs +++ b/tests/src/manager_tests.rs @@ -340,6 +340,191 @@ 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] +#[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]