diff --git a/Cargo.lock b/Cargo.lock index 0182cdc7..0f8e3869 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2466,6 +2466,17 @@ version = "0.2.189" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" +[[package]] +name = "libcontract-sys" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29fff969872cff880a0a328a107528e3065b785df5fa8094fa1002cf0bf6c412" +dependencies = [ + "libc", + "num-derive", + "num-traits", +] + [[package]] name = "libm" version = "0.2.16" @@ -2770,6 +2781,17 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" +[[package]] +name = "num-derive" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "876a53fff98e03a936a674b29568b0e605f06b29372c2489ff4de23f1949743d" +dependencies = [ + "proc-macro2", + "quote", + "syn 1.0.109", +] + [[package]] name = "num-integer" version = "0.1.46" @@ -4895,6 +4917,7 @@ dependencies = [ "hyper", "hyper-util", "libc", + "libcontract-sys", "lru", "memmap2", "p256", diff --git a/Cargo.toml b/Cargo.toml index a1c4f945..30c5e683 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -35,6 +35,7 @@ hyper = { version = "1", features = ["client", "http1", "server"] } hyper-util = { version = "0.1", features = ["tokio"] } indicatif = "0.18" libc = "0.2" +libcontract-sys = "1" lru = "0.18" memmap2 = "0.9" p256 = { version = "0.13", features = ["ecdsa", "pkcs8"] } @@ -68,7 +69,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/Cargo.toml b/server/Cargo.toml index cb8bee95..9701844a 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -62,6 +62,9 @@ tokio-util.workspace = true tokio-tungstenite.workspace = true x509-cert.workspace = true +[target.'cfg(target_os = "illumos")'.dependencies] +libcontract-sys.workspace = true + [dev-dependencies] attest-mock.workspace = true ciborium.workspace = true diff --git a/server/src/contract.rs b/server/src/contract.rs new file mode 100644 index 00000000..c5e5d491 --- /dev/null +++ b/server/src/contract.rs @@ -0,0 +1,135 @@ +// 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/. + +//! Process contracts for jobs; see contract(5) and process(5). +//! +//! `sled-agent` runs in a contract held by `ctrun -o noorphan`, so any job +//! spawned into that contract dies with it. This breaks bare `reboot` as +//! a job: its `SIGTERM` broadcast kills `ctrun`, the contract is orphaned, +//! and the kernel kills `reboot` before it can call `uadmin`, leaving the +//! sled wedged until a power cycle. We therefore spawn each job into a new +//! contract and abandon it, so that jobs may outlive `sled-agent`. The +//! template settings and the abandonment follow what `sshd` does for each +//! login session, in OpenSSH's `openbsd-compat/port-solaris.c`. + +use std::ffi::{CString, c_int}; +use std::fs::File; +use std::io; +use std::os::fd::AsRawFd as _; +use std::ptr::null_mut; + +use libc::ctid_t; +use libcontract_sys::{ + CT_PR_EV_HWERR, CT_PR_PGRPONLY, CTD_COMMON, ct_ctl_abandon, ct_pr_tmpl_set_fatal, + ct_pr_tmpl_set_param, ct_pr_tmpl_set_svc_aux, ct_stathdl_t, ct_status_free, ct_status_get_id, + ct_status_read, ct_tmpl_activate, ct_tmpl_clear, ct_tmpl_set_critical, ct_tmpl_set_informative, +}; +use rustix::fs::{Mode, OFlags}; +use rustix::io::{Errno, FdFlags, fcntl_setfd}; +use rustix::process::Signal; + +use sush_common::jobs::JobId; + +use crate::error::check_errno; + +unsafe extern "C" { + fn sigsend(idtype: libc::idtype_t, id: libc::id_t, sig: c_int) -> c_int; +} + +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)) +} + +/// 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 +/// 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, +) -> io::Result<(T, ctid_t)> { + let template = Template::activate(job_id)?; + let child = spawn()?; + let ctid = template.abandon_latest()?; + Ok((child, ctid)) +} + +/// An active process contract template. While it lives, every child +/// the calling thread creates starts in a new contract. +struct Template(File); + +impl Template { + /// Activate a template for a job's contract on the calling thread. + /// 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 fd = file.as_raw_fd(); + let aux = CString::new(format!("sush job {job_id}")).map_err(io::Error::other)?; + unsafe { + check_errno( + "setting fatal events", + ct_pr_tmpl_set_fatal(fd, CT_PR_EV_HWERR), + )?; + 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_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 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: *mut ct_stathdl_t = null_mut(); + check_errno( + "reading latest contract", + ct_status_read(latest.as_raw_fd(), CTD_COMMON, &mut status), + )?; + let ctid = ct_status_get_id(status); + ct_status_free(status); + ctid + }; + let ctl = open(OFlags::WRONLY, format!("{PROCESS}/{ctid}/ctl"))?; + check_errno("abandoning contract", unsafe { + ct_ctl_abandon(ctl.as_raw_fd()) + })?; + Ok(ctid) + } +} + +impl Drop for Template { + fn drop(&mut self) { + unsafe { ct_tmpl_clear(self.0.as_raw_fd()) }; + } +} 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 9747a682..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; @@ -35,16 +31,32 @@ use sush_common::jobs::{ JobId, JobMode, JobOutputStream, JobStartRequest, ProcessError, VerifiedJob, }; +#[cfg(target_os = "illumos")] +use crate::contract::spawn_in_contract; use crate::io::JobIo; 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. @@ -253,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"; @@ -362,135 +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); - } - 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 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())); } + 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, + ]; - 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(()) - }); - } - - // 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!(cmd.spawn(), "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!(cmd.spawn(), "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() { @@ -528,6 +483,34 @@ async fn job_spawn(launch: Launch) { }); } +/// 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, spawn)?; + debug!(log, "spawned job"; "ctid" => ctid); + Ok(child.in_contract(ctid)) + } + #[cfg(not(target_os = "illumos"))] + { + let _ = (log, job_id); + spawn() + } +} + async fn send_error( log: &Logger, job_id: &JobId, @@ -546,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 09764952..20546ce3 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -11,6 +11,8 @@ extern crate function_name; compile_error!("`test-support` must not be enabled for an embedded server"); pub mod certs; +#[cfg(target_os = "illumos")] +pub mod contract; pub mod error; pub mod executor; pub mod format; @@ -29,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 3e2ce8fc..f0239e3b 100644 --- a/tests/src/manager_tests.rs +++ b/tests/src/manager_tests.rs @@ -270,6 +270,261 @@ async fn jobs() { ); } +/// A job runs in a contract of its own, labeled with its job ID. +#[cfg(target_os = "illumos")] +#[named] +#[tokio::test] +async fn job_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(); + + let job_id = session.next_job_id(); + // We abandon the contract after the spawn returns, so the job may + // see it held at first. + let command = r#" + ctid=$(ps -o ctid= -p $$) + fail() { echo "$1" >&2; ctstat -v -i $ctid >&2; exit 1; } + test $ctid != $(ps -o ctid= -p $PPID) || + fail "job shares contract $ctid with its parent" + for i in 1 2 3 4 5; do + state=$(ctstat -i $ctid | awk 'NR == 2 { print $4 }') + test "$state" = orphan && break + sleep 1 + done + test "$state" = orphan || fail "contract $ctid is $state, not orphan" + status=$(ctstat -v -i $ctid | sed 's/^[[:space:]]*//; s/:[[:space:]]*/: /') + for field in \ + "critical event set: none" \ + "informative event set: none" \ + "fatal event set: hwerr" \ + "parameter set: pgrponly" \ + "aux: sush job $SUSH_JOB_ID" + do + echo "$status" | grep -Fqx "$field" || + fail "contract $ctid lacks \"$field\"" + done"#; + 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 stderr = mgr + .job_output(&authn, &job_id, baseboard_id, Stderr, None) + .await + .unwrap() + .into_bytes() + .await; + assert!(stderr.is_empty(), "{}", String::from_utf8_lossy(&stderr)); + let status = mgr.job_status(&authn, &job_id).await.unwrap()[baseboard_id].clone(); + 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]