Antivirus/crates/hound-supply/src/mcp.rs

//! MCP server audit.
//!
//! An MCP server is a program your coding agent starts and then trusts
//! with tools. The usual way to add one is a line of JSON containing
//! `npx some-package`, which means: fetch code from a registry and run it,
//! with your agent's confidence and your user's permissions.
//!
//! Nothing scans these today. There is no registry review, no reputation
//! signal, and the config file lives somewhere nobody looks after the day
//! they set it up. That is a wide-open door, and it is open on macOS and
//! Windows exactly as much as on Linux.
//!
//! What we check is the shape of the entry, not the behaviour of the
//! server — behaviour needs the runtime watch that Phase 6 adds. Even so,
//! the shape says a lot: whether the code is pinned, whether it is fetched
//! fresh on every launch, what it is handed in its environment, and
//! whether the command line names things a tool server has no business
//! reading.

use crate::{Finding, Severity};

/// One server entry from an MCP config.
#[derive(Debug, Clone, Default)]
pub struct Server {
    pub name: String,
    pub command: String,
    pub args: Vec<String>,
    /// Environment variable names (not values — we never read secrets).
    pub env_keys: Vec<String>,
}

impl Server {
    /// The whole invocation, for pattern matching.
    fn command_line(&self) -> String {
        format!("{} {}", self.command, self.args.join(" ")).to_ascii_lowercase()
    }

    /// Fetched from a registry at launch rather than installed and pinned.
    fn is_fetched_at_launch(&self) -> bool {
        let c = self.command.to_ascii_lowercase();
        let runner = matches!(
            c.rsplit('/').next().unwrap_or(&c),
            "npx" | "bunx" | "uvx" | "pnpx" | "dlx"
        );
        // `npx -y` skips even the "is this what you meant?" prompt.
        runner
    }

    /// A package spec with no version is whatever the registry serves
    /// today, which may not be what it served yesterday.
    fn is_unpinned(&self) -> bool {
        if !self.is_fetched_at_launch() {
            return false;
        }
        // The first argument that is not a flag is the package spec.
        self.args
            .iter()
            .find(|a| !a.starts_with('-'))
            .map(|spec| {
                // scoped names carry a leading @, so only a later @ pins it
                let after_scope = spec.strip_prefix('@').unwrap_or(spec);
                !after_scope.contains('@')
            })
            .unwrap_or(true)
    }
}

const CREDENTIAL_PATHS: &[&str] = &[
    ".ssh", "id_rsa", "id_ed25519", ".aws", ".gnupg", ".netrc", ".npmrc",
    ".pypirc", "credentials", ".kube", ".docker/config",
];

/// Environment keys that hand a server a live secret.
const SECRET_KEYS: &[&str] = &[
    "token", "secret", "password", "passwd", "api_key", "apikey",
    "private_key", "credential", "session",
];

/// Directories broad enough that "filesystem access" means "everything".
const BROAD_ROOTS: &[&str] = &["/", "/home", "$home", "~", "~/", "/etc", "/var"];

