diff --git a/src/executor.rs b/src/executor.rs index cceec24..f938cd7 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,87 @@ 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, 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 + /// `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. + /// + /// [`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 +2332,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 +2581,45 @@ 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 script = channel::StartScript::new()?; + let Resolved { + command: mut cmd, + password_line, + elevated, + } = self.resolve_through( + identity, + channel::START_SHELL, + script.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 { + script: &script, + 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 +2971,50 @@ 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::StartScript::new()?; + let ResolvedRemote { + remote, + password_line, + elevated, + } = self.resolve_through( + identity, + channel::START_SHELL, + script.script(), + Some(script.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 { + script: &script, + 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,940 @@ 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::bounded::TRANSPORT_STDERR_LIMIT; + 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 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"); + 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 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 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())), + )); + 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 + .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 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"); + let lossy = write_script( + dir.path(), + "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" + ), + ); + 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:?}" + ); + } + + #[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 = + "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 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 & 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"); + 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..8994211 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) @@ -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 new file mode 100644 index 0000000..9f2a47d --- /dev/null +++ b/src/executor/channel.rs @@ -0,0 +1,1005 @@ +//! The long-lived channel behind [`Executor::open_channel`]. +//! +//! [`open_started`] spawns a transport — `sudo`, or `ssh` — that runs the +//! 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 +//! 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 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, partial_suffix, + 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. +const STDERR_GRACE: Duration = Duration::from_secs(5); +/// 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 +/// 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. 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. + 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(), + }) + } +} + +/// 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 + /// 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. 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 +/// +/// 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, + ) + .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)) => { + 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 up to + /// the announcement. + 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)?; + 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, 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 mut written = 0; + let mut read = Vec::new(); + let mut chunk = vec![0; READ_CHUNK]; + loop { + if let Some(settled) = judge(&mut read, written >= unwritten.len()) { + return Ok(settled); + } + 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), + } + } + } +} + +/// 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) +} + +/// `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, 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; + 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) + .map_err(|source| spawn_failed(&self.program, source))?; + let code = match (&self.remote_host, drained.remote_code) { + (None, _) => status.code(), + // 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}) {missing}; {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, 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() { + 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); + } }