diff --git a/Cargo.lock b/Cargo.lock index ed97f7d8d..1ff590486 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2916,6 +2916,9 @@ dependencies = [ "cc", "libc", "tempfile", + "thiserror 2.0.18", + "tokio", + "tokio-util", ] [[package]] diff --git a/lib/apps/fabro-cli/src/commands/run/remote_workflow.rs b/lib/apps/fabro-cli/src/commands/run/remote_workflow.rs index 015353490..27b7514a4 100644 --- a/lib/apps/fabro-cli/src/commands/run/remote_workflow.rs +++ b/lib/apps/fabro-cli/src/commands/run/remote_workflow.rs @@ -1,15 +1,15 @@ //! Native Git acquisition owned by the CLI, with no server credential lookup. use std::future::Future; use std::path::{Path, PathBuf}; -use std::process::{ExitStatus, Stdio}; +use std::process::ExitStatus; use std::time::Duration; use anyhow::{Context as _, bail}; use fabro_manifest::CollectedWorkflowClosure; +use fabro_proc::{SupervisedCommand, SupervisedError}; use fabro_types::{GitHubRepositorySlug, GitRunTarget, repository}; -use tokio::io::{AsyncRead, AsyncReadExt as _}; use tokio::process::Command; -use tokio::{fs, signal as tokio_signal, task, time}; +use tokio::{fs, signal as tokio_signal, task}; use tokio_util::sync::CancellationToken; use tokio_util::task::TaskTracker; @@ -18,6 +18,33 @@ use crate::args::RunArgs; const OUTPUT_LIMIT: usize = 64 * 1024; +/// Configuration overrides that keep an untrusted checkout from running code +/// or rewriting bytes: no hooks or fsmonitor, no LFS smudge, no submodule +/// recursion, no `ext::` transport, no background maintenance, no line-ending +/// conversion. +const HARDENED_GIT_CONFIG: &[&str] = &[ + "-c", + "core.hooksPath=/dev/null", + "-c", + "core.fsmonitor=false", + "-c", + "filter.lfs.smudge=", + "-c", + "filter.lfs.process=", + "-c", + "filter.lfs.required=false", + "-c", + "submodule.recurse=false", + "-c", + "protocol.ext.allow=never", + "-c", + "maintenance.auto=0", + "-c", + "gc.auto=0", + "-c", + "core.autocrlf=false", +]; + #[derive(Debug, thiserror::Error)] pub(super) enum RemoteWorkflowError { #[error( @@ -64,34 +91,10 @@ impl NativeGit { args: &[&str], cancel: &CancellationToken, ) -> Result, RemoteWorkflowError> { - if cancel.is_cancelled() { - return Err(RemoteWorkflowError::Cancelled); - } let mut command = Command::new("git"); command .current_dir(cwd) - .args([ - "-c", - "core.hooksPath=/dev/null", - "-c", - "core.fsmonitor=false", - "-c", - "filter.lfs.smudge=", - "-c", - "filter.lfs.process=", - "-c", - "filter.lfs.required=false", - "-c", - "submodule.recurse=false", - "-c", - "protocol.ext.allow=never", - "-c", - "maintenance.auto=0", - "-c", - "gc.auto=0", - "-c", - "core.autocrlf=false", - ]) + .args(HARDENED_GIT_CONFIG) .args(args) .env("GIT_TERMINAL_PROMPT", "0") .env("GIT_LFS_SKIP_SMUDGE", "1") @@ -99,48 +102,30 @@ impl NativeGit { .env_remove("GIT_WORK_TREE") .env_remove("GIT_INDEX_FILE") .env_remove("GIT_OBJECT_DIRECTORY") - .env_remove("GIT_ALTERNATE_OBJECT_DIRECTORIES") - .stdin(Stdio::null()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .kill_on_drop(true); - #[cfg(unix)] - command.process_group(0); + .env_remove("GIT_ALTERNATE_OBJECT_DIRECTORIES"); #[cfg(test)] command.envs(self.environment.iter().cloned()); - let mut child = command.spawn()?; - let process_id = child.id(); - let mut stdout = child.stdout.take().expect("piped Git stdout exists"); - let mut stderr = child.stderr.take().expect("piped Git stderr exists"); - let result = tokio::select! { - biased; - () = cancel.cancelled() => Err(RemoteWorkflowError::Cancelled), - () = time::sleep(self.timeout) => Err(RemoteWorkflowError::Timeout), - result = async { - let (status, (stdout, overflow), _) = tokio::try_join!( - child.wait(), capture(&mut stdout), capture(&mut stderr) - )?; - if !status.success() { - // Output may contain arbitrary helper/config secrets, even after pattern - // redaction. Never retain it in an error/cause chain or tracing event. - return Err(RemoteWorkflowError::Process { operation, status }); - } - if overflow { return Err(RemoteWorkflowError::OutputLimit); } - Ok(stdout) - } => result, - }; - // A timed-out, cancelled, or failed Git may leave helpers running in - // its process group; terminate the group, then reap Git. A successful - // Git has closed its pipes, and helpers it deliberately left behind - // (such as `credential-cache--daemon`) keep serving later commands. - if result.is_err() { - #[cfg(unix)] - if let Some(id) = process_id { - fabro_proc::sigkill_process_group(id); - } - child.kill().await?; + let output = SupervisedCommand::new(command, self.timeout) + .output_limit(OUTPUT_LIMIT) + .run(cancel) + .await + .map_err(|error| match error { + SupervisedError::Timeout { .. } => RemoteWorkflowError::Timeout, + SupervisedError::Cancelled => RemoteWorkflowError::Cancelled, + SupervisedError::Io(source) => RemoteWorkflowError::Io(source), + })?; + if !output.status.success() { + // Output may contain arbitrary helper/config secrets, even after pattern + // redaction. Never retain it in an error/cause chain or tracing event. + return Err(RemoteWorkflowError::Process { + operation, + status: output.status, + }); } - result + if output.stdout_truncated { + return Err(RemoteWorkflowError::OutputLimit); + } + Ok(output.stdout) } /// Create and initialize an empty scratch repository. Temporary files are @@ -360,21 +345,6 @@ impl NativeGit { } } -async fn capture(reader: &mut (impl AsyncRead + Unpin)) -> std::io::Result<(Vec, bool)> { - let mut captured = Vec::new(); - let mut overflow = false; - let mut buffer = vec![0; 8192]; - loop { - let count = reader.read(&mut buffer).await?; - if count == 0 { - return Ok((captured, overflow)); - } - let keep = count.min(OUTPUT_LIMIT - captured.len()); - captured.extend_from_slice(&buffer[..keep]); - overflow |= keep != count; - } -} - fn exact_record(records: &str, reference: &str) -> anyhow::Result> { let mut found = None; for line in records.lines() { @@ -544,6 +514,7 @@ mod tests { use nix::sys::signal::{self, Signal}; use nix::sys::stat::Mode; use nix::unistd; + use tokio::time; use super::super::test_support::{commit_all, write_workflow}; use super::*; diff --git a/lib/apps/fabro-server/src/git_checkout.rs b/lib/apps/fabro-server/src/git_checkout.rs index 9a2e8c6f0..2f1c64925 100644 --- a/lib/apps/fabro-server/src/git_checkout.rs +++ b/lib/apps/fabro-server/src/git_checkout.rs @@ -4,10 +4,12 @@ use std::time::Duration; use base64::Engine as _; use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; +use fabro_proc::{SupervisedCommand, SupervisedError}; use fabro_store::KeyedMutex; use fabro_types::{GitHubRepositorySlug, GitRunTarget}; +use tokio::fs; use tokio::process::Command; -use tokio::{fs, time}; +use tokio_util::sync::CancellationToken; const GIT_CLONE_TIMEOUT: Duration = Duration::from_mins(2); const GIT_FETCH_TIMEOUT: Duration = Duration::from_mins(1); @@ -483,17 +485,25 @@ async fn run_git_plan(plan: GitCommandPlan) -> Result, GitCommandError> if let Some(current_dir) = plan.current_dir.as_ref() { command.current_dir(current_dir); } - command.kill_on_drop(true); - let output = time::timeout(plan.timeout, command.output()) + // Server-side checkouts have no caller to cancel them; the timeout is the + // only deadline. + let output = SupervisedCommand::new(command, plan.timeout) + .run(&CancellationToken::new()) .await - .map_err(|_| GitCommandError::Timeout { - command: safe_command_label(&plan), - timeout_secs: plan.timeout.as_secs(), - })? - .map_err(|err| GitCommandError::Spawn { - command: safe_command_label(&plan), - source: err, + .map_err(|error| match error { + SupervisedError::Timeout { .. } => GitCommandError::Timeout { + command: safe_command_label(&plan), + timeout_secs: plan.timeout.as_secs(), + }, + SupervisedError::Cancelled => GitCommandError::Spawn { + command: safe_command_label(&plan), + source: std::io::Error::new(std::io::ErrorKind::Interrupted, "cancelled"), + }, + SupervisedError::Io(source) => GitCommandError::Spawn { + command: safe_command_label(&plan), + source, + }, })?; if output.status.success() { diff --git a/lib/foundation/fabro-proc/Cargo.toml b/lib/foundation/fabro-proc/Cargo.toml index 4ecdb8ab8..8c032dc91 100644 --- a/lib/foundation/fabro-proc/Cargo.toml +++ b/lib/foundation/fabro-proc/Cargo.toml @@ -9,6 +9,11 @@ description = "Safe wrappers for process management primitives (signals, pre-exe [lints] workspace = true +[dependencies] +thiserror.workspace = true +tokio.workspace = true +tokio-util.workspace = true + [target.'cfg(unix)'.dependencies] libc = "0.2" diff --git a/lib/foundation/fabro-proc/src/lib.rs b/lib/foundation/fabro-proc/src/lib.rs index 5fbadb7bf..3cc110e0d 100644 --- a/lib/foundation/fabro-proc/src/lib.rs +++ b/lib/foundation/fabro-proc/src/lib.rs @@ -8,6 +8,7 @@ mod flock; #[cfg(unix)] mod pre_exec; mod signal; +mod supervised; mod title; #[cfg(unix)] @@ -23,4 +24,5 @@ pub use signal::{process_exists, process_group_alive, process_running, process_r pub use signal::{ sigkill, sigkill_process_group, sigterm, sigterm_process_group, sigusr1, sigusr2, }; +pub use supervised::{SupervisedCommand, SupervisedError, SupervisedOutput}; pub use title::{init as title_init, set as title_set}; diff --git a/lib/foundation/fabro-proc/src/supervised.rs b/lib/foundation/fabro-proc/src/supervised.rs new file mode 100644 index 000000000..fa944fa56 --- /dev/null +++ b/lib/foundation/fabro-proc/src/supervised.rs @@ -0,0 +1,235 @@ +//! Supervised child processes: an owned process group, a wall-clock timeout, +//! cooperative cancellation, and bounded output capture. +//! +//! Callers configure a [`tokio::process::Command`] and hand it over; the +//! supervisor owns stdio, the process group, and teardown. Helpers a child +//! spawns (Git credential helpers, for example) share its group, so an +//! interrupted or timed-out command takes them down with it. +use std::process::{ExitStatus, Stdio}; +use std::time::Duration; + +use tokio::io::{AsyncRead, AsyncReadExt as _}; +use tokio::process::Command; +use tokio::time; +use tokio_util::sync::CancellationToken; + +/// Output of a child that ran to completion, successfully or not. +#[derive(Debug)] +pub struct SupervisedOutput { + pub status: ExitStatus, + pub stdout: Vec, + pub stderr: Vec, + /// Stdout exceeded the configured limit and was cut off. + pub stdout_truncated: bool, + /// Stderr exceeded the configured limit and was cut off. + pub stderr_truncated: bool, +} + +#[derive(Debug, thiserror::Error)] +pub enum SupervisedError { + #[error("command timed out after {}s", timeout.as_secs())] + Timeout { timeout: Duration }, + #[error("command cancelled")] + Cancelled, + #[error("failed to run command")] + Io(#[from] std::io::Error), +} + +/// A command run under supervision. Stdin is closed; stdout and stderr are +/// captured; on Unix the child leads its own process group. +pub struct SupervisedCommand { + command: Command, + timeout: Duration, + output_limit: Option, +} + +impl SupervisedCommand { + pub fn new(command: Command, timeout: Duration) -> Self { + Self { + command, + timeout, + output_limit: None, + } + } + + /// Cap each captured stream at `limit` bytes; the remainder is drained + /// and reported as truncated rather than buffered. + #[must_use] + pub fn output_limit(mut self, limit: usize) -> Self { + self.output_limit = Some(limit); + self + } + + /// Run the child to completion. A timeout, cancellation, or I/O failure + /// kills the child's whole process group before returning; a child that + /// exits on its own, with any status, leaves the helpers it deliberately + /// left behind alone. + pub async fn run( + self, + cancel: &CancellationToken, + ) -> Result { + let Self { + mut command, + timeout, + output_limit, + } = self; + if cancel.is_cancelled() { + return Err(SupervisedError::Cancelled); + } + command + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + #[cfg(unix)] + command.process_group(0); + let mut child = command.spawn()?; + let process_id = child.id(); + let mut stdout = child.stdout.take().expect("piped stdout exists"); + let mut stderr = child.stderr.take().expect("piped stderr exists"); + let result = tokio::select! { + biased; + () = cancel.cancelled() => Err(SupervisedError::Cancelled), + () = time::sleep(timeout) => Err(SupervisedError::Timeout { timeout }), + result = async { + let (status, (stdout, stdout_truncated), (stderr, stderr_truncated)) = tokio::try_join!( + child.wait(), + capture(&mut stdout, output_limit), + capture(&mut stderr, output_limit), + )?; + Ok(SupervisedOutput { status, stdout, stderr, stdout_truncated, stderr_truncated }) + } => result, + }; + if result.is_err() { + #[cfg(unix)] + if let Some(id) = process_id { + crate::sigkill_process_group(id); + } + // The original error is the useful one; `kill_on_drop` backs this up. + let _reaped = child.kill().await; + } + result + } +} + +async fn capture( + reader: &mut (impl AsyncRead + Unpin), + limit: Option, +) -> std::io::Result<(Vec, bool)> { + let mut captured = Vec::new(); + let mut truncated = false; + let mut buffer = vec![0; 8192]; + loop { + let count = reader.read(&mut buffer).await?; + if count == 0 { + return Ok((captured, truncated)); + } + let keep = limit.map_or(count, |limit| count.min(limit - captured.len())); + captured.extend_from_slice(&buffer[..keep]); + truncated |= keep != count; + } +} + +#[cfg(all(test, unix))] +mod tests { + #![expect( + clippy::disallowed_methods, + reason = "supervisor tests read small pid files written by fixture scripts" + )] + + use super::*; + + fn shell(script: &str) -> Command { + let mut command = Command::new("/bin/sh"); + command.args(["-c", script]); + command + } + + #[tokio::test] + async fn supervised_command_captures_both_streams_and_any_status() { + let output = SupervisedCommand::new( + shell("printf out; printf err >&2; exit 3"), + Duration::from_secs(5), + ) + .run(&CancellationToken::new()) + .await + .unwrap(); + assert_eq!(output.status.code(), Some(3)); + assert_eq!(output.stdout, b"out"); + assert_eq!(output.stderr, b"err"); + assert!(!output.stdout_truncated && !output.stderr_truncated); + } + + #[tokio::test] + async fn supervised_command_drains_past_the_output_limit() { + let output = SupervisedCommand::new( + shell("i=0; while [ $i -lt 9000 ]; do printf 'sixteen-bytes---\\n'; i=$((i+1)); done"), + Duration::from_secs(10), + ) + .output_limit(1024) + .run(&CancellationToken::new()) + .await + .unwrap(); + assert!(output.status.success()); + assert_eq!(output.stdout.len(), 1024); + assert!(output.stdout_truncated); + assert!(!output.stderr_truncated); + } + + #[tokio::test] + async fn supervised_command_kills_the_group_on_timeout_and_cancel() { + for timeout in [true, false] { + let dir = tempfile::tempdir().unwrap(); + let pid_file = dir.path().join("pid"); + let mut command = shell("printf '%s' $$ > pid; exec /bin/sleep 60"); + command.current_dir(dir.path()); + let cancel = CancellationToken::new(); + let trigger = async { + if !timeout { + time::timeout(Duration::from_secs(5), async { + while !pid_file.exists() { + time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + cancel.cancel(); + } + }; + let supervised = SupervisedCommand::new( + command, + if timeout { + Duration::from_millis(500) + } else { + Duration::from_mins(1) + }, + ); + let (result, ()) = tokio::join!(supervised.run(&cancel), trigger); + let error = result.unwrap_err(); + assert!( + matches!(error, SupervisedError::Timeout { .. }) == timeout, + "{error}" + ); + let pid: u32 = std::fs::read_to_string(&pid_file).unwrap().parse().unwrap(); + assert!(!crate::process_exists(pid)); + } + assert!(matches!( + SupervisedCommand::new(shell("true"), Duration::from_secs(1)) + .run(&CancellationToken::new().tap_cancel()) + .await + .unwrap_err(), + SupervisedError::Cancelled + )); + } + + trait TapCancel { + fn tap_cancel(self) -> Self; + } + + impl TapCancel for CancellationToken { + fn tap_cancel(self) -> Self { + self.cancel(); + self + } + } +}