soma_mcp_client/process/
guard.rs

1use std::collections::BTreeSet;
2use std::path::Path;
3
4use thiserror::Error;
5
6#[derive(Debug, Error, PartialEq, Eq)]
7pub enum SpawnGuardError {
8    #[error("stdio command must be a bare executable name")]
9    PathCommandDenied,
10    #[error("stdio command is not allowlisted")]
11    CommandDenied,
12    #[error("stdio command must not be empty")]
13    EmptyCommand,
14}
15
16#[derive(Debug, Clone)]
17pub struct SpawnGuard {
18    allowed: BTreeSet<String>,
19}
20
21impl Default for SpawnGuard {
22    fn default() -> Self {
23        Self::new([
24            "npx", "uvx", "docker", "node", "python", "python3", "deno", "pipx", "dnx",
25        ])
26    }
27}
28
29impl SpawnGuard {
30    pub fn new(commands: impl IntoIterator<Item = impl Into<String>>) -> Self {
31        Self {
32            allowed: commands.into_iter().map(Into::into).collect(),
33        }
34    }
35
36    pub fn with_extra(mut self, commands: impl IntoIterator<Item = impl Into<String>>) -> Self {
37        self.allowed.extend(commands.into_iter().map(Into::into));
38        self
39    }
40
41    pub fn validate_command(&self, command: &str) -> Result<(), SpawnGuardError> {
42        if command.trim().is_empty() {
43            return Err(SpawnGuardError::EmptyCommand);
44        }
45        if Path::new(command).components().count() > 1 || command.contains(['/', '\\']) {
46            return Err(SpawnGuardError::PathCommandDenied);
47        }
48        if !self.allowed.contains(command) {
49            return Err(SpawnGuardError::CommandDenied);
50        }
51        Ok(())
52    }
53}
54
55#[cfg(test)]
56#[path = "guard_tests.rs"]
57mod tests;
soma_mcp_client/process/guard.rs

soma_mcp_client/process/
guard.rs
