feat(mcp): supply-chain audit subcommand for effective servers

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,

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",
+]

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()

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