ferrin_tool/sandbox/
local.rs

1//! [`LocalProcessSandbox`]: runs on the host without isolation.
2
3use std::io;
4use std::path::Path;
5use std::path::PathBuf;
6use std::sync::Arc;
7
8use bytes::Bytes;
9use ferrin_spec::BoxFuture;
10use futures_util::StreamExt;
11use futures_util::future::Either;
12use tokio::io::AsyncRead;
13use tokio::io::AsyncReadExt;
14use tokio::io::AsyncWriteExt;
15use tokio::process::Child;
16use tokio::process::Command;
17use tokio::task::JoinSet;
18use tokio_util::sync::CancellationToken;
19
20use super::ByteStream;
21use super::ProcessOptions;
22use super::ProcessResult;
23use super::ReadFileOptions;
24use super::ReadTextFileOptions;
25use super::Sandbox;
26use super::SandboxProcess;
27use super::WriteFileOptions;
28
29/// A [`Sandbox`] backed by the local file system and shell.
30///
31/// **Provides no isolation.** Paths resolve relative to `root`, but absolute
32/// paths and `..` are not blocked. Use only in tests and examples.
33#[derive(Debug, Clone)]
34pub struct LocalProcessSandbox {
35    root: PathBuf,
36    shell: Vec<String>,
37    description: String,
38}
39
40impl LocalProcessSandbox {
41    /// Uses `root` as the working directory and the platform shell
42    /// (`/bin/sh -c` or `cmd /C`).
43    #[must_use]
44    pub fn new(root: impl Into<PathBuf>) -> Self {
45        let root = root.into();
46        let shell = if cfg!(windows) {
47            vec!["cmd".to_owned(), "/C".to_owned()]
48        } else {
49            vec!["/bin/sh".to_owned(), "-c".to_owned()]
50        };
51        Self {
52            description: format!(
53                "Local process sandbox (no isolation). Root directory: {}",
54                root.display()
55            ),
56            root,
57            shell,
58        }
59    }
60
61    /// Overrides the shell used to run commands (program followed by the
62    /// arguments that precede the command line).
63    #[must_use]
64    pub fn with_shell(
65        mut self,
66        program: impl Into<String>,
67        args: impl IntoIterator<Item = String>,
68    ) -> Self {
69        self.shell = std::iter::once(program.into()).chain(args).collect();
70        self
71    }
72
73    /// Overrides the description.
74    #[must_use]
75    pub fn with_description(mut self, description: impl Into<String>) -> Self {
76        self.description = description.into();
77        self
78    }
79
80    /// The root directory.
81    #[must_use]
82    pub fn root(&self) -> &Path {
83        &self.root
84    }
85
86    fn resolve(&self, path: &str) -> PathBuf {
87        self.root.join(path)
88    }
89
90    fn command(&self, options: &ProcessOptions) -> Command {
91        let mut command = Command::new(&self.shell[0]);
92        command.args(&self.shell[1..]).arg(&options.command);
93        let directory = options
94            .working_directory
95            .as_deref()
96            .map_or_else(|| self.root.clone(), |dir| self.resolve(dir));
97        command
98            .current_dir(directory)
99            .envs(&options.env)
100            .stdin(std::process::Stdio::null())
101            .stdout(std::process::Stdio::piped())
102            .stderr(std::process::Stdio::piped())
103            .kill_on_drop(true);
104        command
105    }
106}
107
108fn reader_stream(
109    reader: impl AsyncRead + Send + Unpin + 'static,
110    cancellation: CancellationToken,
111) -> ByteStream {
112    Box::pin(futures_util::stream::unfold(
113        Some((reader, cancellation)),
114        |state| async move {
115            let (mut reader, cancellation) = state?;
116            let mut buffer = vec![0u8; 8 * 1024];
117            match with_cancellation(&cancellation, reader.read(&mut buffer)).await {
118                Ok(0) => None,
119                Ok(read) => {
120                    buffer.truncate(read);
121                    Some((Ok(Bytes::from(buffer)), Some((reader, cancellation))))
122                }
123                Err(error) => Some((Err(error), None)),
124            }
125        },
126    ))
127}
128
129fn cancelled() -> io::Error {
130    io::Error::new(io::ErrorKind::Interrupted, "cancelled")
131}
132
133async fn with_cancellation<T>(
134    cancellation: &CancellationToken,
135    future: impl Future<Output = io::Result<T>>,
136) -> io::Result<T> {
137    match futures_util::future::select(Box::pin(cancellation.cancelled()), Box::pin(future)).await {
138        Either::Left(((), _)) => Err(cancelled()),
139        Either::Right((result, _)) => result,
140    }
141}
142
143fn decode_text(bytes: &[u8], encoding: Option<&str>) -> io::Result<String> {
144    match encoding.map(str::to_ascii_lowercase).as_deref() {
145        None | Some("utf-8" | "utf8") => Ok(String::from_utf8_lossy(bytes).into_owned()),
146        Some(other) => Err(io::Error::new(
147            io::ErrorKind::Unsupported,
148            format!("unsupported text encoding \"{other}\""),
149        )),
150    }
151}
152
153fn select_lines(text: &str, start_line: Option<usize>, end_line: Option<usize>) -> String {
154    if start_line.is_none() && end_line.is_none() {
155        return text.to_owned();
156    }
157    let start = start_line.unwrap_or(1).max(1);
158    let lines: Vec<&str> = text.split('\n').collect();
159    let end = end_line.unwrap_or(lines.len()).min(lines.len());
160    if start > end {
161        return String::new();
162    }
163    lines[start - 1..end].join("\n")
164}
165
166async fn open_optional(path: &Path) -> io::Result<Option<tokio::fs::File>> {
167    match tokio::fs::File::open(path).await {
168        Ok(file) => Ok(Some(file)),
169        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(None),
170        Err(error) => Err(error),
171    }
172}
173
174async fn create_with_parents(path: &Path) -> io::Result<tokio::fs::File> {
175    if let Some(parent) = path.parent() {
176        tokio::fs::create_dir_all(parent).await?;
177    }
178    tokio::fs::File::create(path).await
179}
180
181struct LocalProcess {
182    pid: Option<u32>,
183    tasks: JoinSet<io::Result<i32>>,
184    completion: Option<Result<i32, Arc<io::Error>>>,
185    stdout: Option<ByteStream>,
186    stderr: Option<ByteStream>,
187    kill: CancellationToken,
188}
189
190async fn supervise_process(
191    mut child: Child,
192    cancellation: CancellationToken,
193    kill: CancellationToken,
194) -> io::Result<i32> {
195    let waited = with_cancellation(&cancellation, with_cancellation(&kill, child.wait())).await;
196    match waited {
197        Ok(status) => Ok(status.code().unwrap_or(-1)),
198        Err(error) if error.kind() == io::ErrorKind::Interrupted => {
199            match child.kill().await {
200                Ok(()) => {}
201                Err(error) if error.kind() == io::ErrorKind::InvalidInput => {}
202                Err(error) => return Err(error),
203            }
204            if cancellation.is_cancelled() {
205                Err(error)
206            } else {
207                child.wait().await.map(|status| status.code().unwrap_or(-1))
208            }
209        }
210        Err(error) => Err(error),
211    }
212}
213
214fn process_result(result: &Result<i32, Arc<io::Error>>) -> io::Result<i32> {
215    result
216        .clone()
217        .map_err(|error| io::Error::new(error.kind(), error))
218}
219
220impl SandboxProcess for LocalProcess {
221    fn pid(&self) -> Option<u32> {
222        self.pid
223    }
224
225    fn take_stdout(&mut self) -> Option<ByteStream> {
226        self.stdout.take()
227    }
228
229    fn take_stderr(&mut self) -> Option<ByteStream> {
230        self.stderr.take()
231    }
232
233    fn wait(&mut self) -> BoxFuture<'_, io::Result<i32>> {
234        Box::pin(async move {
235            if let Some(completion) = &self.completion {
236                return process_result(completion);
237            }
238            let result = match self.tasks.join_next().await {
239                Some(Ok(result)) => result,
240                Some(Err(error)) => Err(io::Error::other(error)),
241                None => Err(io::Error::other(
242                    "process supervisor ended without a result",
243                )),
244            };
245            self.pid = None;
246            let completion = result.map_err(Arc::new);
247            let result = process_result(&completion);
248            self.completion = Some(completion);
249            result
250        })
251    }
252
253    fn kill(&mut self) -> BoxFuture<'_, io::Result<()>> {
254        Box::pin(async move {
255            self.kill.cancel();
256            match self.wait().await {
257                Ok(_) => Ok(()),
258                Err(error) if error.kind() == io::ErrorKind::Interrupted => Ok(()),
259                Err(error) => Err(error),
260            }
261        })
262    }
263}
264
265impl Sandbox for LocalProcessSandbox {
266    fn description(&self) -> &str {
267        &self.description
268    }
269
270    fn read_file(&self, options: ReadFileOptions) -> BoxFuture<'_, io::Result<Option<ByteStream>>> {
271        Box::pin(async move {
272            let path = self.resolve(&options.path);
273            let file = with_cancellation(&options.cancellation, open_optional(&path)).await?;
274            Ok(file.map(|file| reader_stream(file, options.cancellation)))
275        })
276    }
277
278    fn read_binary_file(
279        &self,
280        options: ReadFileOptions,
281    ) -> BoxFuture<'_, io::Result<Option<Bytes>>> {
282        Box::pin(async move {
283            let path = self.resolve(&options.path);
284            with_cancellation(&options.cancellation, async {
285                match tokio::fs::read(&path).await {
286                    Ok(bytes) => Ok(Some(Bytes::from(bytes))),
287                    Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(None),
288                    Err(error) => Err(error),
289                }
290            })
291            .await
292        })
293    }
294
295    fn read_text_file(
296        &self,
297        options: ReadTextFileOptions,
298    ) -> BoxFuture<'_, io::Result<Option<String>>> {
299        Box::pin(async move {
300            let bytes = self
301                .read_binary_file(ReadFileOptions {
302                    path: options.path,
303                    cancellation: options.cancellation,
304                })
305                .await?;
306            let Some(bytes) = bytes else {
307                return Ok(None);
308            };
309            let text = decode_text(&bytes, options.encoding.as_deref())?;
310            Ok(Some(select_lines(
311                &text,
312                options.start_line,
313                options.end_line,
314            )))
315        })
316    }
317
318    fn write_file(&self, options: WriteFileOptions<ByteStream>) -> BoxFuture<'_, io::Result<()>> {
319        Box::pin(async move {
320            let path = self.resolve(&options.path);
321            let mut content = options.content;
322            with_cancellation(&options.cancellation, async {
323                let mut file = create_with_parents(&path).await?;
324                while let Some(chunk) = content.next().await {
325                    file.write_all(&chunk?).await?;
326                }
327                file.flush().await
328            })
329            .await
330        })
331    }
332
333    fn write_binary_file(&self, options: WriteFileOptions<Bytes>) -> BoxFuture<'_, io::Result<()>> {
334        Box::pin(async move {
335            let path = self.resolve(&options.path);
336            with_cancellation(&options.cancellation, async {
337                let mut file = create_with_parents(&path).await?;
338                file.write_all(&options.content).await?;
339                file.flush().await
340            })
341            .await
342        })
343    }
344
345    fn write_text_file(&self, options: WriteFileOptions<String>) -> BoxFuture<'_, io::Result<()>> {
346        self.write_binary_file(WriteFileOptions {
347            path: options.path,
348            content: Bytes::from(options.content),
349            cancellation: options.cancellation,
350        })
351    }
352
353    fn spawn(&self, options: ProcessOptions) -> BoxFuture<'_, io::Result<Box<dyn SandboxProcess>>> {
354        Box::pin(async move {
355            if options.cancellation.is_cancelled() {
356                return Err(cancelled());
357            }
358            let mut child = self.command(&options).spawn()?;
359            let pid = child.id();
360            let stdout = child
361                .stdout
362                .take()
363                .map(|reader| reader_stream(reader, options.cancellation.clone()));
364            let stderr = child
365                .stderr
366                .take()
367                .map(|reader| reader_stream(reader, options.cancellation.clone()));
368            let kill = CancellationToken::new();
369            let mut tasks = JoinSet::new();
370            tasks.spawn(supervise_process(child, options.cancellation, kill.clone()));
371            Ok(Box::new(LocalProcess {
372                pid,
373                tasks,
374                completion: None,
375                stdout,
376                stderr,
377                kill,
378            }) as Box<dyn SandboxProcess>)
379        })
380    }
381
382    fn run(&self, options: ProcessOptions) -> BoxFuture<'_, io::Result<ProcessResult>> {
383        Box::pin(async move {
384            let cancellation = options.cancellation.clone();
385            if cancellation.is_cancelled() {
386                return Err(cancelled());
387            }
388            let mut child = self.command(&options).spawn()?;
389            let output = with_cancellation(&cancellation, async {
390                let stdout = child.stdout.take();
391                let stderr = child.stderr.take();
392                let (stdout, stderr) =
393                    futures_util::future::join(read_all(stdout), read_all(stderr)).await;
394                let status = child.wait().await?;
395                Ok(ProcessResult {
396                    exit_code: status.code().unwrap_or(-1),
397                    stdout: String::from_utf8_lossy(&stdout?).into_owned(),
398                    stderr: String::from_utf8_lossy(&stderr?).into_owned(),
399                })
400            })
401            .await;
402            if output
403                .as_ref()
404                .is_err_and(|error| error.kind() == io::ErrorKind::Interrupted)
405            {
406                let _ = child.kill().await;
407            }
408            output
409        })
410    }
411}
412
413async fn read_all(reader: Option<impl AsyncRead + Unpin>) -> io::Result<Vec<u8>> {
414    let mut buffer = Vec::new();
415    if let Some(mut reader) = reader {
416        reader.read_to_end(&mut buffer).await?;
417    }
418    Ok(buffer)
419}
ferrin_tool/sandbox/local.rs

ferrin_tool/sandbox/
local.rs
