From 483082edffc062a480c0cbf220752fd5032e3a83 Mon Sep 17 00:00:00 2001 From: sehkone Date: Sun, 27 Sep 2026 16:32:33 +0900 Subject: [PATCH 1/7] Add a bounded, killable run with stdin to Executor roxyd has to drive REview's core-update recovery through REview's own subcommands: one JSON request on stdin, one JSON reply on stdout, under a timeout, as root or as the REview service account. It must not grow a second, private way of running privileged commands beside Executor, so the capability lives here. run_with_input feeds the input, reads stdout and stderr each up to a limit, and kills the command at a deadline, on every transport. The bounds are enforced with poll(2) over non-blocking pipes rather than a reader thread per pipe, since a thread blocked in read cannot be cancelled once a surviving descendant holds a pipe open. A process this one may not signal - a command sudo started as root, or one on the far side of an SSH connection - is killed from the inside by a supervising shell, on the SIGTERM sudo relays and on a deadline of its own. Its cancellation is a SIGKILL to a bare sleep rather than a SIGTERM to a subshell, because dash drops a signal that lands before the subshell installs its trap. The errors get a type of their own so ExecutorError, which bootler matches exhaustively, is unchanged, and the trait method has a refusing default so the executors dependents implement in their tests still compile. Closes #129 --- Cargo.toml | 5 +- src/executor.rs | 1093 +++++++++++++++++++++++++++++++++- src/executor/bounded.rs | 710 ++++++++++++++++++++++ src/executor/test_support.rs | 347 +++++++++++ 4 files changed, 2142 insertions(+), 13 deletions(-) create mode 100644 src/executor/bounded.rs create mode 100644 src/executor/test_support.rs diff --git a/Cargo.toml b/Cargo.toml index e59cd92..a5154df 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,10 @@ hostname = "0.4" num-bigint = "0.4.5" rustls-pki-types = "1" rustls-webpki = { version = "0.103", default-features = false, features = ["aws-lc-rs", "std"] } -rustix = { version = "1", features = ["fs", "process"] } +# `event` adds `poll`, which `Executor::run_with_input` multiplexes the child's +# stdin, stdout and stderr through so each is bounded and the child can be +# killed at a deadline; std offers no readiness wait or read timeout on a pipe. +rustix = { version = "1", features = ["event", "fs", "process"] } serde = { version = "1", features = ["derive"] } serde_json = "1" sha2 = "0.10" diff --git a/src/executor.rs b/src/executor.rs index 6b17210..685f974 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -41,6 +41,14 @@ use std::time::Duration; use crate::durability::sync_dir; use crate::transport::{HostKeyPolicy, Ssh}; +// The bounded, killable run behind `Executor::run_with_input`, shared by every +// transport. +mod bounded; +// A scriptable, recording executor, for this crate's tests and for dependents +// that enable `test-support` under `[dev-dependencies]`. +#[cfg(any(test, feature = "test-support"))] +pub mod test_support; + /// The `sudo` binary name used unless a test overrides it. const SUDO: &str = "sudo"; /// Number of times a spawn is retried when the target reports `ETXTBSY` @@ -623,6 +631,102 @@ impl CommandOutput { } } +/// The bounds one [`Executor::run_with_input`] call runs under. +/// +/// There is no `Default`: how much output a command may write and how long it +/// may take are the caller's decisions about that command, not the executor's. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct RunLimits { + /// The most bytes the command may write to standard output. + pub max_stdout: usize, + /// The most bytes the command may write to standard error. + pub max_stderr: usize, + /// How long the command may run before it is killed. + pub timeout: Duration, +} + +/// One of a command's two output streams. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OutputStream { + /// Standard output. + Stdout, + /// Standard error. + Stderr, +} + +impl std::fmt::Display for OutputStream { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(match self { + OutputStream::Stdout => "stdout", + OutputStream::Stderr => "stderr", + }) + } +} + +/// Errors raised by [`Executor::run_with_input`]. +/// +/// A type of its own rather than more [`ExecutorError`] variants, because the +/// variants below arise from this one method alone: a caller of +/// [`Executor::run`] or any other primitive never has to consider them, and an +/// exhaustive match over [`ExecutorError`] elsewhere stays exhaustive. +#[derive(Debug, thiserror::Error)] +pub enum RunWithInputError { + /// `command` is not an absolute path, or contains `=`, and was refused + /// before anything was spawned. + /// + /// The command runs with no `PATH`, so only an absolute path names one + /// binary. `=` is refused because the command is started through + /// `env -i` wherever a supervisor runs it, and `env` reads an operand + /// containing `=` as a variable assignment rather than the utility. + #[error("command `{command}` is not an absolute path free of `=`")] + InvalidCommand { + /// The command as the caller named it. + command: String, + }, + /// The command wrote more than `limit` bytes to `stream`. It was killed + /// and waited for, or had already exited. + #[error("`{command}` wrote more than {limit} bytes to {stream}")] + OutputLimit { + /// The command that was run. + command: String, + /// The stream that passed its limit. + stream: OutputStream, + /// The limit it passed. + limit: usize, + }, + /// The command was still running when `timeout` passed. It was killed + /// and waited for. + #[error("`{command}` did not finish within {timeout:?}")] + TimedOut { + /// The command that was run. + command: String, + /// The timeout it outlived. + timeout: Duration, + }, + /// The executor does not implement [`Executor::run_with_input`]. + /// + /// Only the trait's default body returns this. [`LocalExecutor`], + /// [`SshExecutor`] and [`InDaemonExecutor`] each implement the method, as + /// does the `RecordingExecutor` the `test-support` feature exposes. + #[error("this executor cannot run a command with bounded input")] + Unsupported, + /// Spawning, the transport or elevation failed, exactly as + /// [`Executor::run`] reports it. + #[error(transparent)] + Executor(#[from] ExecutorError), +} + +/// Refuses a command [`Executor::run_with_input`] cannot run as named. +fn check_bounded_command(command: &str) -> Result<(), RunWithInputError> { + if Path::new(command).is_absolute() && !command.contains('=') { + Ok(()) + } else { + Err(RunWithInputError::InvalidCommand { + command: command.to_string(), + }) + } +} + /// A service account bootler creates and runs components under (RFC 0003 §6). /// /// This is a closed enum rather than a wrapped string, and that is the whole @@ -939,6 +1043,91 @@ pub trait Executor { args: &[&str], ) -> Result; + /// Runs `command` with `args` as `identity`, feeding it `input` on + /// standard input and holding it to `limits`. + /// + /// This is [`Executor::run`] for a command that is handed a request and + /// must answer within bounds: + /// + /// - **`command` must be an absolute path**, free of `=`. Anything else is + /// [`RunWithInputError::InvalidCommand`], refused before anything is + /// spawned. + /// - **`input` is written to the command's standard input**, which is then + /// closed; an empty `input` closes it at once. The bytes pass through + /// every transport verbatim. + /// - **The command runs with an empty environment** — no inherited + /// variable, and so no `PATH` lookup. Where `sudo` or an SSH session + /// stands between this process and the command, the environment either + /// of them sets up is cleared again before the command starts. + /// - **Standard output and standard error are each read up to their + /// limit.** One byte past either, or the command outliving + /// `limits.timeout`, kills the command and waits for it, and is reported + /// as [`RunWithInputError::OutputLimit`] or + /// [`RunWithInputError::TimedOut`] — never as a [`CommandOutput`]. + /// - **A non-zero exit is a [`CommandOutput`]**, as it is from + /// [`Executor::run`]. + /// + /// `identity` resolves exactly as it does for [`Executor::run`], at the + /// same resolution site: no prefix for [`Identity::Operator`], `sudo` for + /// [`Identity::Root`] and `sudo -u ` for [`Identity::Service`] on + /// the elevating transports, with [`SudoAuth`] governing both; and inside + /// the root daemon, no prefix for root, a `sudo -u` descent that never + /// prompts for a service account, and a refusal for the operator. + /// + /// **What "killed" reaches depends on who may signal whom**, so the + /// transports get there differently: + /// + /// - A command this process spawns directly — [`Identity::Operator`] on + /// [`LocalExecutor`], [`Identity::Root`] on [`InDaemonExecutor`] — runs + /// in a process group of its own, which is killed with `SIGKILL`. Every + /// descendant that stayed in the group dies with it. + /// - Where `sudo` stands in between, the command runs under a supervising + /// shell that kills the command's process group from the inside, on the + /// `SIGTERM` `sudo` relays to it and on a deadline of its own, so a + /// command that ignores `SIGTERM` still dies even where this process may + /// not signal it. Only after that is `sudo`'s own group killed. + /// - Over [`SshExecutor`], the local `ssh` process is killed at the + /// deadline. No signal crosses the connection, so the remote command is + /// ended by the same supervising shell on the remote host, at its own + /// deadline — `limits.timeout` rounded up to a whole second, counted + /// from when the remote side started — and this call does not wait to + /// see it happen. A command whose stream passes its limit is ended the + /// same way on the remote side: once `ssh` is gone its output pipes are + /// closed, and the supervisor's deadline ends it if writing does not. + /// + /// A descendant that leaves the process group — a daemon that calls + /// `setsid` — is out of reach on every transport. A supervised command + /// that a signal kills reports `128 + signal` as its exit code rather than + /// none, and starts with `SIGINT` and `SIGQUIT` ignored, as any command a + /// shell runs in the background does. + /// + /// The command's process group is also not the caller's, so a signal the + /// caller's terminal generates does not reach the command; `limits` is + /// what bounds it. + /// + /// The default body refuses with [`RunWithInputError::Unsupported`], so an + /// existing implementation of this trait keeps compiling; every executor + /// this crate ships overrides it. + /// + /// # Errors + /// + /// Returns [`RunWithInputError::InvalidCommand`] for a command that is not + /// an absolute path free of `=`, [`RunWithInputError::OutputLimit`] or + /// [`RunWithInputError::TimedOut`] for a command that was stopped, and + /// [`RunWithInputError::Executor`] carrying whatever [`Executor::run`] + /// would report for a spawn, transport or elevation failure. + fn run_with_input( + &self, + identity: Identity, + command: &str, + args: &[&str], + input: &[u8], + limits: RunLimits, + ) -> Result { + let _ = (identity, command, args, input, limits); + Err(RunWithInputError::Unsupported) + } + /// Writes `contents` to `dest` on the target with the owner, group and mode /// `meta` names (RFC 0003 §9.2). /// @@ -2091,6 +2280,22 @@ impl LocalExecutor { /// than elevating. The account name is passed as a discrete `Command` /// argument, so it reaches `sudo` as exactly one word whatever it contains. fn resolve(&self, identity: Identity, command: &str, args: &[&str]) -> Resolved { + self.resolve_through(identity, SH, &sudo_sentinel_script(), command, args) + } + + /// Resolves an `(identity, Local)` pair as [`LocalExecutor::resolve`] + /// does, with an elevated command run under `shell -c script` rather than + /// the sentinel script — the one knob [`Executor::run_with_input`] turns. + /// Whether `sudo` is involved, and with which flags and descent, is decided + /// here for both methods. + fn resolve_through( + &self, + identity: Identity, + shell: &str, + script: &str, + command: &str, + args: &[&str], + ) -> Resolved { let Some(elevation) = Elevation::of(identity) else { let mut cmd = Command::new(command); cmd.args(args); @@ -2114,11 +2319,7 @@ impl LocalExecutor { if let Elevation::Descend(account) = elevation { cmd.arg("-u").arg(account.as_str()); } - cmd.arg("sh") - .arg("-c") - .arg(sudo_sentinel_script()) - .arg(command) - .args(args); + cmd.arg(shell).arg("-c").arg(script).arg(command).args(args); Resolved { command: cmd, password_line, @@ -2157,6 +2358,58 @@ impl LocalExecutor { } } +impl LocalExecutor { + /// The local half of [`Executor::run_with_input`]: the command itself for + /// the operator, the supervisor under `sudo` otherwise. + fn run_bounded( + &self, + identity: Identity, + command: &str, + args: &[&str], + input: &[u8], + limits: RunLimits, + ) -> Result { + let script = bounded::supervisor_script(true, limits.timeout); + let Resolved { + command: mut cmd, + password_line, + elevated, + } = self.resolve_through(identity, bounded::SUPERVISOR_SHELL, &script, command, args); + // A command spawned directly has its environment cleared here. `sudo` + // keeps the caller's, so it is found exactly as `run` finds it; the + // supervisor clears the command's after `sudo` has set up its own. + if !elevated { + cmd.env_clear(); + } + let program = cmd.get_program().to_string_lossy().into_owned(); + let (framing, kill) = if elevated { + let framing = bounded::Framing { + supervised: true, + sentinel: true, + remote_code: false, + }; + (framing, bounded::Kill::Relay) + } else { + (bounded::Framing::DIRECT, bounded::Kill::Group) + }; + let feed = match password_line { + Some(mut line) => { + line.extend_from_slice(input); + line + } + None => input.to_vec(), + }; + let ended = bounded::run(cmd, &program, &feed, limits, framing, kill)?; + bounded::finish(ended, command, limits, framing, |output| { + if elevated { + classify_elevation(output, Some(&self.auth), &self.host) + } else { + Ok(output) + } + }) + } +} + /// A local invocation resolved from an identity: the command to spawn, the /// password line to feed ahead of any payload, and whether the sudo sentinel /// must be settled afterwards. @@ -2207,6 +2460,18 @@ impl Executor for LocalExecutor { self.spawn_resolved(self.resolve(identity, command, args), None) } + fn run_with_input( + &self, + identity: Identity, + command: &str, + args: &[&str], + input: &[u8], + limits: RunLimits, + ) -> Result { + check_bounded_command(command)?; + self.run_bounded(identity, command, args, input, limits) + } + fn put_file(&self, dest: &Path, contents: &[u8], meta: FileMeta) -> Result<(), ExecutorError> { // One elevated `sh -c`, so the sequence cannot be interleaved with // another elevation; the payload is fed on the script's stdin. @@ -2338,7 +2603,19 @@ impl SshExecutor { let mut cmd = self.ssh_command(); cmd.arg(wrap_with_rc_marker(remote)); let program = self.ssh_bin.to_string_lossy().into_owned(); - let mut output = spawn_capturing(cmd, &program, stdin)?; + let output = spawn_capturing(cmd, &program, stdin)?; + self.settle_remote_code(output) + } + + /// Replaces `output`'s exit code with the remote command's own, read from + /// the [`RC_MARKER`] line, and removes that line from its stderr. + /// + /// A missing marker means the wrapper never ran — the transport failed + /// before the command started — and is an [`ExecutorError::Connection`]. + fn settle_remote_code( + &self, + mut output: CommandOutput, + ) -> Result { let Some((pos, code)) = extract_remote_code(&output.stderr) else { return Err(ExecutorError::Connection { host: self.host.clone(), @@ -2371,16 +2648,41 @@ impl SshExecutor { /// the target's login shell re-parses each as exactly one word without /// re-splitting argument boundaries. fn resolve(&self, identity: Identity, command: &str, args: &[&str]) -> ResolvedRemote { - let payload = std::iter::once(command).chain(args.iter().copied()); + self.resolve_through(identity, SH, &sudo_sentinel_script(), None, command, args) + } + + /// Resolves an `(identity, Ssh)` pair as [`SshExecutor::resolve`] does, + /// with an elevated command run under `shell -c script` rather than the + /// sentinel script, and — where `operator_script` is given — the + /// operator's command run under `shell -c operator_script` rather than + /// bare. Those are the knobs [`Executor::run_with_input`] turns; whether + /// `sudo` is involved, and with which flags and descent, is decided here + /// for both methods. + fn resolve_through( + &self, + identity: Identity, + shell: &str, + script: &str, + operator_script: Option<&str>, + command: &str, + args: &[&str], + ) -> ResolvedRemote { let Some(elevation) = Elevation::of(identity) else { + let remote = match operator_script { + Some(wrapper) => shell_join( + [shell, "-c", wrapper, command] + .into_iter() + .chain(args.iter().copied()), + ), + None => shell_join(std::iter::once(command).chain(args.iter().copied())), + }; return ResolvedRemote { - remote: shell_join(payload), + remote, password_line: None, elevated: false, }; }; - let script = sudo_sentinel_script(); - let wrapped = ["sh", "-c", script.as_str(), command] + let wrapped = [shell, "-c", script, command] .into_iter() .chain(args.iter().copied()) .collect::>(); @@ -2435,6 +2737,62 @@ impl SshExecutor { } } +impl SshExecutor { + /// The SSH half of [`Executor::run_with_input`]: the supervisor on the + /// remote host for every identity, since no signal to the local `ssh` + /// reaches the remote command. + fn run_bounded( + &self, + identity: Identity, + command: &str, + args: &[&str], + input: &[u8], + limits: RunLimits, + ) -> Result { + let elevated_script = bounded::supervisor_script(true, limits.timeout); + let operator_script = bounded::supervisor_script(false, limits.timeout); + let ResolvedRemote { + remote, + password_line, + elevated, + } = self.resolve_through( + identity, + bounded::SUPERVISOR_SHELL, + &elevated_script, + Some(&operator_script), + command, + args, + ); + let feed = match password_line { + Some(mut line) => { + line.extend_from_slice(input); + line + } + None => input.to_vec(), + }; + let framing = bounded::Framing { + supervised: true, + sentinel: elevated, + remote_code: true, + }; + // `ssh` keeps the caller's environment: it needs `HOME` for its + // configuration and `SSH_AUTH_SOCK` for the agent. The remote + // supervisor clears the command's. + let mut cmd = self.ssh_command(); + cmd.arg(wrap_with_rc_marker(&remote)); + let program = self.ssh_bin.to_string_lossy().into_owned(); + let ended = bounded::run(cmd, &program, &feed, limits, framing, bounded::Kill::Group)?; + bounded::finish(ended, command, limits, framing, |output| { + let output = self.settle_remote_code(output)?; + if elevated { + classify_elevation(output, Some(&self.auth), &self.host) + } else { + Ok(output) + } + }) + } +} + /// A remote invocation resolved from an identity: the complete remote command /// line, the password line to feed ahead of any payload, and whether the sudo /// sentinel must be settled afterwards. @@ -2454,6 +2812,18 @@ impl Executor for SshExecutor { self.run_resolved(self.resolve(identity, command, args), None) } + fn run_with_input( + &self, + identity: Identity, + command: &str, + args: &[&str], + input: &[u8], + limits: RunLimits, + ) -> Result { + check_bounded_command(command)?; + self.run_bounded(identity, command, args, input, limits) + } + fn put_file(&self, dest: &Path, contents: &[u8], meta: FileMeta) -> Result<(), ExecutorError> { // The identical script the local transport runs, shell-quoted word by // word so the remote login shell re-parses each as exactly one word. @@ -2547,6 +2917,22 @@ impl InDaemonExecutor { identity: Identity, command: &str, args: &[&str], + ) -> Result<(Command, bool), ExecutorError> { + self.resolve_through(identity, SH, &sudo_sentinel_script(), command, args) + } + + /// Resolves an `(identity, InDaemon)` pair as [`InDaemonExecutor::resolve`] + /// does, with a descended command run under `shell -c script` rather than + /// the sentinel script — the one knob [`Executor::run_with_input`] turns. + /// Which identities refuse, run bare or descend is decided here for both + /// methods. + fn resolve_through( + &self, + identity: Identity, + shell: &str, + script: &str, + command: &str, + args: &[&str], ) -> Result<(Command, bool), ExecutorError> { match identity { Identity::Operator => Err(ExecutorError::NoOperatorIdentity { @@ -2561,9 +2947,9 @@ impl InDaemonExecutor { let mut cmd = Command::new(&self.sudo_bin); cmd.arg("-u") .arg(account.as_str()) - .arg("sh") + .arg(shell) .arg("-c") - .arg(sudo_sentinel_script()) + .arg(script) .arg(command) .args(args); Ok((cmd, true)) @@ -2589,6 +2975,50 @@ impl InDaemonExecutor { } } +impl InDaemonExecutor { + /// The in-daemon half of [`Executor::run_with_input`]: the command itself + /// for root, the supervisor under a `sudo -u` descent for a service + /// account. + fn run_bounded( + &self, + identity: Identity, + command: &str, + args: &[&str], + input: &[u8], + limits: RunLimits, + ) -> Result { + let script = bounded::supervisor_script(true, limits.timeout); + let (mut cmd, elevated) = + self.resolve_through(identity, bounded::SUPERVISOR_SHELL, &script, command, args)?; + // As on the local transport: root's command is cleared here, and a + // descended one by the supervisor, after `sudo` has run as it does + // for `run`. + if !elevated { + cmd.env_clear(); + } + let program = cmd.get_program().to_string_lossy().into_owned(); + let (framing, kill) = if elevated { + let framing = bounded::Framing { + supervised: true, + sentinel: true, + remote_code: false, + }; + (framing, bounded::Kill::Relay) + } else { + (bounded::Framing::DIRECT, bounded::Kill::Group) + }; + // No password line: descent from root never prompts. + let ended = bounded::run(cmd, &program, input, limits, framing, kill)?; + bounded::finish(ended, command, limits, framing, |output| { + if elevated { + classify_elevation(output, None, &self.host) + } else { + Ok(output) + } + }) + } +} + impl Executor for InDaemonExecutor { fn run( &self, @@ -2600,6 +3030,18 @@ impl Executor for InDaemonExecutor { self.spawn_resolved(cmd, elevated, None) } + fn run_with_input( + &self, + identity: Identity, + command: &str, + args: &[&str], + input: &[u8], + limits: RunLimits, + ) -> Result { + check_bounded_command(command)?; + self.run_bounded(identity, command, args, input, limits) + } + fn put_file(&self, dest: &Path, contents: &[u8], meta: FileMeta) -> Result<(), ExecutorError> { // The daemon already is root, so the sequence runs as direct syscalls: // no shell, no `sudo`, and metadata applied through the open descriptor @@ -6240,5 +6682,632 @@ exec sh -c "$script" _ "$source" "$dest""#; assert!(strays(artifact.parent().expect("dir")).is_empty()); } } + + /// [`Executor::run_with_input`] on every `(identity, transport)` pair + /// that runs a command, against stub `sudo` and `ssh` programs. + /// + /// The stubs keep elevation and the connection out of the picture + /// while leaving everything this method adds in it: the supervisor + /// really runs under the `sudo` stub and on the far side of the `ssh` + /// stub, so the empty environment, the verbatim stdin, the stream + /// accounting around the transport's markers and the kill are all + /// exercised on each path. + mod bounded_runs { + use std::path::{Path, PathBuf}; + use std::process::{Command, Stdio}; + use std::time::{Duration, Instant}; + + use rustix::process::Pid; + use tempfile::TempDir; + + use super::super::super::bounded::{TIMEOUT_MARKER, supervisor_script}; + use super::super::super::{ + Executor, ExecutorError, Identity, InDaemonExecutor, LocalExecutor, OutputStream, + RunLimits, RunWithInputError, ServiceAccount, SshExecutor, SshPrompt, SudoAuth, + }; + use super::{failing_ssh, fake_ssh, write_script}; + + /// Limits roomy enough for every test that is not about them. + const ROOMY: RunLimits = RunLimits { + max_stdout: 1 << 20, + max_stderr: 1 << 20, + timeout: Duration::from_secs(30), + }; + /// The largest request the first consumer sends. + const REQUEST_MAX: usize = 65_536; + /// A command that ignores `SIGTERM`, records its own pid and a + /// `SIGTERM`-ignoring child's in the file named by `$1`, writes + /// `$2` bytes of stdout, and then waits on the child forever. + const STUBBORN: &str = r#"trap '' TERM +/bin/sleep 300 & +echo "$!" > "$1.tmp" +echo "$$" >> "$1.tmp" +/bin/mv "$1.tmp" "$1" +/usr/bin/head -c "$2" /dev/zero +wait"#; + /// How long a killed process may take to disappear from the + /// process table, reaped by whoever inherited it. + const REAP_WAIT: Duration = Duration::from_secs(10); + + /// A `sudo` stub that drops its own flags, `-u ` + /// included, and execs the wrapped command. + fn descending_sudo(dir: &Path) -> PathBuf { + write_script( + dir, + "descending-sudo", + r#"#!/bin/sh +while [ "$#" -gt 0 ]; do + case "$1" in + -p|-u) shift 2 ;; + -n|-S) shift ;; + -*) shift ;; + *) break ;; + esac +done +exec "$@" +"#, + ) + } + + /// A `sudo` stub that insists on reading `s3cret` as its first + /// line of stdin, as `sudo -S` does, before execing the wrapped + /// command with the rest of the stream. + fn password_sudo(dir: &Path) -> PathBuf { + write_script( + dir, + "password-sudo", + r#"#!/bin/sh +IFS= read -r line +[ "$line" = s3cret ] || { echo "sudo: wrong password: $line" >&2; exit 1; } +while [ "$#" -gt 0 ]; do + case "$1" in + -p|-u) shift 2 ;; + -n|-S) shift ;; + -*) shift ;; + *) break ;; + esac +done +exec "$@" +"#, + ) + } + + fn ssh_with(dir: &Path, sudo: &Path, auth: SudoAuth) -> SshExecutor { + let config = crate::transport::Ssh { + user: "ops".to_string(), + port: 22, + key: PathBuf::from("/dev/null"), + host_key: crate::transport::HostKeyPolicy::Strict, + }; + SshExecutor::from_config("target", &config, "10.0.0.10", auth, SshPrompt::Deny) + .with_ssh_bin(fake_ssh(dir)) + .with_remote_sudo(sudo.to_string_lossy().into_owned()) + } + + /// Every pair that runs a command, labelled for failure messages. + fn every_pair(dir: &TempDir) -> Vec<(&'static str, Box, Identity)> { + let sudo = descending_sudo(dir.path()); + let service = Identity::Service(ServiceAccount::Security); + let local = || { + LocalExecutor::new("seat", SudoAuth::NonInteractive).with_sudo_bin(sudo.clone()) + }; + let ssh = || ssh_with(dir.path(), &sudo, SudoAuth::NonInteractive); + let daemon = || InDaemonExecutor::new("seat").with_sudo_bin(sudo.clone()); + vec![ + ("local operator", Box::new(local()), Identity::Operator), + ("local root", Box::new(local()), Identity::Root), + ("local service", Box::new(local()), service), + ("ssh operator", Box::new(ssh()), Identity::Operator), + ("ssh root", Box::new(ssh()), Identity::Root), + ("ssh service", Box::new(ssh()), service), + ("daemon root", Box::new(daemon()), Identity::Root), + ("daemon service", Box::new(daemon()), service), + ] + } + + /// Every byte value, repeated out to `len` bytes. + fn pattern(len: usize) -> Vec { + (0..=u8::MAX).cycle().take(len).collect() + } + + /// Reads the pids [`STUBBORN`] recorded. + fn recorded_pids(path: &Path) -> Vec { + let text = std::fs::read_to_string(path).expect("the command recorded its pids"); + text.lines() + .map(|line| { + let raw: i32 = line.trim().parse().expect("a pid"); + Pid::from_raw(raw).expect("a positive pid") + }) + .collect() + } + + /// Waits until no process with any of `pids` exists any more. + /// + /// A killed process lingers as a zombie until whoever inherited it + /// reaps it, and `kill(pid, 0)` still finds a zombie, so this + /// awaits the condition rather than checking it once. + fn assert_gone(label: &str, pids: &[Pid]) { + let deadline = Instant::now() + REAP_WAIT; + for &pid in pids { + while rustix::process::test_kill_process(pid).is_ok() { + assert!( + Instant::now() < deadline, + "{label}: process {pid:?} survived the kill" + ); + std::thread::sleep(Duration::from_millis(10)); + } + } + } + + #[test] + fn stdin_arrives_exactly_on_every_pair() { + let dir = tempfile::tempdir().expect("tempdir"); + for input in [ + Vec::new(), + pattern(REQUEST_MAX), + b"{\"op\":\"snapshot\"}\n".to_vec(), + ] { + for (label, exec, identity) in every_pair(&dir) { + let output = exec + .run_with_input(identity, "/bin/cat", &[], &input, ROOMY) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + assert_eq!(output.code, Some(0), "{label}"); + assert!( + output.stdout == input, + "{label}: {} bytes in, {} bytes out", + input.len(), + output.stdout.len() + ); + assert!(output.stderr.is_empty(), "{label}: {:?}", output.stderr); + } + } + } + + #[test] + fn a_password_line_is_consumed_before_the_input_on_the_elevating_transports() { + let dir = tempfile::tempdir().expect("tempdir"); + let sudo = password_sudo(dir.path()); + let auth = || SudoAuth::Password("s3cret".to_string()); + let input = pattern(REQUEST_MAX); + let pairs: Vec<(&str, Box)> = vec![ + ( + "local", + Box::new(LocalExecutor::new("seat", auth()).with_sudo_bin(sudo.clone())), + ), + ("ssh", Box::new(ssh_with(dir.path(), &sudo, auth()))), + ]; + for (label, exec) in pairs { + for identity in [Identity::Root, Identity::Service(ServiceAccount::Roxyd)] { + let output = exec + .run_with_input(identity, "/bin/cat", &[], &input, ROOMY) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + assert!( + output.stdout == input, + "{label}: the input must follow intact" + ); + } + } + } + + #[test] + fn the_command_sees_an_empty_environment_on_every_pair() { + let dir = tempfile::tempdir().expect("tempdir"); + assert!( + std::env::var_os("PATH").is_some(), + "the test process must have an environment to withhold" + ); + for (label, exec, identity) in every_pair(&dir) { + let output = exec + .run_with_input(identity, "/usr/bin/env", &[], b"", ROOMY) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + assert_eq!(output.code, Some(0), "{label}"); + assert_eq!( + String::from_utf8_lossy(&output.stdout), + "", + "{label}: no variable may reach the command" + ); + } + } + + #[test] + fn a_command_that_is_not_an_absolute_path_is_refused_before_spawning() { + let dir = tempfile::tempdir().expect("tempdir"); + let marker = dir.path().join("spawned"); + let marker_arg = marker.to_string_lossy().into_owned(); + for command in ["touch", "./touch", "/usr/bin/touch=x"] { + for (label, exec, identity) in every_pair(&dir) { + let error = exec + .run_with_input(identity, command, &[&marker_arg], b"", ROOMY) + .expect_err("the command must be refused"); + assert!( + matches!(&error, RunWithInputError::InvalidCommand { command: named } + if named == command), + "{label}: got {error:?}" + ); + assert!(!marker.exists(), "{label}: nothing may have run"); + } + } + // The refusal comes before the identity is looked at too. + let error = InDaemonExecutor::new("seat") + .run_with_input(Identity::Operator, "cat", &[], b"", ROOMY) + .expect_err("refused"); + assert!( + matches!(error, RunWithInputError::InvalidCommand { .. }), + "{error:?}" + ); + } + + #[test] + fn a_nonzero_exit_is_a_command_output_on_every_pair() { + let dir = tempfile::tempdir().expect("tempdir"); + for (label, exec, identity) in every_pair(&dir) { + let output = exec + .run_with_input( + identity, + "/bin/sh", + &["-c", "printf out; printf err >&2; exit 7"], + b"", + ROOMY, + ) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + assert_eq!(output.code, Some(7), "{label}"); + assert_eq!(output.stdout, b"out", "{label}"); + assert_eq!(output.stderr, b"err", "{label}: only the command's stderr"); + } + } + + #[test] + fn each_stream_may_reach_its_limit_but_not_pass_it_on_every_pair() { + const LIMIT: usize = 7; + let dir = tempfile::tempdir().expect("tempdir"); + let limits = RunLimits { + max_stdout: LIMIT, + max_stderr: LIMIT, + timeout: ROOMY.timeout, + }; + for (stream, script) in [ + (OutputStream::Stdout, "/bin/cat"), + (OutputStream::Stderr, "/bin/cat >&2"), + ] { + for (label, exec, identity) in every_pair(&dir) { + let at_limit = pattern(LIMIT); + let output = exec + .run_with_input(identity, "/bin/sh", &["-c", script], &at_limit, limits) + .unwrap_or_else(|error| panic!("{label} {stream}: {error:?}")); + let captured = match stream { + OutputStream::Stdout => &output.stdout, + OutputStream::Stderr => &output.stderr, + }; + assert_eq!(captured, &at_limit, "{label} {stream}"); + + let error = exec + .run_with_input( + identity, + "/bin/sh", + &["-c", script], + &pattern(LIMIT + 1), + limits, + ) + .expect_err("one byte over the limit is an error"); + match error { + RunWithInputError::OutputLimit { + command, + stream: over, + limit, + } => { + assert_eq!(command, "/bin/sh", "{label}"); + assert_eq!(over, stream, "{label}"); + assert_eq!(limit, LIMIT, "{label}"); + } + other => { + panic!("{label} {stream}: expected OutputLimit, got {other:?}") + } + } + } + } + } + + #[test] + fn a_breach_kills_a_command_that_ignores_sigterm_on_every_pair() { + let dir = tempfile::tempdir().expect("tempdir"); + let limits = RunLimits { + max_stdout: 16, + ..ROOMY + }; + for (index, (label, exec, identity)) in every_pair(&dir).into_iter().enumerate() { + let pids = dir.path().join(format!("breach-{index}")); + let started = Instant::now(); + let error = exec + .run_with_input( + identity, + "/bin/sh", + &["-c", STUBBORN, "sh", &pids.to_string_lossy(), "4096"], + b"", + limits, + ) + .expect_err("the flood must be stopped"); + assert!( + matches!( + error, + RunWithInputError::OutputLimit { + stream: OutputStream::Stdout, + .. + } + ), + "{label}: got {error:?}" + ); + assert!( + started.elapsed() < limits.timeout, + "{label}: the breach, not the timeout, must end the run" + ); + assert_gone(label, &recorded_pids(&pids)); + } + } + + #[test] + fn a_timeout_kills_a_command_that_ignores_sigterm_on_every_pair() { + let dir = tempfile::tempdir().expect("tempdir"); + let limits = RunLimits { + timeout: Duration::from_secs(2), + ..ROOMY + }; + for (index, (label, exec, identity)) in every_pair(&dir).into_iter().enumerate() { + let pids = dir.path().join(format!("timeout-{index}")); + let error = exec + .run_with_input( + identity, + "/bin/sh", + &["-c", STUBBORN, "sh", &pids.to_string_lossy(), "0"], + b"", + limits, + ) + .expect_err("the command outlives its timeout"); + match error { + RunWithInputError::TimedOut { command, timeout } => { + assert_eq!(command, "/bin/sh", "{label}"); + assert_eq!(timeout, limits.timeout, "{label}"); + } + other => panic!("{label}: expected TimedOut, got {other:?}"), + } + assert_gone(label, &recorded_pids(&pids)); + } + } + + #[test] + fn a_command_that_closes_its_streams_is_still_held_to_the_timeout() { + let dir = tempfile::tempdir().expect("tempdir"); + let limits = RunLimits { + timeout: Duration::from_millis(500), + ..ROOMY + }; + let pids = dir.path().join("closed"); + let script = format!( + "exec >/dev/null 2>&1 &2\nexit 1\n", + ); + let error = LocalExecutor::new("mgmt", SudoAuth::NonInteractive) + .with_sudo_bin(refusing.clone()) + .run_with_input(Identity::Root, "/bin/cat", &[], b"{}", ROOMY) + .expect_err("sudo refused"); + assert!( + matches!(&error, RunWithInputError::Executor(ExecutorError::Elevation { host }) + if host == "mgmt"), + "got {error:?}" + ); + let error = InDaemonExecutor::new("mgmt") + .with_sudo_bin(refusing) + .run_with_input( + Identity::Service(ServiceAccount::Security), + "/bin/cat", + &[], + b"{}", + ROOMY, + ) + .expect_err("sudo refused"); + assert!( + matches!( + error, + RunWithInputError::Executor(ExecutorError::SudoRefused { .. }) + ), + "got {error:?}" + ); + + let config = crate::transport::Ssh { + user: "ops".to_string(), + port: 22, + key: PathBuf::from("/dev/null"), + host_key: crate::transport::HostKeyPolicy::Strict, + }; + let error = SshExecutor::from_config( + "mgmt", + &config, + "10.0.0.10", + SudoAuth::NonInteractive, + SshPrompt::Deny, + ) + .with_ssh_bin(failing_ssh(dir.path())) + .run_with_input(Identity::Operator, "/bin/cat", &[], b"{}", ROOMY) + .expect_err("the host is unreachable"); + assert!( + matches!(&error, RunWithInputError::Executor(ExecutorError::Connection { host, .. }) + if host == "mgmt"), + "got {error:?}" + ); + + let error = InDaemonExecutor::new("mgmt") + .run_with_input(Identity::Operator, "/bin/cat", &[], b"{}", ROOMY) + .expect_err("no operator inside the daemon"); + assert!( + matches!( + error, + RunWithInputError::Executor(ExecutorError::NoOperatorIdentity { .. }) + ), + "got {error:?}" + ); + } + + /// Spawns the supervisor the way a transport this process cannot + /// signal through would run it, in a process group of its own so + /// its `kill 0` stays inside it, over [`STUBBORN`]. + fn spawn_supervised(timeout: Duration, pids: &Path) -> std::process::Child { + use std::os::unix::process::CommandExt; + + Command::new("/bin/sh") + .arg("-c") + .arg(supervisor_script(false, timeout)) + .args(["/bin/sh", "-c", STUBBORN, "sh"]) + .arg(pids) + .arg("0") + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .process_group(0) + .spawn() + .expect("spawn the supervisor") + } + + #[test] + fn the_supervisor_kills_the_command_at_its_own_deadline() { + // What ends a command on the far side of an SSH connection, + // where no signal from here arrives: nothing signals the + // supervisor, and its own deadline still kills everything. + let dir = tempfile::tempdir().expect("tempdir"); + let pids = dir.path().join("pids"); + let child = spawn_supervised(Duration::from_millis(200), &pids); + let output = child.wait_with_output().expect("the supervisor ends"); + assert!(!output.status.success()); + assert!( + String::from_utf8_lossy(&output.stderr).contains(TIMEOUT_MARKER), + "the deadline announces itself: {:?}", + output.stderr + ); + assert_gone("remote deadline", &recorded_pids(&pids)); + } + + #[test] + fn a_relayed_sigterm_makes_the_supervisor_kill_the_command() { + // What `sudo` relays when this process terminates it: a + // SIGTERM to the supervisor alone, which the command ignores, + // and which must still end the command. + let dir = tempfile::tempdir().expect("tempdir"); + let pids = dir.path().join("pids"); + let mut child = spawn_supervised(Duration::from_secs(300), &pids); + let deadline = Instant::now() + REAP_WAIT; + while !pids.exists() { + assert!(Instant::now() < deadline, "the command never started"); + std::thread::sleep(Duration::from_millis(10)); + } + let supervisor = Pid::from_child(&child); + rustix::process::kill_process(supervisor, rustix::process::Signal::TERM) + .expect("signal the supervisor"); + let status = child.wait().expect("the supervisor ends"); + assert!(!status.success()); + assert_gone("relayed SIGTERM", &recorded_pids(&pids)); + } + + #[test] + fn identities_resolve_through_sudo_exactly_as_run_resolves_them() { + // A stub `sudo` that prints its argv and grants, so the words + // ahead of the shell — the flags and the descent — can be + // compared between the two methods on each transport. + let dir = tempfile::tempdir().expect("tempdir"); + let recording = write_script( + dir.path(), + "recording-sudo", + &format!( + "#!/bin/sh\nfor arg in \"$@\"; do printf '%s\\n' \"$arg\"; done\n\ + printf '%s' '{}' >&2\n", + super::super::super::SUDO_OK_SENTINEL + ), + ); + let local = LocalExecutor::new("seat", SudoAuth::NonInteractive) + .with_sudo_bin(recording.clone()); + let ssh = ssh_with(dir.path(), &recording, SudoAuth::NonInteractive); + let daemon = InDaemonExecutor::new("seat").with_sudo_bin(recording.clone()); + let prefix = |argv: &[u8], shell: &str| -> Vec { + String::from_utf8_lossy(argv) + .lines() + .take_while(|word| *word != shell) + .map(str::to_string) + .collect() + }; + let pairs: Vec<(&str, &dyn Executor, Identity)> = vec![ + ("local root", &local, Identity::Root), + ( + "local service", + &local, + Identity::Service(ServiceAccount::Insight), + ), + ("ssh root", &ssh, Identity::Root), + ( + "ssh service", + &ssh, + Identity::Service(ServiceAccount::Insight), + ), + ( + "daemon service", + &daemon, + Identity::Service(ServiceAccount::Insight), + ), + ]; + for (label, exec, identity) in pairs { + let run = exec + .run(identity, "/usr/bin/printf", &["%s", "marker"]) + .expect("run"); + let bounded = exec + .run_with_input(identity, "/usr/bin/printf", &["%s", "marker"], b"", ROOMY) + .expect("run_with_input"); + let run_prefix = prefix(&run.stdout, "sh"); + assert!(!run_prefix.is_empty(), "{label}: sudo must be involved"); + assert_eq!( + prefix(&bounded.stdout, "/bin/sh"), + run_prefix, + "{label}: the elevation must match run's" + ); + } + // The identities that involve no `sudo` run the command bare. + for (label, exec, identity) in [ + ( + "local operator", + &local as &dyn Executor, + Identity::Operator, + ), + ("ssh operator", &ssh, Identity::Operator), + ("daemon root", &daemon, Identity::Root), + ] { + let bounded = exec + .run_with_input(identity, "/usr/bin/printf", &["%s", "marker"], b"", ROOMY) + .expect("run_with_input"); + assert_eq!( + bounded.stdout, b"marker", + "{label}: no sudo may be involved" + ); + } + } + } } } diff --git a/src/executor/bounded.rs b/src/executor/bounded.rs new file mode 100644 index 0000000..eaacb35 --- /dev/null +++ b/src/executor/bounded.rs @@ -0,0 +1,710 @@ +//! The bounded, killable run behind [`Executor::run_with_input`]. +//! +//! [`run`] is the engine every transport drives: it feeds the child's standard +//! input, reads its standard output and standard error each up to a limit, and +//! kills the child at a deadline. It is one thread multiplexing three pipes +//! through `poll(2)` rather than a thread per pipe, because a thread blocked in +//! `read` cannot be cancelled: a pipe some descendant of a killed child still +//! holds open would pin that thread, and the caller's scope with it. Here the +//! pipes are non-blocking and are simply dropped once the run is decided. +//! +//! What the engine cannot do by itself is reach a process it may not signal — +//! a command `sudo` started as root, or one running on the far side of an SSH +//! connection. Those transports run the command under [`supervisor_script`], +//! which kills the command's process group from inside, on the relayed signal +//! and on a deadline of its own. +//! +//! [`Executor::run_with_input`]: super::Executor::run_with_input + +use std::io::{ErrorKind, Read, Write}; +use std::os::unix::process::CommandExt; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +use rustix::event::{PollFd, PollFlags, Timespec}; +use rustix::io::Errno; +use rustix::process::{Pid, Signal, WaitId, WaitIdOptions}; + +use super::{ + CommandOutput, ExecutorError, OutputStream, RC_MARKER, RunLimits, RunWithInputError, + SUDO_OK_SENTINEL, spawn_retrying_text_busy, +}; + +/// Bytes read from one output pipe per readiness. +const READ_CHUNK: usize = 8192; +/// Interval at which the exit of a child whose pipes have all closed is polled. +/// `std` offers no timed wait on a child, and the pipes, the only thing +/// `poll(2)` can wait on, are gone by then. +const EXIT_POLL: Duration = Duration::from_millis(5); +/// How long a supervised child is given to die after the terminating signal +/// is relayed to it, before its process group is killed outright. +const RELAY_GRACE: Duration = Duration::from_secs(5); +/// How long a child is given to be reaped after `SIGKILL` reached its group. +const KILL_GRACE: Duration = Duration::from_secs(5); +/// Marker [`supervisor_script`]'s own deadline prints on stderr before it kills +/// the command, so a run the supervisor ended is reported as a timeout rather +/// than read as the command dying of a signal. +pub(super) const TIMEOUT_MARKER: &str = "__BOOTLER_TIMEOUT__"; +/// The shell [`supervisor_script`] runs under, named absolutely so the +/// invocation depends on no `PATH`. +pub(super) const SUPERVISOR_SHELL: &str = "/bin/sh"; +/// The `env` the supervisor clears the command's environment with. Its path is +/// the one fixed location both Linux and macOS guarantee. +const ENV: &str = "/usr/bin/env"; + +/// How a run that must stop early stops its child. +#[derive(Debug, Clone, Copy)] +pub(super) enum Kill { + /// `SIGKILL` the child's process group at once. Right where every process + /// in the group is this process's to signal: a command spawned directly, + /// and the local half of an SSH connection. + Group, + /// `SIGTERM` the child alone, give it [`RELAY_GRACE`] to die, then + /// `SIGKILL` its group. Right where the child is `sudo`: `sudo` relays the + /// `SIGTERM` to [`supervisor_script`], which kills the command from the + /// inside, and this process may not be able to signal the command at all. + Relay, +} + +/// What the invocation itself writes on stderr around the command's own +/// bytes, so the stderr limit is held against the command and not against the +/// transport's markers. +#[derive(Debug, Clone, Copy)] +pub(super) struct Framing { + /// [`supervisor_script`] runs the command, so [`TIMEOUT_MARKER`] may + /// appear. + pub(super) supervised: bool, + /// `sudo` runs the supervisor, which prints [`SUDO_OK_SENTINEL`] first. + pub(super) sentinel: bool, + /// The SSH wrapper appends the remote exit status after [`RC_MARKER`]. + pub(super) remote_code: bool, +} + +impl Framing { + /// A command spawned directly: every byte on stderr is its own. + pub(super) const DIRECT: Self = Self { + supervised: false, + sentinel: false, + remote_code: false, + }; + + /// Returns how many bytes of `stderr` the command itself wrote. + /// + /// A marker already seen in full is discounted, and so is a trailing + /// fragment that may still grow into one — a stream is read in chunks, and + /// a marker split across two of them must not count against the limit + /// while only its first half has arrived. Nothing else is discounted, so a + /// command that writes one byte past its limit is caught at that byte. + fn command_len(self, stderr: &[u8]) -> usize { + let mut framing = 0; + let mut tail = stderr; + let mut pending = 0; + if self.sentinel { + match find(stderr, SUDO_OK_SENTINEL.as_bytes()) { + Some(at) => { + framing += SUDO_OK_SENTINEL.len(); + tail = stderr + .get(at + SUDO_OK_SENTINEL.len()..) + .unwrap_or_default(); + } + None => pending = partial_suffix(stderr, SUDO_OK_SENTINEL.as_bytes()), + } + } + if self.supervised { + if find(tail, TIMEOUT_MARKER.as_bytes()).is_some() { + framing += TIMEOUT_MARKER.len(); + } else { + pending = pending.max(partial_suffix(tail, TIMEOUT_MARKER.as_bytes())); + } + } + if self.remote_code { + pending = pending.max(remote_code_suffix(tail)); + } + stderr.len().saturating_sub(framing + pending) + } +} + +/// How the engine's run ended. +#[derive(Debug)] +pub(super) enum Ended { + /// The child exited, or was killed by something other than this run, with + /// its output captured raw — any transport framing is still in it. + Exited(CommandOutput), + /// A stream passed its limit and the child was killed. + Breach(OutputStream), + /// The deadline passed and the child was killed. + TimedOut, +} + +/// Returns the `sh -c` script that supervises a command on a transport this +/// process cannot signal the command through. +/// +/// Invoked as `sh -c SCRIPT `, so the command and every +/// argument arrive positionally and are never spliced into the script text. +/// With `sentinel`, it first prints [`SUDO_OK_SENTINEL`], so a `sudo` that +/// refused is told apart from a command that failed exactly as +/// [`Executor::run`](super::Executor::run) tells them apart. +/// +/// - **The environment is cleared.** The command is started through +/// `env -i`, so nothing `sudo`'s environment reset or the SSH session put +/// back reaches it — no variable at all, and so no `PATH` either. +/// - **Only the command writes to standard error.** The script keeps the +/// real standard error aside for the command and for [`TIMEOUT_MARKER`], +/// and points its own at `/dev/null`: a shell reports a job a signal killed +/// — bash prints `Killed: 9` for the cancelled `sleep` below — and that +/// report is not the command's output. +/// - **Standard input is handed to the command.** A shell gives a command it +/// runs in the background `/dev/null` unless told otherwise, so the script +/// moves its standard input aside and redirects it back onto the command +/// explicitly. +/// - **The command runs in the background of the script**, because a shell +/// runs a trap only once the foreground command it is waiting on returns, +/// and a command that ignores `SIGTERM` never would. `wait` is interrupted +/// by a trapped signal, so the trap fires at once. +/// - **`SIGTERM`, `SIGHUP` and `SIGINT` kill the whole process group** with +/// `SIGKILL`. This is what `sudo` relays [`Kill::Relay`]'s `SIGTERM` to, and +/// the group is the command and all of its descendants that did not leave +/// it. +/// - **The deadline is enforced from the inside too**: a `sleep` of +/// `timeout`, rounded up to a whole second, started before the command. +/// The script waits on it; the subshell running the command kills it with +/// `SIGKILL` the moment the command exits. A `sleep` that ran out means the +/// deadline passed first, and the script prints [`TIMEOUT_MARKER`] and kills +/// the group. This is what ends the command on the far side of an SSH +/// connection, which a signal to the local `ssh` never reaches, and it is +/// the backstop wherever the relay fails to arrive. +/// +/// The cancellation is `SIGKILL` to a process that is nothing but `sleep` +/// deliberately. A cancelling `SIGTERM` to a subshell can be lost: one that +/// arrives after the fork but before the subshell has installed its own +/// trap is dropped by some shells, dash among them, and the deadline would +/// then fire over a command that finished long before. `SIGKILL` passes +/// through no trap, and `sleep`'s pid cannot be reused while it waits +/// unreaped for the script's `wait`. +/// +/// Two things differ from a command run directly. A command a signal killed +/// reports `128 + signal` rather than no exit code, since the script exits +/// with the status its subshell returned; and a command started in the +/// background of a shell without job control starts with `SIGINT` and +/// `SIGQUIT` ignored. +pub(super) fn supervisor_script(sentinel: bool, timeout: Duration) -> String { + let seconds = timeout + .as_secs() + .saturating_add(u64::from(timeout.subsec_nanos() > 0)); + let announce = if sentinel { + format!("printf '%s' '{SUDO_OK_SENTINEL}' >&2\n") + } else { + String::new() + }; + format!( + r#"trap 'kill -KILL 0' TERM HUP INT +{announce}exec 3<&0 0&2 2>/dev/null +sleep {seconds} /dev/null 3<&- 4>&- & +nap=$! +{{ + {ENV} -i "$0" "$@" 2>&4 4>&- + status=$? + kill -KILL "$nap" + exit "$status" +}} 0<&3 3<&- & +run=$! +exec 3<&- +if wait "$nap"; then + printf '%s' '{TIMEOUT_MARKER}' >&4 + kill -KILL 0 +fi +wait "$run""# + ) +} + +/// Runs `command` to completion under `limits`, feeding it `feed` on stdin. +/// +/// The child is placed in a process group of its own, so that killing it +/// reaches every descendant that did not leave the group, and so that a +/// supervisor's `kill 0` can never reach this process. `program` names the +/// binary for error reporting. +/// +/// A stream passing its limit — stderr measured by `framing` — or the deadline +/// passing kills the child by `kill` and waits for it; neither is an error +/// here, but an [`Ended`] the caller turns into one. +/// +/// # Errors +/// +/// Returns [`ExecutorError::Spawn`] when the child cannot be spawned, or when +/// feeding or reading it fails for a reason other than the child going away; +/// the child is killed first. +pub(super) fn run( + mut command: Command, + program: &str, + feed: &[u8], + limits: RunLimits, + framing: Framing, + kill: Kill, +) -> Result { + let failed = |source: std::io::Error| ExecutorError::Spawn { + command: program.to_string(), + source, + }; + command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .process_group(0); + let mut child = spawn_retrying_text_busy(&mut command).map_err(failed)?; + let deadline = Deadline::after(limits.timeout); + let mut pipes = Pipes { + stdin: child.stdin.take(), + stdout: child.stdout.take(), + stderr: child.stderr.take(), + }; + if let Err(source) = pipes.set_nonblocking() { + terminate(&mut child, kill); + return Err(failed(source)); + } + if feed.is_empty() { + pipes.stdin = None; + } + let pumped = pump(&mut pipes, feed, limits, framing, &deadline); + drop(pipes); + match pumped { + Ok(Pumped::Closed { stdout, stderr }) => { + await_exit(child, kill, &deadline, stdout, stderr).map_err(failed) + } + Ok(Pumped::Stopped(ended)) => { + terminate(&mut child, kill); + Ok(ended) + } + Err(source) => { + terminate(&mut child, kill); + Err(failed(source)) + } + } +} + +/// The child's three pipes, each dropped — closed — once it is finished with. +struct Pipes { + stdin: Option, + stdout: Option, + stderr: Option, +} + +impl Pipes { + /// Makes every pipe non-blocking, so a read or a write never waits past + /// what `poll(2)` reported ready and the deadline is always observed. + fn set_nonblocking(&self) -> std::io::Result<()> { + if let Some(pipe) = &self.stdin { + rustix::io::ioctl_fionbio(pipe, true)?; + } + if let Some(pipe) = &self.stdout { + rustix::io::ioctl_fionbio(pipe, true)?; + } + if let Some(pipe) = &self.stderr { + rustix::io::ioctl_fionbio(pipe, true)?; + } + Ok(()) + } + + fn is_empty(&self) -> bool { + self.stdin.is_none() && self.stdout.is_none() && self.stderr.is_none() + } +} + +/// What [`pump`] stopped on. +enum Pumped { + /// Every pipe closed, with what the child wrote on the two it reads. + Closed { stdout: Vec, stderr: Vec }, + /// The run must stop early; nothing it captured is returned. + Stopped(Ended), +} + +/// Moves bytes until every pipe has closed or the run must stop early. +fn pump( + pipes: &mut Pipes, + feed: &[u8], + limits: RunLimits, + framing: Framing, + deadline: &Deadline, +) -> std::io::Result { + let mut written = 0; + let mut stdout = Vec::new(); + let mut stderr = Vec::new(); + let mut chunk = vec![0; READ_CHUNK]; + while !pipes.is_empty() { + let Some(left) = deadline.remaining() else { + return Ok(Pumped::Stopped(Ended::TimedOut)); + }; + // A time left too large for a `timespec` is no bound at all. + let timeout = Timespec::try_from(left).ok(); + let ready = match wait_ready(pipes, timeout.as_ref()) { + Ok(ready) => ready, + Err(Errno::INTR) => continue, + Err(errno) => return Err(errno.into()), + }; + if ready.stdin + && let Some(pipe) = &mut pipes.stdin + { + match pipe.write(feed.get(written..).unwrap_or_default()) { + Ok(count) => { + written += count; + if written >= feed.len() { + pipes.stdin = None; + } + } + Err(error) if is_transient(&error) => {} + // The child closed its stdin without reading all of it — + // `sudo` refusing, or a command that needed less. What it + // wrote carries the story, so this is not a failure. + Err(error) if error.kind() == ErrorKind::BrokenPipe => pipes.stdin = None, + Err(error) => return Err(error), + } + } + if ready.stdout && drain(&mut pipes.stdout, &mut stdout, &mut chunk)? { + pipes.stdout = None; + } + if stdout.len() > limits.max_stdout { + return Ok(Pumped::Stopped(Ended::Breach(OutputStream::Stdout))); + } + if ready.stderr && drain(&mut pipes.stderr, &mut stderr, &mut chunk)? { + pipes.stderr = None; + } + if framing.command_len(&stderr) > limits.max_stderr { + return Ok(Pumped::Stopped(Ended::Breach(OutputStream::Stderr))); + } + } + Ok(Pumped::Closed { stdout, stderr }) +} + +/// Which pipes `poll(2)` reported ready. +struct Ready { + stdin: bool, + stdout: bool, + stderr: bool, +} + +/// Waits until at least one open pipe is ready or `timeout` passes. +fn wait_ready(pipes: &Pipes, timeout: Option<&Timespec>) -> Result { + let mut fds = Vec::with_capacity(3); + let mut slots = [None; 3]; + if let Some(pipe) = &pipes.stdin { + slots[0] = Some(fds.len()); + fds.push(PollFd::new(pipe, PollFlags::OUT)); + } + if let Some(pipe) = &pipes.stdout { + slots[1] = Some(fds.len()); + fds.push(PollFd::new(pipe, PollFlags::IN)); + } + if let Some(pipe) = &pipes.stderr { + slots[2] = Some(fds.len()); + fds.push(PollFd::new(pipe, PollFlags::IN)); + } + rustix::event::poll(&mut fds, timeout)?; + // A hang-up or an error is readiness too: the read or write it wakes + // reports what happened, and a pipe that only ever reported those would + // otherwise be polled forever. + let ready = |slot: Option| { + slot.and_then(|index| fds.get(index)) + .is_some_and(|fd| !fd.revents().is_empty()) + }; + Ok(Ready { + stdin: ready(slots[0]), + stdout: ready(slots[1]), + stderr: ready(slots[2]), + }) +} + +/// Reads what one ready output pipe holds into `into`, returning whether the +/// pipe reached end of file. +fn drain( + pipe: &mut Option, + into: &mut Vec, + chunk: &mut [u8], +) -> std::io::Result { + let Some(reader) = pipe else { + return Ok(false); + }; + match reader.read(chunk) { + Ok(0) => Ok(true), + Ok(count) => { + into.extend_from_slice(chunk.get(..count).unwrap_or_default()); + Ok(false) + } + Err(error) if is_transient(&error) => Ok(false), + Err(error) => Err(error), + } +} + +/// Reports whether an I/O error only means "not now". +fn is_transient(error: &std::io::Error) -> bool { + matches!(error.kind(), ErrorKind::WouldBlock | ErrorKind::Interrupted) +} + +/// Waits for a child whose pipes have all closed to exit, killing it at the +/// deadline. +/// +/// The pipes closing is not the child exiting: a command can close its +/// standard streams and keep running, and nothing then remains to `poll(2)` +/// on, so the exit is polled. +fn await_exit( + mut child: Child, + kill: Kill, + deadline: &Deadline, + stdout: Vec, + stderr: Vec, +) -> std::io::Result { + loop { + if let Some(status) = child.try_wait()? { + return Ok(Ended::Exited(CommandOutput { + code: status.code(), + stdout, + stderr, + })); + } + let Some(timeout) = deadline.remaining() else { + terminate(&mut child, kill); + return Ok(Ended::TimedOut); + }; + std::thread::sleep(timeout.min(EXIT_POLL)); + } +} + +/// Kills `child` and everything left in its process group, and reaps it. +/// +/// The group is killed even where the child has already exited, since a +/// descendant can outlive it; a group already empty is `ESRCH`, which is the +/// outcome wanted. The group is always killed **before** the child is reaped: +/// until then the child's pid, and so the group's id, cannot be reused, so the +/// `SIGKILL` cannot reach an unrelated group that took the number over. A +/// child this process may not signal — `sudo` running as root under an +/// unprivileged caller, whose relay did not end it within [`RELAY_GRACE`] — is +/// left unreaped after [`KILL_GRACE`] rather than waited on without bound. +fn terminate(child: &mut Child, kill: Kill) { + let pid = Pid::from_child(child); + if matches!(kill, Kill::Relay) { + let _ = rustix::process::kill_process(pid, Signal::TERM); + await_unreaped(pid, RELAY_GRACE); + } + let _ = rustix::process::kill_process_group(pid, Signal::KILL); + reap_within(child, KILL_GRACE); +} + +/// Waits up to `grace` for the child `pid` to exit, leaving it unreaped. +fn await_unreaped(pid: Pid, grace: Duration) { + let deadline = Deadline::after(grace); + loop { + let options = WaitIdOptions::EXITED | WaitIdOptions::NOHANG | WaitIdOptions::NOWAIT; + match rustix::process::waitid(WaitId::Pid(pid), options) { + Ok(None) | Err(Errno::INTR) => {} + // Exited and waiting to be reaped, or not a child to wait on. + Ok(Some(_)) | Err(_) => return, + } + let Some(left) = deadline.remaining() else { + return; + }; + std::thread::sleep(left.min(EXIT_POLL)); + } +} + +/// Waits up to `grace` for `child` to exit and reaps it. +fn reap_within(child: &mut Child, grace: Duration) { + let deadline = Deadline::after(grace); + // An error is a child already reaped, or not ours to reap: nothing more + // to wait for. + while let Ok(None) = child.try_wait() { + let Some(left) = deadline.remaining() else { + return; + }; + std::thread::sleep(left.min(EXIT_POLL)); + } +} + +/// When a run must end. `None` is a timeout too large to represent as an +/// instant, which never passes. +struct Deadline(Option); + +impl Deadline { + fn after(timeout: Duration) -> Self { + Self(Instant::now().checked_add(timeout)) + } + + /// Returns the time left, or `None` once the deadline has passed. + fn remaining(&self) -> Option { + match self.0 { + Some(at) => at + .checked_duration_since(Instant::now()) + .filter(|left| !left.is_zero()), + None => Some(Duration::MAX), + } + } +} + +/// Returns the position of the first occurrence of `needle` in `haystack`. +fn find(haystack: &[u8], needle: &[u8]) -> Option { + haystack + .windows(needle.len()) + .position(|window| window == needle) +} + +/// Reports whether `haystack` contains `needle`. +pub(super) fn contains(haystack: &[u8], needle: &str) -> bool { + find(haystack, needle.as_bytes()).is_some() +} + +/// Returns the length of the longest proper prefix of `marker` that `bytes` +/// ends with: a marker that may still be arriving. +fn partial_suffix(bytes: &[u8], marker: &[u8]) -> usize { + let longest = bytes.len().min(marker.len().saturating_sub(1)); + (1..=longest) + .rev() + .find(|&len| bytes.ends_with(marker.get(..len).unwrap_or_default())) + .unwrap_or(0) +} + +/// Returns the length of the trailing bytes of `bytes` that are, or may still +/// grow into, the SSH wrapper's exit-status line: a newline, [`RC_MARKER`], up +/// to three digits, and a newline. +fn remote_code_suffix(bytes: &[u8]) -> usize { + /// Digits in the largest exit status a shell reports, `255`. + const MAX_DIGITS: usize = 3; + let head_len = 1 + RC_MARKER.len(); + let longest = bytes.len().min(head_len + MAX_DIGITS + 1); + (1..=longest) + .rev() + .find(|&len| { + let tail = bytes.get(bytes.len() - len..).unwrap_or_default(); + is_remote_code_prefix(tail, MAX_DIGITS) + }) + .unwrap_or(0) +} + +/// Reports whether `tail` is a prefix of an exit-status line. +fn is_remote_code_prefix(tail: &[u8], max_digits: usize) -> bool { + let Some((&first, rest)) = tail.split_first() else { + return false; + }; + if first != b'\n' { + return false; + } + let marker = RC_MARKER.as_bytes(); + if rest.len() <= marker.len() { + return marker.starts_with(rest); + } + let Some(code) = rest.strip_prefix(marker) else { + return false; + }; + let digits = code.strip_suffix(b"\n").unwrap_or(code); + !digits.is_empty() + && digits.len() <= max_digits + && digits.iter().all(u8::is_ascii_digit) + && (digits.len() == code.len() || digits.len() + 1 == code.len()) +} + +/// Settles how the engine's run ended into the result +/// [`Executor::run_with_input`](super::Executor::run_with_input) returns. +/// +/// A run that exited is first checked for the supervisor's own deadline +/// having ended it, then handed to `settle` — the transport's framing, removed +/// exactly as [`Executor::run`](super::Executor::run) removes it — and what is +/// left is held against the limits once more, now with nothing but the +/// command's own bytes in it. +pub(super) fn finish( + ended: Ended, + command: &str, + limits: RunLimits, + framing: Framing, + settle: impl FnOnce(CommandOutput) -> Result, +) -> Result { + let output = match ended { + Ended::Exited(output) => output, + Ended::Breach(stream) => return Err(breach(command, stream, limits)), + Ended::TimedOut => return Err(timed_out(command, limits)), + }; + if framing.supervised && contains(&output.stderr, TIMEOUT_MARKER) { + return Err(timed_out(command, limits)); + } + let output = settle(output)?; + if output.stdout.len() > limits.max_stdout { + return Err(breach(command, OutputStream::Stdout, limits)); + } + if output.stderr.len() > limits.max_stderr { + return Err(breach(command, OutputStream::Stderr, limits)); + } + Ok(output) +} + +fn breach(command: &str, stream: OutputStream, limits: RunLimits) -> RunWithInputError { + RunWithInputError::OutputLimit { + command: command.to_string(), + stream, + limit: match stream { + OutputStream::Stdout => limits.max_stdout, + OutputStream::Stderr => limits.max_stderr, + }, + } +} + +fn timed_out(command: &str, limits: RunLimits) -> RunWithInputError { + RunWithInputError::TimedOut { + command: command.to_string(), + timeout: limits.timeout, + } +} + +#[cfg(test)] +mod tests { + use super::{Framing, TIMEOUT_MARKER, partial_suffix, remote_code_suffix}; + use crate::executor::{RC_MARKER, SUDO_OK_SENTINEL}; + + const SUDO_SSH: Framing = Framing { + supervised: true, + sentinel: true, + remote_code: true, + }; + + #[test] + fn a_direct_run_counts_every_stderr_byte() { + let stderr = format!("{SUDO_OK_SENTINEL}abc\n{RC_MARKER}0\n"); + assert_eq!(Framing::DIRECT.command_len(stderr.as_bytes()), stderr.len()); + } + + #[test] + fn the_sentinel_and_the_exit_status_line_are_not_the_commands() { + let stderr = format!("{SUDO_OK_SENTINEL}abc\n{RC_MARKER}255\n"); + assert_eq!(SUDO_SSH.command_len(stderr.as_bytes()), 3); + } + + #[test] + fn a_marker_still_arriving_is_not_counted_yet() { + // Half a sentinel, before the command has written anything. + let half = &SUDO_OK_SENTINEL[..7]; + assert_eq!(SUDO_SSH.command_len(half.as_bytes()), 0); + // Half an exit-status line after the command's own bytes. + let stderr = format!("{SUDO_OK_SENTINEL}abc\n{}", &RC_MARKER[..4]); + assert_eq!(SUDO_SSH.command_len(stderr.as_bytes()), 3); + let stderr = format!("{SUDO_OK_SENTINEL}abc\n{RC_MARKER}2"); + assert_eq!(SUDO_SSH.command_len(stderr.as_bytes()), 3); + let stderr = format!("{SUDO_OK_SENTINEL}abc{}", &TIMEOUT_MARKER[..5]); + assert_eq!(SUDO_SSH.command_len(stderr.as_bytes()), 3); + } + + #[test] + fn bytes_that_cannot_become_a_marker_are_counted() { + let stderr = format!("{SUDO_OK_SENTINEL}abc\n{RC_MARKER}2x"); + assert_eq!( + SUDO_SSH.command_len(stderr.as_bytes()), + stderr.len() - SUDO_OK_SENTINEL.len() + ); + let stderr = format!("{SUDO_OK_SENTINEL}abc\n{RC_MARKER}2555"); + assert_eq!( + SUDO_SSH.command_len(stderr.as_bytes()), + stderr.len() - SUDO_OK_SENTINEL.len() + ); + } + + #[test] + fn suffix_helpers_find_the_longest_candidate() { + assert_eq!(partial_suffix(b"xx__BOOT", b"__BOOTLER"), 6); + assert_eq!(partial_suffix(b"__BOOTLER", b"__BOOTLER"), 0); + assert_eq!(remote_code_suffix(b"abc\n"), 1); + assert_eq!(remote_code_suffix(b"abc"), 0); + } +} diff --git a/src/executor/test_support.rs b/src/executor/test_support.rs new file mode 100644 index 0000000..d4dbb0c --- /dev/null +++ b/src/executor/test_support.rs @@ -0,0 +1,347 @@ +//! A recording, scriptable [`Executor`] for tests. +//! +//! [`RecordingExecutor`] runs nothing. It records every call made to it — +//! the identity, the argument vector and, for +//! [`Executor::run_with_input`], the exact standard-input bytes and the +//! [`RunLimits`] — and answers each call with the next outcome a test +//! scripted for that method. A test of code built on the executor can then +//! assert what was asked for and drive every outcome the real transports +//! produce, including a stream passing its limit and a timeout, without a +//! process being spawned. +//! +//! It holds commands to the same rule the real transports do: a +//! [`Executor::run_with_input`] call naming a command that is not an absolute +//! path free of `=` is recorded and refused with +//! [`RunWithInputError::InvalidCommand`], and consumes no scripted outcome. +//! +//! This module is compiled only for this crate's tests and under the +//! `test-support` feature. Enable that feature in a dependent's +//! `[dev-dependencies]` only — never under `[dependencies]` — so none of it +//! reaches a release build. + +use std::collections::VecDeque; +use std::path::{Path, PathBuf}; +use std::sync::{Mutex, MutexGuard, PoisonError}; + +use super::{ + CommandOutput, Executor, ExecutorError, FileMeta, Identity, OutputStream, RunLimits, + RunWithInputError, check_bounded_command, +}; + +/// One call a [`RecordingExecutor`] received. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RecordedCall { + /// [`Executor::run`], including the calls the trait's default methods + /// make through it. + Run { + /// Who the command was to run as. + identity: Identity, + /// The command. + command: String, + /// Every argument, in order. + args: Vec, + }, + /// [`Executor::run_with_input`]. + RunWithInput { + /// Who the command was to run as. + identity: Identity, + /// The command. + command: String, + /// Every argument, in order. + args: Vec, + /// The bytes to be fed on standard input, exactly. + input: Vec, + /// The limits the command was to run under. + limits: RunLimits, + }, + /// [`Executor::put_file`]. + PutFile { + /// The destination. + dest: PathBuf, + /// The contents, exactly. + contents: Vec, + /// The owner, group and mode. + meta: FileMeta, + }, +} + +/// The outcome a test scripts for one [`Executor::run_with_input`] call. +#[derive(Debug)] +pub enum ScriptedRun { + /// The command ran and exited, with this output — a non-zero exit + /// included, as the real transports report one. + Output(CommandOutput), + /// The command wrote past its limit on this stream. Answered with + /// [`RunWithInputError::OutputLimit`] naming the call's command and the + /// call's limit for the stream. + OutputLimit(OutputStream), + /// The command outlived its timeout. Answered with + /// [`RunWithInputError::TimedOut`] naming the call's command and the + /// call's timeout. + TimedOut, + /// Any other error, returned as it is. + Error(RunWithInputError), +} + +/// An [`Executor`] that records every call and answers from a script. +/// +/// Scripted outcomes are consumed first in, first out, one queue per method. +/// [`Executor::put_file`] needs no script and always succeeds. +#[derive(Debug, Default)] +pub struct RecordingExecutor { + state: Mutex, +} + +#[derive(Debug, Default)] +struct State { + calls: Vec, + runs: VecDeque>, + bounded_runs: VecDeque, +} + +impl RecordingExecutor { + /// Creates an executor with nothing recorded and nothing scripted. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Queues the outcome of the next unanswered [`Executor::run`] call. + pub fn script_run(&self, outcome: Result) { + self.state().runs.push_back(outcome); + } + + /// Queues the outcome of the next unanswered + /// [`Executor::run_with_input`] call. + pub fn script_run_with_input(&self, outcome: ScriptedRun) { + self.state().bounded_runs.push_back(outcome); + } + + /// Returns every call received so far, in order. + #[must_use] + pub fn calls(&self) -> Vec { + self.state().calls.clone() + } + + /// A poisoned lock means a test already panicked while holding it; the + /// record is still the best account of what happened. + fn state(&self) -> MutexGuard<'_, State> { + self.state.lock().unwrap_or_else(PoisonError::into_inner) + } +} + +/// Copies an argument vector into the owned form a record keeps. +fn owned(args: &[&str]) -> Vec { + args.iter().map(|arg| (*arg).to_string()).collect() +} + +impl Executor for RecordingExecutor { + /// Records the call and returns the next scripted [`Executor::run`] + /// outcome. + /// + /// # Panics + /// + /// Panics when no outcome is scripted: the test did not anticipate the + /// call, and inventing an answer would hide that. + fn run( + &self, + identity: Identity, + command: &str, + args: &[&str], + ) -> Result { + let mut state = self.state(); + state.calls.push(RecordedCall::Run { + identity, + command: command.to_string(), + args: owned(args), + }); + match state.runs.pop_front() { + Some(outcome) => outcome, + None => panic!("RecordingExecutor: no outcome scripted for run `{command}` {args:?}"), + } + } + + /// Records the call and returns the next scripted + /// [`Executor::run_with_input`] outcome, after refusing a command the real + /// transports would refuse. + /// + /// # Panics + /// + /// Panics when no outcome is scripted for a command that is not refused: + /// the test did not anticipate the call, and inventing an answer would + /// hide that. + fn run_with_input( + &self, + identity: Identity, + command: &str, + args: &[&str], + input: &[u8], + limits: RunLimits, + ) -> Result { + let mut state = self.state(); + state.calls.push(RecordedCall::RunWithInput { + identity, + command: command.to_string(), + args: owned(args), + input: input.to_vec(), + limits, + }); + check_bounded_command(command)?; + let Some(outcome) = state.bounded_runs.pop_front() else { + panic!("RecordingExecutor: no outcome scripted for run_with_input `{command}` {args:?}") + }; + match outcome { + ScriptedRun::Output(output) => Ok(output), + ScriptedRun::OutputLimit(stream) => Err(RunWithInputError::OutputLimit { + command: command.to_string(), + stream, + limit: match stream { + OutputStream::Stdout => limits.max_stdout, + OutputStream::Stderr => limits.max_stderr, + }, + }), + ScriptedRun::TimedOut => Err(RunWithInputError::TimedOut { + command: command.to_string(), + timeout: limits.timeout, + }), + ScriptedRun::Error(error) => Err(error), + } + } + + fn put_file(&self, dest: &Path, contents: &[u8], meta: FileMeta) -> Result<(), ExecutorError> { + self.state().calls.push(RecordedCall::PutFile { + dest: dest.to_path_buf(), + contents: contents.to_vec(), + meta, + }); + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::{RecordedCall, RecordingExecutor, ScriptedRun}; + use crate::executor::{ + CommandOutput, Executor, ExecutorError, Identity, OutputStream, RunLimits, + RunWithInputError, ServiceAccount, + }; + + const LIMITS: RunLimits = RunLimits { + max_stdout: 65_536, + max_stderr: 4_096, + timeout: Duration::from_secs(30), + }; + + fn output(code: i32, stdout: &[u8]) -> CommandOutput { + CommandOutput { + code: Some(code), + stdout: stdout.to_vec(), + stderr: Vec::new(), + } + } + + #[test] + fn every_argument_of_a_bounded_run_is_recorded() { + let exec = RecordingExecutor::new(); + exec.script_run_with_input(ScriptedRun::Output(output(0, b"{}"))); + let input = vec![0xA5; 65_536]; + let identity = Identity::Service(ServiceAccount::Security); + let result = exec + .run_with_input( + identity, + "/usr/lib/review/review", + &["core-update", "restore"], + &input, + LIMITS, + ) + .expect("the scripted output"); + assert_eq!(result.stdout, b"{}"); + assert_eq!( + exec.calls(), + vec![RecordedCall::RunWithInput { + identity, + command: "/usr/lib/review/review".to_string(), + args: vec!["core-update".to_string(), "restore".to_string()], + input, + limits: LIMITS, + }] + ); + } + + #[test] + fn each_outcome_can_be_scripted() { + let exec = RecordingExecutor::new(); + exec.script_run_with_input(ScriptedRun::Output(output(3, b""))); + exec.script_run_with_input(ScriptedRun::OutputLimit(OutputStream::Stderr)); + exec.script_run_with_input(ScriptedRun::TimedOut); + exec.script_run_with_input(ScriptedRun::Error(RunWithInputError::Executor( + ExecutorError::Elevation { + host: "seat".to_string(), + }, + ))); + let call = || exec.run_with_input(Identity::Root, "/bin/snapshot", &[], b"", LIMITS); + + assert_eq!(call().expect("non-zero exit is an output").code, Some(3)); + match call() { + Err(RunWithInputError::OutputLimit { + command, + stream, + limit, + }) => { + assert_eq!(command, "/bin/snapshot"); + assert_eq!(stream, OutputStream::Stderr); + assert_eq!(limit, LIMITS.max_stderr); + } + other => panic!("expected OutputLimit, got {other:?}"), + } + match call() { + Err(RunWithInputError::TimedOut { command, timeout }) => { + assert_eq!(command, "/bin/snapshot"); + assert_eq!(timeout, LIMITS.timeout); + } + other => panic!("expected TimedOut, got {other:?}"), + } + assert!(matches!( + call(), + Err(RunWithInputError::Executor(ExecutorError::Elevation { .. })) + )); + } + + #[test] + fn a_relative_command_is_refused_without_consuming_the_script() { + let exec = RecordingExecutor::new(); + exec.script_run_with_input(ScriptedRun::TimedOut); + let error = exec + .run_with_input(Identity::Root, "review", &[], b"", LIMITS) + .expect_err("a relative command is refused"); + assert!( + matches!(error, RunWithInputError::InvalidCommand { ref command } if command == "review"), + "got: {error:?}" + ); + assert_eq!(exec.calls().len(), 1, "the refused call is still recorded"); + assert!(matches!( + exec.run_with_input(Identity::Root, "/bin/review", &[], b"", LIMITS), + Err(RunWithInputError::TimedOut { .. }) + )); + } + + #[test] + fn run_and_the_default_methods_are_recorded_and_scripted() { + let exec = RecordingExecutor::new(); + exec.script_run(Ok(output(0, b"bytes"))); + let read = exec + .fetch_file(Identity::Root, std::path::Path::new("/etc/x")) + .expect("the scripted read"); + assert_eq!(read, b"bytes"); + assert_eq!( + exec.calls(), + vec![RecordedCall::Run { + identity: Identity::Root, + command: "cat".to_string(), + args: vec!["/etc/x".to_string()], + }] + ); + } +} From 51442d2c5c352fa6753c177be2968e4fcb1e7d57 Mon Sep 17 00:00:00 2001 From: sehkone Date: Sun, 27 Sep 2026 16:36:51 +0900 Subject: [PATCH 2/7] Cap the supervisor's deadline at what sleep accepts macOS's sleep refuses a duration past i32::MAX seconds and exits at once. The supervisor read that as the deadline never arriving, so a very long timeout lost the remote backstop, and the cancelling kill went to a pid already reaped and free for reuse. Clamp the sleep to i32::MAX seconds, which is no bound in practice. Also name the recording executor among the test-support fixtures in the README and AGENTS.md. Part of #129 --- AGENTS.md | 5 +++-- README.md | 6 ++++-- src/executor.rs | 17 ++++++++++++++++ src/executor/bounded.rs | 43 +++++++++++++++++++++++++++++++++++++++-- 4 files changed, 65 insertions(+), 6 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 6c0abbc..f0c4884 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -489,8 +489,9 @@ Where the crate handles key material or secrets: `test-support` exists only to expose test-only fixtures across the crate boundary, since `cfg(test)` is invisible to dependents: the account fixtures (`Principal::Fixture`, `ServiceAccount::Fixture`), -payload fixtures, and the synthetic image archive builder in -`image::test_support`. Enable it in a dependent's `[dev-dependencies]` +payload fixtures, the synthetic image archive builder in +`image::test_support`, and the recording, scriptable executor in +`executor::test_support`. Enable it in a dependent's `[dev-dependencies]` only — never under `[dependencies]`, so the fixtures stay out of every release build. diff --git a/README.md b/README.md index ff07164..ef1376e 100644 --- a/README.md +++ b/README.md @@ -169,8 +169,10 @@ The `test-support` feature exposes test-only account fixtures (`Principal::Fixture` / `ServiceAccount::Fixture`), payload fixtures such as `payload::widen_envelope_blocks`, and `image::test_support` — a builder for synthetic image archives the image validator accepts, and a classifier that -holds any image archive against a declaration — so a **dependent** crate's -tests can construct them across the crate boundary. With +holds any image archive against a declaration — and +`executor::test_support::RecordingExecutor`, which records every call it +receives and answers each from a script, so a **dependent** crate's tests can +construct them across the crate boundary. With `package::prepare_sign_finalize` and a test key minted per test, those archives become genuinely signed format-6 packages that pass every final check; `tests/signed_fixtures.rs` builds one through the public API alone. diff --git a/src/executor.rs b/src/executor.rs index 685f974..e683726 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -7073,6 +7073,23 @@ exec "$@" } } + #[test] + fn an_unbounded_timeout_runs_the_command_to_completion_on_every_pair() { + let dir = tempfile::tempdir().expect("tempdir"); + let limits = RunLimits { + timeout: Duration::MAX, + ..ROOMY + }; + for (label, exec, identity) in every_pair(&dir) { + let output = exec + .run_with_input(identity, "/bin/cat", &[], b"{}", limits) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + assert_eq!(output.code, Some(0), "{label}"); + assert_eq!(output.stdout, b"{}", "{label}"); + assert!(output.stderr.is_empty(), "{label}: {:?}", output.stderr); + } + } + #[test] fn a_command_that_closes_its_streams_is_still_held_to_the_timeout() { let dir = tempfile::tempdir().expect("tempdir"); diff --git a/src/executor/bounded.rs b/src/executor/bounded.rs index eaacb35..c80ec72 100644 --- a/src/executor/bounded.rs +++ b/src/executor/bounded.rs @@ -51,6 +51,12 @@ pub(super) const SUPERVISOR_SHELL: &str = "/bin/sh"; /// The `env` the supervisor clears the command's environment with. Its path is /// the one fixed location both Linux and macOS guarantee. const ENV: &str = "/usr/bin/env"; +/// The longest deadline [`supervisor_script`] hands `sleep`, `i32::MAX` +/// seconds — some 68 years, which is no bound in practice. macOS's `sleep` +/// refuses anything longer and exits at once, which the script would read as +/// the command's deadline never arriving: the backstop would be gone, and the +/// cancelling `kill` would be sent to a pid already reaped and free for reuse. +const MAX_SLEEP_SECS: u64 = 2_147_483_647; /// How a run that must stop early stops its child. #[derive(Debug, Clone, Copy)] @@ -190,7 +196,8 @@ pub(super) enum Ended { pub(super) fn supervisor_script(sentinel: bool, timeout: Duration) -> String { let seconds = timeout .as_secs() - .saturating_add(u64::from(timeout.subsec_nanos() > 0)); + .saturating_add(u64::from(timeout.subsec_nanos() > 0)) + .min(MAX_SLEEP_SECS); let announce = if sentinel { format!("printf '%s' '{SUDO_OK_SENTINEL}' >&2\n") } else { @@ -651,7 +658,12 @@ fn timed_out(command: &str, limits: RunLimits) -> RunWithInputError { #[cfg(test)] mod tests { - use super::{Framing, TIMEOUT_MARKER, partial_suffix, remote_code_suffix}; + use std::time::Duration; + + use super::{ + Framing, MAX_SLEEP_SECS, TIMEOUT_MARKER, partial_suffix, remote_code_suffix, + supervisor_script, + }; use crate::executor::{RC_MARKER, SUDO_OK_SENTINEL}; const SUDO_SSH: Framing = Framing { @@ -700,6 +712,33 @@ mod tests { ); } + #[test] + fn the_supervisor_deadline_is_whole_seconds_rounded_up_and_capped() { + let sleeps = |timeout| { + let script = supervisor_script(false, timeout); + let line = script + .lines() + .find(|line| line.starts_with("sleep ")) + .expect("the deadline's sleep") + .to_string(); + line.split_whitespace() + .nth(1) + .expect("a duration") + .parse::() + .expect("whole seconds") + }; + assert_eq!(sleeps(Duration::from_secs(2)), 2); + assert_eq!(sleeps(Duration::from_millis(2001)), 3); + assert_eq!(sleeps(Duration::from_millis(1)), 1); + // A timeout past what every `sleep` accepts is held to the cap rather + // than handed on and refused. + assert_eq!( + sleeps(Duration::from_secs(MAX_SLEEP_SECS + 1)), + MAX_SLEEP_SECS + ); + assert_eq!(sleeps(Duration::MAX), MAX_SLEEP_SECS); + } + #[test] fn suffix_helpers_find_the_longest_candidate() { assert_eq!(partial_suffix(b"xx__BOOT", b"__BOOTLER"), 6); From 555a755540fdabc67c2963f450b0dc4629d9f5de Mon Sep 17 00:00:00 2001 From: sehkone Date: Sun, 27 Sep 2026 16:54:42 +0900 Subject: [PATCH 3/7] Test unread input and the default bounded body A command that exits without reading an input larger than a pipe buffer exercises the path where feeding it meets a closed pipe, which must still yield the command's own exit on every transport. The trait's default body is what keeps dependents' executors compiling, so its refusal is pinned too. Part of #129 --- src/executor.rs | 57 +++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 55 insertions(+), 2 deletions(-) diff --git a/src/executor.rs b/src/executor.rs index e683726..3395c3e 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -6702,8 +6702,9 @@ exec sh -c "$script" _ "$source" "$dest""#; use super::super::super::bounded::{TIMEOUT_MARKER, supervisor_script}; use super::super::super::{ - Executor, ExecutorError, Identity, InDaemonExecutor, LocalExecutor, OutputStream, - RunLimits, RunWithInputError, ServiceAccount, SshExecutor, SshPrompt, SudoAuth, + CommandOutput, Executor, ExecutorError, FileMeta, Identity, InDaemonExecutor, + LocalExecutor, OutputStream, RunLimits, RunWithInputError, ServiceAccount, + SshExecutor, SshPrompt, SudoAuth, }; use super::{failing_ssh, fake_ssh, write_script}; @@ -6956,6 +6957,58 @@ exec "$@" } } + #[test] + fn a_command_that_leaves_its_input_unread_still_reports_its_exit_on_every_pair() { + // More than a pipe buffer holds, so feeding it cannot finish + // before the command exits and the write meets a closed pipe. + let input = pattern(4 * REQUEST_MAX); + let dir = tempfile::tempdir().expect("tempdir"); + for (label, exec, identity) in every_pair(&dir) { + let output = exec + .run_with_input( + identity, + "/bin/sh", + &["-c", "printf done; exit 3"], + &input, + ROOMY, + ) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + assert_eq!(output.code, Some(3), "{label}"); + assert_eq!(output.stdout, b"done", "{label}"); + assert!(output.stderr.is_empty(), "{label}: {:?}", output.stderr); + } + } + + #[test] + fn an_executor_that_does_not_implement_it_refuses_as_unsupported() { + struct RunOnly; + impl Executor for RunOnly { + fn run( + &self, + _identity: Identity, + command: &str, + _args: &[&str], + ) -> Result { + panic!("`{command}` must not be run through `run`") + } + fn put_file( + &self, + dest: &Path, + _contents: &[u8], + _meta: FileMeta, + ) -> Result<(), ExecutorError> { + panic!("`{}` must not be written", dest.display()) + } + } + let error = RunOnly + .run_with_input(Identity::Root, "/bin/cat", &[], b"{}", ROOMY) + .expect_err("the default body refuses"); + assert!( + matches!(error, RunWithInputError::Unsupported), + "got {error:?}" + ); + } + #[test] fn each_stream_may_reach_its_limit_but_not_pass_it_on_every_pair() { const LIMIT: usize = 7; From 80b2d7ef39f15aae8583369d036ed1326c1f7b05 Mon Sep 17 00:00:00 2001 From: sehkone Date: Sun, 27 Sep 2026 17:07:58 +0900 Subject: [PATCH 4/7] Name the supervisor's sleep by absolute path The supervisor names its shell and `env` absolutely so it depends on no `PATH`, but its deadline `sleep` was still looked up through one. A `PATH` that cannot resolve it makes `sleep` exit at once: the supervisor then loses the deadline that ends a command on the far side of an SSH connection, and its cancelling `kill` goes to a pid already reaped. The supervisor test now runs under a `PATH` holding no utility and fails without this change. Part of #129 --- src/executor.rs | 8 +++++++- src/executor/bounded.rs | 10 +++++++--- 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/src/executor.rs b/src/executor.rs index 3395c3e..98cd1f9 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -7242,11 +7242,17 @@ exec "$@" /// Spawns the supervisor the way a transport this process cannot /// signal through would run it, in a process group of its own so - /// its `kill 0` stays inside it, over [`STUBBORN`]. + /// its `kill 0` stays inside it, over [`STUBBORN`]. Its `PATH` + /// names only the directory holding `pids`, where no utility lives, + /// so its deadline is shown to depend on no `PATH` lookup — an empty + /// environment would not show it, since a shell then falls back to + /// a default `PATH` of its own. fn spawn_supervised(timeout: Duration, pids: &Path) -> std::process::Child { use std::os::unix::process::CommandExt; Command::new("/bin/sh") + .env_clear() + .env("PATH", pids.parent().expect("pids lives in a directory")) .arg("-c") .arg(supervisor_script(false, timeout)) .args(["/bin/sh", "-c", STUBBORN, "sh"]) diff --git a/src/executor/bounded.rs b/src/executor/bounded.rs index c80ec72..91bd33d 100644 --- a/src/executor/bounded.rs +++ b/src/executor/bounded.rs @@ -51,6 +51,10 @@ pub(super) const SUPERVISOR_SHELL: &str = "/bin/sh"; /// The `env` the supervisor clears the command's environment with. Its path is /// the one fixed location both Linux and macOS guarantee. const ENV: &str = "/usr/bin/env"; +/// The `sleep` the supervisor's deadline runs, named absolutely for the same +/// reason: one `PATH` could not resolve would exit at once, and the script +/// would lose its deadline exactly as it would to a `sleep` that refused it. +const SLEEP: &str = "/bin/sleep"; /// The longest deadline [`supervisor_script`] hands `sleep`, `i32::MAX` /// seconds — some 68 years, which is no bound in practice. macOS's `sleep` /// refuses anything longer and exits at once, which the script would read as @@ -206,7 +210,7 @@ pub(super) fn supervisor_script(sentinel: bool, timeout: Duration) -> String { format!( r#"trap 'kill -KILL 0' TERM HUP INT {announce}exec 3<&0 0&2 2>/dev/null -sleep {seconds} /dev/null 3<&- 4>&- & +{SLEEP} {seconds} /dev/null 3<&- 4>&- & nap=$! {{ {ENV} -i "$0" "$@" 2>&4 4>&- @@ -661,7 +665,7 @@ mod tests { use std::time::Duration; use super::{ - Framing, MAX_SLEEP_SECS, TIMEOUT_MARKER, partial_suffix, remote_code_suffix, + Framing, MAX_SLEEP_SECS, SLEEP, TIMEOUT_MARKER, partial_suffix, remote_code_suffix, supervisor_script, }; use crate::executor::{RC_MARKER, SUDO_OK_SENTINEL}; @@ -718,7 +722,7 @@ mod tests { let script = supervisor_script(false, timeout); let line = script .lines() - .find(|line| line.starts_with("sleep ")) + .find(|line| line.starts_with(&format!("{SLEEP} "))) .expect("the deadline's sleep") .to_string(); line.split_whitespace() From af7be84edb34c0adbf1f31ff9de6bec0cad7761f Mon Sep 17 00:00:00 2001 From: sehkone Date: Sun, 27 Sep 2026 17:21:29 +0900 Subject: [PATCH 5/7] Test that a complete timeout marker is not counted The stderr accounting discounts the supervisor's timeout marker once it has arrived in full, but only a half-arrived marker was under test, so a regression would have charged the marker's 19 bytes against max_stderr on every timed-out supervised run. Part of #129 --- src/executor/bounded.rs | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/executor/bounded.rs b/src/executor/bounded.rs index 91bd33d..4c70517 100644 --- a/src/executor/bounded.rs +++ b/src/executor/bounded.rs @@ -688,6 +688,14 @@ mod tests { assert_eq!(SUDO_SSH.command_len(stderr.as_bytes()), 3); } + #[test] + fn the_timeout_marker_is_not_the_commands() { + let stderr = format!("{SUDO_OK_SENTINEL}abc{TIMEOUT_MARKER}"); + assert_eq!(SUDO_SSH.command_len(stderr.as_bytes()), 3); + let stderr = format!("{SUDO_OK_SENTINEL}abc{TIMEOUT_MARKER}\n{RC_MARKER}124\n"); + assert_eq!(SUDO_SSH.command_len(stderr.as_bytes()), 3); + } + #[test] fn a_marker_still_arriving_is_not_counted_yet() { // Half a sentinel, before the command has written anything. From ef7be89459fec14ef7859ea00780e8aa6e3fbbf7 Mon Sep 17 00:00:00 2001 From: sehkone Date: Sun, 27 Sep 2026 17:40:33 +0900 Subject: [PATCH 6/7] Frame the supervisor's markers apart from output A command that printed the supervisor's fixed timeout marker and then exited normally was reported as timed out, and those bytes were not counted against its stderr limit. Each supervisor now draws a random nonce for its marker, so only that run's own deadline matches it. The stderr limit also applied before a transport failure was classified, so under a small limit a sudo refusal or an SSH connection diagnostic came back as an output-limit breach. Every supervisor now prints the start sentinel, including the SSH operator's. Stderr written before the sentinel is the transport's: it is held to a fixed cap instead of max_stderr and is left out of the returned output, and the failure classifies exactly as run classifies it. Part of #129 --- src/executor.rs | 267 ++++++++++++++++------ src/executor/bounded.rs | 481 +++++++++++++++++++++++++++++++--------- 2 files changed, 576 insertions(+), 172 deletions(-) diff --git a/src/executor.rs b/src/executor.rs index 98cd1f9..9654d2b 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -1064,6 +1064,10 @@ pub trait Executor { /// `limits.timeout`, kills the command and waits for it, and is reported /// as [`RunWithInputError::OutputLimit`] or /// [`RunWithInputError::TimedOut`] — never as a [`CommandOutput`]. + /// The limits bound the command's own bytes: what `sudo` or `ssh` + /// writes before the command starts is neither counted nor returned, + /// so a refusal or a failed connection is reported as the error + /// [`Executor::run`] reports for it however small `max_stderr` is. /// - **A non-zero exit is a [`CommandOutput`]**, as it is from /// [`Executor::run`]. /// @@ -2369,12 +2373,18 @@ impl LocalExecutor { input: &[u8], limits: RunLimits, ) -> Result { - let script = bounded::supervisor_script(true, limits.timeout); + let supervisor = bounded::Supervisor::new(limits.timeout)?; let Resolved { command: mut cmd, password_line, elevated, - } = self.resolve_through(identity, bounded::SUPERVISOR_SHELL, &script, command, args); + } = self.resolve_through( + identity, + bounded::SUPERVISOR_SHELL, + supervisor.script(), + command, + args, + ); // A command spawned directly has its environment cleared here. `sudo` // keeps the caller's, so it is found exactly as `run` finds it; the // supervisor clears the command's after `sudo` has set up its own. @@ -2383,12 +2393,7 @@ impl LocalExecutor { } let program = cmd.get_program().to_string_lossy().into_owned(); let (framing, kill) = if elevated { - let framing = bounded::Framing { - supervised: true, - sentinel: true, - remote_code: false, - }; - (framing, bounded::Kill::Relay) + (supervisor.framing(false), bounded::Kill::Relay) } else { (bounded::Framing::DIRECT, bounded::Kill::Group) }; @@ -2749,8 +2754,7 @@ impl SshExecutor { input: &[u8], limits: RunLimits, ) -> Result { - let elevated_script = bounded::supervisor_script(true, limits.timeout); - let operator_script = bounded::supervisor_script(false, limits.timeout); + let supervisor = bounded::Supervisor::new(limits.timeout)?; let ResolvedRemote { remote, password_line, @@ -2758,8 +2762,8 @@ impl SshExecutor { } = self.resolve_through( identity, bounded::SUPERVISOR_SHELL, - &elevated_script, - Some(&operator_script), + supervisor.script(), + Some(supervisor.script()), command, args, ); @@ -2770,11 +2774,7 @@ impl SshExecutor { } None => input.to_vec(), }; - let framing = bounded::Framing { - supervised: true, - sentinel: elevated, - remote_code: true, - }; + let framing = supervisor.framing(true); // `ssh` keeps the caller's environment: it needs `HOME` for its // configuration and `SSH_AUTH_SOCK` for the agent. The remote // supervisor clears the command's. @@ -2783,10 +2783,14 @@ impl SshExecutor { let program = self.ssh_bin.to_string_lossy().into_owned(); let ended = bounded::run(cmd, &program, &feed, limits, framing, bounded::Kill::Group)?; bounded::finish(ended, command, limits, framing, |output| { - let output = self.settle_remote_code(output)?; + let mut output = self.settle_remote_code(output)?; if elevated { classify_elevation(output, Some(&self.auth), &self.host) } else { + // The operator's supervisor announces itself too; with no + // `sudo` to have refused, its absence is only a supervisor + // that never ran, and the exit status says why. + take_sudo_sentinel(&mut output.stderr); Ok(output) } }) @@ -2987,9 +2991,14 @@ impl InDaemonExecutor { input: &[u8], limits: RunLimits, ) -> Result { - let script = bounded::supervisor_script(true, limits.timeout); - let (mut cmd, elevated) = - self.resolve_through(identity, bounded::SUPERVISOR_SHELL, &script, command, args)?; + let supervisor = bounded::Supervisor::new(limits.timeout)?; + let (mut cmd, elevated) = self.resolve_through( + identity, + bounded::SUPERVISOR_SHELL, + supervisor.script(), + command, + args, + )?; // As on the local transport: root's command is cleared here, and a // descended one by the supervisor, after `sudo` has run as it does // for `run`. @@ -2998,12 +3007,7 @@ impl InDaemonExecutor { } let program = cmd.get_program().to_string_lossy().into_owned(); let (framing, kill) = if elevated { - let framing = bounded::Framing { - supervised: true, - sentinel: true, - remote_code: false, - }; - (framing, bounded::Kill::Relay) + (supervisor.framing(false), bounded::Kill::Relay) } else { (bounded::Framing::DIRECT, bounded::Kill::Group) }; @@ -6700,7 +6704,7 @@ exec sh -c "$script" _ "$source" "$dest""#; use rustix::process::Pid; use tempfile::TempDir; - use super::super::super::bounded::{TIMEOUT_MARKER, supervisor_script}; + use super::super::super::bounded::Supervisor; use super::super::super::{ CommandOutput, Executor, ExecutorError, FileMeta, Identity, InDaemonExecutor, LocalExecutor, OutputStream, RunLimits, RunWithInputError, ServiceAccount, @@ -6957,6 +6961,46 @@ exec "$@" } } + #[test] + fn a_command_that_prints_a_timeout_marker_still_exits_on_every_pair() { + // The fixed marker earlier revisions announced a timeout with, + // and one shaped like the current marker with another run's + // nonce: a command that prints either and exits has exited, + // and each byte counts against its stderr limit. + let printed = "__BOOTLER_TIMEOUT__\ + __BOOTLER_TIMEOUT_00112233445566778899aabbccddeeff__"; + let exact = RunLimits { + max_stderr: printed.len(), + ..ROOMY + }; + let under = RunLimits { + max_stderr: printed.len() - 1, + ..ROOMY + }; + let dir = tempfile::tempdir().expect("tempdir"); + for (label, exec, identity) in every_pair(&dir) { + let args = ["-c", "printf '%s' \"$0\" >&2", printed]; + let output = exec + .run_with_input(identity, "/bin/sh", &args, b"", exact) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + assert_eq!(output.code, Some(0), "{label}"); + assert_eq!(output.stderr, printed.as_bytes(), "{label}"); + let error = exec + .run_with_input(identity, "/bin/sh", &args, b"", under) + .expect_err("one byte over"); + assert!( + matches!( + error, + RunWithInputError::OutputLimit { + stream: OutputStream::Stderr, + .. + } + ), + "{label}: {error:?}" + ); + } + } + #[test] fn a_command_that_leaves_its_input_unread_still_reports_its_exit_on_every_pair() { // More than a pipe buffer holds, so feeding it cannot finish @@ -7171,62 +7215,117 @@ exec "$@" assert_gone("closed streams", &recorded_pids(&pids)); } + // One table of failures across every transport, each checked under + // both limits; split up, the shared stubs would be built per part. + #[allow(clippy::too_many_lines)] #[test] fn elevation_and_transport_failures_are_reported_as_run_reports_them() { + // Under no stderr allowance at all, too: what `sudo` or `ssh` + // writes when it fails is not the command's output, so it + // cannot pass the command's limit before it is classified. + let tight = RunLimits { + max_stdout: 0, + max_stderr: 0, + ..ROOMY + }; let dir = tempfile::tempdir().expect("tempdir"); let refusing = write_script( dir.path(), "refusing-sudo", "#!/bin/sh\necho 'sudo: a password is required' >&2\nexit 1\n", ); - let error = LocalExecutor::new("mgmt", SudoAuth::NonInteractive) - .with_sudo_bin(refusing.clone()) - .run_with_input(Identity::Root, "/bin/cat", &[], b"{}", ROOMY) - .expect_err("sudo refused"); - assert!( - matches!(&error, RunWithInputError::Executor(ExecutorError::Elevation { host }) - if host == "mgmt"), - "got {error:?}" - ); - let error = InDaemonExecutor::new("mgmt") - .with_sudo_bin(refusing) - .run_with_input( - Identity::Service(ServiceAccount::Security), - "/bin/cat", - &[], - b"{}", - ROOMY, - ) - .expect_err("sudo refused"); - assert!( - matches!( - error, - RunWithInputError::Executor(ExecutorError::SudoRefused { .. }) - ), - "got {error:?}" + let denying = write_script( + dir.path(), + "denying-sudo", + "#!/bin/sh\necho 'ops is not in the sudoers file.' >&2\nexit 1\n", ); - let config = crate::transport::Ssh { user: "ops".to_string(), port: 22, key: PathBuf::from("/dev/null"), host_key: crate::transport::HostKeyPolicy::Strict, }; - let error = SshExecutor::from_config( + let unreachable = SshExecutor::from_config( "mgmt", &config, "10.0.0.10", SudoAuth::NonInteractive, SshPrompt::Deny, ) - .with_ssh_bin(failing_ssh(dir.path())) - .run_with_input(Identity::Operator, "/bin/cat", &[], b"{}", ROOMY) - .expect_err("the host is unreachable"); - assert!( - matches!(&error, RunWithInputError::Executor(ExecutorError::Connection { host, .. }) - if host == "mgmt"), - "got {error:?}" - ); + .with_ssh_bin(failing_ssh(dir.path())); + let remote_refusing = SshExecutor::from_config( + "mgmt", + &config, + "10.0.0.10", + SudoAuth::NonInteractive, + SshPrompt::Deny, + ) + .with_ssh_bin(fake_ssh(dir.path())) + .with_remote_sudo(refusing.to_string_lossy().into_owned()); + let password = password_sudo(dir.path()); + let wrong_password = + LocalExecutor::new("mgmt", SudoAuth::Password("wrong".to_string())) + .with_sudo_bin(password); + for limits in [ROOMY, tight] { + let error = LocalExecutor::new("mgmt", SudoAuth::NonInteractive) + .with_sudo_bin(refusing.clone()) + .run_with_input(Identity::Root, "/bin/cat", &[], b"{}", limits) + .expect_err("sudo refused"); + assert!( + matches!(&error, RunWithInputError::Executor(ExecutorError::Elevation { host }) + if host == "mgmt"), + "{limits:?}: got {error:?}" + ); + let error = wrong_password + .run_with_input(Identity::Root, "/bin/cat", &[], b"{}", limits) + .expect_err("sudo rejected the password"); + assert!( + matches!(&error, RunWithInputError::Executor(ExecutorError::SudoRefused { reason, .. }) + if reason.contains("wrong password")), + "{limits:?}: got {error:?}" + ); + let error = InDaemonExecutor::new("mgmt") + .with_sudo_bin(denying.clone()) + .run_with_input( + Identity::Service(ServiceAccount::Security), + "/bin/cat", + &[], + b"{}", + limits, + ) + .expect_err("sudo refused"); + assert!( + matches!(&error, RunWithInputError::Executor(ExecutorError::SudoRefused { reason, .. }) + if reason.contains("sudoers")), + "{limits:?}: got {error:?}" + ); + let error = unreachable + .run_with_input(Identity::Operator, "/bin/cat", &[], b"{}", limits) + .expect_err("the host is unreachable"); + assert!( + matches!(&error, RunWithInputError::Executor(ExecutorError::Connection { host, reason }) + if host == "mgmt" && reason.contains("Connection refused")), + "{limits:?}: got {error:?}" + ); + let error = unreachable + .run_with_input(Identity::Root, "/bin/cat", &[], b"{}", limits) + .expect_err("the host is unreachable"); + assert!( + matches!( + &error, + RunWithInputError::Executor(ExecutorError::Connection { .. }) + ), + "{limits:?}: got {error:?}" + ); + let error = remote_refusing + .run_with_input(Identity::Root, "/bin/cat", &[], b"{}", limits) + .expect_err("the remote sudo refused"); + assert!( + matches!(&error, RunWithInputError::Executor(ExecutorError::Elevation { host }) + if host == "mgmt"), + "{limits:?}: got {error:?}" + ); + } let error = InDaemonExecutor::new("mgmt") .run_with_input(Identity::Operator, "/bin/cat", &[], b"{}", ROOMY) @@ -7240,6 +7339,36 @@ exec "$@" ); } + #[test] + fn a_transport_that_floods_stderr_before_the_command_starts_is_refused() { + // A `sudo` that never grants, writes far more than any real + // diagnostic, and then hangs: the run is abandoned well before + // the timeout, and still classifies as the refusal it is. + let dir = tempfile::tempdir().expect("tempdir"); + let flooding = write_script( + dir.path(), + "flooding-sudo", + "#!/bin/sh\n/usr/bin/head -c 1048576 /dev/zero | /usr/bin/tr '\\0' x >&2\n\ + exec /bin/sleep 300\n", + ); + let started = Instant::now(); + let error = LocalExecutor::new("mgmt", SudoAuth::NonInteractive) + .with_sudo_bin(flooding) + .run_with_input(Identity::Root, "/bin/cat", &[], b"", ROOMY) + .expect_err("sudo never granted"); + assert!( + matches!( + error, + RunWithInputError::Executor(ExecutorError::SudoRefused { .. }) + ), + "got {error:?}" + ); + assert!( + started.elapsed() < ROOMY.timeout, + "abandoned, not timed out" + ); + } + /// Spawns the supervisor the way a transport this process cannot /// signal through would run it, in a process group of its own so /// its `kill 0` stays inside it, over [`STUBBORN`]. Its `PATH` @@ -7247,14 +7376,14 @@ exec "$@" /// so its deadline is shown to depend on no `PATH` lookup — an empty /// environment would not show it, since a shell then falls back to /// a default `PATH` of its own. - fn spawn_supervised(timeout: Duration, pids: &Path) -> std::process::Child { + fn spawn_supervised(supervisor: &Supervisor, pids: &Path) -> std::process::Child { use std::os::unix::process::CommandExt; Command::new("/bin/sh") .env_clear() .env("PATH", pids.parent().expect("pids lives in a directory")) .arg("-c") - .arg(supervisor_script(false, timeout)) + .arg(supervisor.script()) .args(["/bin/sh", "-c", STUBBORN, "sh"]) .arg(pids) .arg("0") @@ -7273,11 +7402,12 @@ exec "$@" // supervisor, and its own deadline still kills everything. let dir = tempfile::tempdir().expect("tempdir"); let pids = dir.path().join("pids"); - let child = spawn_supervised(Duration::from_millis(200), &pids); + let supervisor = Supervisor::new(Duration::from_millis(200)).expect("a supervisor"); + let child = spawn_supervised(&supervisor, &pids); let output = child.wait_with_output().expect("the supervisor ends"); assert!(!output.status.success()); assert!( - String::from_utf8_lossy(&output.stderr).contains(TIMEOUT_MARKER), + String::from_utf8_lossy(&output.stderr).contains(supervisor.timeout_marker()), "the deadline announces itself: {:?}", output.stderr ); @@ -7291,7 +7421,8 @@ exec "$@" // and which must still end the command. let dir = tempfile::tempdir().expect("tempdir"); let pids = dir.path().join("pids"); - let mut child = spawn_supervised(Duration::from_secs(300), &pids); + let supervisor = Supervisor::new(Duration::from_secs(300)).expect("a supervisor"); + let mut child = spawn_supervised(&supervisor, &pids); let deadline = Instant::now() + REAP_WAIT; while !pids.exists() { assert!(Instant::now() < deadline, "the command never started"); diff --git a/src/executor/bounded.rs b/src/executor/bounded.rs index 4c70517..0507399 100644 --- a/src/executor/bounded.rs +++ b/src/executor/bounded.rs @@ -10,9 +10,9 @@ //! //! What the engine cannot do by itself is reach a process it may not signal — //! a command `sudo` started as root, or one running on the far side of an SSH -//! connection. Those transports run the command under [`supervisor_script`], -//! which kills the command's process group from inside, on the relayed signal -//! and on a deadline of its own. +//! connection. Those transports run the command under a [`Supervisor`], which +//! kills the command's process group from inside, on the relayed signal and on +//! a deadline of its own. //! //! [`Executor::run_with_input`]: super::Executor::run_with_input @@ -21,6 +21,7 @@ use std::os::unix::process::CommandExt; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; +use aws_lc_rs::rand::SecureRandom; use rustix::event::{PollFd, PollFlags, Timespec}; use rustix::io::Errno; use rustix::process::{Pid, Signal, WaitId, WaitIdOptions}; @@ -41,11 +42,20 @@ const EXIT_POLL: Duration = Duration::from_millis(5); const RELAY_GRACE: Duration = Duration::from_secs(5); /// How long a child is given to be reaped after `SIGKILL` reached its group. const KILL_GRACE: Duration = Duration::from_secs(5); -/// Marker [`supervisor_script`]'s own deadline prints on stderr before it kills -/// the command, so a run the supervisor ended is reported as a timeout rather -/// than read as the command dying of a signal. -pub(super) const TIMEOUT_MARKER: &str = "__BOOTLER_TIMEOUT__"; -/// The shell [`supervisor_script`] runs under, named absolutely so the +/// What opens every [`Supervisor`]'s timeout marker; the run's own random +/// nonce follows it. +const TIMEOUT_MARKER_PREFIX: &str = "__BOOTLER_TIMEOUT_"; +/// What closes every [`Supervisor`]'s timeout marker. +const TIMEOUT_MARKER_SUFFIX: &str = "__"; +/// Random bytes in a timeout marker's nonce. +const TIMEOUT_NONCE_LEN: usize = 16; +/// The most stderr a supervised run's transport may write before the command +/// starts — `sudo`'s refusal, `ssh`'s connection diagnostic. None of it is the +/// command's, so `max_stderr` does not bound it; this does, so that memory +/// stays bounded while the transport fails. Real diagnostics are a line or +/// two. +const TRANSPORT_STDERR_LIMIT: usize = 64 * 1024; +/// The shell a [`Supervisor`] script runs under, named absolutely so the /// invocation depends on no `PATH`. pub(super) const SUPERVISOR_SHELL: &str = "/bin/sh"; /// The `env` the supervisor clears the command's environment with. Its path is @@ -55,7 +65,7 @@ const ENV: &str = "/usr/bin/env"; /// reason: one `PATH` could not resolve would exit at once, and the script /// would lose its deadline exactly as it would to a `sleep` that refused it. const SLEEP: &str = "/bin/sleep"; -/// The longest deadline [`supervisor_script`] hands `sleep`, `i32::MAX` +/// The longest deadline a [`Supervisor`] script hands `sleep`, `i32::MAX` /// seconds — some 68 years, which is no bound in practice. macOS's `sleep` /// refuses anything longer and exits at once, which the script would read as /// the command's deadline never arriving: the backstop would be gone, and the @@ -71,66 +81,77 @@ pub(super) enum Kill { Group, /// `SIGTERM` the child alone, give it [`RELAY_GRACE`] to die, then /// `SIGKILL` its group. Right where the child is `sudo`: `sudo` relays the - /// `SIGTERM` to [`supervisor_script`], which kills the command from the + /// `SIGTERM` to the [`Supervisor`], which kills the command from the /// inside, and this process may not be able to signal the command at all. Relay, } /// What the invocation itself writes on stderr around the command's own /// bytes, so the stderr limit is held against the command and not against the -/// transport's markers. +/// transport. #[derive(Debug, Clone, Copy)] -pub(super) struct Framing { - /// [`supervisor_script`] runs the command, so [`TIMEOUT_MARKER`] may - /// appear. - pub(super) supervised: bool, - /// `sudo` runs the supervisor, which prints [`SUDO_OK_SENTINEL`] first. - pub(super) sentinel: bool, +pub(super) struct Framing<'a> { + /// The run's timeout marker where a [`Supervisor`] runs the command. The + /// supervisor prints [`SUDO_OK_SENTINEL`] before it starts the command, + /// so whatever precedes the sentinel is the transport's, and the marker + /// may follow the command's bytes. + pub(super) timeout_marker: Option<&'a str>, /// The SSH wrapper appends the remote exit status after [`RC_MARKER`]. pub(super) remote_code: bool, } -impl Framing { +/// Whose stderr a supervised run has written so far. +#[derive(Debug, PartialEq, Eq)] +enum Attributed { + /// The command has not started; this many bytes are the transport's. + Transport(usize), + /// The command has started and has written this many bytes of its own. + Command(usize), +} + +impl Framing<'static> { /// A command spawned directly: every byte on stderr is its own. pub(super) const DIRECT: Self = Self { - supervised: false, - sentinel: false, + timeout_marker: None, remote_code: false, }; +} - /// Returns how many bytes of `stderr` the command itself wrote. +impl Framing<'_> { + /// Attributes the bytes of `stderr` read so far. /// - /// A marker already seen in full is discounted, and so is a trailing - /// fragment that may still grow into one — a stream is read in chunks, and - /// a marker split across two of them must not count against the limit - /// while only its first half has arrived. Nothing else is discounted, so a - /// command that writes one byte past its limit is caught at that byte. - fn command_len(self, stderr: &[u8]) -> usize { + /// Until [`SUDO_OK_SENTINEL`] has arrived, nothing on a supervised run's + /// stderr is the command's: it is `sudo` refusing, `ssh` failing to + /// connect, or the sentinel itself still arriving, and it is left for the + /// transport's failure to be classified exactly as + /// [`Executor::run`](super::Executor::run) classifies it. After the + /// sentinel, a timeout marker already seen in full is discounted, and so + /// is a trailing fragment that may still grow into it or into the + /// exit-status line — a stream is read in chunks, and a marker split + /// across two of them must not count against the limit while only its + /// first half has arrived. Nothing else is discounted, so a command that + /// writes one byte past its limit is caught at that byte. + fn attribute(self, stderr: &[u8]) -> Attributed { + let Some(marker) = self.timeout_marker else { + return Attributed::Command(stderr.len()); + }; + let Some(at) = find(stderr, SUDO_OK_SENTINEL.as_bytes()) else { + return Attributed::Transport(stderr.len()); + }; + let tail = stderr + .get(at + SUDO_OK_SENTINEL.len()..) + .unwrap_or_default(); let mut framing = 0; - let mut tail = stderr; let mut pending = 0; - if self.sentinel { - match find(stderr, SUDO_OK_SENTINEL.as_bytes()) { - Some(at) => { - framing += SUDO_OK_SENTINEL.len(); - tail = stderr - .get(at + SUDO_OK_SENTINEL.len()..) - .unwrap_or_default(); - } - None => pending = partial_suffix(stderr, SUDO_OK_SENTINEL.as_bytes()), - } - } - if self.supervised { - if find(tail, TIMEOUT_MARKER.as_bytes()).is_some() { - framing += TIMEOUT_MARKER.len(); - } else { - pending = pending.max(partial_suffix(tail, TIMEOUT_MARKER.as_bytes())); - } + if find(tail, marker.as_bytes()).is_some() { + framing += marker.len(); + } else { + pending = partial_suffix(tail, marker.as_bytes()); } if self.remote_code { pending = pending.max(remote_code_suffix(tail)); } - stderr.len().saturating_sub(framing + pending) + Attributed::Command(tail.len().saturating_sub(framing + pending)) } } @@ -140,26 +161,34 @@ pub(super) enum Ended { /// The child exited, or was killed by something other than this run, with /// its output captured raw — any transport framing is still in it. Exited(CommandOutput), + /// The transport wrote more than [`TRANSPORT_STDERR_LIMIT`] before the + /// command started, and the child was killed. What it wrote is kept raw, + /// so the failure is classified as an exit would be. + Abandoned(CommandOutput), /// A stream passed its limit and the child was killed. Breach(OutputStream), /// The deadline passed and the child was killed. TimedOut, } -/// Returns the `sh -c` script that supervises a command on a transport this -/// process cannot signal the command through. +/// The `sh -c` script that supervises a command on a transport this process +/// cannot signal the command through, with the timeout marker of its one run. /// /// Invoked as `sh -c SCRIPT `, so the command and every /// argument arrive positionally and are never spliced into the script text. -/// With `sentinel`, it first prints [`SUDO_OK_SENTINEL`], so a `sudo` that -/// refused is told apart from a command that failed exactly as -/// [`Executor::run`](super::Executor::run) tells them apart. +/// +/// - **It first prints [`SUDO_OK_SENTINEL`]**, so a `sudo` that refused is +/// told apart from a command that failed exactly as +/// [`Executor::run`](super::Executor::run) tells them apart, and so what the +/// transport wrote before the command started is never counted as the +/// command's. It prints it on every transport, `sudo` or not: over SSH it +/// is also where `ssh`'s own diagnostics end. /// /// - **The environment is cleared.** The command is started through /// `env -i`, so nothing `sudo`'s environment reset or the SSH session put /// back reaches it — no variable at all, and so no `PATH` either. /// - **Only the command writes to standard error.** The script keeps the -/// real standard error aside for the command and for [`TIMEOUT_MARKER`], +/// real standard error aside for the command and for its timeout marker, /// and points its own at `/dev/null`: a shell reports a job a signal killed /// — bash prints `Killed: 9` for the cancelled `sleep` below — and that /// report is not the command's output. @@ -179,8 +208,8 @@ pub(super) enum Ended { /// `timeout`, rounded up to a whole second, started before the command. /// The script waits on it; the subshell running the command kills it with /// `SIGKILL` the moment the command exits. A `sleep` that ran out means the -/// deadline passed first, and the script prints [`TIMEOUT_MARKER`] and kills -/// the group. This is what ends the command on the far side of an SSH +/// deadline passed first, and the script prints its timeout marker and +/// kills the group. This is what ends the command on the far side of an SSH /// connection, which a signal to the local `ssh` never reaches, and it is /// the backstop wherever the relay fails to arrive. /// @@ -192,24 +221,79 @@ pub(super) enum Ended { /// through no trap, and `sleep`'s pid cannot be reused while it waits /// unreaped for the script's `wait`. /// +/// The timeout marker shares standard error with the command, so it is not a +/// fixed string a command could happen to print: it carries a nonce of +/// [`TIMEOUT_NONCE_LEN`] bytes drawn from the system's secure random source +/// for this run alone. A command could find it only by reading its +/// supervisor's arguments on purpose, and all that would win it is being +/// reported as timed out — which it could as well have had by not exiting. +/// /// Two things differ from a command run directly. A command a signal killed /// reports `128 + signal` rather than no exit code, since the script exits /// with the status its subshell returned; and a command started in the /// background of a shell without job control starts with `SIGINT` and /// `SIGQUIT` ignored. -pub(super) fn supervisor_script(sentinel: bool, timeout: Duration) -> String { +pub(super) struct Supervisor { + script: String, + timeout_marker: String, +} + +impl Supervisor { + /// Creates the supervisor of one run held to `timeout`. + /// + /// # Errors + /// + /// Returns [`ExecutorError::Spawn`] when the system's random source fails + /// to draw the timeout marker's nonce: without it, nothing can be spawned + /// under this supervisor. + pub(super) fn new(timeout: Duration) -> Result { + let mut nonce = [0u8; TIMEOUT_NONCE_LEN]; + aws_lc_rs::rand::SystemRandom::new() + .fill(&mut nonce) + .map_err(|_| ExecutorError::Spawn { + command: SUPERVISOR_SHELL.to_string(), + source: std::io::Error::other("system random source failed"), + })?; + let hex = crate::payload::to_hex(&nonce); + let timeout_marker = format!("{TIMEOUT_MARKER_PREFIX}{hex}{TIMEOUT_MARKER_SUFFIX}"); + Ok(Self { + script: script(&timeout_marker, timeout), + timeout_marker, + }) + } + + /// Returns the script to run as `sh -c SCRIPT `. + pub(super) fn script(&self) -> &str { + &self.script + } + + /// Returns the marker this supervisor prints at its own deadline. + #[cfg(test)] + pub(super) fn timeout_marker(&self) -> &str { + &self.timeout_marker + } + + /// Returns the stderr framing of a run under this supervisor, with the + /// SSH wrapper's exit-status line after it where `remote_code`. + pub(super) fn framing(&self, remote_code: bool) -> Framing<'_> { + Framing { + timeout_marker: Some(&self.timeout_marker), + remote_code, + } + } +} + +/// Returns a [`Supervisor`]'s script, printing `timeout_marker` at a deadline +/// of `timeout` rounded up to a whole second. +fn script(timeout_marker: &str, timeout: Duration) -> String { let seconds = timeout .as_secs() .saturating_add(u64::from(timeout.subsec_nanos() > 0)) .min(MAX_SLEEP_SECS); - let announce = if sentinel { - format!("printf '%s' '{SUDO_OK_SENTINEL}' >&2\n") - } else { - String::new() - }; format!( r#"trap 'kill -KILL 0' TERM HUP INT -{announce}exec 3<&0 0&2 2>/dev/null +printf '%s' '{SUDO_OK_SENTINEL}' >&2 +exec 3<&0 0&2 2>/dev/null {SLEEP} {seconds} /dev/null 3<&- 4>&- & nap=$! {{ @@ -221,7 +305,7 @@ nap=$! run=$! exec 3<&- if wait "$nap"; then - printf '%s' '{TIMEOUT_MARKER}' >&4 + printf '%s' '{timeout_marker}' >&4 kill -KILL 0 fi wait "$run""# @@ -235,7 +319,7 @@ wait "$run""# /// supervisor's `kill 0` can never reach this process. `program` names the /// binary for error reporting. /// -/// A stream passing its limit — stderr measured by `framing` — or the deadline +/// A stream passing its limit — stderr attributed by `framing` — or the deadline /// passing kills the child by `kill` and waits for it; neither is an error /// here, but an [`Ended`] the caller turns into one. /// @@ -249,7 +333,7 @@ pub(super) fn run( program: &str, feed: &[u8], limits: RunLimits, - framing: Framing, + framing: Framing<'_>, kill: Kill, ) -> Result { let failed = |source: std::io::Error| ExecutorError::Spawn { @@ -324,16 +408,22 @@ impl Pipes { enum Pumped { /// Every pipe closed, with what the child wrote on the two it reads. Closed { stdout: Vec, stderr: Vec }, - /// The run must stop early; nothing it captured is returned. + /// The run must stop early; only an [`Ended::Abandoned`] returns what it + /// captured. Stopped(Ended), } /// Moves bytes until every pipe has closed or the run must stop early. +/// +/// The stderr limit is held against what [`Framing::attribute`] credits to the +/// command; what the transport writes before the command starts is held to +/// [`TRANSPORT_STDERR_LIMIT`] instead, and passing that abandons the run with +/// the transport's output kept for classification. fn pump( pipes: &mut Pipes, feed: &[u8], limits: RunLimits, - framing: Framing, + framing: Framing<'_>, deadline: &Deadline, ) -> std::io::Result { let mut written = 0; @@ -378,8 +468,18 @@ fn pump( if ready.stderr && drain(&mut pipes.stderr, &mut stderr, &mut chunk)? { pipes.stderr = None; } - if framing.command_len(&stderr) > limits.max_stderr { - return Ok(Pumped::Stopped(Ended::Breach(OutputStream::Stderr))); + match framing.attribute(&stderr) { + Attributed::Command(len) if len > limits.max_stderr => { + return Ok(Pumped::Stopped(Ended::Breach(OutputStream::Stderr))); + } + Attributed::Transport(len) if len > TRANSPORT_STDERR_LIMIT => { + return Ok(Pumped::Stopped(Ended::Abandoned(CommandOutput { + code: None, + stdout, + stderr, + }))); + } + Attributed::Command(_) | Attributed::Transport(_) => {} } } Ok(Pumped::Closed { stdout, stderr }) @@ -612,25 +712,33 @@ fn is_remote_code_prefix(tail: &[u8], max_digits: usize) -> bool { /// Settles how the engine's run ended into the result /// [`Executor::run_with_input`](super::Executor::run_with_input) returns. /// -/// A run that exited is first checked for the supervisor's own deadline -/// having ended it, then handed to `settle` — the transport's framing, removed -/// exactly as [`Executor::run`](super::Executor::run) removes it — and what is -/// left is held against the limits once more, now with nothing but the -/// command's own bytes in it. +/// A run that exited is first checked for its supervisor's own deadline +/// having ended it, and stripped of whatever the transport wrote before the +/// command started. It is then handed to `settle` — the transport's framing, +/// removed exactly as [`Executor::run`](super::Executor::run) removes it, and +/// its failure classified the same way — and what is left is held against the +/// limits once more, now with nothing but the command's own bytes in it. The +/// transport's bytes are kept where [`SUDO_OK_SENTINEL`] never arrived: the +/// command did not start, and they are the diagnostic `settle` reports. pub(super) fn finish( ended: Ended, command: &str, limits: RunLimits, - framing: Framing, + framing: Framing<'_>, settle: impl FnOnce(CommandOutput) -> Result, ) -> Result { - let output = match ended { - Ended::Exited(output) => output, + let mut output = match ended { + Ended::Exited(output) | Ended::Abandoned(output) => output, Ended::Breach(stream) => return Err(breach(command, stream, limits)), Ended::TimedOut => return Err(timed_out(command, limits)), }; - if framing.supervised && contains(&output.stderr, TIMEOUT_MARKER) { - return Err(timed_out(command, limits)); + if let Some(marker) = framing.timeout_marker { + if contains(&output.stderr, marker) { + return Err(timed_out(command, limits)); + } + if let Some(at) = find(&output.stderr, SUDO_OK_SENTINEL.as_bytes()) { + output.stderr.drain(..at); + } } let output = settle(output)?; if output.stdout.len() > limits.max_stdout { @@ -665,70 +773,235 @@ mod tests { use std::time::Duration; use super::{ - Framing, MAX_SLEEP_SECS, SLEEP, TIMEOUT_MARKER, partial_suffix, remote_code_suffix, - supervisor_script, + Attributed, CommandOutput, Ended, Framing, MAX_SLEEP_SECS, SLEEP, Supervisor, + TIMEOUT_MARKER_PREFIX, TIMEOUT_MARKER_SUFFIX, finish, partial_suffix, remote_code_suffix, }; - use crate::executor::{RC_MARKER, SUDO_OK_SENTINEL}; + use crate::executor::{RC_MARKER, RunLimits, RunWithInputError, SUDO_OK_SENTINEL}; - const SUDO_SSH: Framing = Framing { - supervised: true, - sentinel: true, + const MARKER: &str = "__BOOTLER_TIMEOUT_00112233445566778899aabbccddeeff__"; + const SUDO_SSH: Framing<'static> = Framing { + timeout_marker: Some(MARKER), remote_code: true, }; + const LIMITS: RunLimits = RunLimits { + max_stdout: 16, + max_stderr: 16, + timeout: Duration::from_secs(5), + }; + + fn exited(stderr: &[u8]) -> Ended { + Ended::Exited(CommandOutput { + code: Some(0), + stdout: Vec::new(), + stderr: stderr.to_vec(), + }) + } + + /// What `settle` does for a `sudo` that granted: strip the sentinel. + fn strip_sentinel( + mut output: CommandOutput, + ) -> Result { + let at = super::find(&output.stderr, SUDO_OK_SENTINEL.as_bytes()).ok_or_else(|| { + crate::executor::ExecutorError::SudoRefused { + host: "test".to_string(), + reason: "no sentinel".to_string(), + } + })?; + output.stderr.drain(at..at + SUDO_OK_SENTINEL.len()); + Ok(output) + } #[test] fn a_direct_run_counts_every_stderr_byte() { let stderr = format!("{SUDO_OK_SENTINEL}abc\n{RC_MARKER}0\n"); - assert_eq!(Framing::DIRECT.command_len(stderr.as_bytes()), stderr.len()); + assert_eq!( + Framing::DIRECT.attribute(stderr.as_bytes()), + Attributed::Command(stderr.len()) + ); } #[test] fn the_sentinel_and_the_exit_status_line_are_not_the_commands() { let stderr = format!("{SUDO_OK_SENTINEL}abc\n{RC_MARKER}255\n"); - assert_eq!(SUDO_SSH.command_len(stderr.as_bytes()), 3); + assert_eq!( + SUDO_SSH.attribute(stderr.as_bytes()), + Attributed::Command(3) + ); + } + + #[test] + fn what_precedes_the_sentinel_is_the_transports() { + // `sudo` refusing, or `ssh` failing to connect: however long, none of + // it is held against the command's limit. + let refusal = "sudo: a password is required\n"; + assert_eq!( + SUDO_SSH.attribute(refusal.as_bytes()), + Attributed::Transport(refusal.len()) + ); + // Half a sentinel is the transport's too, until the rest arrives. + let half = format!("{refusal}{}", &SUDO_OK_SENTINEL[..7]); + assert_eq!( + SUDO_SSH.attribute(half.as_bytes()), + Attributed::Transport(half.len()) + ); + // Once it has arrived, a warning ahead of it is still not counted. + let stderr = format!("Warning: added host key\n{SUDO_OK_SENTINEL}abc"); + assert_eq!( + SUDO_SSH.attribute(stderr.as_bytes()), + Attributed::Command(3) + ); + } + + #[test] + fn the_runs_own_timeout_marker_is_not_the_commands() { + let stderr = format!("{SUDO_OK_SENTINEL}abc{MARKER}"); + assert_eq!( + SUDO_SSH.attribute(stderr.as_bytes()), + Attributed::Command(3) + ); + let stderr = format!("{SUDO_OK_SENTINEL}abc{MARKER}\n{RC_MARKER}124\n"); + assert_eq!( + SUDO_SSH.attribute(stderr.as_bytes()), + Attributed::Command(3) + ); } #[test] - fn the_timeout_marker_is_not_the_commands() { - let stderr = format!("{SUDO_OK_SENTINEL}abc{TIMEOUT_MARKER}"); - assert_eq!(SUDO_SSH.command_len(stderr.as_bytes()), 3); - let stderr = format!("{SUDO_OK_SENTINEL}abc{TIMEOUT_MARKER}\n{RC_MARKER}124\n"); - assert_eq!(SUDO_SSH.command_len(stderr.as_bytes()), 3); + fn another_runs_timeout_marker_is_the_commands() { + // The fixed marker earlier revisions used, and one well formed with + // another nonce: both are bytes the command wrote, and count. + for other in [ + "__BOOTLER_TIMEOUT__".to_string(), + format!( + "{TIMEOUT_MARKER_PREFIX}ffeeddccbbaa99887766554433221100{TIMEOUT_MARKER_SUFFIX}" + ), + ] { + // A trailing byte, since a trailing `_` may still grow into + // this run's marker and is not counted until it cannot. + let stderr = format!("{SUDO_OK_SENTINEL}{other}."); + assert_eq!( + SUDO_SSH.attribute(stderr.as_bytes()), + Attributed::Command(other.len() + 1), + "{other}" + ); + } } #[test] fn a_marker_still_arriving_is_not_counted_yet() { - // Half a sentinel, before the command has written anything. - let half = &SUDO_OK_SENTINEL[..7]; - assert_eq!(SUDO_SSH.command_len(half.as_bytes()), 0); // Half an exit-status line after the command's own bytes. let stderr = format!("{SUDO_OK_SENTINEL}abc\n{}", &RC_MARKER[..4]); - assert_eq!(SUDO_SSH.command_len(stderr.as_bytes()), 3); + assert_eq!( + SUDO_SSH.attribute(stderr.as_bytes()), + Attributed::Command(3) + ); let stderr = format!("{SUDO_OK_SENTINEL}abc\n{RC_MARKER}2"); - assert_eq!(SUDO_SSH.command_len(stderr.as_bytes()), 3); - let stderr = format!("{SUDO_OK_SENTINEL}abc{}", &TIMEOUT_MARKER[..5]); - assert_eq!(SUDO_SSH.command_len(stderr.as_bytes()), 3); + assert_eq!( + SUDO_SSH.attribute(stderr.as_bytes()), + Attributed::Command(3) + ); + let stderr = format!("{SUDO_OK_SENTINEL}abc{}", &MARKER[..5]); + assert_eq!( + SUDO_SSH.attribute(stderr.as_bytes()), + Attributed::Command(3) + ); } #[test] fn bytes_that_cannot_become_a_marker_are_counted() { let stderr = format!("{SUDO_OK_SENTINEL}abc\n{RC_MARKER}2x"); assert_eq!( - SUDO_SSH.command_len(stderr.as_bytes()), - stderr.len() - SUDO_OK_SENTINEL.len() + SUDO_SSH.attribute(stderr.as_bytes()), + Attributed::Command(stderr.len() - SUDO_OK_SENTINEL.len()) ); let stderr = format!("{SUDO_OK_SENTINEL}abc\n{RC_MARKER}2555"); assert_eq!( - SUDO_SSH.command_len(stderr.as_bytes()), - stderr.len() - SUDO_OK_SENTINEL.len() + SUDO_SSH.attribute(stderr.as_bytes()), + Attributed::Command(stderr.len() - SUDO_OK_SENTINEL.len()) ); } + #[test] + fn only_the_runs_own_marker_settles_as_a_timeout() { + let framing = Framing { + timeout_marker: Some(MARKER), + remote_code: false, + }; + let stderr = format!("{SUDO_OK_SENTINEL}{MARKER}"); + let error = finish( + exited(stderr.as_bytes()), + "/bin/x", + LIMITS, + framing, + strip_sentinel, + ) + .expect_err("the supervisor's deadline ended it"); + assert!( + matches!(error, RunWithInputError::TimedOut { .. }), + "{error:?}" + ); + // A command that printed the fixed marker of earlier revisions and + // exited is a command that exited. + let stderr = format!("{SUDO_OK_SENTINEL}__BOOTLER_TIMEOUT__"); + let output = finish( + exited(stderr.as_bytes()), + "/bin/x", + RunLimits { + max_stderr: 19, + ..LIMITS + }, + framing, + strip_sentinel, + ) + .expect("the command's own output"); + assert_eq!(output.code, Some(0)); + assert_eq!(output.stderr, b"__BOOTLER_TIMEOUT__"); + } + + #[test] + fn what_the_transport_wrote_before_the_command_is_not_returned() { + let framing = Framing { + timeout_marker: Some(MARKER), + remote_code: false, + }; + let stderr = format!("Warning: added host key\n{SUDO_OK_SENTINEL}abc"); + let output = finish( + exited(stderr.as_bytes()), + "/bin/x", + RunLimits { + max_stderr: 3, + ..LIMITS + }, + framing, + strip_sentinel, + ) + .expect("the command's own output"); + assert_eq!(output.stderr, b"abc"); + } + + #[test] + fn each_supervisor_draws_its_own_marker() { + let first = Supervisor::new(Duration::from_secs(1)).expect("a supervisor"); + let second = Supervisor::new(Duration::from_secs(1)).expect("a supervisor"); + assert_ne!(first.timeout_marker(), second.timeout_marker()); + for supervisor in [&first, &second] { + let marker = supervisor.timeout_marker(); + let nonce = marker + .strip_prefix(TIMEOUT_MARKER_PREFIX) + .and_then(|rest| rest.strip_suffix(TIMEOUT_MARKER_SUFFIX)) + .expect("prefix, nonce, suffix"); + assert_eq!(nonce.len(), 32); + assert!(nonce.bytes().all(|byte| byte.is_ascii_hexdigit())); + assert!(supervisor.script().contains(marker)); + } + } + #[test] fn the_supervisor_deadline_is_whole_seconds_rounded_up_and_capped() { let sleeps = |timeout| { - let script = supervisor_script(false, timeout); - let line = script + let supervisor = Supervisor::new(timeout).expect("a supervisor"); + let line = supervisor + .script() .lines() .find(|line| line.starts_with(&format!("{SLEEP} "))) .expect("the deadline's sleep") From b47a0fed70f9cd2ec3e79a40cb55bf693ada8e00 Mon Sep 17 00:00:00 2001 From: sehkone Date: Sun, 27 Sep 2026 17:53:56 +0900 Subject: [PATCH 7/7] Count undecided stderr once its grace runs out A trailing stderr fragment that might still grow into the supervisor's timeout marker or the SSH exit-status line was left out of the count indefinitely. A command could write one such byte past its limit, keep running, and be reported as timed out instead of being killed for the breach. Such bytes are now held as pending. When counting them would pass the limit, they get a one-second grace to become framing, which is written whole and followed by the stream closing; after that they count and the run ends as an output-limit breach. Part of #129 --- src/executor.rs | 55 ++++++++++ src/executor/bounded.rs | 226 ++++++++++++++++++++++++++++++++-------- 2 files changed, 239 insertions(+), 42 deletions(-) diff --git a/src/executor.rs b/src/executor.rs index 9654d2b..cceec24 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -1068,6 +1068,11 @@ pub trait Executor { /// writes before the command starts is neither counted nor returned, /// so a refusal or a failed connection is reported as the error /// [`Executor::run`] reports for it however small `max_stderr` is. + /// Where a supervising shell or the SSH exit-status line shares + /// standard error with the command, a byte past `max_stderr` that could + /// still be the start of that framing is given a second to become it + /// before it counts, so a command that stops on such a byte is killed + /// up to a second after it passed its limit. /// - **A non-zero exit is a [`CommandOutput`]**, as it is from /// [`Executor::run`]. /// @@ -7001,6 +7006,56 @@ exec "$@" } } + #[test] + fn a_byte_over_the_limit_that_could_open_a_marker_kills_a_running_command_on_every_pair() + { + // `_` could open the timeout marker and a newline the SSH + // exit-status line, but a command still running after either + // wrote it itself, and one byte over is one byte over. + let dir = tempfile::tempdir().expect("tempdir"); + let limits = RunLimits { + max_stderr: 0, + ..ROOMY + }; + // Written only once the pids are recorded, since a direct + // run is stopped at the byte. + let flood = "/usr/bin/head -c \"$2\" /dev/zero"; + assert!(STUBBORN.contains(flood)); + let script = STUBBORN.replace(flood, "printf '%s' \"$3\" >&2"); + for (fragment, name) in [("_", "underscore"), ("\n", "newline")] { + for (index, (label, exec, identity)) in every_pair(&dir).into_iter().enumerate() + { + let pids = dir.path().join(format!("{name}-{index}")); + let started = Instant::now(); + let error = exec + .run_with_input( + identity, + "/bin/sh", + &["-c", &script, "sh", &pids.to_string_lossy(), "0", fragment], + b"", + limits, + ) + .expect_err("the byte must be counted"); + assert!( + matches!( + error, + RunWithInputError::OutputLimit { + stream: OutputStream::Stderr, + limit: 0, + .. + } + ), + "{label} {name}: got {error:?}" + ); + assert!( + started.elapsed() < limits.timeout, + "{label} {name}: the breach, not the timeout, must end the run" + ); + assert_gone(label, &recorded_pids(&pids)); + } + } + } + #[test] fn a_command_that_leaves_its_input_unread_still_reports_its_exit_on_every_pair() { // More than a pipe buffer holds, so feeding it cannot finish diff --git a/src/executor/bounded.rs b/src/executor/bounded.rs index 0507399..f8fc6c7 100644 --- a/src/executor/bounded.rs +++ b/src/executor/bounded.rs @@ -42,6 +42,14 @@ const EXIT_POLL: Duration = Duration::from_millis(5); const RELAY_GRACE: Duration = Duration::from_secs(5); /// How long a child is given to be reaped after `SIGKILL` reached its group. const KILL_GRACE: Duration = Duration::from_secs(5); +/// How long stderr bytes that may yet prove to be framing — the start of a +/// timeout marker, or of the SSH wrapper's exit-status line — are given to +/// prove it, once counting them as the command's would pass `max_stderr`. +/// Framing is written whole, a line in one write with the stream closing +/// right behind it, so it resolves within moments; this bounds only how long +/// a command that wrote a look-alike fragment and went on running outlives +/// its limit. +const FRAMING_GRACE: Duration = Duration::from_secs(1); /// What opens every [`Supervisor`]'s timeout marker; the run's own random /// nonce follows it. const TIMEOUT_MARKER_PREFIX: &str = "__BOOTLER_TIMEOUT_"; @@ -105,8 +113,10 @@ pub(super) struct Framing<'a> { enum Attributed { /// The command has not started; this many bytes are the transport's. Transport(usize), - /// The command has started and has written this many bytes of its own. - Command(usize), + /// The command has started. `own` bytes are certainly its own; the + /// `pending` bytes after them are too, unless they turn out to be framing + /// still arriving. + Command { own: usize, pending: usize }, } impl Framing<'static> { @@ -125,15 +135,20 @@ impl Framing<'_> { /// connect, or the sentinel itself still arriving, and it is left for the /// transport's failure to be classified exactly as /// [`Executor::run`](super::Executor::run) classifies it. After the - /// sentinel, a timeout marker already seen in full is discounted, and so - /// is a trailing fragment that may still grow into it or into the - /// exit-status line — a stream is read in chunks, and a marker split - /// across two of them must not count against the limit while only its - /// first half has arrived. Nothing else is discounted, so a command that - /// writes one byte past its limit is caught at that byte. + /// sentinel, a timeout marker already seen in full is discounted: its + /// nonce makes it this run's alone. A trailing fragment that may still + /// grow into that marker, or that is or may grow into the exit-status + /// line, is left pending — a stream is read in chunks, a marker split + /// across two of them has arrived only in half, and the exit-status line + /// is a fixed string a command could print itself. Nothing else is + /// discounted or deferred, so a command that writes one byte past its + /// limit is caught at that byte unless the byte could still be framing. fn attribute(self, stderr: &[u8]) -> Attributed { let Some(marker) = self.timeout_marker else { - return Attributed::Command(stderr.len()); + return Attributed::Command { + own: stderr.len(), + pending: 0, + }; }; let Some(at) = find(stderr, SUDO_OK_SENTINEL.as_bytes()) else { return Attributed::Transport(stderr.len()); @@ -151,7 +166,10 @@ impl Framing<'_> { if self.remote_code { pending = pending.max(remote_code_suffix(tail)); } - Attributed::Command(tail.len().saturating_sub(framing + pending)) + Attributed::Command { + own: tail.len().saturating_sub(framing + pending), + pending, + } } } @@ -362,8 +380,12 @@ pub(super) fn run( let pumped = pump(&mut pipes, feed, limits, framing, &deadline); drop(pipes); match pumped { - Ok(Pumped::Closed { stdout, stderr }) => { - await_exit(child, kill, &deadline, stdout, stderr).map_err(failed) + Ok(Pumped::Closed { + stdout, + stderr, + undecided, + }) => { + await_exit(child, kill, &deadline, undecided.as_ref(), stdout, stderr).map_err(failed) } Ok(Pumped::Stopped(ended)) => { terminate(&mut child, kill); @@ -406,8 +428,14 @@ impl Pipes { /// What [`pump`] stopped on. enum Pumped { - /// Every pipe closed, with what the child wrote on the two it reads. - Closed { stdout: Vec, stderr: Vec }, + /// Every pipe closed, with what the child wrote on the two it reads, and + /// when stderr bytes that may be framing must have proved it, where + /// counting them as the command's passes the limit. + Closed { + stdout: Vec, + stderr: Vec, + undecided: Option, + }, /// The run must stop early; only an [`Ended::Abandoned`] returns what it /// captured. Stopped(Ended), @@ -418,7 +446,10 @@ enum Pumped { /// The stderr limit is held against what [`Framing::attribute`] credits to the /// command; what the transport writes before the command starts is held to /// [`TRANSPORT_STDERR_LIMIT`] instead, and passing that abandons the run with -/// the transport's output kept for classification. +/// the transport's output kept for classification. Where only bytes still +/// pending as possible framing pass the limit, they are given +/// [`FRAMING_GRACE`] to become framing, and count as the command's once it +/// runs out. fn pump( pipes: &mut Pipes, feed: &[u8], @@ -430,10 +461,14 @@ fn pump( let mut stdout = Vec::new(); let mut stderr = Vec::new(); let mut chunk = vec![0; READ_CHUNK]; + let mut undecided: Option = None; while !pipes.is_empty() { - let Some(left) = deadline.remaining() else { + let Some(mut left) = deadline.remaining() else { return Ok(Pumped::Stopped(Ended::TimedOut)); }; + if let Some(grace) = &undecided { + left = left.min(grace.remaining().unwrap_or_default()); + } // A time left too large for a `timespec` is no bound at all. let timeout = Timespec::try_from(left).ok(); let ready = match wait_ready(pipes, timeout.as_ref()) { @@ -469,9 +504,17 @@ fn pump( pipes.stderr = None; } match framing.attribute(&stderr) { - Attributed::Command(len) if len > limits.max_stderr => { + Attributed::Command { own, .. } if own > limits.max_stderr => { return Ok(Pumped::Stopped(Ended::Breach(OutputStream::Stderr))); } + Attributed::Command { own, pending } + if own.saturating_add(pending) > limits.max_stderr => + { + let grace = undecided.get_or_insert_with(|| Deadline::after(FRAMING_GRACE)); + if grace.remaining().is_none() { + return Ok(Pumped::Stopped(Ended::Breach(OutputStream::Stderr))); + } + } Attributed::Transport(len) if len > TRANSPORT_STDERR_LIMIT => { return Ok(Pumped::Stopped(Ended::Abandoned(CommandOutput { code: None, @@ -479,10 +522,14 @@ fn pump( stderr, }))); } - Attributed::Command(_) | Attributed::Transport(_) => {} + Attributed::Command { .. } | Attributed::Transport(_) => undecided = None, } } - Ok(Pumped::Closed { stdout, stderr }) + Ok(Pumped::Closed { + stdout, + stderr, + undecided, + }) } /// Which pipes `poll(2)` reported ready. @@ -550,15 +597,18 @@ fn is_transient(error: &std::io::Error) -> bool { } /// Waits for a child whose pipes have all closed to exit, killing it at the -/// deadline. +/// deadline — or, where stderr ended on bytes that pass the limit unless they +/// are framing, at the end of their `undecided` grace. /// /// The pipes closing is not the child exiting: a command can close its /// standard streams and keep running, and nothing then remains to `poll(2)` -/// on, so the exit is polled. +/// on, so the exit is polled. Framing is followed by the exit at once, so a +/// child still running when the grace runs out wrote those bytes itself. fn await_exit( mut child: Child, kill: Kill, deadline: &Deadline, + undecided: Option<&Deadline>, stdout: Vec, stderr: Vec, ) -> std::io::Result { @@ -570,10 +620,17 @@ fn await_exit( stderr, })); } - let Some(timeout) = deadline.remaining() else { + let Some(mut timeout) = deadline.remaining() else { terminate(&mut child, kill); return Ok(Ended::TimedOut); }; + if let Some(grace) = undecided { + let Some(left) = grace.remaining() else { + terminate(&mut child, kill); + return Ok(Ended::Breach(OutputStream::Stderr)); + }; + timeout = timeout.min(left); + } std::thread::sleep(timeout.min(EXIT_POLL)); } } @@ -770,13 +827,17 @@ fn timed_out(command: &str, limits: RunLimits) -> RunWithInputError { #[cfg(test)] mod tests { - use std::time::Duration; + use std::process::Command; + use std::time::{Duration, Instant}; use super::{ - Attributed, CommandOutput, Ended, Framing, MAX_SLEEP_SECS, SLEEP, Supervisor, - TIMEOUT_MARKER_PREFIX, TIMEOUT_MARKER_SUFFIX, finish, partial_suffix, remote_code_suffix, + Attributed, CommandOutput, Ended, FRAMING_GRACE, Framing, Kill, MAX_SLEEP_SECS, SLEEP, + Supervisor, TIMEOUT_MARKER_PREFIX, TIMEOUT_MARKER_SUFFIX, finish, partial_suffix, + remote_code_suffix, run, + }; + use crate::executor::{ + OutputStream, RC_MARKER, RunLimits, RunWithInputError, SUDO_OK_SENTINEL, }; - use crate::executor::{RC_MARKER, RunLimits, RunWithInputError, SUDO_OK_SENTINEL}; const MARKER: &str = "__BOOTLER_TIMEOUT_00112233445566778899aabbccddeeff__"; const SUDO_SSH: Framing<'static> = Framing { @@ -816,16 +877,25 @@ mod tests { let stderr = format!("{SUDO_OK_SENTINEL}abc\n{RC_MARKER}0\n"); assert_eq!( Framing::DIRECT.attribute(stderr.as_bytes()), - Attributed::Command(stderr.len()) + Attributed::Command { + own: stderr.len(), + pending: 0 + } ); } #[test] - fn the_sentinel_and_the_exit_status_line_are_not_the_commands() { - let stderr = format!("{SUDO_OK_SENTINEL}abc\n{RC_MARKER}255\n"); + fn the_sentinel_is_not_the_commands_and_the_exit_status_line_is_pending() { + // The exit-status line is a fixed string the command could print + // itself, so it is never discounted outright. + let line = format!("\n{RC_MARKER}255\n"); + let stderr = format!("{SUDO_OK_SENTINEL}abc{line}"); assert_eq!( SUDO_SSH.attribute(stderr.as_bytes()), - Attributed::Command(3) + Attributed::Command { + own: 3, + pending: line.len() + } ); } @@ -848,7 +918,7 @@ mod tests { let stderr = format!("Warning: added host key\n{SUDO_OK_SENTINEL}abc"); assert_eq!( SUDO_SSH.attribute(stderr.as_bytes()), - Attributed::Command(3) + Attributed::Command { own: 3, pending: 0 } ); } @@ -857,12 +927,16 @@ mod tests { let stderr = format!("{SUDO_OK_SENTINEL}abc{MARKER}"); assert_eq!( SUDO_SSH.attribute(stderr.as_bytes()), - Attributed::Command(3) + Attributed::Command { own: 3, pending: 0 } ); - let stderr = format!("{SUDO_OK_SENTINEL}abc{MARKER}\n{RC_MARKER}124\n"); + let line = format!("\n{RC_MARKER}124\n"); + let stderr = format!("{SUDO_OK_SENTINEL}abc{MARKER}{line}"); assert_eq!( SUDO_SSH.attribute(stderr.as_bytes()), - Attributed::Command(3) + Attributed::Command { + own: 3, + pending: line.len() + } ); } @@ -877,33 +951,39 @@ mod tests { ), ] { // A trailing byte, since a trailing `_` may still grow into - // this run's marker and is not counted until it cannot. + // this run's marker and is pending until it cannot. let stderr = format!("{SUDO_OK_SENTINEL}{other}."); assert_eq!( SUDO_SSH.attribute(stderr.as_bytes()), - Attributed::Command(other.len() + 1), + Attributed::Command { + own: other.len() + 1, + pending: 0 + }, "{other}" ); } } #[test] - fn a_marker_still_arriving_is_not_counted_yet() { + fn a_marker_still_arriving_is_pending() { // Half an exit-status line after the command's own bytes. let stderr = format!("{SUDO_OK_SENTINEL}abc\n{}", &RC_MARKER[..4]); assert_eq!( SUDO_SSH.attribute(stderr.as_bytes()), - Attributed::Command(3) + Attributed::Command { own: 3, pending: 5 } ); let stderr = format!("{SUDO_OK_SENTINEL}abc\n{RC_MARKER}2"); assert_eq!( SUDO_SSH.attribute(stderr.as_bytes()), - Attributed::Command(3) + Attributed::Command { + own: 3, + pending: RC_MARKER.len() + 2 + } ); let stderr = format!("{SUDO_OK_SENTINEL}abc{}", &MARKER[..5]); assert_eq!( SUDO_SSH.attribute(stderr.as_bytes()), - Attributed::Command(3) + Attributed::Command { own: 3, pending: 5 } ); } @@ -912,12 +992,18 @@ mod tests { let stderr = format!("{SUDO_OK_SENTINEL}abc\n{RC_MARKER}2x"); assert_eq!( SUDO_SSH.attribute(stderr.as_bytes()), - Attributed::Command(stderr.len() - SUDO_OK_SENTINEL.len()) + Attributed::Command { + own: stderr.len() - SUDO_OK_SENTINEL.len(), + pending: 0 + } ); let stderr = format!("{SUDO_OK_SENTINEL}abc\n{RC_MARKER}2555"); assert_eq!( SUDO_SSH.attribute(stderr.as_bytes()), - Attributed::Command(stderr.len() - SUDO_OK_SENTINEL.len()) + Attributed::Command { + own: stderr.len() - SUDO_OK_SENTINEL.len(), + pending: 0 + } ); } @@ -1024,6 +1110,62 @@ mod tests { assert_eq!(sleeps(Duration::MAX), MAX_SLEEP_SECS); } + /// Runs `script` under `sh`, directly and in its own group, framed as a + /// supervised run over SSH and held to no stderr at all. + fn run_framed(script: &str) -> (Ended, Duration) { + let mut command = Command::new("/bin/sh"); + command.args(["-c", script, "sh", SUDO_OK_SENTINEL]); + let limits = RunLimits { + max_stderr: 0, + timeout: Duration::from_secs(30), + ..LIMITS + }; + let started = Instant::now(); + let ended = + run(command, "/bin/sh", b"", limits, SUDO_SSH, Kill::Group).expect("the run completes"); + (ended, started.elapsed()) + } + + #[test] + fn a_pending_fragment_past_the_limit_is_the_commands_once_its_grace_runs_out() { + // A byte that could open either marker, then a command that goes on + // running: with its streams open, and with them closed. + for fragment in ["_", "\\n"] { + for after in ["", "exec >&- 2>&-;"] { + let script = format!( + "printf '%s{fragment}' \"$1\" >&2; {after} trap '' TERM; exec {SLEEP} 300" + ); + let (ended, elapsed) = run_framed(&script); + assert!( + matches!(ended, Ended::Breach(OutputStream::Stderr)), + "{fragment:?} {after:?}: {ended:?}" + ); + assert!( + elapsed >= FRAMING_GRACE && elapsed < Duration::from_secs(10), + "{fragment:?} {after:?}: ended after {elapsed:?}" + ); + } + } + } + + #[test] + fn a_pending_fragment_that_becomes_framing_is_not_the_commands() { + // The exit-status line split across two writes, and the stream then + // closing: the command wrote nothing of its own. + let script = + format!("printf '%s\\n__BOOT' \"$1\" >&2; {SLEEP} 0.2; printf 'LER_RC__:0\\n' >&2"); + assert_eq!(RC_MARKER, "__BOOTLER_RC__:"); + let (ended, _) = run_framed(&script); + let Ended::Exited(output) = ended else { + panic!("the command exited: {ended:?}"); + }; + assert_eq!(output.code, Some(0)); + assert_eq!( + output.stderr, + format!("{SUDO_OK_SENTINEL}\n{RC_MARKER}0\n").as_bytes() + ); + } + #[test] fn suffix_helpers_find_the_longest_candidate() { assert_eq!(partial_suffix(b"xx__BOOT", b"__BOOTLER"), 6);