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

sim_lib_exec/
exec.rs
