From ac8b268475b0b09f8725f7e320475f49def6cb9c Mon Sep 17 00:00:00 2001 From: sehkone Date: Sun, 27 Sep 2026 22:03:45 +0900 Subject: [PATCH 1/6] Add a long-lived elevated channel to Executor bootler's recorded recovery runs every operation for a host over one framed session to a root helper, and needs that helper started locally or over SSH as a long-lived child with piped stdio, under this crate's SSH and elevation policy. run and run_with_input return only once the command exits, and ssh_command and resolve_through are private, so nothing public could do it. open_channel resolves an identity exactly as run does, at the same resolution site, proves the start by reading the sudo sentinel before returning, and hands the caller stdin and stdout. Stderr is drained on a joined, poll-driven thread for the channel's whole life, so the command never blocks on it, and over SSH the exit-status line supplies the command's own code. InDaemonExecutor keeps the default Unsupported body, and the recording executor records and scripts the call. Closes #132 --- src/executor.rs | 1001 +++++++++++++++++++++++++++++++++- src/executor/bounded.rs | 26 +- src/executor/channel.rs | 828 ++++++++++++++++++++++++++++ src/executor/test_support.rs | 172 +++++- 4 files changed, 1996 insertions(+), 31 deletions(-) create mode 100644 src/executor/channel.rs diff --git a/src/executor.rs b/src/executor.rs index cceec24..302febc 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -38,12 +38,15 @@ use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; use std::time::Duration; +pub use self::channel::{Channel, ChannelError, ChannelExit, ChannelLimits}; 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; +// The long-lived channel behind `Executor::open_channel`. +mod channel; // 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"))] @@ -716,9 +719,16 @@ pub enum RunWithInputError { Executor(#[from] ExecutorError), } +/// Reports whether `command` is an absolute path free of `=`: the one form +/// that names a single binary with no `PATH`, and that `env` cannot read as an +/// assignment. +fn is_absolute_and_plain(command: &str) -> bool { + Path::new(command).is_absolute() && !command.contains('=') +} + /// 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('=') { + if is_absolute_and_plain(command) { Ok(()) } else { Err(RunWithInputError::InvalidCommand { @@ -1137,6 +1147,90 @@ pub trait Executor { Err(RunWithInputError::Unsupported) } + /// Starts `command` with `args` as `identity` and returns a [`Channel`] + /// to it that lives until the caller ends it. + /// + /// This is the long-lived counterpart of [`Executor::run_with_input`]: + /// where that call feeds a request and collects an answer within bounds, + /// this one hands the caller the command's standard input and standard + /// output for as long as it runs, with no size or time bound from this + /// crate. + /// + /// - **`command` must be an absolute path**, free of `=`. Anything else is + /// [`ChannelError::InvalidCommand`], refused before anything is spawned. + /// `args` stay discrete words on every transport. + /// - **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 with `env -i` as the command starts. + /// - **The start is proven before this returns.** Where `sudo` or SSH + /// stands in between — [`Identity::Root`] and [`Identity::Service`] on + /// [`LocalExecutor`], every identity on [`SshExecutor`] — the command + /// starts under a fixed `sh -c` script that announces the start on + /// standard error and then replaces itself with the command. Under + /// [`SudoAuth::Password`] the password line is written first, and + /// standard input then stays open for the caller, so the command's first + /// byte of input is the caller's first byte. Standard error is read until + /// the announcement, and standard output not at all. A transport that + /// ends first, or writes more than 64 KiB first, is killed and classified + /// as [`Executor::run`] classifies that output; one that does neither + /// within `limits.elevation_timeout` is killed and reaped. The operator on + /// [`LocalExecutor`] is spawned directly, and this returns at once. + /// - **Standard error stays with the channel**, drained for its whole + /// life so the command never blocks on it: the first + /// `limits.max_stderr` bytes of the command's own standard error are + /// kept for [`Channel::wait`], the rest discarded and noted. Over SSH the + /// wrapper's trailing exit-status line is removed and supplies the exit + /// code, so a remote `255` is the command's. + /// + /// `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`], + /// with [`SudoAuth`] choosing `-n` or `-S -p ""`; over SSH, the same `ssh` + /// invocation — key, port, host-key policy, and `BatchMode=yes` under + /// [`SshPrompt::Deny`] — with no terminal requested. The transport is + /// spawned in the caller's process group, as [`Executor::run`] spawns it, + /// so a passphrase prompt [`SshPrompt::Allow`] permits still reaches the + /// terminal. + /// + /// Under [`SudoAuth::Password`], a `sudo` that does not ask for the + /// password — a `NOPASSWD` rule, or credentials it still has cached — + /// leaves the password line unread, and the command reads it as its first + /// line of input. This is `sudo`'s choice, and the same one + /// [`Executor::run_with_input`] meets; [`SudoAuth::NonInteractive`] is + /// the setting for a host where `sudo` does not ask. + /// + /// [`Channel`] states what ending the channel reaches: the local + /// transport process, not a command started through `sudo` or over SSH, + /// and never a descendant. + /// + /// The default body refuses with [`ChannelError::Unsupported`], so an + /// existing implementation of this trait keeps compiling. [`LocalExecutor`] + /// and [`SshExecutor`] override it; [`InDaemonExecutor`] does not. + /// + /// # Errors + /// + /// Returns [`ChannelError::InvalidCommand`] for a command that is not an + /// absolute path free of `=`, [`ChannelError::ElevationTimedOut`] for a + /// start not proven within `limits.elevation_timeout`, and + /// [`ChannelError::Executor`] carrying whatever [`Executor::run`] would + /// report for a spawn, transport or elevation failure — + /// [`ExecutorError::Connection`], [`ExecutorError::Elevation`] or + /// [`ExecutorError::SudoRefused`]. An SSH transport whose remote shell + /// ran but did not start the command is reported as + /// [`ExecutorError::SudoRefused`] carrying the remote diagnostic, for the + /// operator too. + fn open_channel( + &self, + identity: Identity, + command: &str, + args: &[&str], + limits: ChannelLimits, + ) -> Result { + let _ = (identity, command, args, limits); + Err(ChannelError::Unsupported) + } + /// Writes `contents` to `dest` on the target with the owner, group and mode /// `meta` names (RFC 0003 §9.2). /// @@ -2241,15 +2335,23 @@ fn classify_elevation( if granted { return Ok(output); } - if matches!(auth, Some(SudoAuth::NonInteractive)) && sudo_needs_password(&output.stderr) { - return Err(ExecutorError::Elevation { + Err(elevation_refusal(&output.stderr, auth, host)) +} + +/// Classifies what `sudo` wrote on `stderr` when the sentinel never arrived, +/// as [`classify_elevation`] reports it: [`ExecutorError::Elevation`] for a +/// non-interactive `sudo` that wanted a password, [`ExecutorError::SudoRefused`] +/// carrying the diagnostic otherwise. +fn elevation_refusal(stderr: &[u8], auth: Option<&SudoAuth>, host: &str) -> ExecutorError { + if matches!(auth, Some(SudoAuth::NonInteractive)) && sudo_needs_password(stderr) { + return ExecutorError::Elevation { host: host.to_string(), - }); + }; } - Err(ExecutorError::SudoRefused { + ExecutorError::SudoRefused { host: host.to_string(), - reason: String::from_utf8_lossy(&output.stderr).trim().to_string(), - }) + reason: String::from_utf8_lossy(stderr).trim().to_string(), + } } /// An [`Executor`] that acts on the local (seat) machine. @@ -2482,6 +2584,43 @@ impl Executor for LocalExecutor { self.run_bounded(identity, command, args, input, limits) } + fn open_channel( + &self, + identity: Identity, + command: &str, + args: &[&str], + limits: ChannelLimits, + ) -> Result { + channel::check_command(command)?; + let Resolved { + command: mut cmd, + password_line, + elevated, + } = self.resolve_through( + identity, + channel::START_SHELL, + &channel::start_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 + // start script clears the command's after `sudo` has set up its own. + if !elevated { + cmd.env_clear(); + return channel::open_direct(cmd, limits.max_stderr); + } + let start = channel::Start { + host: &self.host, + password_line, + remote_code: false, + limits, + }; + channel::open_started(cmd, &start, |output| { + elevation_refusal(&output.stderr, Some(&self.auth), &self.host) + }) + } + 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. @@ -2833,6 +2972,49 @@ impl Executor for SshExecutor { self.run_bounded(identity, command, args, input, limits) } + fn open_channel( + &self, + identity: Identity, + command: &str, + args: &[&str], + limits: ChannelLimits, + ) -> Result { + channel::check_command(command)?; + let script = channel::start_script(); + let ResolvedRemote { + remote, + password_line, + elevated, + } = self.resolve_through( + identity, + channel::START_SHELL, + &script, + Some(&script), + command, + args, + ); + // `ssh` keeps the caller's environment, as it does for `run`; the + // remote start script clears the command's. + let mut cmd = self.ssh_command(); + cmd.arg(wrap_with_rc_marker(&remote)); + let start = channel::Start { + host: &self.host, + password_line, + remote_code: true, + limits, + }; + channel::open_started(cmd, &start, |output| { + match self.settle_remote_code(output) { + Err(connection) => connection, + // The remote shell ran and the command did not start: `sudo` + // refused it, or, for the operator, the shell could not. + Ok(output) => { + elevation_refusal(&output.stderr, elevated.then_some(&self.auth), &self.host) + } + } + }) + } + 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. @@ -2892,6 +3074,9 @@ impl Executor for SshExecutor { /// line. The sudo sentinel still wraps the invocation, because `sudo -u` can /// still refuse — an unknown or non-descendable account — and that refusal must /// classify as an elevation failure rather than a command failure. +/// +/// It does not implement [`Executor::open_channel`]: that call returns the +/// trait default's [`ChannelError::Unsupported`] here. #[derive(Debug, Clone)] pub struct InDaemonExecutor { host: String, @@ -6741,7 +6926,7 @@ wait"#; /// A `sudo` stub that drops its own flags, `-u ` /// included, and execs the wrapped command. - fn descending_sudo(dir: &Path) -> PathBuf { + pub(super) fn descending_sudo(dir: &Path) -> PathBuf { write_script( dir, "descending-sudo", @@ -6762,7 +6947,7 @@ 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 { + pub(super) fn password_sudo(dir: &Path) -> PathBuf { write_script( dir, "password-sudo", @@ -6782,7 +6967,7 @@ exec "$@" ) } - fn ssh_with(dir: &Path, sudo: &Path, auth: SudoAuth) -> SshExecutor { + pub(super) fn ssh_with(dir: &Path, sudo: &Path, auth: SudoAuth) -> SshExecutor { let config = crate::transport::Ssh { user: "ops".to_string(), port: 22, @@ -7571,5 +7756,801 @@ exec "$@" } } } + + /// [`Executor::open_channel`] on every pair that implements it. + mod channels { + use std::io::{Read, Write}; + use std::path::{Path, PathBuf}; + use std::time::{Duration, Instant}; + + use rustix::process::Pid; + use tempfile::TempDir; + + use super::super::super::{ + Channel, ChannelError, ChannelLimits, CommandOutput, Executor, ExecutorError, + FileMeta, Identity, InDaemonExecutor, LocalExecutor, RC_MARKER, SUDO_OK_SENTINEL, + ServiceAccount, SshExecutor, SshPrompt, SudoAuth, spawn_retrying_text_busy, + }; + use super::bounded_runs::{descending_sudo, password_sudo, ssh_with}; + use super::{failing_ssh, fake_ssh, write_script}; + + /// Limits roomy enough for every test that is not about them. + const ROOMY: ChannelLimits = ChannelLimits { + elevation_timeout: Duration::from_secs(30), + max_stderr: 1 << 20, + }; + /// More than any pipe buffer holds, so an echo of it cannot + /// complete unless both directions move at once. + const ECHO_LEN: usize = 256 * 1024; + /// 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); + + /// Every pair that implements the channel, 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); + 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), + ] + } + + /// Every byte value, repeated out to `len` bytes. + fn pattern(len: usize) -> Vec { + (0..=u8::MAX).cycle().take(len).collect() + } + + /// Writes `input` to the channel's standard input and closes it, + /// while reading its standard output to the end. + fn echo(channel: &mut Channel, input: &[u8]) -> Vec { + let mut stdin = channel.take_stdin().expect("stdin is the caller's"); + let mut stdout = channel.take_stdout().expect("stdout is the caller's"); + std::thread::scope(|scope| { + let writer = scope.spawn(move || stdin.write_all(input)); + let mut echoed = Vec::new(); + stdout.read_to_end(&mut echoed).expect("read stdout"); + writer + .join() + .expect("the writer does not panic") + .expect("write stdin"); + echoed + }) + } + + /// Reads a channel's standard output to the end, closing its + /// standard input first. + fn read_out(channel: &mut Channel) -> Vec { + echo(channel, b"") + } + + /// Waits until `path` exists, then reads the pid written in it. + fn recorded_pid(path: &Path) -> Pid { + let deadline = Instant::now() + REAP_WAIT; + while !path.exists() { + assert!(Instant::now() < deadline, "the command never started"); + std::thread::sleep(Duration::from_millis(10)); + } + let text = std::fs::read_to_string(path).expect("the recorded pid"); + pid(text.trim()) + } + + fn pid(text: &str) -> Pid { + Pid::from_raw(text.parse().expect("a pid")).expect("a positive pid") + } + + /// Waits until no process with `pid` exists any more. + fn assert_gone(label: &str, pid: Pid) { + let deadline = Instant::now() + REAP_WAIT; + while rustix::process::test_kill_process(pid).is_ok() { + assert!( + Instant::now() < deadline, + "{label}: process {pid:?} survived" + ); + std::thread::sleep(Duration::from_millis(10)); + } + } + + #[test] + fn bytes_pass_verbatim_both_ways_on_every_pair() { + let dir = tempfile::tempdir().expect("tempdir"); + let input = pattern(ECHO_LEN); + for (label, exec, identity) in every_pair(&dir) { + let mut channel = exec + .open_channel(identity, "/bin/cat", &[], ROOMY) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + let echoed = echo(&mut channel, &input); + assert!( + echoed == input, + "{label}: {} bytes in, {} bytes out", + input.len(), + echoed.len() + ); + let exit = channel + .wait() + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + assert_eq!(exit.code, Some(0), "{label}"); + assert!(exit.stderr.is_empty(), "{label}: {:?}", exit.stderr); + assert!(!exit.stderr_truncated, "{label}"); + } + } + + #[test] + fn stderr_is_drained_and_held_to_its_limit_while_the_echo_runs_on_every_pair() { + const KEPT: usize = 1000; + let dir = tempfile::tempdir().expect("tempdir"); + let limits = ChannelLimits { + max_stderr: KEPT, + ..ROOMY + }; + let input = pattern(ECHO_LEN); + let flood = "/usr/bin/head -c 1048576 /dev/zero >&2 & /bin/cat; wait"; + for (label, exec, identity) in every_pair(&dir) { + let mut channel = exec + .open_channel(identity, "/bin/sh", &["-c", flood], limits) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + let echoed = echo(&mut channel, &input); + assert!(echoed == input, "{label}: the echo must complete intact"); + let exit = channel + .wait() + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + assert_eq!(exit.code, Some(0), "{label}"); + assert_eq!(exit.stderr, vec![0; KEPT], "{label}"); + assert!(exit.stderr_truncated, "{label}"); + } + } + + #[test] + fn the_commands_own_code_and_stderr_come_back_on_every_pair() { + let dir = tempfile::tempdir().expect("tempdir"); + for code in [0, 3, 255] { + let script = format!("printf err >&2; exit {code}"); + for (label, exec, identity) in every_pair(&dir) { + let mut channel = exec + .open_channel(identity, "/bin/sh", &["-c", &script], ROOMY) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + assert!(read_out(&mut channel).is_empty(), "{label}"); + let exit = channel + .wait() + .unwrap_or_else(|error| panic!("{label} {code}: {error:?}")); + assert_eq!(exit.code, Some(code), "{label}"); + assert_eq!( + exit.stderr, b"err", + "{label}: neither the sentinel nor the exit-status line" + ); + assert!(!exit.stderr_truncated, "{label}"); + } + } + } + + #[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 mut channel = exec + .open_channel(identity, "/usr/bin/env", &[], ROOMY) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + let printed = read_out(&mut channel); + assert_eq!( + String::from_utf8_lossy(&printed), + "", + "{label}: no variable may reach the command" + ); + let exit = channel + .wait() + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + assert_eq!(exit.code, Some(0), "{label}"); + } + } + + #[test] + fn the_start_consumes_no_standard_output() { + // A transport that writes on stdout before the command starts: + // every byte of it is still there for the caller. + let dir = tempfile::tempdir().expect("tempdir"); + let early = write_script( + dir.path(), + "early-sudo", + &format!( + "#!/bin/sh\nprintf early\n{}", + std::fs::read_to_string(descending_sudo(dir.path())) + .expect("the stub") + .trim_start_matches("#!/bin/sh\n") + ), + ); + let pairs: Vec<(&str, Box)> = vec![ + ( + "local", + Box::new( + LocalExecutor::new("seat", SudoAuth::NonInteractive) + .with_sudo_bin(early.clone()), + ), + ), + ( + "ssh", + Box::new(ssh_with(dir.path(), &early, SudoAuth::NonInteractive)), + ), + ]; + for (label, exec) in pairs { + let mut channel = exec + .open_channel(Identity::Root, "/bin/cat", &[], ROOMY) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + assert_eq!(echo(&mut channel, b"-late"), b"early-late", "{label}"); + assert_eq!(channel.wait().expect("wait").code, Some(0), "{label}"); + } + } + + #[test] + fn identities_resolve_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' '{SUDO_OK_SENTINEL}' >&2\n" + ), + ); + let local = LocalExecutor::new("seat", SudoAuth::NonInteractive) + .with_sudo_bin(recording.clone()); + let ssh = ssh_with(dir.path(), &recording, SudoAuth::NonInteractive); + let password = LocalExecutor::new("seat", SudoAuth::Password("pw".to_string())) + .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 service = Identity::Service(ServiceAccount::Insight); + let pairs: Vec<(&str, &dyn Executor, Identity)> = vec![ + ("local root", &local, Identity::Root), + ("local service", &local, service), + ("local root, password", &password, Identity::Root), + ("local service, password", &password, service), + ("ssh root", &ssh, Identity::Root), + ("ssh service", &ssh, service), + ]; + for (label, exec, identity) in pairs { + let run = exec + .run(identity, "/usr/bin/printf", &["%s", "marker"]) + .expect("run"); + let mut channel = exec + .open_channel(identity, "/usr/bin/printf", &["%s", "marker"], ROOMY) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + let argv = read_out(&mut channel); + assert_eq!(channel.wait().expect("wait").code, Some(0), "{label}"); + let run_prefix = prefix(&run.stdout, "sh"); + assert!(!run_prefix.is_empty(), "{label}: sudo must be involved"); + assert_eq!( + prefix(&argv, "/bin/sh"), + run_prefix, + "{label}: the elevation must match run's" + ); + let words: Vec<&str> = + std::str::from_utf8(&argv).expect("utf-8").lines().collect(); + let shell = words.iter().position(|word| *word == "/bin/sh"); + assert_eq!( + shell.and_then(|at| words.get(at + 1..at + 2)), + Some(&["-c"][..]), + "{label}: {words:?}" + ); + assert!( + words.ends_with(&["/usr/bin/printf", "%s", "marker"]), + "{label}: the command and its arguments stay discrete words: {words:?}" + ); + } + // The identity that involves no `sudo` runs the command bare. + for (label, exec) in [("local", &local as &dyn Executor), ("ssh", &ssh)] { + let mut channel = exec + .open_channel( + Identity::Operator, + "/usr/bin/printf", + &["%s", "marker"], + ROOMY, + ) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + assert_eq!( + read_out(&mut channel), + b"marker", + "{label}: no sudo may be involved" + ); + assert_eq!(channel.wait().expect("wait").code, Some(0), "{label}"); + } + } + + #[test] + fn the_ssh_invocation_is_run_s_with_no_terminal() { + // A stub `ssh` that prints its argv and answers as the remote + // start script and wrapper would, so both methods complete. + let dir = tempfile::tempdir().expect("tempdir"); + let recording = write_script( + dir.path(), + "channel-recording-ssh", + &format!( + "#!/bin/sh\nfor arg in \"$@\"; do printf '%s\\n' \"$arg\"; done\n\ + printf '%s\\n{RC_MARKER}0\\n' '{SUDO_OK_SENTINEL}' >&2\n" + ), + ); + let config = crate::transport::Ssh { + user: "ops".to_string(), + port: 2222, + key: PathBuf::from("/keys/id_ed25519"), + host_key: crate::transport::HostKeyPolicy::AcceptNew, + }; + for prompt in [SshPrompt::Deny, SshPrompt::Allow] { + for identity in [ + Identity::Operator, + Identity::Root, + Identity::Service(ServiceAccount::Roxyd), + ] { + let exec = |bin: &Path| { + SshExecutor::from_config( + "target", + &config, + "10.0.0.10", + SudoAuth::NonInteractive, + prompt, + ) + .with_ssh_bin(bin.to_path_buf()) + }; + let run = exec(&recording) + .run(identity, "/bin/cat", &[]) + .expect("run over recording ssh"); + let mut channel = exec(&recording) + .open_channel(identity, "/bin/cat", &[], ROOMY) + .unwrap_or_else(|error| panic!("{prompt:?} {identity:?}: {error:?}")); + let argv = read_out(&mut channel); + assert_eq!(channel.wait().expect("wait").code, Some(0)); + let words = |argv: &[u8]| -> Vec { + let mut words: Vec = String::from_utf8_lossy(argv) + .lines() + .map(str::to_string) + .collect(); + // The remote command line, which differs by design, + // spans the lines after the target. + let target = words + .iter() + .position(|word| word == "ops@10.0.0.10") + .expect("the target"); + words.truncate(target + 1); + words + }; + let channel_words = words(&argv); + assert_eq!( + channel_words, + words(&run.stdout), + "{prompt:?} {identity:?}: the ssh prefix must match run's" + ); + assert_eq!( + channel_words.iter().any(|word| word == "BatchMode=yes"), + prompt == SshPrompt::Deny, + "{prompt:?}: {channel_words:?}" + ); + assert!( + !channel_words + .iter() + .any(|word| word == "-t" || word == "-tt"), + "no terminal is requested: {channel_words:?}" + ); + } + } + } + + #[test] + fn a_password_line_is_consumed_before_the_callers_bytes() { + let dir = tempfile::tempdir().expect("tempdir"); + let sudo = password_sudo(dir.path()); + let auth = || SudoAuth::Password("s3cret".to_string()); + let input = pattern(ECHO_LEN); + 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 mut channel = exec + .open_channel(identity, "/bin/cat", &[], ROOMY) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + let echoed = echo(&mut channel, &input); + assert!( + echoed == input, + "{label}: the caller's first byte must be the command's first" + ); + assert_eq!(channel.wait().expect("wait").code, Some(0), "{label}"); + } + } + } + + #[test] + fn failures_before_the_start_are_reported_as_run_reports_them() { + 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 denying = write_script( + dir.path(), + "denying-sudo", + "#!/bin/sh\necho 'ops is not in the sudoers file.' >&2\nexit 1\n", + ); + let open = |exec: &dyn Executor, identity| { + exec.open_channel(identity, "/bin/cat", &[], ROOMY) + .expect_err("the start fails") + }; + + let error = open( + &LocalExecutor::new("mgmt", SudoAuth::NonInteractive) + .with_sudo_bin(refusing.clone()), + Identity::Root, + ); + assert!( + matches!(&error, ChannelError::Executor(ExecutorError::Elevation { host }) + if host == "mgmt"), + "got {error:?}" + ); + let error = open( + &LocalExecutor::new("mgmt", SudoAuth::NonInteractive) + .with_sudo_bin(denying.clone()), + Identity::Service(ServiceAccount::Security), + ); + assert!( + matches!(&error, ChannelError::Executor(ExecutorError::SudoRefused { host, reason }) + if host == "mgmt" && reason.contains("sudoers")), + "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 ssh = || { + SshExecutor::from_config( + "mgmt", + &config, + "10.0.0.10", + SudoAuth::NonInteractive, + SshPrompt::Deny, + ) + }; + let unreachable = ssh().with_ssh_bin(failing_ssh(dir.path())); + for identity in [Identity::Operator, Identity::Root] { + let error = open(&unreachable, identity); + assert!( + matches!(&error, ChannelError::Executor(ExecutorError::Connection { host, reason }) + if host == "mgmt" && reason.contains("Connection refused")), + "{identity:?}: got {error:?}" + ); + } + let remote_refusing = ssh() + .with_ssh_bin(fake_ssh(dir.path())) + .with_remote_sudo(refusing.to_string_lossy().into_owned()); + let error = open(&remote_refusing, Identity::Root); + assert!( + matches!(&error, ChannelError::Executor(ExecutorError::Elevation { host }) + if host == "mgmt"), + "got {error:?}" + ); + let remote_denying = ssh() + .with_ssh_bin(fake_ssh(dir.path())) + .with_remote_sudo(denying.to_string_lossy().into_owned()); + let error = open(&remote_denying, Identity::Root); + assert!( + matches!(&error, ChannelError::Executor(ExecutorError::SudoRefused { reason, .. }) + if reason.contains("sudoers")), + "got {error:?}" + ); + } + + #[test] + fn a_start_not_proven_in_time_is_killed_and_reaped() { + let dir = tempfile::tempdir().expect("tempdir"); + let silent = write_script( + dir.path(), + "silent-sudo", + "#!/bin/sh\n[ \"$1\" = warm ] && exit 0\n\ + printf 'sudo: pid %s\\n' \"$$\" >&2\nexec /bin/sleep 300\n", + ); + // Run once first, so the first run of a fresh executable — + // which macOS scans — is not what the short timeout measures. + let warmed = + spawn_retrying_text_busy(std::process::Command::new(&silent).arg("warm")) + .and_then(|mut child| child.wait()) + .expect("warm the stub"); + assert!(warmed.success()); + let limits = ChannelLimits { + elevation_timeout: Duration::from_millis(200), + ..ROOMY + }; + let started = Instant::now(); + let error = LocalExecutor::new("mgmt", SudoAuth::NonInteractive) + .with_sudo_bin(silent) + .open_channel(Identity::Root, "/bin/cat", &[], limits) + .expect_err("the sentinel never arrives"); + assert!( + started.elapsed() < REAP_WAIT, + "took {:?}", + started.elapsed() + ); + let ChannelError::ElevationTimedOut { + host, + timeout, + diagnostic, + } = error + else { + panic!("expected ElevationTimedOut, got {error:?}"); + }; + assert_eq!(host, "mgmt"); + assert_eq!(timeout, limits.elevation_timeout); + let transport = diagnostic + .strip_prefix("sudo: pid ") + .unwrap_or_else(|| panic!("the preamble read so far: {diagnostic:?}")); + let transport = pid(transport); + assert!( + rustix::process::test_kill_process(transport).is_err(), + "the transport was reaped before the error returned" + ); + } + + #[test] + fn a_rejected_password_times_out_with_sudos_complaint() { + // `sudo -S` asks again rather than exiting on a wrong password. + let dir = tempfile::tempdir().expect("tempdir"); + let asking = write_script( + dir.path(), + "asking-sudo", + "#!/bin/sh\nwhile IFS= read -r line; do\n \ + [ \"$line\" = s3cret ] && exit 0\n \ + echo 'Sorry, try again.' >&2\ndone\n", + ); + let limits = ChannelLimits { + elevation_timeout: Duration::from_millis(500), + ..ROOMY + }; + let error = LocalExecutor::new("mgmt", SudoAuth::Password("wrong".to_string())) + .with_sudo_bin(asking) + .open_channel(Identity::Root, "/bin/cat", &[], limits) + .expect_err("the password was rejected"); + assert!( + matches!(&error, ChannelError::ElevationTimedOut { diagnostic, .. } + if diagnostic.contains("Sorry, try again")), + "got {error:?}" + ); + } + + #[test] + fn a_transport_that_floods_stderr_before_the_start_is_killed_and_classified() { + 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) + .open_channel(Identity::Root, "/bin/cat", &[], ROOMY) + .expect_err("sudo never granted"); + assert!( + matches!( + error, + ChannelError::Executor(ExecutorError::SudoRefused { .. }) + ), + "got {error:?}" + ); + assert!( + started.elapsed() < ROOMY.elevation_timeout, + "abandoned, not timed out" + ); + } + + #[test] + fn an_ssh_channel_that_loses_its_exit_status_has_no_code() { + let dir = tempfile::tempdir().expect("tempdir"); + let lossy = write_script( + dir.path(), + "lossy-ssh", + &format!( + "#!/bin/sh\nprintf '%s' '{SUDO_OK_SENTINEL}' >&2\n\ + /bin/cat\necho 'Connection to 10.0.0.10 closed.' >&2\nexit 255\n" + ), + ); + let config = crate::transport::Ssh { + user: "ops".to_string(), + port: 22, + key: PathBuf::from("/dev/null"), + host_key: crate::transport::HostKeyPolicy::Strict, + }; + let exec = SshExecutor::from_config( + "mgmt", + &config, + "10.0.0.10", + SudoAuth::NonInteractive, + SshPrompt::Deny, + ) + .with_ssh_bin(lossy); + let mut channel = exec + .open_channel(Identity::Root, "/bin/cat", &[], ROOMY) + .expect("the start was announced"); + assert_eq!(echo(&mut channel, b"frame"), b"frame"); + let error = channel.wait().expect_err("no exit-status line"); + assert!( + matches!(&error, ChannelError::ExitUnknown { host, reason } + if host == "mgmt" && reason.contains("closed")), + "got {error:?}" + ); + } + + /// A command that records its pid in the file named by `$1`, then + /// blocks on standard input. + const BLOCKING: &str = + "echo \"$$\" > \"$1.tmp\"; /bin/mv \"$1.tmp\" \"$1\"; exec /bin/cat"; + + #[test] + fn kill_and_drop_end_the_transport_and_the_command_sees_eof_on_every_pair() { + let dir = tempfile::tempdir().expect("tempdir"); + for how in ["kill", "drop"] { + for (index, (label, exec, identity)) in every_pair(&dir).into_iter().enumerate() + { + let pid_file = dir.path().join(format!("{how}-{index}")); + let channel = exec + .open_channel( + identity, + "/bin/sh", + &["-c", BLOCKING, "sh", &pid_file.to_string_lossy()], + ROOMY, + ) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + let command = recorded_pid(&pid_file); + let transport = pid(&channel.transport_pid().to_string()); + match how { + "kill" => channel + .kill() + .unwrap_or_else(|error| panic!("{label}: {error:?}")), + _ => drop(channel), + } + assert!( + rustix::process::test_kill_process(transport).is_err(), + "{label} {how}: the transport was reaped" + ); + assert_gone(&format!("{label} {how}"), command); + } + } + } + + #[test] + fn wait_gives_stderr_a_grace_when_a_descendant_holds_it() { + // A descendant that outlives the command keeps stderr open; + // `wait` returns once the grace passes, and says so. + let dir = tempfile::tempdir().expect("tempdir"); + let exec = LocalExecutor::new("seat", SudoAuth::NonInteractive) + .with_sudo_bin(descending_sudo(dir.path())); + let started = Instant::now(); + let channel = exec + .open_channel( + Identity::Operator, + "/bin/sh", + &["-c", "printf err >&2; /bin/sleep 8 >/dev/null & exit 4"], + ROOMY, + ) + .expect("open"); + let exit = channel.wait().expect("wait"); + let elapsed = started.elapsed(); + assert_eq!(exit.code, Some(4)); + assert_eq!(exit.stderr, b"err"); + assert!(exit.stderr_truncated, "the stream had not ended"); + assert!( + elapsed >= Duration::from_secs(5) && elapsed < Duration::from_secs(8), + "waited {elapsed:?}" + ); + } + + #[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 marking = write_script( + dir.path(), + "marking-stub", + &format!("#!/bin/sh\n: > '{}'\nexit 1\n", marker.display()), + ); + let config = crate::transport::Ssh { + user: "ops".to_string(), + port: 22, + key: PathBuf::from("/dev/null"), + host_key: crate::transport::HostKeyPolicy::Strict, + }; + let local = LocalExecutor::new("seat", SudoAuth::NonInteractive) + .with_sudo_bin(marking.clone()); + let ssh = SshExecutor::from_config( + "target", + &config, + "10.0.0.10", + SudoAuth::NonInteractive, + SshPrompt::Deny, + ) + .with_ssh_bin(marking); + for command in ["cat", "/bin/a=b"] { + for exec in [&local as &dyn Executor, &ssh] { + for identity in [ + Identity::Operator, + Identity::Root, + Identity::Service(ServiceAccount::Security), + ] { + let error = exec + .open_channel(identity, command, &[], ROOMY) + .expect_err("the command must be refused"); + assert!( + matches!(&error, ChannelError::InvalidCommand { command: named } + if named == command), + "{identity:?}: got {error:?}" + ); + assert!(!marker.exists(), "nothing may have been spawned"); + } + } + } + } + + #[test] + fn the_default_body_and_the_daemon_are_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()) + } + } + for (label, exec) in [ + ("default", &RunOnly as &dyn Executor), + ("daemon", &InDaemonExecutor::new("seat")), + ] { + for identity in [Identity::Root, Identity::Service(ServiceAccount::Roxyd)] { + let error = exec + .open_channel(identity, "/bin/cat", &[], ROOMY) + .expect_err("unsupported"); + assert!( + matches!(error, ChannelError::Unsupported), + "{label}: got {error:?}" + ); + } + } + } + } } } diff --git a/src/executor/bounded.rs b/src/executor/bounded.rs index f8fc6c7..905a4c4 100644 --- a/src/executor/bounded.rs +++ b/src/executor/bounded.rs @@ -41,7 +41,7 @@ const EXIT_POLL: Duration = Duration::from_millis(5); /// 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); +pub(super) 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`. @@ -57,18 +57,18 @@ const TIMEOUT_MARKER_PREFIX: &str = "__BOOTLER_TIMEOUT_"; 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 most stderr a supervised run's transport, or a channel's, 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. +pub(super) 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 /// the one fixed location both Linux and macOS guarantee. -const ENV: &str = "/usr/bin/env"; +pub(super) 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. @@ -673,7 +673,7 @@ fn await_unreaped(pid: Pid, grace: Duration) { } /// Waits up to `grace` for `child` to exit and reaps it. -fn reap_within(child: &mut Child, grace: Duration) { +pub(super) 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. @@ -687,15 +687,15 @@ fn reap_within(child: &mut Child, grace: Duration) { /// When a run must end. `None` is a timeout too large to represent as an /// instant, which never passes. -struct Deadline(Option); +pub(super) struct Deadline(Option); impl Deadline { - fn after(timeout: Duration) -> Self { + pub(super) 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 { + pub(super) fn remaining(&self) -> Option { match self.0 { Some(at) => at .checked_duration_since(Instant::now()) @@ -706,7 +706,7 @@ impl Deadline { } /// Returns the position of the first occurrence of `needle` in `haystack`. -fn find(haystack: &[u8], needle: &[u8]) -> Option { +pub(super) fn find(haystack: &[u8], needle: &[u8]) -> Option { haystack .windows(needle.len()) .position(|window| window == needle) diff --git a/src/executor/channel.rs b/src/executor/channel.rs new file mode 100644 index 0000000..91be7e1 --- /dev/null +++ b/src/executor/channel.rs @@ -0,0 +1,828 @@ +//! The long-lived channel behind [`Executor::open_channel`]. +//! +//! [`open_started`] spawns a transport — `sudo`, or `ssh` — that runs the +//! command under [`start_script`], and returns only once that script has +//! announced the command's start with [`SUDO_OK_SENTINEL`]. [`open_direct`] +//! spawns a command with nothing in between and returns at once. Either way +//! the caller then owns standard input and standard output, while standard +//! error stays with the [`Channel`] and is drained on a thread of its own for +//! the channel's whole life, so the command never blocks on a full pipe. +//! +//! That thread cannot sit in a blocking `read`: a thread blocked there +//! cannot be cancelled, and a pipe some descendant of the command still holds +//! open would pin it past the channel's end. So, as in [`bounded`], the pipe +//! is non-blocking and waited on with `poll(2)`, together with a wake pipe +//! whose closing tells the thread to stop. +//! +//! [`Executor::open_channel`]: super::Executor::open_channel +//! [`bounded`]: super::bounded + +use std::io::{ErrorKind, PipeReader, PipeWriter, Read, Write}; +use std::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command, Stdio}; +use std::sync::mpsc::{self, RecvTimeoutError}; +use std::thread::JoinHandle; +use std::time::Duration; + +use rustix::event::{PollFd, PollFlags, Timespec}; +use rustix::io::Errno; + +use super::bounded::{ + Deadline, ENV, KILL_GRACE, SUPERVISOR_SHELL, TRANSPORT_STDERR_LIMIT, find, reap_within, +}; +use super::{CommandOutput, ExecutorError, RC_MARKER, SUDO_OK_SENTINEL, spawn_retrying_text_busy}; + +/// How long [`Channel::wait`] waits for standard error to end once the local +/// transport process has exited. A descendant of the command can hold the +/// pipe open indefinitely; past this, what was read is what is returned, and +/// the rest is noted as truncated. +pub(super) const STDERR_GRACE: Duration = Duration::from_secs(5); +/// Bytes read from standard error per readiness. +const READ_CHUNK: usize = 8192; +/// Bytes at the end of an SSH channel's standard error held back from the +/// command's until the stream ends, because they may be — or hold — the +/// wrapper's exit-status line. Far more than the line itself needs, so that +/// the end of the stream can also explain a missing line. +const STATUS_HOLD: usize = 512; + +/// The bounds one [`Executor::open_channel`](super::Executor::open_channel) +/// call opens under. +/// +/// There is no `Default`: how long elevation may take and how much of the +/// command's standard error is worth keeping are the caller's decisions about +/// that command, not the executor's. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ChannelLimits { + /// How long the start may take to prove — from spawning the transport to + /// reading its announcement that the command has started. Where nothing + /// stands between this process and the command, the start is proven by + /// the spawn itself and this is unused. + pub elevation_timeout: Duration, + /// How many bytes of the command's own standard error are kept for + /// [`ChannelExit::stderr`]. The rest is read and discarded. + pub max_stderr: usize, +} + +/// How a [`Channel`]'s command ended. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ChannelExit { + /// The command's own exit code. `None` only where the local transport + /// process — the command itself for [`Identity::Operator`] on + /// [`LocalExecutor`], else `sudo` — was ended by a signal; over + /// [`SshExecutor`] an exit code that did not arrive is + /// [`ChannelError::ExitUnknown`] instead. + /// + /// [`Identity::Operator`]: super::Identity::Operator + /// [`LocalExecutor`]: super::LocalExecutor + /// [`SshExecutor`]: super::SshExecutor + pub code: Option, + /// At most [`ChannelLimits::max_stderr`] bytes of the command's own + /// standard error, from its start: nothing the transport wrote before the + /// command started, and not the SSH exit-status line. + pub stderr: Vec, + /// Whether standard error held more than [`ChannelExit::stderr`] keeps — + /// bytes past [`ChannelLimits::max_stderr`] were discarded, or the stream + /// had not ended within 5 seconds of the transport exiting. + pub stderr_truncated: bool, +} + +/// Errors raised by [`Executor::open_channel`](super::Executor::open_channel) +/// and by the [`Channel`] it returns. +/// +/// A type of its own rather than more [`ExecutorError`] variants, for the +/// reason [`RunWithInputError`](super::RunWithInputError) is one: these arise +/// from the channel alone. +#[derive(Debug, thiserror::Error)] +pub enum ChannelError { + /// `command` is not an absolute path, or contains `=`, and was refused + /// before anything was spawned. + /// + /// The command runs with an empty environment, so with no `PATH`, and is + /// started through `env -i` wherever `sudo` or SSH stands in between — + /// where an operand containing `=` would be read as an assignment. + #[error("command `{command}` is not an absolute path free of `=`")] + InvalidCommand { + /// The command as the caller named it. + command: String, + }, + /// The transport had not announced the command's start within + /// [`ChannelLimits::elevation_timeout`]. It was killed and reaped. + /// + /// A `sudo -S` that rejected the password waits for another line rather + /// than exiting, so a wrong password ends here, its `Sorry, try again.` + /// in `diagnostic`. + #[error("host `{host}`: the channel did not start within {timeout:?}: {diagnostic}")] + ElevationTimedOut { + /// The host whose transport did not start the command. + host: String, + /// The timeout it outlived. + timeout: Duration, + /// What the transport wrote on standard error before it was killed. + diagnostic: String, + }, + /// The SSH transport ended without reporting the remote command's exit + /// status, so the command's outcome is not known — the connection was + /// lost, or `ssh` was ended some other way. Never a guessed code. + #[error("host `{host}`: the channel's exit status is unknown: {reason}")] + ExitUnknown { + /// The host whose command's outcome is unknown. + host: String, + /// Why, with the end of what the transport wrote on standard error. + reason: String, + }, + /// The executor does not implement + /// [`Executor::open_channel`](super::Executor::open_channel). + /// + /// The trait's default body returns this, and so does + /// [`InDaemonExecutor`](super::InDaemonExecutor), which keeps it. + #[error("this executor cannot open a channel")] + Unsupported, + /// Spawning, the transport or elevation failed, exactly as + /// [`Executor::run`](super::Executor::run) reports it. + #[error(transparent)] + Executor(#[from] ExecutorError), +} + +/// Refuses a command a channel cannot start as named. +pub(super) fn check_command(command: &str) -> Result<(), ChannelError> { + if super::is_absolute_and_plain(command) { + Ok(()) + } else { + Err(ChannelError::InvalidCommand { + command: command.to_string(), + }) + } +} + +/// Returns the `sh -c` script a channel's command starts under wherever +/// `sudo` or SSH stands in between: it announces the start on standard error +/// with [`SUDO_OK_SENTINEL`], then replaces itself with the command, run with +/// an empty environment. Invoked as `sh -c SCRIPT `, so the +/// command and every argument arrive positionally and are never spliced into +/// the script text. +pub(super) fn start_script() -> String { + format!("printf '%s' '{SUDO_OK_SENTINEL}' >&2; exec {ENV} -i \"$0\" \"$@\"") +} + +/// The shell [`start_script`] runs under, named absolutely so the start +/// depends on no `PATH`. +pub(super) const START_SHELL: &str = SUPERVISOR_SHELL; + +/// What [`open_started`] needs to know beyond the command it spawns. +pub(super) struct Start<'a> { + /// The host the transport reaches, for the errors that name it. + pub(super) host: &'a str, + /// The line to write on standard input before anything else — `sudo -S`'s + /// password — or `None`. + pub(super) password_line: Option>, + /// Whether the transport appends the SSH wrapper's exit-status line. + pub(super) remote_code: bool, + /// The caller's limits. + pub(super) limits: ChannelLimits, +} + +/// Spawns `command` — a transport running the command under +/// [`start_script`] — and returns the channel once the start is proven. +/// +/// The password line, if any, is written first. Standard error is then read, +/// and standard output never, until [`SUDO_OK_SENTINEL`] arrives; what follows +/// it is the command's. A transport that ends first, or writes more than +/// [`TRANSPORT_STDERR_LIMIT`] first, is killed and reaped, and what it wrote +/// is handed to `refusal` to be classified as +/// [`Executor::run`](super::Executor::run) classifies it. One that does +/// neither within the elevation timeout is killed and reaped too. +/// +/// # Errors +/// +/// Returns [`ChannelError::ElevationTimedOut`] for a start that did not prove +/// itself in time, [`ChannelError::Executor`] carrying `refusal`'s error for +/// one that failed, and [`ExecutorError::Spawn`] when the transport cannot be +/// spawned or its pipes fail. +pub(super) fn open_started( + mut command: Command, + start: &Start<'_>, + refusal: impl FnOnce(CommandOutput) -> ExecutorError, +) -> Result { + let program = command.get_program().to_string_lossy().into_owned(); + let mut spawned = spawn(&mut command, &program)?; + let settled = settle_start( + &mut spawned.stdin, + &mut spawned.stderr, + start.password_line.as_deref(), + start.limits.elevation_timeout, + ); + let started = match settled { + Ok(Settled::Started(started)) => started, + Ok(Settled::Ended(stderr)) => { + let _ = kill_and_reap(&mut spawned.child); + return Err(refusal(CommandOutput { + code: None, + stdout: Vec::new(), + stderr, + }) + .into()); + } + Ok(Settled::TimedOut(stderr)) => { + let _ = kill_and_reap(&mut spawned.child); + return Err(ChannelError::ElevationTimedOut { + host: start.host.to_string(), + timeout: start.limits.elevation_timeout, + diagnostic: String::from_utf8_lossy(&stderr).trim().to_string(), + }); + } + Err(source) => { + let _ = kill_and_reap(&mut spawned.child); + return Err(spawn_failed(&program, source)); + } + }; + Channel::assemble( + spawned, + started, + start.limits.max_stderr, + start.remote_code.then(|| start.host.to_string()), + program, + ) +} + +/// Spawns `command` with nothing standing between this process and it, and +/// returns its channel at once: the spawn is the start. +/// +/// The caller has already set the command's environment. +/// +/// # Errors +/// +/// Returns [`ExecutorError::Spawn`] when the command cannot be spawned or its +/// standard error cannot be drained. +pub(super) fn open_direct( + mut command: Command, + max_stderr: usize, +) -> Result { + let program = command.get_program().to_string_lossy().into_owned(); + let spawned = spawn(&mut command, &program)?; + Channel::assemble(spawned, Vec::new(), max_stderr, None, program) +} + +/// A spawned transport and its three pipes. +struct Spawned { + child: Child, + stdin: ChildStdin, + stdout: ChildStdout, + stderr: ChildStderr, +} + +/// Spawns `command` with all three streams piped, in the caller's process +/// group, and takes the pipes. +fn spawn(command: &mut Command, program: &str) -> Result { + command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + let mut child = + spawn_retrying_text_busy(command).map_err(|source| spawn_failed(program, source))?; + let (Some(stdin), Some(stdout), Some(stderr)) = + (child.stdin.take(), child.stdout.take(), child.stderr.take()) + else { + let _ = kill_and_reap(&mut child); + return Err(spawn_failed( + program, + std::io::Error::other("a standard stream was not piped"), + )); + }; + Ok(Spawned { + child, + stdin, + stdout, + stderr, + }) +} + +fn spawn_failed(program: &str, source: std::io::Error) -> ChannelError { + ChannelError::Executor(ExecutorError::Spawn { + command: program.to_string(), + source, + }) +} + +/// How the start of a channel settled. +enum Settled { + /// The start was announced; these bytes followed the announcement and are + /// the command's. + Started(Vec), + /// The transport's standard error ended, or passed + /// [`TRANSPORT_STDERR_LIMIT`], before the announcement: all of it. + Ended(Vec), + /// The elevation timeout passed first, with what had been written by then. + TimedOut(Vec), +} + +/// Writes `password` and reads standard error until the start is announced, +/// the transport gives up, or `timeout` passes. Standard output is not read. +/// +/// Standard input is non-blocking only while the password is written, so a +/// transport that never reads it cannot hold the start past `timeout`; the +/// caller writes to it with ordinary blocking writes afterwards. +fn settle_start( + stdin: &mut ChildStdin, + stderr: &mut ChildStderr, + password: Option<&[u8]>, + timeout: Duration, +) -> std::io::Result { + rustix::io::ioctl_fionbio(&*stderr, true)?; + if password.is_none() { + return feed_and_settle(stdin, stderr, b"", timeout); + } + rustix::io::ioctl_fionbio(&*stdin, true)?; + let settled = feed_and_settle(stdin, stderr, password.unwrap_or_default(), timeout)?; + rustix::io::ioctl_fionbio(&*stdin, false)?; + Ok(settled) +} + +/// [`settle_start`]'s loop, over pipes already made non-blocking. +fn feed_and_settle( + stdin: &mut ChildStdin, + stderr: &mut ChildStderr, + unwritten: &[u8], + timeout: Duration, +) -> std::io::Result { + let deadline = Deadline::after(timeout); + let sentinel = SUDO_OK_SENTINEL.as_bytes(); + let mut written = 0; + let mut read = Vec::new(); + let mut chunk = vec![0; READ_CHUNK]; + loop { + if let Some(at) = find(&read, sentinel) { + return Ok(Settled::Started(read.split_off(at + sentinel.len()))); + } + if read.len() > TRANSPORT_STDERR_LIMIT { + return Ok(Settled::Ended(read)); + } + let Some(left) = deadline.remaining() else { + return Ok(Settled::TimedOut(read)); + }; + // A time left too large for a `timespec` is no bound at all. + let wait = Timespec::try_from(left).ok(); + let feeding = written < unwritten.len(); + let (err_ready, in_ready) = { + let mut fds = vec![PollFd::new(&*stderr, PollFlags::IN)]; + if feeding { + fds.push(PollFd::new(&*stdin, PollFlags::OUT)); + } + match rustix::event::poll(&mut fds, wait.as_ref()) { + Ok(_) => {} + Err(Errno::INTR) => continue, + Err(errno) => return Err(errno.into()), + } + // A hang-up or an error is readiness too: the read or write it + // wakes reports what happened. + let ready = |index: usize| fds.get(index).is_some_and(|fd| !fd.revents().is_empty()); + (ready(0), ready(1)) + }; + if in_ready { + match stdin.write(unwritten.get(written..).unwrap_or_default()) { + Ok(count) => written += count, + Err(error) if is_transient(&error) => {} + // The transport closed its standard input without reading + // the password; what it wrote on standard error says why. + Err(error) if error.kind() == ErrorKind::BrokenPipe => written = unwritten.len(), + Err(error) => return Err(error), + } + } + if err_ready { + match stderr.read(&mut chunk) { + Ok(0) => return Ok(Settled::Ended(read)), + Ok(count) => read.extend_from_slice(chunk.get(..count).unwrap_or_default()), + Err(error) if is_transient(&error) => {} + Err(error) => return 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) +} + +/// `SIGKILL`s `child` — the process alone, never its group, which is the +/// caller's — and reaps it. A child this process may not signal is given +/// [`KILL_GRACE`] to exit on its own rather than waited on without bound. +fn kill_and_reap(child: &mut Child) -> std::io::Result<()> { + match child.kill() { + Ok(()) => child.wait().map(drop), + Err(error) => { + reap_within(child, KILL_GRACE); + Err(error) + } + } +} + +/// A long-lived command started by +/// [`Executor::open_channel`](super::Executor::open_channel), with its +/// standard input and standard output for the caller to take and standard +/// error drained for it. +/// +/// Bytes pass verbatim both ways, with no size or time bound from this crate. +/// The channel ends by [`Channel::wait`] or [`Channel::kill`]; one dropped +/// without either is killed as [`Channel::kill`] kills it. +/// +/// **What a kill reaches is limited.** It `SIGKILL`s the local transport +/// process — the command itself for [`Identity::Operator`] on +/// [`LocalExecutor`], else `sudo` or `ssh` — and reaps it. A command started +/// through `sudo` or over SSH is not signalled: it learns of the end only by +/// end of file on its standard input and `EPIPE` on its standard output. +/// Descendants of the command are not tracked on any transport. Whatever +/// survives is the caller's to contain. +/// +/// [`Identity::Operator`]: super::Identity::Operator +/// [`LocalExecutor`]: super::LocalExecutor +#[derive(Debug)] +pub struct Channel { + child: Child, + stdin: Option, + stdout: Option, + drain: Option, + /// The host an SSH channel reaches, whose exit code arrives in the + /// exit-status line; `None` where the local transport's status is the + /// code. + remote_host: Option, + program: String, + /// Whether [`Channel::wait`] or [`Channel::kill`] has ended the channel, + /// so dropping it has nothing left to do. + ended: bool, +} + +impl Channel { + /// Starts draining the spawned transport's standard error, whose first + /// bytes past the start are `first`, and assembles the channel. The + /// transport is killed and reaped if the drain cannot start. + fn assemble( + spawned: Spawned, + first: Vec, + max_stderr: usize, + remote_host: Option, + program: String, + ) -> Result { + let Spawned { + mut child, + stdin, + stdout, + stderr, + } = spawned; + let hold = if remote_host.is_some() { + STATUS_HOLD + } else { + 0 + }; + match Drain::start(stderr, first, Sink::new(max_stderr, hold)) { + Ok(drain) => Ok(Self { + child, + stdin: Some(stdin), + stdout: Some(stdout), + drain: Some(drain), + remote_host, + program, + ended: false, + }), + Err(source) => { + let _ = kill_and_reap(&mut child); + Err(spawn_failed(&program, source)) + } + } + } + + /// Takes the command's standard input. Returns `None` once taken. + /// + /// Dropping it closes the pipe, which is how the command learns its input + /// has ended. + pub fn take_stdin(&mut self) -> Option { + self.stdin.take() + } + + /// Takes the command's standard output. Returns `None` once taken. + /// + /// Until it is taken the channel holds it open and nothing reads it, so a + /// command that writes more than a pipe holds blocks. + pub fn take_stdout(&mut self) -> Option { + self.stdout.take() + } + + /// Waits for the command to end and returns how it ended. + /// + /// Standard input is closed first if it was never taken. The local + /// transport process is then waited for with no time bound, and standard + /// error for at most 5 seconds more, since a descendant of the command + /// can hold it open; past that, [`ChannelExit::stderr_truncated`] is set. + /// + /// On [`LocalExecutor`](super::LocalExecutor) the code is the local + /// process's own status, which `sudo` passes the command's through. Over + /// [`SshExecutor`](super::SshExecutor) it is the remote command's own, read + /// from the wrapper's exit-status line — a remote `255` included, which + /// is the command's and never the transport's. + /// + /// # Errors + /// + /// Returns [`ChannelError::ExitUnknown`] when an SSH channel ended without + /// its exit-status line, and [`ExecutorError::Spawn`] when waiting for the + /// transport or reading standard error fails. + pub fn wait(mut self) -> Result { + self.stdin = None; + let status = match self.child.wait() { + Ok(status) => status, + // Dropping `self` kills and reaps what may still be running. + Err(source) => return Err(spawn_failed(&self.program, source)), + }; + self.ended = true; + let drained = self.stop_drain(STDERR_GRACE); + let drained = drained.map_err(|source| spawn_failed(&self.program, source))?; + let code = match (&self.remote_host, drained.remote_code) { + (None, _) => status.code(), + (Some(_), Some(code)) => Some(code), + (Some(host), None) => { + let ending = if drained.ended { + "standard error ended" + } else { + "standard error had not ended 5 seconds later" + }; + return Err(ChannelError::ExitUnknown { + host: host.clone(), + reason: format!( + "`{}` exited ({status}) without reporting the remote exit status; \ + {ending} with: {}", + self.program, + String::from_utf8_lossy(&drained.tail).trim() + ), + }); + } + }; + Ok(ChannelExit { + code, + stderr: drained.stderr, + stderr_truncated: drained.truncated || !drained.ended, + }) + } + + /// Ends the channel at once: closes the pipe ends it still holds, + /// `SIGKILL`s the local transport process and reaps it, and stops + /// draining standard error. + /// + /// See the type's documentation for what this does not reach. + /// + /// # Errors + /// + /// Returns [`ExecutorError::Spawn`] when the transport cannot be killed or + /// reaped — a `sudo` this process may not signal. The drain is stopped + /// either way. + pub fn kill(mut self) -> Result<(), ChannelError> { + self.ended = true; + self.shut_down() + .map_err(|source| spawn_failed(&self.program, source)) + } + + /// Closes the held pipes, kills and reaps the transport, and stops and + /// joins the drain. + fn shut_down(&mut self) -> std::io::Result<()> { + self.stdin = None; + self.stdout = None; + let killed = kill_and_reap(&mut self.child); + // Stopped rather than waited for: nothing more is wanted from it, + // and a drain that panicked has nothing to report here either. + if let Some(mut drain) = self.drain.take() { + drain.wake = None; + let _ = drain.handle.join(); + } + killed + } + + /// Gives the drain up to `grace` to reach the end of standard error, then + /// stops it and joins it. + fn stop_drain(&mut self, grace: Duration) -> std::io::Result { + let Some(mut drain) = self.drain.take() else { + return Err(std::io::Error::other("standard error was already drained")); + }; + // The drain never sends; it hangs up by ending. + let _: Result<(), RecvTimeoutError> = drain.done.recv_timeout(grace); + drain.wake = None; + match drain.handle.join() { + Ok(drained) => drained, + Err(panic) => std::panic::resume_unwind(panic), + } + } + + /// Returns the local transport process's id. + #[cfg(test)] + pub(super) fn transport_pid(&self) -> u32 { + self.child.id() + } +} + +impl Drop for Channel { + fn drop(&mut self) { + if !self.ended { + let _ = self.shut_down(); + } + } +} + +/// The thread draining a channel's standard error, and what stops it. +#[derive(Debug)] +struct Drain { + handle: JoinHandle>, + /// Dropping this wakes the thread and stops it. + wake: Option, + /// Hangs up when the thread ends. + done: mpsc::Receiver<()>, +} + +impl Drain { + fn start(stderr: ChildStderr, first: Vec, mut sink: Sink) -> std::io::Result { + rustix::io::ioctl_fionbio(&stderr, true)?; + let (asleep, wake) = std::io::pipe()?; + let (done_tx, done) = mpsc::channel(); + let handle = std::thread::Builder::new() + .name("channel-stderr".to_string()) + .spawn(move || { + let _done = done_tx; + sink.push(&first); + drain(stderr, &asleep, sink) + })?; + Ok(Self { + handle, + wake: Some(wake), + done, + }) + } +} + +/// What a drain read. +#[derive(Debug)] +struct Drained { + /// The command's own standard error, held to its limit. + stderr: Vec, + /// Whether bytes past the limit were discarded. + truncated: bool, + /// Whether standard error reached its end, rather than the drain being + /// stopped first. + ended: bool, + /// The remote exit code, where the stream ended on the exit-status line. + remote_code: Option, + /// The last bytes of the stream, to explain a missing exit-status line. + tail: Vec, +} + +/// Reads `stderr` into `sink` until it ends or `asleep` is woken. +fn drain(mut stderr: ChildStderr, asleep: &PipeReader, mut sink: Sink) -> std::io::Result { + let mut chunk = vec![0; READ_CHUNK]; + loop { + let (stop, ready) = { + let mut fds = [ + PollFd::new(asleep, PollFlags::IN), + PollFd::new(&stderr, PollFlags::IN), + ]; + match rustix::event::poll(&mut fds, None) { + Ok(_) => {} + Err(Errno::INTR) => continue, + Err(errno) => return Err(errno.into()), + } + let [woken, readable] = &fds; + (!woken.revents().is_empty(), !readable.revents().is_empty()) + }; + // Woken before reading, so a stream that never stops cannot keep the + // drain from stopping. + if stop { + return Ok(sink.finish(false)); + } + if ready { + match stderr.read(&mut chunk) { + Ok(0) => return Ok(sink.finish(true)), + Ok(count) => sink.push(chunk.get(..count).unwrap_or_default()), + Err(error) if is_transient(&error) => {} + Err(error) => return Err(error), + } + } + } +} + +/// Where the command's standard error goes: kept up to its limit, with a +/// trailing window held back where the exit-status line may end the stream. +struct Sink { + max: usize, + kept: Vec, + truncated: bool, + hold: usize, + held: Vec, +} + +impl Sink { + fn new(max: usize, hold: usize) -> Self { + Self { + max, + kept: Vec::new(), + truncated: false, + hold, + held: Vec::new(), + } + } + + fn push(&mut self, bytes: &[u8]) { + if self.hold == 0 { + self.keep_bytes(bytes); + return; + } + self.held.extend_from_slice(bytes); + let excess = self.held.len().saturating_sub(self.hold); + if excess > 0 { + let rest = self.held.split_off(excess); + let committed = std::mem::replace(&mut self.held, rest); + self.keep_bytes(&committed); + } + } + + fn keep_bytes(&mut self, bytes: &[u8]) { + let room = self.max.saturating_sub(self.kept.len()); + let (kept, discarded) = bytes.split_at(room.min(bytes.len())); + self.kept.extend_from_slice(kept); + self.truncated |= !discarded.is_empty(); + } + + /// Settles what was held: the exit-status line, where the stream ends on + /// one, is removed and read; everything else is the command's. + fn finish(mut self, ended: bool) -> Drained { + let held = std::mem::take(&mut self.held); + let (own, remote_code) = match split_status_line(&held) { + Some((cut, code)) => (held.get(..cut).unwrap_or_default(), Some(code)), + None => (held.as_slice(), None), + }; + self.keep_bytes(own); + Drained { + stderr: self.kept, + truncated: self.truncated, + ended, + remote_code, + tail: held, + } + } +} + +/// Finds the SSH wrapper's exit-status line — an optional newline, +/// [`RC_MARKER`], the code, a newline — at the very end of `tail`, returning +/// where the line starts and the code. +fn split_status_line(tail: &[u8]) -> Option<(usize, i32)> { + let body = tail.strip_suffix(b"\n")?; + let marker = RC_MARKER.as_bytes(); + let at = body + .windows(marker.len()) + .rposition(|window| window == marker)?; + let digits = body.get(at + marker.len()..)?; + if digits.is_empty() || !digits.iter().all(u8::is_ascii_digit) { + return None; + } + let code = std::str::from_utf8(digits).ok()?.parse::().ok()?; + let cut = match at.checked_sub(1) { + Some(before) if body.get(before) == Some(&b'\n') => before, + _ => at, + }; + Some((cut, code)) +} + +#[cfg(test)] +mod tests { + use super::{RC_MARKER, Sink, split_status_line}; + + #[test] + fn the_status_line_is_read_only_at_the_end_of_the_stream() { + let line = format!("own\n{RC_MARKER}255\n"); + assert_eq!(split_status_line(line.as_bytes()), Some((3, 255))); + let line = format!("{RC_MARKER}0\n"); + assert_eq!(split_status_line(line.as_bytes()), Some((0, 0))); + let line = format!("\n{RC_MARKER}3\nmore"); + assert_eq!(split_status_line(line.as_bytes()), None); + let line = format!("\n{RC_MARKER}\n"); + assert_eq!(split_status_line(line.as_bytes()), None); + let line = format!("\n{RC_MARKER}3"); + assert_eq!(split_status_line(line.as_bytes()), None); + } + + #[test] + fn a_sink_keeps_its_limit_and_removes_the_status_line() { + let mut sink = Sink::new(4, 32); + sink.push(b"abcdefghabcdefghabcdefghabcdefgh"); + sink.push(format!("ij\n{RC_MARKER}7\n").as_bytes()); + let drained = sink.finish(true); + assert_eq!(drained.stderr, b"abcd"); + assert!(drained.truncated); + assert_eq!(drained.remote_code, Some(7)); + + let mut sink = Sink::new(16, 64); + sink.push(format!("ab\n{RC_MARKER}1").as_bytes()); + sink.push(b"2\n"); + let drained = sink.finish(true); + assert_eq!(drained.stderr, b"ab"); + assert!(!drained.truncated); + assert_eq!(drained.remote_code, Some(12)); + + let mut sink = Sink::new(2, 0); + sink.push(b"ab"); + let drained = sink.finish(true); + assert_eq!(drained.stderr, b"ab"); + assert!(!drained.truncated, "reaching the limit is not passing it"); + } +} diff --git a/src/executor/test_support.rs b/src/executor/test_support.rs index d4dbb0c..c2cd846 100644 --- a/src/executor/test_support.rs +++ b/src/executor/test_support.rs @@ -10,9 +10,14 @@ //! 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. +//! [`Executor::run_with_input`] or [`Executor::open_channel`] call naming a +//! command that is not an absolute path free of `=` is recorded and refused +//! with [`RunWithInputError::InvalidCommand`] or +//! [`ChannelError::InvalidCommand`], and consumes no scripted outcome. +//! +//! [`Executor::open_channel`] is the one call that can start a process: a +//! scripted [`ScriptedChannel::Spawn`] runs the program it names, so a test can +//! drive a real [`Channel`] without any transport in between. //! //! This module is compiled only for this crate's tests and under the //! `test-support` feature. Enable that feature in a dependent's @@ -21,11 +26,12 @@ use std::collections::VecDeque; use std::path::{Path, PathBuf}; +use std::process::Command; use std::sync::{Mutex, MutexGuard, PoisonError}; use super::{ - CommandOutput, Executor, ExecutorError, FileMeta, Identity, OutputStream, RunLimits, - RunWithInputError, check_bounded_command, + Channel, ChannelError, ChannelLimits, CommandOutput, Executor, ExecutorError, FileMeta, + Identity, OutputStream, RunLimits, RunWithInputError, channel, check_bounded_command, }; /// One call a [`RecordingExecutor`] received. @@ -54,6 +60,17 @@ pub enum RecordedCall { /// The limits the command was to run under. limits: RunLimits, }, + /// [`Executor::open_channel`]. + OpenChannel { + /// Who the command was to run as. + identity: Identity, + /// The command. + command: String, + /// Every argument, in order. + args: Vec, + /// The limits the channel was to open under. + limits: ChannelLimits, + }, /// [`Executor::put_file`]. PutFile { /// The destination. @@ -83,6 +100,22 @@ pub enum ScriptedRun { Error(RunWithInputError), } +/// The outcome a test scripts for one [`Executor::open_channel`] call. +#[derive(Debug)] +pub enum ScriptedChannel { + /// Spawns `program` with `args` directly — no `sudo`, piped standard + /// streams, an empty environment — and returns it as the [`Channel`], in + /// place of the command the call named. + Spawn { + /// The program to spawn. + program: PathBuf, + /// Its arguments, in order. + args: Vec, + }, + /// Returns this error as it is. + Fail(ChannelError), +} + /// An [`Executor`] that records every call and answers from a script. /// /// Scripted outcomes are consumed first in, first out, one queue per method. @@ -97,6 +130,7 @@ struct State { calls: Vec, runs: VecDeque>, bounded_runs: VecDeque, + channels: VecDeque, } impl RecordingExecutor { @@ -117,6 +151,12 @@ impl RecordingExecutor { self.state().bounded_runs.push_back(outcome); } + /// Queues the outcome of the next unanswered [`Executor::open_channel`] + /// call. + pub fn script_open_channel(&self, outcome: ScriptedChannel) { + self.state().channels.push_back(outcome); + } + /// Returns every call received so far, in order. #[must_use] pub fn calls(&self) -> Vec { @@ -208,6 +248,55 @@ impl Executor for RecordingExecutor { } } + /// Records the call and answers it with the next scripted + /// [`ScriptedChannel`], after refusing a command the real transports + /// would refuse. + /// + /// # Errors + /// + /// Returns [`ChannelError::InvalidCommand`] for a command that is not an + /// absolute path free of `=`, the scripted error for + /// [`ScriptedChannel::Fail`], and [`ExecutorError::Spawn`] when a + /// [`ScriptedChannel::Spawn`] program cannot be spawned. + /// + /// # 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 open_channel( + &self, + identity: Identity, + command: &str, + args: &[&str], + limits: ChannelLimits, + ) -> Result { + let outcome = { + let mut state = self.state(); + state.calls.push(RecordedCall::OpenChannel { + identity, + command: command.to_string(), + args: owned(args), + limits, + }); + channel::check_command(command)?; + let Some(outcome) = state.channels.pop_front() else { + panic!( + "RecordingExecutor: no outcome scripted for open_channel `{command}` {args:?}" + ) + }; + outcome + }; + match outcome { + ScriptedChannel::Spawn { program, args } => { + let mut spawned = Command::new(program); + spawned.args(args).env_clear(); + channel::open_direct(spawned, limits.max_stderr) + } + ScriptedChannel::Fail(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(), @@ -220,12 +309,14 @@ impl Executor for RecordingExecutor { #[cfg(test)] mod tests { + use std::io::{Read, Write}; + use std::path::PathBuf; use std::time::Duration; - use super::{RecordedCall, RecordingExecutor, ScriptedRun}; + use super::{RecordedCall, RecordingExecutor, ScriptedChannel, ScriptedRun}; use crate::executor::{ - CommandOutput, Executor, ExecutorError, Identity, OutputStream, RunLimits, - RunWithInputError, ServiceAccount, + ChannelError, ChannelLimits, CommandOutput, Executor, ExecutorError, Identity, + OutputStream, RunLimits, RunWithInputError, ServiceAccount, }; const LIMITS: RunLimits = RunLimits { @@ -233,6 +324,10 @@ mod tests { max_stderr: 4_096, timeout: Duration::from_secs(30), }; + const CHANNEL: ChannelLimits = ChannelLimits { + elevation_timeout: Duration::from_secs(10), + max_stderr: 4_096, + }; fn output(code: i32, stdout: &[u8]) -> CommandOutput { CommandOutput { @@ -344,4 +439,65 @@ mod tests { }] ); } + + #[test] + fn a_scripted_spawn_is_a_channel_to_that_program() { + let exec = RecordingExecutor::new(); + exec.script_open_channel(ScriptedChannel::Spawn { + program: PathBuf::from("/bin/cat"), + args: Vec::new(), + }); + let identity = Identity::Service(ServiceAccount::Roxyd); + let mut channel = exec + .open_channel(identity, "/usr/lib/helper", &["__attempt-launch"], CHANNEL) + .expect("the scripted spawn"); + let mut stdin = channel.take_stdin().expect("stdin"); + let mut stdout = channel.take_stdout().expect("stdout"); + stdin.write_all(b"frame").expect("write"); + drop(stdin); + let mut echoed = Vec::new(); + stdout.read_to_end(&mut echoed).expect("read"); + assert_eq!(echoed, b"frame"); + let exit = channel.wait().expect("wait"); + assert_eq!(exit.code, Some(0)); + assert!(exit.stderr.is_empty()); + assert_eq!( + exec.calls(), + vec![RecordedCall::OpenChannel { + identity, + command: "/usr/lib/helper".to_string(), + args: vec!["__attempt-launch".to_string()], + limits: CHANNEL, + }] + ); + } + + #[test] + fn a_scripted_failure_is_returned_and_an_invalid_command_consumes_nothing() { + let exec = RecordingExecutor::new(); + exec.script_open_channel(ScriptedChannel::Fail(ChannelError::Executor( + ExecutorError::Elevation { + host: "seat".to_string(), + }, + ))); + let error = exec + .open_channel(Identity::Root, "helper", &[], CHANNEL) + .expect_err("a relative command is refused"); + assert!( + matches!(error, ChannelError::InvalidCommand { ref command } if command == "helper"), + "got: {error:?}" + ); + assert_eq!(exec.calls().len(), 1, "the refused call is still recorded"); + let error = exec + .open_channel(Identity::Root, "/usr/lib/helper", &[], CHANNEL) + .expect_err("the scripted failure"); + assert!( + matches!( + error, + ChannelError::Executor(ExecutorError::Elevation { .. }) + ), + "got: {error:?}" + ); + assert_eq!(exec.calls().len(), 2); + } } From eb88898c214a31ef797210e10b5822b6e8969922 Mon Sep 17 00:00:00 2001 From: sehkone Date: Sun, 27 Sep 2026 22:10:41 +0900 Subject: [PATCH 2/6] Tighten the channel module's constants STDERR_GRACE is read only inside the channel module, so it no longer needs to be visible to its parent, and START_SHELL now sits with the module's other constants at the top of the file. settle_start takes the password with a let-else rather than testing for None and then unwrapping a default it could never reach. Part of #132 --- src/executor/channel.rs | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/src/executor/channel.rs b/src/executor/channel.rs index 91be7e1..efc1b4a 100644 --- a/src/executor/channel.rs +++ b/src/executor/channel.rs @@ -35,7 +35,10 @@ use super::{CommandOutput, ExecutorError, RC_MARKER, SUDO_OK_SENTINEL, spawn_ret /// transport process has exited. A descendant of the command can hold the /// pipe open indefinitely; past this, what was read is what is returned, and /// the rest is noted as truncated. -pub(super) const STDERR_GRACE: Duration = Duration::from_secs(5); +const STDERR_GRACE: Duration = Duration::from_secs(5); +/// The shell [`start_script`] runs under, named absolutely so the start +/// depends on no `PATH`. +pub(super) const START_SHELL: &str = SUPERVISOR_SHELL; /// Bytes read from standard error per readiness. const READ_CHUNK: usize = 8192; /// Bytes at the end of an SSH channel's standard error held back from the @@ -163,10 +166,6 @@ pub(super) fn start_script() -> String { format!("printf '%s' '{SUDO_OK_SENTINEL}' >&2; exec {ENV} -i \"$0\" \"$@\"") } -/// The shell [`start_script`] runs under, named absolutely so the start -/// depends on no `PATH`. -pub(super) const START_SHELL: &str = SUPERVISOR_SHELL; - /// What [`open_started`] needs to know beyond the command it spawns. pub(super) struct Start<'a> { /// The host the transport reaches, for the errors that name it. @@ -327,11 +326,11 @@ fn settle_start( timeout: Duration, ) -> std::io::Result { rustix::io::ioctl_fionbio(&*stderr, true)?; - if password.is_none() { + let Some(password) = password else { return feed_and_settle(stdin, stderr, b"", timeout); - } + }; rustix::io::ioctl_fionbio(&*stdin, true)?; - let settled = feed_and_settle(stdin, stderr, password.unwrap_or_default(), timeout)?; + let settled = feed_and_settle(stdin, stderr, password, timeout)?; rustix::io::ioctl_fionbio(&*stdin, false)?; Ok(settled) } @@ -531,8 +530,9 @@ impl Channel { Err(source) => return Err(spawn_failed(&self.program, source)), }; self.ended = true; - let drained = self.stop_drain(STDERR_GRACE); - let drained = drained.map_err(|source| spawn_failed(&self.program, source))?; + let drained = self + .stop_drain(STDERR_GRACE) + .map_err(|source| spawn_failed(&self.program, source))?; let code = match (&self.remote_host, drained.remote_code) { (None, _) => status.code(), (Some(_), Some(code)) => Some(code), From e1fea13527ac8754045cf4a6ab2c3535013a99f8 Mon Sep 17 00:00:00 2001 From: sehkone Date: Sun, 27 Sep 2026 22:17:35 +0900 Subject: [PATCH 3/6] Stop the descendant the stderr-grace test starts The test's `/bin/sleep 8` held standard error open past the grace and was then left to run out on its own, outliving the test. It now records its pid, and the test kills it and waits for it to be gone once the channel has returned. Part of #132 --- src/executor.rs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/src/executor.rs b/src/executor.rs index 302febc..8c80cdb 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -8449,17 +8449,27 @@ exec "$@" let dir = tempfile::tempdir().expect("tempdir"); let exec = LocalExecutor::new("seat", SudoAuth::NonInteractive) .with_sudo_bin(descending_sudo(dir.path())); + let pid_file = dir.path().join("descendant"); let started = Instant::now(); let channel = exec .open_channel( Identity::Operator, "/bin/sh", - &["-c", "printf err >&2; /bin/sleep 8 >/dev/null & exit 4"], + &[ + "-c", + "printf err >&2; /bin/sleep 8 >/dev/null & echo \"$!\" > \"$1\"; exit 4", + "sh", + &pid_file.to_string_lossy(), + ], ROOMY, ) .expect("open"); let exit = channel.wait().expect("wait"); let elapsed = started.elapsed(); + // The descendant is this test's to stop, not left to run out. + let descendant = recorded_pid(&pid_file); + let _ = rustix::process::kill_process(descendant, rustix::process::Signal::KILL); + assert_gone("the descendant", descendant); assert_eq!(exit.code, Some(4)); assert_eq!(exit.stderr, b"err"); assert!(exit.stderr_truncated, "the stream had not ended"); From 822de2f87292618e2e2cb2d258c2921404621bf8 Mon Sep 17 00:00:00 2001 From: sehkone Date: Sun, 27 Sep 2026 22:36:24 +0900 Subject: [PATCH 4/6] Test a channel's untaken stdin and signalled end Two points of the open_channel contract had no test: wait closes a standard input the caller never took, so a command reading to end of input still ends, and a local transport ended by a signal reports no code rather than an invented one, while over SSH the remote shell's status stands. Part of #132 --- src/executor.rs | 37 +++++++++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/src/executor.rs b/src/executor.rs index 8c80cdb..eb449ee 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -7932,6 +7932,43 @@ exec "$@" } } + #[test] + fn wait_closes_standard_input_that_was_never_taken_on_every_pair() { + let dir = tempfile::tempdir().expect("tempdir"); + for (label, exec, identity) in every_pair(&dir) { + let channel = exec + .open_channel(identity, "/bin/cat", &[], ROOMY) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + let exit = channel + .wait() + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + assert_eq!(exit.code, Some(0), "{label}: `cat` saw end of input"); + } + } + + #[test] + fn a_local_transport_ended_by_a_signal_has_no_code() { + // Locally the stub `sudo` execs its way to the command, so the + // signal ends the local transport process itself. Over SSH the + // remote shell reports the signalled command's status as a code. + let dir = tempfile::tempdir().expect("tempdir"); + for (label, exec, identity) in every_pair(&dir) { + let mut channel = exec + .open_channel(identity, "/bin/sh", &["-c", "kill -9 $$"], ROOMY) + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + assert!(read_out(&mut channel).is_empty(), "{label}"); + let exit = channel + .wait() + .unwrap_or_else(|error| panic!("{label}: {error:?}")); + let expected = if label.starts_with("local") { + None + } else { + Some(128 + 9) + }; + assert_eq!(exit.code, expected, "{label}"); + } + } + #[test] fn the_command_sees_an_empty_environment_on_every_pair() { let dir = tempfile::tempdir().expect("tempdir"); From 242b25488fa4b4f0a9003a37fa45eebfd1cb778e Mon Sep 17 00:00:00 2001 From: sehkone Date: Sun, 27 Sep 2026 22:57:16 +0900 Subject: [PATCH 5/6] Keep a channel's stdin and preamble to their bounds Under SudoAuth::Password a sudo that does not ask for the password -- a NOPASSWD rule, or cached credentials -- left the password line unread, and the command read it ahead of the caller's first byte. The start script now reads standard input up to a handoff line carrying a random nonce, written only once the sentinel has arrived and sudo is done with stdin, and discards whatever precedes it before exec'ing the command. The start loop also found the sentinel before checking the 64 KiB preamble limit, so a transport that flooded past the limit and then printed the sentinel in the same read was accepted. What precedes the sentinel is now held to the limit wherever the sentinel lands. Part of #132 --- src/executor.rs | 102 +++++++++++++----- src/executor/bounded.rs | 2 +- src/executor/channel.rs | 222 +++++++++++++++++++++++++++++++++++----- 3 files changed, 271 insertions(+), 55 deletions(-) diff --git a/src/executor.rs b/src/executor.rs index eb449ee..4d0168b 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -1167,14 +1167,18 @@ pub trait Executor { /// stands in between — [`Identity::Root`] and [`Identity::Service`] on /// [`LocalExecutor`], every identity on [`SshExecutor`] — the command /// starts under a fixed `sh -c` script that announces the start on - /// standard error and then replaces itself with the command. Under - /// [`SudoAuth::Password`] the password line is written first, and - /// standard input then stays open for the caller, so the command's first - /// byte of input is the caller's first byte. Standard error is read until - /// the announcement, and standard output not at all. A transport that - /// ends first, or writes more than 64 KiB first, is killed and classified - /// as [`Executor::run`] classifies that output; one that does neither - /// within `limits.elevation_timeout` is killed and reaped. The operator on + /// standard error, reads standard input up to a line written after the + /// announcement, and then replaces itself with the command. Under + /// [`SudoAuth::Password`] the password line is written first; a `sudo` + /// that does not ask for it — a `NOPASSWD` rule, or credentials it still + /// has cached — leaves it for the script, which discards it. Either way + /// standard input then stays open for the caller, and the command's + /// first byte of input is the caller's first byte. Standard error is + /// read until the announcement, and standard output not at all. A + /// transport that ends first, or writes more than 64 KiB ahead of the + /// announcement, is killed and classified as [`Executor::run`] + /// classifies that output; one that does neither within + /// `limits.elevation_timeout` is killed and reaped. The operator on /// [`LocalExecutor`] is spawned directly, and this returns at once. /// - **Standard error stays with the channel**, drained for its whole /// life so the command never blocks on it: the first @@ -1193,13 +1197,6 @@ pub trait Executor { /// so a passphrase prompt [`SshPrompt::Allow`] permits still reaches the /// terminal. /// - /// Under [`SudoAuth::Password`], a `sudo` that does not ask for the - /// password — a `NOPASSWD` rule, or credentials it still has cached — - /// leaves the password line unread, and the command reads it as its first - /// line of input. This is `sudo`'s choice, and the same one - /// [`Executor::run_with_input`] meets; [`SudoAuth::NonInteractive`] is - /// the setting for a host where `sudo` does not ask. - /// /// [`Channel`] states what ending the channel reaches: the local /// transport process, not a command started through `sudo` or over SSH, /// and never a descendant. @@ -2592,6 +2589,7 @@ impl Executor for LocalExecutor { limits: ChannelLimits, ) -> Result { channel::check_command(command)?; + let script = channel::StartScript::new()?; let Resolved { command: mut cmd, password_line, @@ -2599,7 +2597,7 @@ impl Executor for LocalExecutor { } = self.resolve_through( identity, channel::START_SHELL, - &channel::start_script(), + script.script(), command, args, ); @@ -2611,6 +2609,7 @@ impl Executor for LocalExecutor { return channel::open_direct(cmd, limits.max_stderr); } let start = channel::Start { + script: &script, host: &self.host, password_line, remote_code: false, @@ -2980,7 +2979,7 @@ impl Executor for SshExecutor { limits: ChannelLimits, ) -> Result { channel::check_command(command)?; - let script = channel::start_script(); + let script = channel::StartScript::new()?; let ResolvedRemote { remote, password_line, @@ -2988,8 +2987,8 @@ impl Executor for SshExecutor { } = self.resolve_through( identity, channel::START_SHELL, - &script, - Some(&script), + script.script(), + Some(script.script()), command, args, ); @@ -2998,6 +2997,7 @@ impl Executor for SshExecutor { let mut cmd = self.ssh_command(); cmd.arg(wrap_with_rc_marker(&remote)); let start = channel::Start { + script: &script, host: &self.host, password_line, remote_code: true, @@ -7766,6 +7766,7 @@ exec "$@" use rustix::process::Pid; use tempfile::TempDir; + use super::super::super::bounded::TRANSPORT_STDERR_LIMIT; use super::super::super::{ Channel, ChannelError, ChannelLimits, CommandOutput, Executor, ExecutorError, FileMeta, Identity, InDaemonExecutor, LocalExecutor, RC_MARKER, SUDO_OK_SENTINEL, @@ -8194,16 +8195,23 @@ exec "$@" #[test] fn a_password_line_is_consumed_before_the_callers_bytes() { let dir = tempfile::tempdir().expect("tempdir"); - let sudo = password_sudo(dir.path()); + let asking = password_sudo(dir.path()); + // A `NOPASSWD` rule, or cached credentials: `sudo` leaves the + // password line unread. + let not_asking = descending_sudo(dir.path()); let auth = || SudoAuth::Password("s3cret".to_string()); let input = pattern(ECHO_LEN); - let pairs: Vec<(&str, Box)> = vec![ - ( - "local", + let mut pairs: Vec<(String, Box)> = Vec::new(); + for (how, sudo) in [("asking", &asking), ("not asking", ¬_asking)] { + pairs.push(( + format!("local, {how}"), Box::new(LocalExecutor::new("seat", auth()).with_sudo_bin(sudo.clone())), - ), - ("ssh", Box::new(ssh_with(dir.path(), &sudo, auth()))), - ]; + )); + pairs.push(( + format!("ssh, {how}"), + Box::new(ssh_with(dir.path(), sudo, auth())), + )); + } for (label, exec) in pairs { for identity in [Identity::Root, Identity::Service(ServiceAccount::Roxyd)] { let mut channel = exec @@ -8405,6 +8413,47 @@ exec "$@" ); } + #[test] + fn a_start_at_the_transport_limit_opens_and_one_past_it_is_refused() { + let dir = tempfile::tempdir().expect("tempdir"); + // Writes `$FLOOD` bytes, drops `-n`, and execs the start + // script, whose sentinel follows them directly. + let flooding_then_granting = |name: &str, flood: usize| { + write_script( + dir.path(), + name, + &format!( + "#!/bin/sh\n/usr/bin/head -c {flood} /dev/zero | \ + /usr/bin/tr '\\0' x >&2\nshift\nexec \"$@\"\n" + ), + ) + }; + let at_limit = flooding_then_granting("at-limit-sudo", TRANSPORT_STDERR_LIMIT); + let mut channel = LocalExecutor::new("mgmt", SudoAuth::NonInteractive) + .with_sudo_bin(at_limit) + .open_channel(Identity::Root, "/bin/cat", &[], ROOMY) + .expect("the limit itself is not passed"); + assert_eq!(echo(&mut channel, b"frame"), b"frame"); + let exit = channel.wait().expect("wait"); + assert_eq!(exit.code, Some(0)); + assert!(exit.stderr.is_empty(), "{:?}", exit.stderr.len()); + + let past_limit = + flooding_then_granting("past-limit-sudo", TRANSPORT_STDERR_LIMIT + 1); + let error = LocalExecutor::new("mgmt", SudoAuth::NonInteractive) + .with_sudo_bin(past_limit) + .open_channel(Identity::Root, "/bin/cat", &[], ROOMY) + .expect_err("the sentinel came one byte past the limit"); + let ChannelError::Executor(ExecutorError::SudoRefused { reason, .. }) = &error + else { + panic!("expected SudoRefused, got {error:?}"); + }; + assert!( + !reason.contains(SUDO_OK_SENTINEL), + "only the transport's own bytes are classified" + ); + } + #[test] fn an_ssh_channel_that_loses_its_exit_status_has_no_code() { let dir = tempfile::tempdir().expect("tempdir"); @@ -8413,6 +8462,7 @@ exec "$@" "lossy-ssh", &format!( "#!/bin/sh\nprintf '%s' '{SUDO_OK_SENTINEL}' >&2\n\ + IFS= read -r handoff\n\ /bin/cat\necho 'Connection to 10.0.0.10 closed.' >&2\nexit 255\n" ), ); diff --git a/src/executor/bounded.rs b/src/executor/bounded.rs index 905a4c4..8994211 100644 --- a/src/executor/bounded.rs +++ b/src/executor/bounded.rs @@ -719,7 +719,7 @@ pub(super) fn contains(haystack: &[u8], needle: &str) -> bool { /// 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 { +pub(super) fn partial_suffix(bytes: &[u8], marker: &[u8]) -> usize { let longest = bytes.len().min(marker.len().saturating_sub(1)); (1..=longest) .rev() diff --git a/src/executor/channel.rs b/src/executor/channel.rs index efc1b4a..1a3e353 100644 --- a/src/executor/channel.rs +++ b/src/executor/channel.rs @@ -1,8 +1,9 @@ //! The long-lived channel behind [`Executor::open_channel`]. //! //! [`open_started`] spawns a transport — `sudo`, or `ssh` — that runs the -//! command under [`start_script`], and returns only once that script has -//! announced the command's start with [`SUDO_OK_SENTINEL`]. [`open_direct`] +//! command under a [`StartScript`], and returns only once that script has +//! announced the command's start with [`SUDO_OK_SENTINEL`] and been handed +//! standard input. [`open_direct`] //! spawns a command with nothing in between and returns at once. Either way //! the caller then owns standard input and standard output, while standard //! error stays with the [`Channel`] and is drained on a thread of its own for @@ -23,11 +24,13 @@ use std::sync::mpsc::{self, RecvTimeoutError}; use std::thread::JoinHandle; use std::time::Duration; +use aws_lc_rs::rand::SecureRandom; use rustix::event::{PollFd, PollFlags, Timespec}; use rustix::io::Errno; use super::bounded::{ - Deadline, ENV, KILL_GRACE, SUPERVISOR_SHELL, TRANSPORT_STDERR_LIMIT, find, reap_within, + Deadline, ENV, KILL_GRACE, SUPERVISOR_SHELL, TRANSPORT_STDERR_LIMIT, find, partial_suffix, + reap_within, }; use super::{CommandOutput, ExecutorError, RC_MARKER, SUDO_OK_SENTINEL, spawn_retrying_text_busy}; @@ -36,9 +39,16 @@ use super::{CommandOutput, ExecutorError, RC_MARKER, SUDO_OK_SENTINEL, spawn_ret /// pipe open indefinitely; past this, what was read is what is returned, and /// the rest is noted as truncated. const STDERR_GRACE: Duration = Duration::from_secs(5); -/// The shell [`start_script`] runs under, named absolutely so the start +/// The shell a [`StartScript`] runs under, named absolutely so the start /// depends on no `PATH`. pub(super) const START_SHELL: &str = SUPERVISOR_SHELL; +/// What opens every [`StartScript`]'s handoff line; the start's own random +/// nonce follows it. +const HANDOFF_PREFIX: &str = "__BOOTLER_STDIN_"; +/// What closes every [`StartScript`]'s handoff line. +const HANDOFF_SUFFIX: &str = "__"; +/// Random bytes in a handoff line's nonce. +const HANDOFF_NONCE_LEN: usize = 16; /// Bytes read from standard error per readiness. const READ_CHUNK: usize = 8192; /// Bytes at the end of an SSH channel's standard error held back from the @@ -156,18 +166,82 @@ pub(super) fn check_command(command: &str) -> Result<(), ChannelError> { } } -/// Returns the `sh -c` script a channel's command starts under wherever -/// `sudo` or SSH stands in between: it announces the start on standard error -/// with [`SUDO_OK_SENTINEL`], then replaces itself with the command, run with -/// an empty environment. Invoked as `sh -c SCRIPT `, so the -/// command and every argument arrive positionally and are never spliced into -/// the script text. -pub(super) fn start_script() -> String { - format!("printf '%s' '{SUDO_OK_SENTINEL}' >&2; exec {ENV} -i \"$0\" \"$@\"") +/// The `sh -c` script a channel's command starts under wherever `sudo` or SSH +/// stands in between, with the handoff line of its one start. +/// +/// Invoked as `sh -c SCRIPT `, so the command and every +/// argument arrive positionally and are never spliced into the script text. +/// +/// - **It first announces the start** on standard error with +/// [`SUDO_OK_SENTINEL`]: `sudo` has granted it, and has read standard input +/// for the last time. +/// - **It then takes standard input up to its handoff line**, which +/// [`open_started`] writes only once the announcement has arrived, and +/// discards it. Under [`SudoAuth::Password`] the password line comes first +/// on standard input, and a `sudo` that did not ask for it — a `NOPASSWD` +/// rule, or credentials it still has cached — leaves it unread; this is +/// where it goes instead of to the command. A `sudo` that did ask has +/// consumed it, and only the handoff line is left. A shell reads a pipe a +/// byte at a time, so the command's standard input begins exactly after +/// the handoff line, at the caller's first byte. +/// - **It replaces itself with the command**, run through `env -i` with an +/// empty environment. Standard input that ends before the handoff line +/// means the transport broke, and the script exits `1` without starting it. +/// +/// The handoff line shares standard input with what precedes it, so it is not +/// a fixed string a password could happen to equal: it carries a nonce of +/// [`HANDOFF_NONCE_LEN`] bytes drawn from the system's secure random source +/// for this start alone. +/// +/// [`SudoAuth::Password`]: super::SudoAuth::Password +pub(super) struct StartScript { + script: String, + handoff: String, +} + +impl StartScript { + /// Creates the script of one start, with a handoff line of its own. + /// + /// # Errors + /// + /// Returns [`ExecutorError::Spawn`] when the system's random source fails + /// to draw the handoff line's nonce: without it, nothing can be started + /// under this script. + pub(super) fn new() -> Result { + let mut nonce = [0u8; HANDOFF_NONCE_LEN]; + aws_lc_rs::rand::SystemRandom::new() + .fill(&mut nonce) + .map_err(|_| ExecutorError::Spawn { + command: START_SHELL.to_string(), + source: std::io::Error::other("system random source failed"), + })?; + let hex = crate::payload::to_hex(&nonce); + let handoff = format!("{HANDOFF_PREFIX}{hex}{HANDOFF_SUFFIX}"); + let script = format!( + "printf '%s' '{SUDO_OK_SENTINEL}' >&2; \ + while IFS= read -r line; do \ + [ \"$line\" = '{handoff}' ] && exec {ENV} -i \"$0\" \"$@\"; \ + done; exit 1" + ); + Ok(Self { script, handoff }) + } + + /// Returns the script to run as `sh -c SCRIPT `. + pub(super) fn script(&self) -> &str { + &self.script + } + + /// Returns the line [`open_started`] writes once the start is announced, + /// newline included. + fn handoff_line(&self) -> Vec { + format!("{}\n", self.handoff).into_bytes() + } } /// What [`open_started`] needs to know beyond the command it spawns. pub(super) struct Start<'a> { + /// The script the transport runs the command under. + pub(super) script: &'a StartScript, /// The host the transport reaches, for the errors that name it. pub(super) host: &'a str, /// The line to write on standard input before anything else — `sudo -S`'s @@ -179,16 +253,18 @@ pub(super) struct Start<'a> { pub(super) limits: ChannelLimits, } -/// Spawns `command` — a transport running the command under -/// [`start_script`] — and returns the channel once the start is proven. +/// Spawns `command` — a transport running the command under `start.script` — +/// and returns the channel once the start is proven. /// /// The password line, if any, is written first. Standard error is then read, /// and standard output never, until [`SUDO_OK_SENTINEL`] arrives; what follows -/// it is the command's. A transport that ends first, or writes more than -/// [`TRANSPORT_STDERR_LIMIT`] first, is killed and reaped, and what it wrote -/// is handed to `refusal` to be classified as -/// [`Executor::run`](super::Executor::run) classifies it. One that does -/// neither within the elevation timeout is killed and reaped too. +/// it is the command's. The script's handoff line is written then, so that +/// standard input passes to the caller with nothing ahead of the caller's +/// bytes. A transport that ends first, or writes more than +/// [`TRANSPORT_STDERR_LIMIT`] ahead of the sentinel, is killed and reaped, +/// and what it wrote ahead of the sentinel is handed to `refusal` to be +/// classified as [`Executor::run`](super::Executor::run) classifies it. One +/// that does neither within the elevation timeout is killed and reaped too. /// /// # Errors /// @@ -208,7 +284,13 @@ pub(super) fn open_started( &mut spawned.stderr, start.password_line.as_deref(), start.limits.elevation_timeout, - ); + ) + .and_then(|settled| { + if matches!(settled, Settled::Started(_)) { + hand_off(&mut spawned.stdin, &start.script.handoff_line())?; + } + Ok(settled) + }); let started = match settled { Ok(Settled::Started(started)) => started, Ok(Settled::Ended(stderr)) => { @@ -307,7 +389,8 @@ enum Settled { /// the command's. Started(Vec), /// The transport's standard error ended, or passed - /// [`TRANSPORT_STDERR_LIMIT`], before the announcement: all of it. + /// [`TRANSPORT_STDERR_LIMIT`], before the announcement: all of it up to + /// the announcement. Ended(Vec), /// The elevation timeout passed first, with what had been written by then. TimedOut(Vec), @@ -343,16 +426,12 @@ fn feed_and_settle( timeout: Duration, ) -> std::io::Result { let deadline = Deadline::after(timeout); - let sentinel = SUDO_OK_SENTINEL.as_bytes(); let mut written = 0; let mut read = Vec::new(); let mut chunk = vec![0; READ_CHUNK]; loop { - if let Some(at) = find(&read, sentinel) { - return Ok(Settled::Started(read.split_off(at + sentinel.len()))); - } - if read.len() > TRANSPORT_STDERR_LIMIT { - return Ok(Settled::Ended(read)); + if let Some(settled) = judge(&mut read, written >= unwritten.len()) { + return Ok(settled); } let Some(left) = deadline.remaining() else { return Ok(Settled::TimedOut(read)); @@ -396,6 +475,45 @@ fn feed_and_settle( } } +/// Returns what the standard error `read` so far says of the start, or `None` +/// while it is still undecided. `fed` is whether the password line, if any, +/// has been written in full. +/// +/// What precedes the sentinel is the transport's, and is held to +/// [`TRANSPORT_STDERR_LIMIT`] wherever the sentinel lands — in the same read +/// that passes the limit included. Before the sentinel has arrived, only a +/// trailing fragment that may still grow into it is not yet counted. +fn judge(read: &mut Vec, fed: bool) -> Option { + let sentinel = SUDO_OK_SENTINEL.as_bytes(); + match find(read, sentinel) { + Some(at) if at > TRANSPORT_STDERR_LIMIT => { + read.truncate(at); + Some(Settled::Ended(std::mem::take(read))) + } + // A password `sudo` did not read is still written in full first, so + // that the start script can discard it as one whole line. + Some(at) if fed => Some(Settled::Started(read.split_off(at + sentinel.len()))), + None if read.len() - partial_suffix(read, sentinel) > TRANSPORT_STDERR_LIMIT => { + Some(Settled::Ended(std::mem::take(read))) + } + Some(_) | None => None, + } +} + +/// Writes the start script's handoff line, after which standard input is the +/// caller's. +/// +/// Nothing is left in the pipe but, at most, a password line `sudo` did not +/// read, so the line fits and the write does not block. A transport that has +/// closed standard input since announcing the start is not a failure here: +/// [`Channel::wait`] reports how it ended. +fn hand_off(stdin: &mut ChildStdin, line: &[u8]) -> std::io::Result<()> { + match stdin.write_all(line) { + Err(error) if error.kind() == ErrorKind::BrokenPipe => Ok(()), + written => written, + } +} + /// 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) @@ -785,7 +903,55 @@ fn split_status_line(tail: &[u8]) -> Option<(usize, i32)> { #[cfg(test)] mod tests { - use super::{RC_MARKER, Sink, split_status_line}; + use super::{ + RC_MARKER, SUDO_OK_SENTINEL, Settled, Sink, TRANSPORT_STDERR_LIMIT, judge, + split_status_line, + }; + + /// `before` bytes of the transport's, then `after`. + fn preamble(before: usize, after: &str) -> Vec { + let mut read = vec![b'x'; before]; + read.extend_from_slice(after.as_bytes()); + read + } + + #[test] + fn the_transport_limit_holds_wherever_the_sentinel_lands() { + let mut read = preamble(TRANSPORT_STDERR_LIMIT, &format!("{SUDO_OK_SENTINEL}own")); + assert!( + matches!(judge(&mut read, true), Some(Settled::Started(own)) if own == b"own"), + "the limit itself is not passed" + ); + + // The sentinel arriving in the read that passes the limit does not + // excuse what precedes it. + let mut read = preamble( + TRANSPORT_STDERR_LIMIT + 1, + &format!("{SUDO_OK_SENTINEL}own"), + ); + assert!( + matches!(judge(&mut read, true), + Some(Settled::Ended(ended)) if ended == vec![b'x'; TRANSPORT_STDERR_LIMIT + 1]), + "only the transport's bytes are kept for classification" + ); + + let mut read = preamble(TRANSPORT_STDERR_LIMIT + 1, ""); + assert!(matches!(judge(&mut read, true), Some(Settled::Ended(_)))); + + // A fragment that may still become the sentinel is not yet counted. + let fragment = SUDO_OK_SENTINEL + .get(..SUDO_OK_SENTINEL.len() - 1) + .expect("the sentinel is longer than one byte"); + let mut read = preamble(TRANSPORT_STDERR_LIMIT, fragment); + assert!(judge(&mut read, true).is_none()); + } + + #[test] + fn the_start_waits_for_the_password_to_be_written_in_full() { + let mut read = preamble(0, SUDO_OK_SENTINEL); + assert!(judge(&mut read, false).is_none()); + assert!(matches!(judge(&mut read, true), Some(Settled::Started(own)) if own.is_empty())); + } #[test] fn the_status_line_is_read_only_at_the_end_of_the_stream() { From 588cad0921df3c9b25e145c0e893560d1374623e Mon Sep 17 00:00:00 2001 From: sehkone Date: Sun, 27 Sep 2026 23:06:57 +0900 Subject: [PATCH 6/6] Distrust an exit-status line from a failed ssh The SSH wrapper prints the exit-status line as its last act and so exits the remote shell zero. A local ssh that exited unsuccessfully therefore never delivered that line: anything shaped like it at the end of stderr was written by the command itself before the connection dropped. Accepting it would hand bootler a zero it treats as proof that cleanup succeeded. Such a channel now ends in ExitUnknown. Part of #132 --- src/executor.rs | 42 +++++++++++++++++++++++++++++++++++++++++ src/executor/channel.rs | 23 ++++++++++++++++------ 2 files changed, 59 insertions(+), 6 deletions(-) diff --git a/src/executor.rs b/src/executor.rs index 4d0168b..f938cd7 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -8492,6 +8492,48 @@ exec "$@" ); } + #[test] + fn a_failed_ssh_is_not_trusted_for_a_status_line_the_command_forged() { + let dir = tempfile::tempdir().expect("tempdir"); + // The command's own stderr ends in a well-formed exit-status + // line; the connection then drops before the wrapper reports. + let forging = write_script( + dir.path(), + "forging-ssh", + &format!( + "#!/bin/sh\nprintf '%s' '{SUDO_OK_SENTINEL}' >&2\n\ + IFS= read -r handoff\n\ + /bin/cat\nprintf '\\n{RC_MARKER}0\\n' >&2\nexit 255\n" + ), + ); + let config = crate::transport::Ssh { + user: "ops".to_string(), + port: 22, + key: PathBuf::from("/dev/null"), + host_key: crate::transport::HostKeyPolicy::Strict, + }; + let exec = SshExecutor::from_config( + "mgmt", + &config, + "10.0.0.10", + SudoAuth::NonInteractive, + SshPrompt::Deny, + ) + .with_ssh_bin(forging); + let mut channel = exec + .open_channel(Identity::Root, "/bin/cat", &[], ROOMY) + .expect("the start was announced"); + assert_eq!(echo(&mut channel, b"frame"), b"frame"); + let error = channel + .wait() + .expect_err("a failed ssh delivered no exit-status line"); + assert!( + matches!(&error, ChannelError::ExitUnknown { host, reason } + if host == "mgmt" && reason.contains("255")), + "got {error:?}" + ); + } + /// A command that records its pid in the file named by `$1`, then /// blocks on standard input. const BLOCKING: &str = diff --git a/src/executor/channel.rs b/src/executor/channel.rs index 1a3e353..9f2a47d 100644 --- a/src/executor/channel.rs +++ b/src/executor/channel.rs @@ -134,7 +134,9 @@ pub enum ChannelError { }, /// The SSH transport ended without reporting the remote command's exit /// status, so the command's outcome is not known — the connection was - /// lost, or `ssh` was ended some other way. Never a guessed code. + /// lost, or `ssh` was ended some other way. An `ssh` that exits + /// unsuccessfully is this too, whatever its standard error ends with. + /// Never a guessed code. #[error("host `{host}`: the channel's exit status is unknown: {reason}")] ExitUnknown { /// The host whose command's outcome is unknown. @@ -638,7 +640,8 @@ impl Channel { /// # Errors /// /// Returns [`ChannelError::ExitUnknown`] when an SSH channel ended without - /// its exit-status line, and [`ExecutorError::Spawn`] when waiting for the + /// its exit-status line, or when `ssh` itself exited unsuccessfully — the + /// wrapper exits zero once it reports, so a line then is not its — and [`ExecutorError::Spawn`] when waiting for the /// transport or reading standard error fails. pub fn wait(mut self) -> Result { self.stdin = None; @@ -653,18 +656,26 @@ impl Channel { .map_err(|source| spawn_failed(&self.program, source))?; let code = match (&self.remote_host, drained.remote_code) { (None, _) => status.code(), - (Some(_), Some(code)) => Some(code), - (Some(host), None) => { + // The wrapper's last act is printing the line, which exits the + // remote shell zero, so only a successful `ssh` delivered it: a + // failed one leaves marker-shaped text the command wrote itself. + (Some(_), Some(code)) if status.success() => Some(code), + (Some(host), remote_code) => { let ending = if drained.ended { "standard error ended" } else { "standard error had not ended 5 seconds later" }; + let missing = if remote_code.is_some() { + "after standard error ended on an exit-status line the wrapper did not \ + deliver" + } else { + "without reporting the remote exit status" + }; return Err(ChannelError::ExitUnknown { host: host.clone(), reason: format!( - "`{}` exited ({status}) without reporting the remote exit status; \ - {ending} with: {}", + "`{}` exited ({status}) {missing}; {ending} with: {}", self.program, String::from_utf8_lossy(&drained.tail).trim() ),