fix(mcp): block exfil-shaped stdio server configs (#46083)

hermes_cli/config.py

@@ -4121,7 +4121,7 @@ _KNOWN_ROOT_KEYS = {
     "fallback_providers", "credential_pool_strategies", "toolsets",
     "agent", "terminal", "display", "compression", "delegation",
     "auxiliary", "custom_providers", "context", "memory", "gateway",
-    "sessions", "streaming", "updates",
+    "sessions", "streaming", "updates", "mcp_servers",
 }
 
 # Valid fields inside a custom_providers list entry
@@ -4829,6 +4829,38 @@ def migrate_config(interactive: bool = True, quiet: bool = False) -> Dict[str, A
             if not quiet:
                 print("  ✓ Renamed write_mode → write_approval (boolean gate)")
 
+    # ── Post-migration: disable exfiltration-shaped MCP stdio entries ──
+    # Users can hand-edit mcp_servers, and older installs may already contain a
+    # malicious entry. Preserve the stanza for auditability but mark it
+    # disabled so the next startup will not spawn it. (#45620)
+    config = read_raw_config()
+    raw_mcp_servers = config.get("mcp_servers")
+    if isinstance(raw_mcp_servers, dict):
+        try:
+            from hermes_cli.mcp_security import validate_mcp_server_entry
+        except Exception:
+            validate_mcp_server_entry = None
+        if validate_mcp_server_entry:
+            mcp_touched = False
+            for server_name, entry in raw_mcp_servers.items():
+                if not isinstance(entry, dict):
+                    continue
+                issues = validate_mcp_server_entry(server_name, entry)
+                if not issues:
+                    continue
+                entry["enabled"] = False
+                mcp_touched = True
+                results["warnings"].append(
+                    f"Disabled suspicious MCP server '{server_name}'"
+                )
+                if not quiet:
+                    for issue in issues:
+                        print(f"  ⚠ {issue}")
+                    print(f"  ⚠ Disabled MCP server '{server_name}' pending review")
+            if mcp_touched:
+                config["mcp_servers"] = raw_mcp_servers
+                save_config(config)
+
     if current_ver < latest_ver and not quiet:
         print(f"Config version: {current_ver} → {latest_ver}")
     

hermes_cli/doctor.py

@@ -556,6 +556,30 @@ def run_doctor(args):
     except Exception as e:
         # Never let a bug in the advisory check block the rest of doctor.
         check_warn(f"Security advisory check failed: {e}")
+
+    _section("MCP Server Security")
+    try:
+        from hermes_cli.config import load_config
+        from hermes_cli.mcp_security import validate_mcp_server_entry
+
+        servers = load_config().get("mcp_servers") or {}
+        suspicious = 0
+        if isinstance(servers, dict):
+            for name, entry in sorted(servers.items()):
+                if not isinstance(entry, dict):
+                    continue
+                issues_found = validate_mcp_server_entry(name, entry)
+                if not issues_found:
+                    continue
+                suspicious += 1
+                check_warn(f"MCP server '{name}' has suspicious stdio command", "; ".join(issues_found))
+                manual_issues.append(
+                    f"Review/remove mcp_servers.{name} in config.yaml; rotate any credentials that may have been exposed."
+                )
+        if suspicious == 0:
+            check_ok("No suspicious MCP stdio commands")
+    except Exception as e:
+        check_warn(f"MCP security check failed: {e}")
     
     _section("Python Environment")
     py_version = sys.version_info

hermes_cli/mcp_catalog.py

@@ -730,9 +730,12 @@ def install_entry(entry: CatalogEntry, *, enable: bool = True) -> None:
     server_cfg = _build_server_config(entry, install_dir)
     server_cfg["enabled"] = enable
 
-    cfg = load_config()
-    cfg.setdefault("mcp_servers", {})[entry.name] = server_cfg
-    save_config(cfg)
+    from hermes_cli.mcp_config import _save_mcp_server
+
+    if not _save_mcp_server(entry.name, server_cfg):
+        raise CatalogError(
+            f"catalog entry '{entry.name}' rejected: suspicious command/args configuration"
+        )
 
     # ── Probe + tool selection ──────────────────────────────────────────
     _apply_tool_selection(entry, prior_selection=prior_selection)

hermes_cli/mcp_config.py

@@ -25,6 +25,7 @@ from hermes_cli.config import (
 )
 from hermes_cli.colors import Colors, color
 from hermes_constants import display_hermes_home
+from hermes_cli.mcp_security import validate_mcp_server_entry
 from tools.mcp_tool import _ENV_VAR_PATTERN
 
 logger = logging.getLogger(__name__)
@@ -84,11 +85,23 @@ def _get_mcp_servers(config: Optional[dict] = None) -> Dict[str, dict]:
     return servers
 
 
-def _save_mcp_server(name: str, server_config: dict):
-    """Add or update a server entry in config.yaml."""
+def _save_mcp_server(name: str, server_config: dict) -> bool:
+    """Add or update a server entry in config.yaml.
+
+    Returns False when a high-signal exfiltration-shaped stdio command is
+    rejected. MCP stdio servers are user-chosen local commands, so this blocks
+    shell+egress payloads rather than whitelisting command families.
+    """
+    issues = validate_mcp_server_entry(name, server_config)
+    if issues:
+        for issue in issues:
+            _warning(issue)
+        _warning(f"Server '{name}' was NOT saved due to suspicious configuration.")
+        return False
     config = load_config()
     config.setdefault("mcp_servers", {})[name] = server_config
     save_config(config)
+    return True
 
 
 def _remove_mcp_server(name: str) -> bool:
@@ -403,16 +416,16 @@ def cmd_mcp_add(args):
         _error(f"Failed to connect: {exc}")
         if _confirm("Save config anyway (you can test later)?", default=False):
             server_config["enabled"] = False
-            _save_mcp_server(name, server_config)
-            _success(f"Saved '{name}' to config (disabled)")
-            _info("Fix the issue, then: hermes mcp test " + name)
+            if _save_mcp_server(name, server_config):
+                _success(f"Saved '{name}' to config (disabled)")
+                _info("Fix the issue, then: hermes mcp test " + name)
         return
 
     if not tools:
         _warning("Server connected but reported no tools.")
         if _confirm("Save config anyway?", default=True):
-            _save_mcp_server(name, server_config)
-            _success(f"Saved '{name}' to config")
+            if _save_mcp_server(name, server_config):
+                _success(f"Saved '{name}' to config")
         return
 
     # ── Tool selection ────────────────────────────────────────────────
@@ -469,11 +482,10 @@ def cmd_mcp_add(args):
     # ── Save ──────────────────────────────────────────────────────────
 
     server_config["enabled"] = True
-    _save_mcp_server(name, server_config)
-
-    print()
-    _success(f"Saved '{name}' to {display_hermes_home()}/config.yaml ({tool_count}/{total} tools enabled)")
-    _info("Start a new session to use these tools.")
+    if _save_mcp_server(name, server_config):
+        print()
+        _success(f"Saved '{name}' to {display_hermes_home()}/config.yaml ({tool_count}/{total} tools enabled)")
+        _info("Start a new session to use these tools.")
 
 
 # ─── hermes mcp remove ───────────────────────────────────────────────────────

hermes_cli/mcp_security.py

@@ -0,0 +1,96 @@
+"""Security checks for user-configured MCP server entries.
+
+MCP stdio transports intentionally support arbitrary local commands so users can
+run custom servers. This module does not try to sandbox that capability. It only
+blocks the high-signal exfiltration shape from #45620: a shell interpreter whose
+inline script invokes network egress tooling.
+"""
+from __future__ import annotations
+
+import os
+import re
+import shlex
+from typing import Any
+
+_SHELL_INTERPRETERS = frozenset({
+    "bash",
+    "sh",
+    "zsh",
+    "dash",
+    "fish",
+    "cmd",
+    "cmd.exe",
+    "powershell",
+    "powershell.exe",
+    "pwsh",
+    "pwsh.exe",
+})
+
+_EGRESS_PATTERN = re.compile(
+    r"(?<![\w.-])(?:curl|wget|nc|ncat|socat)(?![\w.-])"
+    r"|/dev/tcp/"
+    r"|\bInvoke-WebRequest\b"
+    r"|\bInvoke-RestMethod\b"
+    r"|\bSystem\.Net\.WebClient\b",
+    re.IGNORECASE,
+)
+
+_EXFIL_HINT_PATTERN = re.compile(
+    r"\.env\b|--data-binary|--data-raw|\b-X\s+POST\b|\bPOST\b|<\s*[^\s]+",
+    re.IGNORECASE,
+)
+
+
+def _command_basename(command: Any) -> str:
+    text = str(command or "").strip()
+    if not text:
+        return ""
+    try:
+        parts = shlex.split(text, posix=(os.name != "nt"))
+    except ValueError:
+        parts = text.split()
+    first = parts[0] if parts else text
+    return os.path.basename(first).lower()
+
+
+def _inline_script(args: Any) -> str:
+    if args is None:
+        return ""
+    if isinstance(args, (list, tuple)):
+        return " ".join(str(item) for item in args)
+    return str(args)
+
+
+def validate_mcp_server_entry(name: str, entry: dict[str, Any]) -> list[str]:
+    """Return security warnings for an MCP server entry.
+
+    Empty return means the entry is not suspicious under the narrow #45620
+    exfiltration heuristic. This is intentionally not a whitelist: legitimate
+    local MCPs can still use custom commands, Python scripts, npx, uvx, etc.
+    """
+    if not isinstance(entry, dict):
+        return []
+
+    command = entry.get("command")
+    basename = _command_basename(command)
+    if basename not in _SHELL_INTERPRETERS:
+        return []
+
+    script = _inline_script(entry.get("args"))
+    if not script:
+        return []
+
+    if not _EGRESS_PATTERN.search(script):
+        return []
+
+    issue = (
+        f"MCP server '{name}' uses shell interpreter '{command}' with network "
+        "egress in args"
+    )
+    if _EXFIL_HINT_PATTERN.search(script):
+        issue += " and exfiltration-shaped arguments"
+    return [issue]
+
+
+def is_mcp_server_entry_suspicious(name: str, entry: dict[str, Any]) -> bool:
+    return bool(validate_mcp_server_entry(name, entry))

hermes_cli/web_server.py

@@ -7134,7 +7134,11 @@ async def add_mcp_server(body: MCPServerCreate, profile: Optional[str] = None):
 
     try:
         with _profile_scope(body.profile or profile):
-            _save_mcp_server(name, server_config)
+            if not _save_mcp_server(name, server_config):
+                raise HTTPException(
+                    status_code=400,
+                    detail=f"Server '{name}' rejected: suspicious command/args configuration",
+                )
     except HTTPException:
         raise
     except Exception as exc:
@@ -8732,6 +8736,7 @@ def _write_profile_mcp_servers(profile_dir: Path, servers: List["MCPServerCreate
     Returns the number of servers written.
     """
     from hermes_constants import set_hermes_home_override, reset_hermes_home_override
+    from hermes_cli.mcp_security import validate_mcp_server_entry
 
     written = 0
     token = set_hermes_home_override(str(profile_dir))
@@ -8757,6 +8762,10 @@ def _write_profile_mcp_servers(profile_dir: Path, servers: List["MCPServerCreate
                 # Nothing usable to write (neither url nor command) — skip
                 # rather than persist an empty, unusable server stanza.
                 continue
+            issues = validate_mcp_server_entry(name, entry)
+            if issues:
+                _log.warning("Profile-create: skipping MCP server '%s': %s", name, "; ".join(issues))
+                continue
             mcp[name] = entry
             written += 1
         if written: