use std::path::Path;
use std::process::Stdio;
use std::time::{Duration, Instant};
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::Command;
use tokio::sync::mpsc;
use crate::security::redact::sanitized_child_env;
use super::stream::{OutputLine, OutputStream};
use super::SubprocessHandle;
const MAX_LINE_LENGTH: usize = 4096;
pub struct SubprocessExecutor;
impl SubprocessExecutor {
pub async fn spawn(
command: &str,
args: &[String],
working_dir: &Path,
timeout: Duration,
) -> anyhow::Result<SubprocessHandle> {
let env = sanitized_child_env();
let mut cmd = Command::new(command);
cmd.args(args)
.current_dir(working_dir)
.env_clear()
.envs(env)
.kill_on_drop(true)
.stdout(Stdio::piped())
.stderr(Stdio::piped());
#[cfg(unix)]
cmd.process_group(0);
let mut child = cmd.spawn()?;
let stdout = child
.stdout
.take()
.ok_or_else(|| anyhow::anyhow!("stdout not captured from subprocess"))?;
let stderr = child
.stderr
.take()
.ok_or_else(|| anyhow::anyhow!("stderr not captured from subprocess"))?;
let (stdout_tx, stdout_rx) = mpsc::channel(10_000);
let (stderr_tx, stderr_rx) = mpsc::channel(10_000);
tokio::spawn(async move {
let reader = BufReader::new(stdout);
let mut lines = reader.lines();
while let Ok(Some(line)) = lines.next_line().await {
let content = if line.len() > MAX_LINE_LENGTH {
let mut end = MAX_LINE_LENGTH;
while end > 0 && !line.is_char_boundary(end) {
end -= 1;
}
let mut truncated = line[..end].to_string();
truncated.push_str(" [truncated]");
truncated
} else {
line
};
let output_line = OutputLine {
content,
timestamp: Instant::now(),
stream: OutputStream::Stdout,
};
if stdout_tx.try_send(output_line).is_err() {
break;
}
}
});
tokio::spawn(async move {
let reader = BufReader::new(stderr);
let mut lines = reader.lines();
while let Ok(Some(line)) = lines.next_line().await {
let content = if line.len() > MAX_LINE_LENGTH {
let mut end = MAX_LINE_LENGTH;
while end > 0 && !line.is_char_boundary(end) {
end -= 1;
}
let mut truncated = line[..end].to_string();
truncated.push_str(" [truncated]");
truncated
} else {
line
};
let output_line = OutputLine {
content,
timestamp: Instant::now(),
stream: OutputStream::Stderr,
};
if stderr_tx.try_send(output_line).is_err() {
break;
}
}
});
Ok(SubprocessHandle::new(child, stdout_rx, stderr_rx, timeout))
}
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
#[tokio::test]
async fn spawn_echo_captures_stdout() {
let tmp = TempDir::new().unwrap();
let mut handle = SubprocessExecutor::spawn(
"echo",
&["hello world".to_string()],
tmp.path(),
Duration::from_secs(10),
)
.await
.unwrap();
let code = handle.wait().await.unwrap();
assert_eq!(code, 0);
tokio::time::sleep(Duration::from_millis(50)).await;
let mut lines = Vec::new();
while let Some(line) = handle.try_recv_stdout() {
lines.push(line.content);
}
assert!(lines.contains(&"hello world".to_string()));
}
#[tokio::test]
async fn spawn_false_returns_nonzero() {
let tmp = TempDir::new().unwrap();
let mut handle =
SubprocessExecutor::spawn("false", &[], tmp.path(), Duration::from_secs(10))
.await
.unwrap();
let code = handle.wait().await.unwrap();
assert_ne!(code, 0);
}
#[tokio::test]
async fn spawn_captures_stderr() {
let tmp = TempDir::new().unwrap();
let mut handle = SubprocessExecutor::spawn(
"sh",
&["-c".to_string(), "echo error >&2".to_string()],
tmp.path(),
Duration::from_secs(10),
)
.await
.unwrap();
let code = handle.wait().await.unwrap();
assert_eq!(code, 0);
tokio::time::sleep(Duration::from_millis(50)).await;
let mut lines = Vec::new();
while let Some(line) = handle.try_recv_stderr() {
lines.push(line.content);
}
assert!(lines.contains(&"error".to_string()));
}
}