From 9fb83fa37dbe7390f92e400a71ffdbc0059e46cb Mon Sep 17 00:00:00 2001 From: Raymond Ginger Date: Wed, 9 Sep 2026 21:43:56 +0800 Subject: [PATCH] feat(mcp): supply-chain audit declarations for MCP servers --- core/mcp/audit.py | 177 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 177 insertions(+) create mode 100644 core/mcp/audit.py diff --git a/core/mcp/audit.py b/core/mcp/audit.py new file mode 100644 index 00000000..1d31d907 --- /dev/null +++ b/core/mcp/audit.py @@ -0,0 +1,177 @@ +"""P2-E3 (GenAI lesson 13): MCP supply-chain audit. + +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 a resolved MCP +plan: for every server, what is being introduced (transport, source, +command/URL), what capabilities it declares (tool count + names), and what +policy constrains it (approval mode, enabled/disabled tools, read-only hints, +allowlist status). Pure mechanism — no network, no execution. + +The output is designed for: (a) a human review before first use, (b) a +regression diff when a config changes (a newly appearing server/tool in the +diff is a supply-chain event worth noticing), and (c) feeding the P1-9 +allowlist decision. +""" + +from __future__ import annotations + +import json +from dataclasses import asdict, dataclass, field +from typing import Any + +from core.mcp.naming import server_allowed + + +@dataclass +class ServerAuditEntry: + """One server's supply-chain declaration.""" + + server_id: str + name: str + source: str # "user" | "project" | "plugin" | ... + transport: str # "stdio" | "http" | "sse" + command: str | None = None # stdio executable (provenance) + url: str | None = None # http endpoint + tool_count: int = 0 + tools: list[str] = field(default_factory=list) + approval_mode: str | None = None + enabled_tools: tuple[str, ...] | None = None + disabled_tools: tuple[str, ...] = field(default_factory=tuple) + allowlisted: bool = True # P1-9: passes DEEPCODE_MCP_SERVER_ALLOWLIST + read_only_tools: int = 0 + notes: list[str] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + payload = asdict(self) + if not payload["tools"]: + payload.pop("tools") + if not payload["disabled_tools"]: + payload.pop("disabled_tools") + if not payload["notes"]: + payload.pop("notes") + return payload + + +@dataclass +class McpAuditReport: + """Aggregate audit of a resolved MCP plan.""" + + 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).""" + risks: list[str] = [] + for server in self.servers: + if not server.allowlisted: + risks.append( + f"server '{server.name}' is NOT on the P1-9 allowlist — " + "it will register no tools" + ) + 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 ("http", "sse") 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 " + f"({server.tool_count}) with no explicit filter" + ) + return risks + + +def audit_plan(plan: Any) -> McpAuditReport: + """Build the audit for a resolved :class:`McpRuntimePlan`. + + Accepts any object exposing ``servers`` (an iterable of resolved servers + with ``definition``) so it stays decoupled from the exact model type. + """ + report = McpAuditReport() + servers = getattr(plan, "servers", None) or [] + for resolved in servers: + server = getattr(resolved, "server", None) or resolved + definition = getattr(server, "definition", None) + if definition is None: + continue + transport = str(getattr(definition, "type", "unknown") or "unknown") + entry = ServerAuditEntry( + server_id=str(getattr(server, "server_id", "") or ""), + name=str(getattr(server, "name", "") or ""), + source=str(getattr(server, "source", "unknown") or "unknown"), + transport=transport, + command=getattr(definition, "command", None), + url=getattr(definition, "url", None), + approval_mode=_mode_name(getattr(definition, "approval_mode", None)), + enabled_tools=getattr(definition, "enabled_tools", None), + disabled_tools=tuple(getattr(definition, "disabled_tools", None) or ()), + allowlisted=server_allowed( + str(getattr(server, "server_id", "") or ""), + str(getattr(server, "name", "") or ""), + ), + ) + # Tool inventory comes from the definition's filters; tool_count is a + # declared capability (the runtime discovers the real set at startup). + declared = _declared_tools(definition) + entry.tools = declared + entry.tool_count = len(declared) + notes = _definition_notes(definition) + entry.notes = notes + report.servers.append(entry) + return report + + +def _mode_name(value: Any) -> str | None: + if value is None: + return None + return getattr(value, "value", None) or str(value) + + +def _declared_tools(definition: Any) -> list[str]: + enabled = getattr(definition, "enabled_tools", None) + if enabled and "*" not in enabled: + return list(enabled) + disabled = tuple(getattr(definition, "disabled_tools", None) or ()) + if disabled: + return ["* (all except: " + ", ".join(disabled) + ")"] + return ["*"] + + +def _definition_notes(definition: Any) -> list[str]: + notes: list[str] = [] + if getattr(definition, "read_only_tools", None): + notes.append("declares read-only tool hints") + if getattr(definition, "required_env_vars", None): + notes.append("requires env vars: " + ", ".join(definition.required_env_vars)) + if getattr(definition, "supports_parallel_tool_calls", None) is False: + notes.append("serial tool calls only") + return notes + + +def _now_iso() -> str: + import time + + return time.strftime("%Y-%m-%dT%H:%M:%S") + + +__all__ = [ + "McpAuditReport", + "ServerAuditEntry", + "audit_plan", +]