/// Audit one server entry.
pub fn scan_server(s: &Server, location: &str) -> Vec<Finding> {
    let mut out = Vec::new();
    let cmdline = s.command_line();

    if CREDENTIAL_PATHS.iter().any(|p| cmdline.contains(p)) {
        out.push(Finding::new(
            "mcp-credential-scope",
            Severity::Critical,
            s.name.clone(),
            location,
            format!(
                "The \"{}\" tool server is started with your credential files in its \
                 arguments. Anything it is given, it can read — and an assistant will \
                 call its tools without asking you first.",
                s.name
            ),
            "hound-mcp-credentials-a",
            "Remove this server unless you are certain you need it, and narrow what it \
             is pointed at. Nothing that talks to an assistant should be handed your keys.",
        ));
    }

    if s.args.iter().any(|a| BROAD_ROOTS.contains(&a.to_ascii_lowercase().as_str())) {
        out.push(Finding::new(
            "mcp-broad-scope",
            Severity::Warning,
            s.name.clone(),
            location,
            format!(
                "The \"{}\" tool server is pointed at your whole home directory or the \
                 root of the filesystem. Whatever it can reach, your assistant can reach \
                 through it.",
                s.name
            ),
            "hound-mcp-overreach-a",
            "Point it at the specific project directory you want it to work in.",
        ));
    }

    if s.is_unpinned() {
        out.push(Finding::new(
            "mcp-unpinned",
            Severity::Warning,
            s.name.clone(),
            location,
            format!(
                "The \"{}\" tool server downloads its code fresh from the internet every \
                 time it starts, and no version is fixed. Whoever controls that package \
                 can change what runs on your machine at any moment, without you \
                 installing anything.",
                s.name
            ),
            "hound-mcp-unpinned-a",
            "Pin a version, or install the server properly and run the installed copy.",
        ));
    }

    let secrets: Vec<&String> = s
        .env_keys
        .iter()
        .filter(|k| {
            let l = k.to_ascii_lowercase();
            SECRET_KEYS.iter().any(|s| l.contains(s))
        })
        .collect();
    if !secrets.is_empty() && s.is_fetched_at_launch() {
        let names: Vec<&str> = secrets.iter().map(|s| s.as_str()).collect();
        out.push(Finding::new(
            "mcp-secret-to-unpinned",
            Severity::Critical,
            s.name.clone(),
            location,
            format!(
                "The \"{}\" tool server is handed {} — and its code is downloaded fresh \
                 from the internet on every launch. A change to that package would hand \
                 your secret to whoever made the change.",
                s.name,
                names.join(", ")
            ),
            "hound-mcp-secret-unpinned-a",
            "Pin the version, or install the server locally. Then rotate the secret if \
             you have been running it unpinned.",
        ));
    }

    out
}

/// Parse and audit an MCP config file.
///
/// Handles both shapes in the wild: a top-level `mcpServers` object
/// (Claude Desktop, Cursor) and a bare `servers` object.
pub fn scan_config(json: &str, location: &str) -> Vec<Finding> {
    let Ok(v) = serde_json::from_str::<serde_json::Value>(json) else {
        return Vec::new();
    };
    let servers = v
        .get("mcpServers")
        .or_else(|| v.get("servers"))
        .and_then(|s| s.as_object());
    let Some(servers) = servers else {
        return Vec::new();
    };

    let mut out = Vec::new();
    for (name, entry) in servers {
        let command = entry
            .get("command")
            .and_then(|c| c.as_str())
            .unwrap_or_default()
            .to_string();
        let args = entry
            .get("args")
            .and_then(|a| a.as_array())
            .map(|a| {
                a.iter()
                    .filter_map(|x| x.as_str().map(str::to_string))
                    .collect()
            })
            .unwrap_or_default();
        let env_keys = entry
            .get("env")
            .and_then(|e| e.as_object())
            .map(|e| e.keys().cloned().collect())
            .unwrap_or_default();

        out.extend(scan_server(
            &Server { name: name.clone(), command, args, env_keys },
            location,
        ));
    }
    out
}

#[cfg(test)]
mod tests {
    use super::*;

    fn server(cmd: &str, args: &[&str]) -> Server {
        Server {
            name: "test-server".into(),
            command: cmd.into(),
            args: args.iter().map(|s| s.to_string()).collect(),
            env_keys: Vec::new(),
        }
    }

    // ── pinning ──

    #[test]
    fn npx_without_a_version_is_unpinned() {
        assert!(server("npx", &["-y", "mcp-github-tools"]).is_unpinned());
    }

    #[test]
    fn npx_with_a_version_is_pinned() {
        assert!(!server("npx", &["-y", "mcp-github-tools@0.3.1"]).is_unpinned());
    }

    #[test]
    fn a_scoped_package_needs_a_version_after_the_scope() {
        assert!(server("npx", &["@acme/mcp-tools"]).is_unpinned());
        assert!(!server("npx", &["@acme/mcp-tools@1.2.3"]).is_unpinned());
    }

