diff --git a/cli/mcp_cli.py b/cli/mcp_cli.py index 1d0f887a..56aa2efc 100644 --- a/cli/mcp_cli.py +++ b/cli/mcp_cli.py @@ -32,6 +32,13 @@ def _parser() -> argparse.ArgumentParser: _workspace_arguments(listing, allow_trust=False) listing.add_argument("--json", action="store_true") + audit = commands.add_parser( + "audit", + help="Render a supply-chain audit of the effective MCP servers.", + ) + _workspace_arguments(audit, allow_trust=False) + audit.add_argument("--json", action="store_true") + presets = commands.add_parser("presets", help="List bundled MCP templates.") _workspace_arguments(presets, allow_trust=False) presets.add_argument("--json", action="store_true") @@ -167,6 +174,11 @@ def run(argv: list[str] | None = None) -> int: if args.command == "list": inventory = application.mcp.list(project_id) result = mcp_inventory_view(inventory) + elif args.command == "audit": + from core.mcp.audit import audit_inventory + + report = audit_inventory(application.mcp.list(project_id)) + result = {**report.to_dict(), "risks": report.risks()} elif args.command == "presets": result = mcp_preset_inventory_view(application.mcp.list_presets(project_id)) elif args.command in {"enable", "disable"}: @@ -249,6 +261,8 @@ def run(argv: list[str] | None = None) -> int: if result["removed"] else "No OAuth credentials were stored." ) + elif args.command == "audit": + _print_audit(result) else: _print_inventory(result) return exit_code @@ -259,6 +273,15 @@ def run(argv: list[str] | None = None) -> int: application.close() +def _print_audit(result: dict[str, Any]) -> None: + print(f"MCP supply-chain audit: {result['server_count']} server(s).") + for server in result["servers"]: + endpoint = server.get("command") or server.get("url") or "-" + print(f" {server['name']:<20} {server['transport']:<14} {endpoint}") + for risk in result["risks"]: + print(f"risk: {risk}") + + def _project_id( application: DeepCodeApplication, workspace: str | None, diff --git a/core/mcp/audit.py b/core/mcp/audit.py new file mode 100644 index 00000000..bbd8ca7d --- /dev/null +++ b/core/mcp/audit.py @@ -0,0 +1,142 @@ +"""MCP supply-chain audit: declaration review for the effective inventory. + +Lesson 13's supply-chain warning: third-party components (Python modules, +external datasets — and for a harness, remote MCP servers) can be compromised. +This module renders a *declaration audit* for the effective MCP inventory: for +every server, what is being introduced (transport, source, command/URL) and +what posture it declares (enabled, tool filter, approval mode). Pure +mechanism — no network, no execution. + +The output is designed for: (a) a human review before first use (``deepcode +mcp audit``), (b) a regression diff when a config changes (a newly appearing +server/tool in the diff is a supply-chain event worth noticing). +""" + +from __future__ import annotations + +import json +import time +from dataclasses import asdict, dataclass, field +from typing import Any + + +@dataclass +class ServerAuditEntry: + """One server's supply-chain declaration.""" + + server_id: str + name: str + transport: str # "stdio" | "sse" | "streamableHttp" + enabled: bool = True + command: str | None = None # stdio executable (provenance) + url: str | None = None # remote endpoint + approval_mode: str | None = None + enabled_tools: tuple[str, ...] | None = None + disabled_tools: tuple[str, ...] = field(default_factory=tuple) + notes: list[str] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + payload = asdict(self) + if not payload["disabled_tools"]: + payload.pop("disabled_tools") + if not payload["notes"]: + payload.pop("notes") + return payload + + +@dataclass +class McpAuditReport: + """Aggregate audit of an effective MCP inventory.""" + + servers: list[ServerAuditEntry] = field(default_factory=list) + generated_at: str = field(default_factory=lambda: _now_iso()) + + def to_dict(self) -> dict[str, Any]: + return { + "generated_at": self.generated_at, + "server_count": len(self.servers), + "servers": [s.to_dict() for s in self.servers], + } + + def to_json(self) -> str: + return json.dumps(self.to_dict(), ensure_ascii=False, indent=2, default=str) + + def risks(self) -> list[str]: + """Human-facing risk lines (supply-chain review checklist). + + Disabled servers register nothing at runtime, so they introduce no + supply-chain exposure and produce no risk lines (mirrors the posture + rules in ``cli.mcp_cli._posture_summary``). + """ + risks: list[str] = [] + for server in self.servers: + if not server.enabled: + continue + if server.transport == "stdio" and server.command: + risks.append( + f"server '{server.name}' executes local command " + f"'{server.command}' (verify provenance)" + ) + if server.transport in ("sse", "streamableHttp") and server.url: + risks.append( + f"server '{server.name}' talks to remote endpoint " + f"'{server.url}' (verify trust)" + ) + if not server.enabled_tools and not server.disabled_tools: + risks.append( + f"server '{server.name}' exposes ALL its tools " + "with no explicit filter" + ) + return risks + + +def audit_inventory(inventory: Any) -> McpAuditReport: + """Build the audit for an effective MCP inventory. + + Accepts any object exposing ``servers`` (an iterable of entries with the + ``McpServerInfo`` field names: id/name/transport/command/url/enabled/ + enabled_tools/disabled_tools/approval_mode/missing_env_keys) so it stays + decoupled from the exact model type. + """ + report = McpAuditReport() + for server in getattr(inventory, "servers", None) or []: + report.servers.append( + ServerAuditEntry( + server_id=str(getattr(server, "id", "") or ""), + name=str(getattr(server, "name", "") or ""), + transport=str(getattr(server, "transport", "unknown") or "unknown"), + enabled=bool(getattr(server, "enabled", True)), + command=getattr(server, "command", None), + url=getattr(server, "url", None), + approval_mode=_mode_name(getattr(server, "approval_mode", None)), + enabled_tools=getattr(server, "enabled_tools", None), + disabled_tools=tuple(getattr(server, "disabled_tools", None) or ()), + notes=_server_notes(server), + ) + ) + return report + + +def _mode_name(value: Any) -> str | None: + if value is None: + return None + return getattr(value, "value", None) or str(value) + + +def _server_notes(server: Any) -> list[str]: + notes: list[str] = [] + missing = tuple(getattr(server, "missing_env_keys", None) or ()) + if missing: + notes.append("missing env keys: " + ", ".join(missing)) + return notes + + +def _now_iso() -> str: + return time.strftime("%Y-%m-%dT%H:%M:%S") + + +__all__ = [ + "McpAuditReport", + "ServerAuditEntry", + "audit_inventory", +] diff --git a/tests/test_mcp_audit.py b/tests/test_mcp_audit.py new file mode 100644 index 00000000..9f067970 --- /dev/null +++ b/tests/test_mcp_audit.py @@ -0,0 +1,79 @@ +"""Supply-chain audit of the effective MCP inventory (pure mechanism).""" + +from __future__ import annotations + +from types import SimpleNamespace + +from core.mcp.audit import audit_inventory + + +def _server(**overrides): + base = { + "id": "srv-1", + "name": "fixture", + "transport": "stdio", + "command": "python3", + "args": ("server.py",), + "url": None, + "enabled": True, + "enabled_tools": None, + "disabled_tools": (), + "approval_mode": "auto", + "missing_env_keys": (), + } + base.update(overrides) + return SimpleNamespace(**base) + + +def _inventory(*servers): + return SimpleNamespace(servers=servers) + + +def test_stdio_server_reports_local_command_provenance() -> None: + report = audit_inventory(_inventory(_server())) + risks = report.risks() + assert any("executes local command 'python3'" in risk for risk in risks) + + +def test_remote_server_reports_endpoint_trust() -> None: + report = audit_inventory( + _inventory( + _server( + transport="streamableHttp", + command=None, + url="https://mcp.example/rpc", + ) + ) + ) + assert any( + "remote endpoint 'https://mcp.example/rpc'" in risk for risk in report.risks() + ) + + +def test_unfiltered_server_is_flagged_and_filtered_is_not() -> None: + unfiltered = audit_inventory(_inventory(_server())) + assert any("exposes ALL its tools" in risk for risk in unfiltered.risks()) + filtered = audit_inventory(_inventory(_server(enabled_tools=("read_file",)))) + assert not any("exposes ALL" in risk for risk in filtered.risks()) + + +def test_disabled_servers_produce_no_risks() -> None: + report = audit_inventory(_inventory(_server(enabled=False))) + assert report.risks() == [] + assert report.servers[0].enabled is False + + +def test_missing_env_keys_appear_as_notes() -> None: + report = audit_inventory(_inventory(_server(missing_env_keys=("API_TOKEN",)))) + assert report.servers[0].notes == ["missing env keys: API_TOKEN"] + + +def test_report_serializes_with_trimmed_empty_fields() -> None: + report = audit_inventory(_inventory(_server())) + payload = report.to_dict() + assert payload["server_count"] == 1 + assert payload["servers"][0]["server_id"] == "srv-1" + assert payload["servers"][0]["transport"] == "stdio" + assert "disabled_tools" not in payload["servers"][0] + assert "notes" not in payload["servers"][0] + assert report.to_json() diff --git a/tests/test_mcp_cli.py b/tests/test_mcp_cli.py index 65d023ed..c88bcc44 100644 --- a/tests/test_mcp_cli.py +++ b/tests/test_mcp_cli.py @@ -272,3 +272,49 @@ def test_mcp_list_names_concerns_and_states_default_posture_once( "Default posture among 3 enabled servers: " "2 expose all tools, 2 add no MCP approval gate." ) + + +def test_mcp_cli_audit_renders_supply_chain_report( + tmp_path: Path, + monkeypatch, + capsys, +) -> None: + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("DEEPCODE_HOME", str(home)) + (home / "deepcode_config.json").write_text( + json.dumps( + { + "mcpServers": { + "plain": { + "type": "stdio", + "command": "python3", + "args": ["plain.py"], + }, + "retired": { + "type": "stdio", + "command": "python3", + "args": ["old.py"], + "enabled": False, + }, + } + } + ), + encoding="utf-8", + ) + + assert run(["audit", "--json"]) == 0 + report = _json_output(capsys) + assert report["server_count"] == 2 + by_name = {server["name"]: server for server in report["servers"]} + assert by_name["plain"]["command"] == "python3" + assert by_name["retired"]["enabled"] is False + # Only the enabled server introduces supply-chain exposure. + assert len(report["risks"]) == 2 + assert any("executes local command 'python3'" in risk for risk in report["risks"]) + assert any("exposes ALL its tools" in risk for risk in report["risks"]) + + assert run(["audit"]) == 0 + human = capsys.readouterr().out + assert "MCP supply-chain audit: 2 server(s)." in human + assert "risk:" in human