sim_lib_exec/
exec.rs

1use std::{
2    io::{self, Read, Write},
3    path::PathBuf,
4    process::{Child, Command, ExitStatus, Stdio},
5    sync::{
6        Arc, Mutex,
7        mpsc::{self, Receiver, RecvTimeoutError, Sender},
8    },
9    thread,
10    time::{Duration, Instant},
11};
12
13#[cfg(unix)]
14use std::os::unix::process::CommandExt as _;
15
16use sim_kernel::{CapabilityName, Cx, Error, Expr, NumberLiteral, Result, Symbol};
17
18use crate::timeout::terminate_timed_out_child;
19
20/// Returns the capability required by [`exec`].
21pub fn exec_capability() -> CapabilityName {
22    CapabilityName::new("exec")
23}
24
25/// Returns the constructor symbol used by [`ProcResult::to_constructor_expr`].
26pub fn proc_result_symbol() -> Symbol {
27    Symbol::new("ProcResult")
28}
29
30/// Bounded process execution options.
31#[derive(Clone, Debug, PartialEq, Eq)]
32pub struct ExecOptions {
33    /// Directory the child runs in.
34    ///
35    /// When set, the directory must canonicalize inside [`root`](Self::root).
36    pub cwd: Option<PathBuf>,
37    /// Root that confines [`cwd`](Self::cwd).
38    ///
39    /// When `cwd` is set and this is absent, the current process directory is
40    /// the confinement root. When this is set and `cwd` is absent, the child
41    /// runs in this root directory.
42    pub root: Option<PathBuf>,
43    /// Mandatory timeout in milliseconds. Zero is rejected before spawning.
44    pub timeout_ms: u64,
45    /// Maximum total captured stdout plus stderr bytes.
46    pub max_output_bytes: usize,
47    /// Optional stdin bytes written to the child.
48    pub stdin: Option<Vec<u8>>,
49}
50
51impl ExecOptions {
52    /// Builds process options with a timeout and output cap.
53    pub fn new(timeout_ms: u64, max_output_bytes: usize) -> Self {
54        Self {
55            cwd: None,
56            root: None,
57            timeout_ms,
58            max_output_bytes,
59            stdin: None,
60        }
61    }
62
63    /// Returns options with a confined working directory.
64    pub fn with_cwd(mut self, cwd: impl Into<PathBuf>, root: impl Into<PathBuf>) -> Self {
65        self.cwd = Some(cwd.into());
66        self.root = Some(root.into());
67        self
68    }
69
70    /// Returns options with stdin bytes.
71    pub fn with_stdin(mut self, stdin: impl Into<Vec<u8>>) -> Self {
72        self.stdin = Some(stdin.into());
73        self
74    }
75}
76
77/// Result of a bounded process run.
78#[derive(Clone, Debug, PartialEq, Eq)]
79pub struct ProcResult {
80    /// Captured stdout, lossily decoded as UTF-8 after byte capping.
81    pub stdout: String,
82    /// Captured stderr, lossily decoded as UTF-8 after byte capping.
83    pub stderr: String,
84    /// Process exit code, or `-1` when the host reports no numeric code.
85    pub exit_code: i32,
86    /// Whether stdout or stderr exceeded the shared output byte cap.
87    pub truncated: bool,
88}
89
90impl ProcResult {
91    /// Encodes this result as `#(ProcResult stdout stderr exit_code truncated)`.
92    pub fn to_constructor_expr(&self) -> Expr {
93        Expr::Call {
94            operator: Box::new(Expr::Symbol(proc_result_symbol())),
95            args: vec![
96                Expr::String(self.stdout.clone()),
97                Expr::String(self.stderr.clone()),
98                Expr::Number(NumberLiteral {
99                    domain: Symbol::qualified("numbers", "i64"),
100                    canonical: self.exit_code.to_string(),
101                }),
102                Expr::Bool(self.truncated),
103            ],
104        }
105    }
106}
107
108/// Runs one host process with explicit argv and bounded output.
109///
110/// The caller must hold [`exec_capability`]. The argv list must be non-empty;
111/// the first element is the program and the remaining elements are passed
112/// verbatim as arguments. No shell is inserted by this function.
113pub fn exec(cx: &mut Cx, argv: &[String], options: &ExecOptions) -> Result<ProcResult> {
114    cx.require(&exec_capability())?;
115    validate_request(argv, options)?;
116
117    #[cfg(not(unix))]
118    {
119        return Err(Error::HostError(
120            "exec is unavailable on this platform until process-tree timeout enforcement is implemented"
121                .to_owned(),
122        ));
123    }
124
125    let mut command = Command::new(&argv[0]);
126    command.args(&argv[1..]);
127    command.stdin(Stdio::piped());
128    command.stdout(Stdio::piped());
129    command.stderr(Stdio::piped());
130    #[cfg(unix)]
131    command.process_group(0);
132    if let Some(cwd) = confined_cwd(options)? {
133        command.current_dir(cwd);
134    }
135
136    let mut child = command
137        .spawn()
138        .map_err(|err| Error::HostError(format!("exec spawn {}: {err}", argv[0])))?;
139    run_child(&mut child, options)
140}
141
142fn validate_request(argv: &[String], options: &ExecOptions) -> Result<()> {
143    if argv.is_empty() {
144        return Err(Error::Eval(
145            "exec requires a non-empty argv list".to_owned(),
146        ));
147    }
148    if options.timeout_ms == 0 {
149        return Err(Error::Eval(
150            "exec requires a non-zero timeout_ms".to_owned(),
151        ));
152    }
153    Ok(())
154}
155
156fn confined_cwd(options: &ExecOptions) -> Result<Option<PathBuf>> {
157    if options.cwd.is_none() && options.root.is_none() {
158        return Ok(None);
159    }
160
161    let root = match &options.root {
162        Some(root) => root.clone(),
163        None => std::env::current_dir()
164            .map_err(|err| Error::HostError(format!("exec current dir: {err}")))?,
165    };
166    let cwd = options.cwd.clone().unwrap_or_else(|| root.clone());
167    let root = canonicalize_path(root, "exec root")?;
168    let cwd = canonicalize_path(cwd, "exec cwd")?;
169    if !cwd.starts_with(&root) {
170        return Err(Error::HostError(format!(
171            "exec cwd {} escapes root {}",
172            cwd.display(),
173            root.display()
174        )));
175    }
176    Ok(Some(cwd))
177}
178
179fn canonicalize_path(path: PathBuf, label: &'static str) -> Result<PathBuf> {
180    path.canonicalize()
181        .map_err(|err| Error::HostError(format!("{label} {}: {err}", path.display())))
182}
183
184fn run_child(child: &mut Child, options: &ExecOptions) -> Result<ProcResult> {
185    let stdout = child
186        .stdout
187        .take()
188        .ok_or_else(|| Error::HostError("exec stdout pipe missing".to_owned()))?;
189    let stderr = child
190        .stderr
191        .take()
192        .ok_or_else(|| Error::HostError("exec stderr pipe missing".to_owned()))?;
193    let stdin = child.stdin.take();
194
195    let budget = Arc::new(Mutex::new(CaptureBudget::new(options.max_output_bytes)));
196    let deadline = Instant::now()
197        .checked_add(Duration::from_millis(options.timeout_ms))
198        .ok_or_else(|| Error::Eval("exec timeout is too large".to_owned()))?;
199
200    let (tx, rx) = mpsc::channel();
201    spawn_reader(
202        stdout,
203        Arc::clone(&budget),
204        CaptureStream::Stdout,
205        tx.clone(),
206    );
207    spawn_reader(
208        stderr,
209        Arc::clone(&budget),
210        CaptureStream::Stderr,
211        tx.clone(),
212    );
213    let stdin_pending = if let Some(stdin) = stdin {
214        spawn_writer(stdin, options.stdin.clone(), tx.clone());
215        true
216    } else {
217        false
218    };
219    drop(tx);
220
221    let completion = wait_for_completion(child, &rx, deadline, stdin_pending);
222    match completion {
223        Ok(ChildCompletion {
224            status,
225            stdout,
226            stderr,
227            stdin,
228        }) => {
229            stdin?;
230            let stdout = stdout?;
231            let stderr = stderr?;
232            let truncated = budget
233                .lock()
234                .map_err(|_| Error::PoisonedLock("exec output budget"))?
235                .truncated;
236            Ok(ProcResult {
237                stdout: String::from_utf8_lossy(&stdout).into_owned(),
238                stderr: String::from_utf8_lossy(&stderr).into_owned(),
239                exit_code: exit_code(status),
240                truncated,
241            })
242        }
243        Err(WaitError::Timeout { child_exited }) => {
244            Err(timeout_error(child, options.timeout_ms, child_exited))
245        }
246        Err(WaitError::Host(err)) => Err(err),
247    }
248}
249
250fn spawn_reader<R>(
251    reader: R,
252    budget: Arc<Mutex<CaptureBudget>>,
253    stream: CaptureStream,
254    tx: Sender<ChildEvent>,
255) where
256    R: Read + Send + 'static,
257{
258    thread::spawn(move || {
259        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
260            read_capped(reader, budget, stream.name())
261        }))
262        .unwrap_or_else(|_| {
263            Err(Error::HostError(format!(
264                "exec {} thread panicked",
265                stream.name()
266            )))
267        });
268        let _ = tx.send(ChildEvent::Capture { stream, result });
269    });
270}
271
272fn spawn_writer(
273    mut stdin: std::process::ChildStdin,
274    input: Option<Vec<u8>>,
275    tx: Sender<ChildEvent>,
276) {
277    thread::spawn(move || {
278        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
279            write_stdin(&mut stdin, input)
280        }))
281        .unwrap_or_else(|_| Err(Error::HostError("exec stdin thread panicked".to_owned())));
282        let _ = tx.send(ChildEvent::Stdin(result));
283    });
284}
285
286fn write_stdin(stdin: &mut std::process::ChildStdin, input: Option<Vec<u8>>) -> Result<()> {
287    let Some(input) = input else {
288        return Ok(());
289    };
290    match stdin.write_all(&input) {
291        Ok(()) => Ok(()),
292        Err(err) if err.kind() == io::ErrorKind::BrokenPipe => Ok(()),
293        Err(err) => Err(Error::HostError(format!("exec stdin write: {err}"))),
294    }
295}
296
297fn wait_for_completion(
298    child: &mut Child,
299    rx: &Receiver<ChildEvent>,
300    deadline: Instant,
301    stdin_pending: bool,
302) -> std::result::Result<ChildCompletion, WaitError> {
303    let mut status = None;
304    let mut stdout = None;
305    let mut stderr = None;
306    let mut stdin = if stdin_pending { None } else { Some(Ok(())) };
307
308    loop {
309        poll_child_status(child, &mut status)?;
310        drain_child_events(rx, &mut stdout, &mut stderr, &mut stdin)?;
311        if status.is_some() && stdout.is_some() && stderr.is_some() && stdin.is_some() {
312            return Ok(ChildCompletion {
313                status: status.take().expect("status checked above"),
314                stdout: stdout.take().expect("stdout checked above"),
315                stderr: stderr.take().expect("stderr checked above"),
316                stdin: stdin.take().expect("stdin checked above"),
317            });
318        }
319
320        let now = Instant::now();
321        if now >= deadline {
322            return Err(WaitError::Timeout {
323                child_exited: status.is_some(),
324            });
325        }
326
327        match rx.recv_timeout((deadline - now).min(Duration::from_millis(10))) {
328            Ok(event) => record_child_event(event, &mut stdout, &mut stderr, &mut stdin),
329            Err(RecvTimeoutError::Timeout) => {}
330            Err(RecvTimeoutError::Disconnected) if status.is_some() => {
331                return Err(WaitError::Host(Error::HostError(
332                    "exec capture thread ended without result".to_owned(),
333                )));
334            }
335            Err(RecvTimeoutError::Disconnected) => {
336                thread::sleep((deadline - now).min(Duration::from_millis(10)));
337            }
338        }
339    }
340}
341
342fn poll_child_status(
343    child: &mut Child,
344    status: &mut Option<ExitStatus>,
345) -> std::result::Result<(), WaitError> {
346    if status.is_some() {
347        return Ok(());
348    }
349    *status = child
350        .try_wait()
351        .map_err(|err| WaitError::Host(Error::HostError(format!("exec wait: {err}"))))?;
352    Ok(())
353}
354
355fn drain_child_events(
356    rx: &Receiver<ChildEvent>,
357    stdout: &mut Option<Result<Vec<u8>>>,
358    stderr: &mut Option<Result<Vec<u8>>>,
359    stdin: &mut Option<Result<()>>,
360) -> std::result::Result<(), WaitError> {
361    loop {
362        match rx.try_recv() {
363            Ok(event) => record_child_event(event, stdout, stderr, stdin),
364            Err(mpsc::TryRecvError::Empty) => return Ok(()),
365            Err(mpsc::TryRecvError::Disconnected) => return Ok(()),
366        }
367    }
368}
369
370fn record_child_event(
371    event: ChildEvent,
372    stdout: &mut Option<Result<Vec<u8>>>,
373    stderr: &mut Option<Result<Vec<u8>>>,
374    stdin: &mut Option<Result<()>>,
375) {
376    match event {
377        ChildEvent::Capture {
378            stream: CaptureStream::Stdout,
379            result,
380        } => *stdout = Some(result),
381        ChildEvent::Capture {
382            stream: CaptureStream::Stderr,
383            result,
384        } => *stderr = Some(result),
385        ChildEvent::Stdin(result) => *stdin = Some(result),
386    }
387}
388
389fn timeout_error(child: &mut Child, timeout_ms: u64, child_exited: bool) -> Error {
390    let kill_result = terminate_timed_out_child(child, child_exited);
391    let wait_result = child.wait();
392    let mut message = format!("exec timed out after {timeout_ms} ms");
393    if let Err(err) = kill_result {
394        message.push_str(&format!("; kill failed: {err}"));
395    }
396    if let Err(err) = wait_result {
397        message.push_str(&format!("; wait failed: {err}"));
398    }
399    Error::HostError(message)
400}
401
402struct ChildCompletion {
403    status: ExitStatus,
404    stdout: Result<Vec<u8>>,
405    stderr: Result<Vec<u8>>,
406    stdin: Result<()>,
407}
408
409enum WaitError {
410    Timeout { child_exited: bool },
411    Host(Error),
412}
413
414#[derive(Clone, Copy)]
415enum CaptureStream {
416    Stdout,
417    Stderr,
418}
419
420impl CaptureStream {
421    fn name(self) -> &'static str {
422        match self {
423            Self::Stdout => "stdout",
424            Self::Stderr => "stderr",
425        }
426    }
427}
428
429enum ChildEvent {
430    Capture {
431        stream: CaptureStream,
432        result: Result<Vec<u8>>,
433    },
434    Stdin(Result<()>),
435}
436
437fn read_capped<R>(
438    mut reader: R,
439    budget: Arc<Mutex<CaptureBudget>>,
440    name: &'static str,
441) -> Result<Vec<u8>>
442where
443    R: Read,
444{
445    let mut captured = Vec::new();
446    let mut chunk = [0_u8; 4096];
447    loop {
448        let read = reader
449            .read(&mut chunk)
450            .map_err(|err| Error::HostError(format!("exec read {name}: {err}")))?;
451        if read == 0 {
452            return Ok(captured);
453        }
454        let keep = {
455            let mut budget = budget
456                .lock()
457                .map_err(|_| Error::PoisonedLock("exec output budget"))?;
458            budget.claim(read)
459        };
460        captured.extend_from_slice(&chunk[..keep]);
461    }
462}
463
464fn exit_code(status: ExitStatus) -> i32 {
465    status.code().unwrap_or(-1)
466}
467
468#[derive(Debug)]
469struct CaptureBudget {
470    remaining: usize,
471    truncated: bool,
472}
473
474impl CaptureBudget {
475    fn new(max_output_bytes: usize) -> Self {
476        Self {
477            remaining: max_output_bytes,
478            truncated: false,
479        }
480    }
481
482    fn claim(&mut self, read: usize) -> usize {
483        let keep = read.min(self.remaining);
484        self.remaining -= keep;
485        if keep < read {
486            self.truncated = true;
487        }
488        keep
489    }
490}
sim_lib_exec/exec.rs

sim_lib_exec/
exec.rs
