#!/usr/bin/env python3 """mcp-audit: read every MCP tool description your coding agent trusts, before it does. Part 3 of "Hacking AI Agents" (kernel.panic.0x11) shows a calculator tool whose description told the agent to read ~/.ssh/id_rsa and pass it along in a "sidenote" argument. The math came back right. The credentials went to the attacker. Most agents read tool descriptions as trusted text. What this script does 1. Finds the MCP servers configured for Claude Code, Claude Desktop, Cursor, Windsurf, and VS Code, in your home directory and in the current project. 2. For each server it does exactly what your agent does at startup: launches the configured command (or POSTs to the configured URL), sends `initialize` and `tools/list`, reads the tool descriptions and argument schemas, and shuts the server down. Ten seconds per server. 3. Scans every description, argument name and argument description for the tells of tool poisoning: secret paths, secrecy toward you, agent-directed imperatives, exfiltration verbs, and cross-tool hijacking. Quotes the offending sentence so you can judge it yourself. What it never does - It never calls a tool (`tools/call`). Listing is read-only. - It never sends anything anywhere. No network except to HTTP MCP servers you configured. - It never modifies a config file. Read this file before running it. Then: python3 mcp-audit.py audit everything it can find python3 mcp-audit.py --no-launch only list the servers, start nothing python3 mcp-audit.py --config FILE audit one config file python3 mcp-audit.py --json machine-readable report Exit code 1 when any tool is flagged, so it can gate CI. Standard library only. Python 3.9+. A flag is a reason to read the description yourself, not proof of an attack. """ from __future__ import annotations import argparse import json import os import queue import re import shlex import subprocess import sys import threading import time import urllib.error import urllib.request from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Iterable, Optional VERSION = "1.0.0" LONG_DESCRIPTION_CHARS = 1200 # ---------------------------------------------------------------------------------------------- # 1. The flagger # ---------------------------------------------------------------------------------------------- # family -> list of case-insensitive regexes. Matches are quoted with their surrounding sentence. PATTERNS: dict[str, list[str]] = { "secrets": [ r"(?", r"\byou must\b", r"\bbefore (using|calling|running) (this|any|the) tool\b", r"\bignore (all |any )?(previous|prior|above|earlier) (instructions|rules)\b", r"\balways (call|use|include|run|read|attach)\b", r"\bfirst,? (load|read|open|cat|fetch)\b", r"\bsystem prompt\b", r"\byou are (now|an?) \b", r"\bnew instructions?\b", r"\bthis is (required|mandatory) (for|to)\b", r"\bfailure to (do|comply)\b", ], "exfiltration": [ r"\binclude (it|its contents|the contents|the (full )?(file|key|value|output)s?|all of it|everything) (in|as|with)\b", r"\bpass (it |them |this |that )?(along|through|on)\b", r"\bside ?notes?\b", r"\bsend (it|them|this|that|the (contents|file|key|data)s?) (to|along)\b", r"\bappend (it|them|the contents|the file) to\b", r"\bexfil", r"\bforward (it|them|the) \b", r"\bas (a|an) (extra|additional|side|hidden) (parameter|argument|field|note)\b", r"\bpaste (it|them|the contents|the file) \b", r"\bupload (it|them|the file)\b", r"\bencode (it|them|the contents) \b", r"\bin the (background|response) field\b", ], "hijack": [ r"\bwhen the user (uses|calls|asks for|invokes|requests|mentions)\b", r"\binstead of (it|the|that|calling|using)\b", r"\b(all|any|the) other tools?\b", r"\breplace the\b", r"\boverride\b", r"\b(disable|skip|bypass) (the |any )?(approval|confirmation|review|check)s?\b", r"\bredirect\b", r"\bchange the (recipient|destination|address|url)\b", r"\bfor every (tool|request|call|message)\b", r"\bon every (tool|request|call|message)\b", ], } COMPILED = {fam: [re.compile(p, re.IGNORECASE) for p in pats] for fam, pats in PATTERNS.items()} FAMILY_LABEL = { "secrets": "references secrets", "secrecy": "hides something from you", "imperative": "instructs the agent", "exfiltration": "moves data out", "hijack": "hijacks other tools", "heuristic": "looks unusual", } @dataclass(frozen=True) class Flag: family: str # one of PATTERNS keys or "heuristic" where: str # "description" or "argument " match: str # the sentence (or summary) that tripped the flag pattern: str = "" # the regex, for the curious # a sentence ends at . ! or ? followed by whitespace or end of text, or at a line break. # A dot inside a path like ~/.ssh/id_rsa or a version like 1.2.3 is not a sentence end. _BOUNDARY = re.compile(r"[.!?](?=\s|$)|[\r\n]") def _sentence_around(text: str, start: int, end: int, limit: int = 160) -> str: """The sentence containing [start, end), trimmed, so a reader can judge the phrase in context.""" before = [m.end() for m in _BOUNDARY.finditer(text, 0, start)] s = before[-1] if before else 0 m = _BOUNDARY.search(text, end) e = m.end() if m else len(text) sentence = " ".join(text[s:e].split()) if len(sentence) > limit: # keep the match visible when the sentence is long rel = start - s lo = max(0, rel - limit // 3) sentence = ("…" if lo else "") + " ".join(text[s + lo:s + lo + limit].split()) + "…" return sentence.strip() def _scan_text(text: str, where: str) -> list[Flag]: flags: list[Flag] = [] if not text: return flags for fam, regs in COMPILED.items(): for rx in regs: for m in rx.finditer(text): flags.append(Flag(fam, where, _sentence_around(text, m.start(), m.end()), rx.pattern)) return flags def scan_tool(tool: dict) -> list[Flag]: """All flags for one MCP tool record (as returned by tools/list). Empty list means clean.""" name = str(tool.get("name", "?")) desc = str(tool.get("description") or "") schema = tool.get("inputSchema") or tool.get("input_schema") or {} props = schema.get("properties") or {} if isinstance(schema, dict) else {} flags = _scan_text(desc, "description") if len(desc) > LONG_DESCRIPTION_CHARS: flags.append(Flag("heuristic", "description", f"description is {len(desc)} characters; that is a lot of text for the agent to obey")) text_args_without_docs: list[str] = [] for arg, spec in props.items(): spec = spec if isinstance(spec, dict) else {} where = f"argument {arg}" flags += _scan_text(str(arg), where) flags += _scan_text(str(spec.get("description") or ""), where) if spec.get("type") == "string" and not spec.get("description") and not spec.get("enum"): text_args_without_docs.append(str(arg)) if text_args_without_docs and any(f.family == "secrets" for f in flags): flags.append(Flag("heuristic", "arguments", f"free-text argument(s) {', '.join(text_args_without_docs)} with no description, next to " f"secret references: a place to carry data out")) # dedupe, keep first occurrence order seen: set[tuple[str, str, str]] = set() out: list[Flag] = [] for f in flags: key = (f.family, f.where, f.match) if key not in seen: seen.add(key) out.append(f) _ = name return out # ---------------------------------------------------------------------------------------------- # 2. Config discovery # ---------------------------------------------------------------------------------------------- @dataclass class ConfigSource: client: str path: Path servers: dict = field(default_factory=dict) # name -> spec (command/args/env or url/type) error: Optional[str] = None def _read_json(path: Path) -> Optional[dict]: try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError): return None return data if isinstance(data, dict) else None def _servers_from(data: dict, key: str = "mcpServers") -> dict: servers = data.get(key) return {str(k): v for k, v in servers.items() if isinstance(v, dict)} if isinstance(servers, dict) else {} def _same_path(a: str, b: Path) -> bool: try: return a == str(b) or Path(a).resolve() == b.resolve() except OSError: return False def discover_configs(home: Optional[Path] = None, cwd: Optional[Path] = None) -> list[ConfigSource]: """Every MCP config this machine's agents might load, for the given home and project.""" home = Path(home or Path.home()) cwd = Path(cwd or Path.cwd()) out: list[ConfigSource] = [] def add(client: str, path: Path, key: str = "mcpServers", extra: Optional[dict] = None): if not path.is_file(): return data = _read_json(path) if data is None: out.append(ConfigSource(client, path, {}, "could not parse (not valid JSON)")) return servers = _servers_from(data, key) if extra: servers.update(extra) out.append(ConfigSource(client, path, servers)) # Claude Code: global servers plus the servers registered for this project cc = home / ".claude.json" if cc.is_file(): data = _read_json(cc) if data is None: out.append(ConfigSource("Claude Code", cc, {}, "could not parse (not valid JSON)")) else: servers = _servers_from(data) projects = data.get("projects") if isinstance(data.get("projects"), dict) else {} for proj_path, proj in projects.items(): if isinstance(proj, dict) and _same_path(str(proj_path), cwd): for n, spec in _servers_from(proj).items(): servers.setdefault(n, spec) out.append(ConfigSource("Claude Code", cc, servers)) add("Claude Code (project)", cwd / ".mcp.json") add("Claude Desktop", home / "Library" / "Application Support" / "Claude" / "claude_desktop_config.json") add("Claude Desktop", home / ".config" / "Claude" / "claude_desktop_config.json") add("Cursor", home / ".cursor" / "mcp.json") add("Cursor (project)", cwd / ".cursor" / "mcp.json") add("Windsurf", home / ".codeium" / "windsurf" / "mcp_config.json") add("VS Code (project)", cwd / ".vscode" / "mcp.json", key="servers") return out # ---------------------------------------------------------------------------------------------- # 3. Talking to servers: initialize + tools/list, nothing else # ---------------------------------------------------------------------------------------------- def _rpc(msg_id: Optional[int], method: str, params: Optional[dict] = None) -> dict: msg: dict = {"jsonrpc": "2.0", "method": method} if params is not None: msg["params"] = params if msg_id is not None: msg["id"] = msg_id return msg INIT_PARAMS = {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "mcp-audit", "version": VERSION}} def _transport(spec: dict) -> str: t = str(spec.get("type") or spec.get("transport") or "").lower() if t in ("http", "streamable-http", "streamable_http", "sse"): return "http" if spec.get("url") and not spec.get("command"): return "http" return "stdio" def _describe(spec: dict) -> str: if _transport(spec) == "http": return str(spec.get("url", "?")) parts = [str(spec.get("command", "?"))] + [str(a) for a in (spec.get("args") or [])] return " ".join(shlex.quote(p) for p in parts) def list_tools_stdio(spec: dict, timeout: float = 10.0) -> list[dict]: """Launch the server the way an agent would, read its tool list, shut it down. Never calls a tool.""" cmd = [str(spec.get("command", ""))] + [str(a) for a in (spec.get("args") or [])] if not cmd[0]: raise RuntimeError("no command configured") env = dict(os.environ) env.update({str(k): str(v) for k, v in (spec.get("env") or {}).items()}) cwd = spec.get("cwd") or None proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, encoding="utf-8", errors="replace", env=env, cwd=cwd, bufsize=1) lines: "queue.Queue[Optional[str]]" = queue.Queue() def pump(): try: for line in proc.stdout: # type: ignore[union-attr] lines.put(line) finally: lines.put(None) threading.Thread(target=pump, daemon=True).start() deadline = time.monotonic() + timeout def send(msg: dict): assert proc.stdin is not None proc.stdin.write(json.dumps(msg) + "\n") proc.stdin.flush() def wait_for(msg_id: int) -> dict: while True: remaining = deadline - time.monotonic() if remaining <= 0: raise TimeoutError(f"no response to id {msg_id} within {timeout:.0f}s") try: line = lines.get(timeout=remaining) except queue.Empty: raise TimeoutError(f"no response to id {msg_id} within {timeout:.0f}s") if line is None: rc = proc.poll() raise RuntimeError(f"server exited before answering (exit code {rc})") line = line.strip() if not line.startswith("{"): continue # log noise on stdout try: msg = json.loads(line) except json.JSONDecodeError: continue if msg.get("id") == msg_id: if "error" in msg: err = msg["error"] raise RuntimeError(f"server error {err.get('code')}: {err.get('message')}") return msg.get("result") or {} try: send(_rpc(1, "initialize", INIT_PARAMS)) wait_for(1) send(_rpc(None, "notifications/initialized")) tools: list[dict] = [] cursor = None msg_id = 2 while True: params = {"cursor": cursor} if cursor else {} send(_rpc(msg_id, "tools/list", params)) result = wait_for(msg_id) tools += [t for t in (result.get("tools") or []) if isinstance(t, dict)] cursor = result.get("nextCursor") msg_id += 1 if not cursor or msg_id > 50: break return tools finally: try: if proc.stdin: proc.stdin.close() except OSError: pass proc.terminate() try: proc.wait(timeout=2) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=2) if proc.stdout: proc.stdout.close() def _http_post(url: str, headers: dict, body: dict, timeout: float) -> tuple[dict, dict]: req = urllib.request.Request(url, data=json.dumps(body).encode(), method="POST", headers={"Content-Type": "application/json", "Accept": "application/json, text/event-stream", **headers}) with urllib.request.urlopen(req, timeout=timeout) as resp: ctype = resp.headers.get("Content-Type", "") raw = resp.read().decode("utf-8", "replace") resp_headers = {k.lower(): v for k, v in resp.headers.items()} if "text/event-stream" in ctype: for line in raw.splitlines(): if line.startswith("data:"): try: msg = json.loads(line[5:].strip()) except json.JSONDecodeError: continue if msg.get("id") == body.get("id"): return msg, resp_headers raise RuntimeError("no matching response in event stream") return (json.loads(raw) if raw.strip() else {}), resp_headers def list_tools_http(spec: dict, timeout: float = 10.0) -> list[dict]: """Streamable HTTP: POST initialize, then tools/list. Best effort; SSE-only servers may not answer.""" url = str(spec.get("url", "")) if not url: raise RuntimeError("no url configured") headers = {str(k): str(v) for k, v in (spec.get("headers") or {}).items()} msg, rh = _http_post(url, headers, _rpc(1, "initialize", INIT_PARAMS), timeout) if "error" in msg: raise RuntimeError(f"server error: {msg['error'].get('message')}") if rh.get("mcp-session-id"): headers["Mcp-Session-Id"] = rh["mcp-session-id"] try: _http_post(url, headers, _rpc(None, "notifications/initialized"), timeout) except (urllib.error.HTTPError, RuntimeError, json.JSONDecodeError): pass # notifications may legitimately return 202 with an empty body tools: list[dict] = [] cursor = None msg_id = 2 while True: msg, _ = _http_post(url, headers, _rpc(msg_id, "tools/list", {"cursor": cursor} if cursor else {}), timeout) if "error" in msg: raise RuntimeError(f"server error: {msg['error'].get('message')}") result = msg.get("result") or {} tools += [t for t in (result.get("tools") or []) if isinstance(t, dict)] cursor = result.get("nextCursor") msg_id += 1 if not cursor or msg_id > 50: return tools def list_tools(spec: dict, timeout: float = 10.0) -> list[dict]: return list_tools_http(spec, timeout) if _transport(spec) == "http" else list_tools_stdio(spec, timeout) # ---------------------------------------------------------------------------------------------- # 4. Auditing # ---------------------------------------------------------------------------------------------- def explain_error(e: BaseException) -> str: """One plain sentence a reader can act on, instead of an exception dump.""" if isinstance(e, urllib.error.HTTPError): if e.code in (401, 403): return (f"needs authentication the agent holds (HTTP {e.code}); this script has no OAuth token, " f"so read this server's tool list from inside your client") return f"HTTP {e.code} from the server" if isinstance(e, urllib.error.URLError): return f"could not connect: {e.reason}" if isinstance(e, TimeoutError): return f"{e}; the server may expect different arguments or environment than this config gives it" if isinstance(e, FileNotFoundError): return f"command not found: {e.filename or e}" return f"{type(e).__name__}: {e}"[:300] def audit_servers(servers: dict, launch: bool = True, timeout: float = 10.0) -> list[dict]: rows: list[dict] = [] for name, spec in servers.items(): row: dict = {"name": name, "transport": _transport(spec), "command": _describe(spec), "tools": [], "error": None} if launch: try: for t in list_tools(spec, timeout): row["tools"].append({"name": str(t.get("name", "?")), "description": str(t.get("description") or ""), "flags": [asdict(f) for f in scan_tool(t)]}) except (OSError, RuntimeError, TimeoutError, urllib.error.URLError, ValueError) as e: row["error"] = explain_error(e) rows.append(row) return rows def audit_config_file(path: Path, client: str = "config", launch: bool = True, timeout: float = 10.0) -> dict: """Audit one config file (mcpServers or VS Code style servers). Used by --config and by tests.""" data = _read_json(Path(path)) if data is None: return {"client": client, "path": str(path), "servers": [], "error": "could not parse (not valid JSON)"} servers = _servers_from(data) or _servers_from(data, "servers") return {"client": client, "path": str(path), "servers": audit_servers(servers, launch, timeout), "error": None} def audit_sources(sources: Iterable[ConfigSource], launch: bool = True, timeout: float = 10.0) -> dict: clients = [] for src in sources: clients.append({"client": src.client, "path": str(src.path), "servers": audit_servers(src.servers, launch, timeout) if not src.error else [], "error": src.error}) return summarize(clients, launch) def summarize(clients: list[dict], launch: bool) -> dict: tools = [t for c in clients for s in c["servers"] for t in s["tools"]] return { "version": VERSION, "launched": launch, "clients": clients, "servers": sum(len(c["servers"]) for c in clients), "tools_read": len(tools), "flagged_tools": sum(1 for t in tools if t["flags"]), "server_errors": sum(1 for c in clients for s in c["servers"] if s["error"]), } # ---------------------------------------------------------------------------------------------- # 5. Report # ---------------------------------------------------------------------------------------------- def _plural(n: int, word: str) -> str: return f"{n} {word}{'' if n == 1 else 's'}" def render(report: dict) -> str: out = [f"mcp-audit {VERSION} · kernel.panic.0x11 · read the descriptions before your agent does", ""] if not report["clients"]: out.append("No MCP configs found for Claude Code, Claude Desktop, Cursor, Windsurf, or VS Code.") out.append("Point at one directly: python3 mcp-audit.py --config path/to/mcp.json") return "\n".join(out) + "\n" for c in report["clients"]: out.append(f"{c['client']} {c['path']}") if c["error"]: out.append(f" ! {c['error']}") if not c["servers"] and not c["error"]: out.append(" (no servers configured)") for s in c["servers"]: if not report["launched"]: out.append(f" {s['name']:<22} {s['transport']:<6} {s['command']}") continue if s["error"]: out.append(f" {s['name']:<22} ! could not read tools: {s['error']}") continue flagged = [t for t in s["tools"] if t["flags"]] status = "clean" if not flagged else f"⚠ {_plural(sum(len(t['flags']) for t in flagged), 'flag')}" out.append(f" {s['name']:<22} {_plural(len(s['tools']), 'tool'):<10} {status}") for t in flagged: out.append(f" {t['name']}") for f in t["flags"]: out.append(f" [{FAMILY_LABEL.get(f['family'], f['family'])}] {f['where']}: “{f['match']}”") out.append("") if report["launched"]: out.append(f"{_plural(report['tools_read'], 'tool description')} read from {_plural(report['servers'], 'server')}. " f"{_plural(report['flagged_tools'], 'tool')} flagged." + (f" {_plural(report['server_errors'], 'server')} could not be read." if report["server_errors"] else "")) if report["flagged_tools"]: out.append("A flag is a reason to read that description yourself, not proof of an attack. " "If you did not write the server, find out who did.") elif report["tools_read"]: out.append("Nothing tripped. Rerun after every server update; descriptions change without your config changing.") else: out.append(f"{_plural(report['servers'], 'server')} configured. Nothing was launched (--no-launch). " f"Drop the flag to read their tool descriptions.") return "\n".join(out) + "\n" # ---------------------------------------------------------------------------------------------- # 6. CLI # ---------------------------------------------------------------------------------------------- def main(argv: Optional[list[str]] = None) -> int: ap = argparse.ArgumentParser(prog="mcp-audit", description=__doc__.split("\n\n")[0], epilog="Exit code 1 when any tool is flagged.") ap.add_argument("--config", action="append", metavar="FILE", help="audit this config file (repeatable)") ap.add_argument("--no-launch", action="store_true", help="list configured servers without starting any") ap.add_argument("--json", action="store_true", help="machine-readable report") ap.add_argument("--timeout", type=float, default=10.0, metavar="SECONDS", help="per-server timeout (default 10)") ap.add_argument("--version", action="version", version=f"mcp-audit {VERSION}") args = ap.parse_args(argv) launch = not args.no_launch if args.config: clients = [audit_config_file(Path(p), client="config", launch=launch, timeout=args.timeout) for p in args.config] if not launch: for c, p in zip(clients, args.config): data = _read_json(Path(p)) or {} servers = _servers_from(data) or _servers_from(data, "servers") c["servers"] = audit_servers(servers, launch=False) report = summarize(clients, launch) else: report = audit_sources(discover_configs(), launch=launch, timeout=args.timeout) sys.stdout.write(json.dumps(report, indent=2) + "\n" if args.json else render(report)) return 1 if report["flagged_tools"] else 0 if __name__ == "__main__": sys.exit(main())