    #[test]
    fn an_installed_binary_is_not_fetched_at_launch() {
        let s = server("/usr/local/bin/my-mcp-server", &["--root", "/srv/project"]);
        assert!(!s.is_fetched_at_launch());
        assert!(!s.is_unpinned());
    }

    #[test]
    fn other_runners_count_too() {
        for runner in ["bunx", "uvx", "pnpx"] {
            assert!(server(runner, &["thing"]).is_unpinned(), "{runner}");
        }
    }

    // ── must catch ──

    #[test]
    fn credentials_on_the_command_line_are_critical() {
        let f = scan_server(&server("npx", &["mcp-fs", "/home/joe/.ssh"]), "mcp.json");
        assert!(f.iter().any(|x| x.severity == Severity::Critical));
        assert!(f.iter().any(|x| x.kind == "mcp-credential-scope"));
    }

    #[test]
    fn a_server_pointed_at_home_is_flagged() {
        let f = scan_server(&server("npx", &["mcp-filesystem@1.0.0", "$HOME"]), "mcp.json");
        assert!(f.iter().any(|x| x.kind == "mcp-broad-scope"));
    }

    #[test]
    fn a_secret_handed_to_unpinned_code_is_critical() {
        let mut s = server("npx", &["-y", "mcp-github-tools"]);
        s.env_keys = vec!["GITHUB_TOKEN".into()];
        let f = scan_server(&s, "mcp.json");
        let hit = f.iter().find(|x| x.kind == "mcp-secret-to-unpinned").expect("must fire");
        assert_eq!(hit.severity, Severity::Critical);
        assert!(hit.advice.contains("rotate"));
    }

    #[test]
    fn a_secret_handed_to_pinned_local_code_is_not_flagged() {
        let mut s = server("/usr/local/bin/mcp-github", &["--repo", "acme/app"]);
        s.env_keys = vec!["GITHUB_TOKEN".into()];
        let f = scan_server(&s, "mcp.json");
        assert!(
            !f.iter().any(|x| x.kind == "mcp-secret-to-unpinned"),
            "an installed, pinned server holding a token is normal"
        );
    }

    // ── must NOT catch ──

    #[test]
    fn a_well_configured_server_is_clean() {
        let s = server("/usr/local/bin/mcp-project", &["--root", "/home/joe/src/app"]);
        assert!(scan_server(&s, "mcp.json").is_empty());
    }

    // ── config parsing ──

    #[test]
    fn parses_the_claude_desktop_shape() {
        let json = r#"{
          "mcpServers": {
            "github": {
              "command": "npx",
              "args": ["-y", "mcp-github-tools"],
              "env": {"GITHUB_TOKEN": "ghp_x"}
            }
          }
        }"#;
        let f = scan_config(json, "~/.config/mcp/servers.json");
        assert!(f.iter().any(|x| x.kind == "mcp-unpinned"));
        assert!(f.iter().any(|x| x.kind == "mcp-secret-to-unpinned"));
        assert!(f.iter().all(|x| x.subject == "github"));
    }

    #[test]
    fn parses_the_bare_servers_shape() {
        let json = r#"{"servers": {"fs": {"command": "npx", "args": ["mcp-fs", "/"]}}}"#;
        let f = scan_config(json, "mcp.json");
        assert!(f.iter().any(|x| x.kind == "mcp-broad-scope"));
    }

    #[test]
    fn secret_values_are_never_read() {
        // We take env KEYS only. A finding that quoted the token would put
        // the secret in a log file, which is its own vulnerability.
        let json = r#"{"mcpServers":{"g":{"command":"npx","args":["x"],"env":{"API_KEY":"sk-live-SECRET"}}}}"#;
        let f = scan_config(json, "mcp.json");
        for finding in &f {
            assert!(!finding.explanation.contains("sk-live-SECRET"));
            assert!(!finding.subject.contains("sk-live-SECRET"));
        }
    }

    #[test]
    fn malformed_config_does_not_panic_or_accuse() {
        assert!(scan_config("{ not json", "mcp.json").is_empty());
        assert!(scan_config("{}", "mcp.json").is_empty());
        assert!(scan_config("", "mcp.json").is_empty());
    }
}