"""mcp-registry-tool-definition-scan.py - scan the tool definitions shipped in the npm packages listed in the official MCP Registry with the tool-definition auditor from coding-agent-security-prompt-injection-mcp-secrets.py. Two steps, so the numbers can be recomputed offline from a saved copy: python mcp-registry-tool-definition-scan.py fetch --cache mcp-registry-snapshot 1. reads robots.txt for registry.modelcontextprotocol.io and registry.npmjs.org first (it stops if either disallows ClaudeBot, Claude-User, anthropic-ai or *); 2. pages through https://registry.modelcontextprotocol.io/v0/servers?version=latest (100 per page) and saves every page as it arrived; 3. draws a fixed random sample (--sample, default 600; seed 20261029) of the latest, active server entries with an npm package, and for each saves the version manifest from registry.npmjs.org and the package tarball it names (skipped above 15 MB); one request per second, honest user agent, every response logged with URL, bytes, SHA-256 and UTC time in fetch_log.jsonl. Re-running skips saved files. python mcp-registry-tool-definition-scan.py scan --cache mcp-registry-snapshot [--json out.json] [--show N] Offline. Draws the same sample from the saved pages, opens each saved tarball in memory (nothing is unpacked or executed), extracts tool definitions statically from the JavaScript and TypeScript files, and runs audit_tools() from the auditor on each package's list. Prints the coverage (how many packages yielded tools), the findings by code, the share of packages with each code, and the first N findings of each code. --auditor runs another copy of the auditor instead; --tool PACKAGE TOOL prints one extracted definition (parameters appear as p0, p1, ... because the extractor keeps their descriptions, not their names) with its findings, and stops. What "statically extracted" means. The script tokenises each .js/.mjs/.cjs/.ts/.mts/.cts file (string, template and regex literals, comments) and recognises the three ways the TypeScript SDK registers a tool: server.registerTool("name", { title, description, inputSchema: { p: z.string().describe("...") } }, cb) server.tool("name", "description", { p: z.string().describe("...") }, cb) { name: "name", description: "...", inputSchema: { ... } } (the low-level tools/list handler) { name: { description: "...", inputSchema: { ... } } } (the same, keyed by tool name) It keeps the name, the title and description when they are literal strings (adjacent literals joined by + are concatenated; a template literal keeps its fixed text and shows each ${...} as ${}), and every literal `.describe("...")` or `description: "..."` inside the registration as a parameter description. A description built at run time (a variable, a function call, a file read) is not seen. Tools are de-duplicated per package by name (the first definition found wins), because many packages ship both src/ and dist/. Files under test/, tests/, __tests__/, example(s)/ and fixture(s)/, and *.test.* / *.spec.* files, are skipped. A package that registers tools only through code this does not recognise counts as "no tools extracted", not as clean. The definitions a running server sends in tools/list can differ from these strings; this reads what was published, not what a server says at run time. Standard library only (Python 3.11+). The auditor file must sit next to this one. """ from __future__ import annotations import argparse import collections import hashlib import importlib.util import io import json import os import re import sys import tarfile import time import urllib.error import urllib.parse import urllib.request import urllib.robotparser from datetime import datetime, timezone UA = "vibecodetextbook-research/1.0 (+https://vibecodetextbook.com; MCP Registry snapshot for one article)" REGISTRY = "https://registry.modelcontextprotocol.io" NPM = "https://registry.npmjs.org" AGENTS = ("ClaudeBot", "Claude-User", "anthropic-ai", "*") MAX_TARBALL = 15 * 1024 * 1024 SEED = 20261029 SRC_EXT = (".js", ".mjs", ".cjs", ".ts", ".mts", ".cts") HERE = os.path.dirname(os.path.abspath(__file__)) def load_auditor(path=None): path = path or os.path.join(HERE, "coding-agent-security-prompt-injection-mcp-secrets.py") spec = importlib.util.spec_from_file_location("auditor", path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod # ------------------------------------------------------------------ fetch --- class Fetcher: def __init__(self, cache): self.cache = cache self.last = 0.0 self.log = open(os.path.join(cache, "fetch_log.jsonl"), "a", encoding="utf-8") def robots(self, host): req = urllib.request.Request(host + "/robots.txt", headers={"User-Agent": UA, "Accept": "text/plain"}) rp = urllib.robotparser.RobotFileParser() try: body = urllib.request.urlopen(req, timeout=30).read().decode("utf-8", "replace") status = 200 except urllib.error.HTTPError as e: body, status = "", e.code name = urllib.parse.urlsplit(host).netloc.replace(".", "_") with open(os.path.join(self.cache, f"robots_{name}.txt"), "w", encoding="utf-8") as fh: fh.write(f"# status {status} read {datetime.now(timezone.utc).isoformat(timespec='seconds')}\n{body}") # registry.npmjs.org answers /robots.txt with the JSON document of an npm package called robots.txt, # not with crawler rules; a 4xx or a non-robots body means no rules apply. lines = body.splitlines() if status == 200 and not body.lstrip().startswith("{") else [] rp.parse(lines) ok = all(rp.can_fetch(a, host + "/") for a in AGENTS) print(f"robots {host}: HTTP {status}, {'allowed' if ok else 'DISALLOWED'}") return ok def get(self, url, rel): dst = os.path.join(self.cache, rel.replace("/", os.sep)) if os.path.exists(dst): with open(dst, "rb") as fh: return fh.read() wait = self.last + 1.0 - time.time() if wait > 0: time.sleep(wait) req = urllib.request.Request(url, headers={"User-Agent": UA}) status, body = None, b"" try: with urllib.request.urlopen(req, timeout=120) as r: status = r.status length = int(r.headers.get("Content-Length") or 0) if length > MAX_TARBALL: status = "skipped-size" else: body = r.read(MAX_TARBALL + 1) if len(body) > MAX_TARBALL: status, body = "skipped-size", b"" except urllib.error.HTTPError as e: status = e.code except Exception as e: # noqa: BLE001 - logged, the run continues status = repr(e)[:120] self.last = time.time() entry = {"url": url, "file": rel, "status": status, "bytes": len(body), "sha256": hashlib.sha256(body).hexdigest() if body else None, "utc": datetime.now(timezone.utc).isoformat(timespec="seconds")} self.log.write(json.dumps(entry) + "\n") self.log.flush() if status == 200 and body: os.makedirs(os.path.dirname(dst), exist_ok=True) with open(dst, "wb") as fh: fh.write(body) return body return None def registry_entries(cache): """Every server entry from the saved pages, in page order.""" out = [] pages = sorted(f for f in os.listdir(os.path.join(cache, "registry")) if f.startswith("page_")) for f in pages: with open(os.path.join(cache, "registry", f), encoding="utf-8") as fh: out.extend(json.load(fh)["servers"]) return out def npm_targets(entries): """(server name, npm package, version) for latest, active entries with an npm package; first package wins.""" seen, out = set(), [] for e in entries: meta = e.get("_meta", {}).get("io.modelcontextprotocol.registry/official", {}) if not meta.get("isLatest") or meta.get("status", "active") != "active": continue s = e["server"] for p in s.get("packages") or []: if p.get("registryType") == "npm" and p.get("identifier") and p.get("version"): key = (p["identifier"], p["version"]) if key not in seen: seen.add(key) out.append((s["name"], p["identifier"], p["version"])) break return out def sample(targets, n): """The same n targets every time for the same registry pages.""" import random pool = sorted(targets) return pool if n <= 0 or n >= len(pool) else sorted(random.Random(SEED).sample(pool, n)) def safe(name): return re.sub(r"[^A-Za-z0-9._-]", "_", name) def cmd_fetch(a): os.makedirs(os.path.join(a.cache, "registry"), exist_ok=True) f = Fetcher(a.cache) if not (f.robots(REGISTRY) and f.robots(NPM)): return 3 cursor, n = None, 0 while True: q = {"limit": "100", "version": "latest"} if cursor: q["cursor"] = cursor body = f.get(f"{REGISTRY}/v0/servers?{urllib.parse.urlencode(q)}", f"registry/page_{n:04d}.json") if body is None: print("registry page failed; stopping") return 1 cursor = json.loads(body).get("metadata", {}).get("nextCursor") n += 1 if not cursor: break everything = npm_targets(registry_entries(a.cache)) targets = sample(everything, a.sample) print(f"{n} registry pages; {len(everything)} npm packages listed; reading a sample of {len(targets)}", flush=True) for i, (_, pkg, ver) in enumerate(targets): enc = urllib.parse.quote(pkg, safe="@") mbody = f.get(f"{NPM}/{enc}/{urllib.parse.quote(ver)}", f"npm/{safe(pkg)}@{safe(ver)}.json") if not mbody: continue tar = (json.loads(mbody).get("dist") or {}).get("tarball") if tar and tar.startswith(NPM + "/"): f.get(tar, f"npm/{safe(pkg)}@{safe(ver)}.tgz") if i % 50 == 0: print(f" {i}/{len(targets)}", flush=True) return 0 # ------------------------------------------------------------ tokenizer --- PUNCT_BEFORE_REGEX = set("(,=:[!&|?{};+-*%<>~^") KEYWORDS_BEFORE_REGEX = {"return", "typeof", "instanceof", "in", "of", "new", "delete", "void", "throw", "case", "do", "else", "yield", "await"} ESC = {"n": "\n", "t": "\t", "r": "\r", "b": "\b", "f": "\f", "v": "\v", "0": "\0"} def unescape(s): out, i = [], 0 while i < len(s): c = s[i] if c != "\\" or i + 1 >= len(s): out.append(c) i += 1 continue n = s[i + 1] if n in ESC: out.append(ESC[n]); i += 2 elif n == "u" and i + 2 < len(s) and s[i + 2] == "{": j = s.find("}", i) try: out.append(chr(int(s[i + 3:j], 16))) except ValueError: out.append(s[i:j + 1]) i = j + 1 elif n == "u": try: out.append(chr(int(s[i + 2:i + 6], 16))) except ValueError: out.append(s[i:i + 6]) i += 6 elif n == "x": try: out.append(chr(int(s[i + 2:i + 4], 16))) except ValueError: out.append(s[i:i + 4]) i += 4 elif n in "\r\n": i += 2 else: out.append(n); i += 2 return "".join(out) def tokenize(src): """Yield (kind, value) with kind in id, str, punct. Comments and regex literals are dropped.""" i, n, prev = 0, len(src), None toks = [] while i < n: c = src[i] if c in " \t\r\n": i += 1 continue if src.startswith("//", i): j = src.find("\n", i) i = n if j < 0 else j continue if src.startswith("/*", i): j = src.find("*/", i + 2) i = n if j < 0 else j + 2 continue if c in "\"'": j = i + 1 while j < n and src[j] != c and src[j] != "\n": j += 2 if src[j] == "\\" else 1 toks.append(("str", unescape(src[i + 1:j]))) i = j + 1 prev = toks[-1] continue if c == "`": j, depth, parts, start = i + 1, 0, [], i + 1 while j < n: if src[j] == "\\": j += 2 continue if src[j] == "`": break if src.startswith("${", j): parts.append(src[start:j]) k, d = j + 2, 1 while k < n and d: d += {"{": 1, "}": -1}.get(src[k], 0) k += 1 parts.append("${}") j = start = k continue j += 1 parts.append(src[start:j]) toks.append(("str", unescape("".join(parts)))) i = j + 1 prev = toks[-1] continue if c == "/": regex_ok = prev is None or (prev[0] == "punct" and prev[1] in PUNCT_BEFORE_REGEX) or (prev[0] == "id" and prev[1] in KEYWORDS_BEFORE_REGEX) if regex_ok: j, in_class = i + 1, False while j < n and src[j] != "\n": if src[j] == "\\": j += 2 continue if src[j] == "[": in_class = True elif src[j] == "]": in_class = False elif src[j] == "/" and not in_class: break j += 1 j += 1 while j < n and (src[j].isalnum()): j += 1 i = j prev = ("str", "") continue if c.isalnum() or c in "_$": j = i + 1 while j < n and (src[j].isalnum() or src[j] in "_$"): j += 1 toks.append(("id", src[i:j])) i = j prev = toks[-1] continue toks.append(("punct", c)) i += 1 prev = toks[-1] return toks # ------------------------------------------------------------ extraction --- def string_at(toks, k): """Literal string starting at k, joining 'a' + 'b' + `c`; returns (text or None, next index).""" if k >= len(toks) or toks[k][0] != "str": return None, k s, k = toks[k][1], k + 1 while k + 1 < len(toks) and toks[k] == ("punct", "+") and toks[k + 1][0] == "str": s += toks[k + 1][1] k += 2 return s, k def brackets(toks): """match[i] = index of the bracket closing toks[i]; parent[i] = index of the innermost open bracket around i.""" pairs = {")": "(", "}": "{", "]": "["} match, parent, stack = {}, [-1] * len(toks), [] for i, t in enumerate(toks): parent[i] = stack[-1] if stack else -1 if t[0] != "punct": continue if t[1] in "({[": stack.append(i) elif t[1] in pairs and stack and toks[stack[-1]][1] == pairs[t[1]]: match[stack.pop()] = i return match, parent def keyed_string(toks, lo, hi, key, top_only=True): """Value of `key: "literal"` inside toks[lo:hi] at bracket depth 1 (top_only) or anywhere.""" depth = 0 for j in range(lo, hi): t = toks[j] if t[0] == "punct" and t[1] in "({[": depth += 1 elif t[0] == "punct" and t[1] in ")}]": depth -= 1 elif (t == ("id", key) or t == ("str", key)) and j + 1 < hi and toks[j + 1] == ("punct", ":") and (depth == 1 or not top_only): s, _ = string_at(toks, j + 2) if s is not None: return s return None def param_descriptions(toks, lo, hi, skip=None): out = [] for j in range(lo, hi): if toks[j] == ("id", "describe") and j + 2 < hi and toks[j + 1] == ("punct", "("): s, _ = string_at(toks, j + 2) if s is not None: out.append(s) elif toks[j] in (("id", "description"), ("str", "description")) and j + 2 < hi and toks[j + 1] == ("punct", ":"): s, _ = string_at(toks, j + 2) if s is not None and s != skip: out.append(s) return out def make_tool(name, title, desc, params): t = {"name": name} if title is not None: t["title"] = title t["description"] = desc if desc is not None else "" t["inputSchema"] = {"type": "object", "properties": {f"p{i}": {"description": d} for i, d in enumerate(params)}} return t def extract(toks): tools = [] n = len(toks) match, parent = brackets(toks) done = set() for k in range(n - 3): t = toks[k] # server.registerTool("name", {...}, cb) / server.tool("name", "desc"?, {...}?, cb) if t[0] == "id" and t[1] in ("registerTool", "tool") and k > 0 and toks[k - 1] == ("punct", ".") and toks[k + 1] == ("punct", "("): name, j = string_at(toks, k + 2) if name is None or not re.fullmatch(r"[A-Za-z0-9_.\-/]{1,128}", name): continue end = match.get(k + 1, n - 1) if t[1] == "registerTool": if j + 1 < n and toks[j] == ("punct", ",") and toks[j + 1] == ("punct", "{"): oend = match.get(j + 1, n - 1) done.add(j + 1) desc = keyed_string(toks, j + 1, oend, "description") title = keyed_string(toks, j + 1, oend, "title") tools.append(make_tool(name, title, desc, param_descriptions(toks, j + 1, oend, skip=desc))) continue desc = None if j + 1 < n and toks[j] == ("punct", ","): desc, _ = string_at(toks, j + 1) if desc is None and not (j + 1 < n and toks[j] == ("punct", ",") and toks[j + 1][1] in ("{", "(")): continue # not an SDK .tool() call shape tools.append(make_tool(name, None, desc, param_descriptions(toks, j, end, skip=desc))) # { name: "x", description: "...", inputSchema: {...} } (the low-level tools/list handler) elif t in (("id", "inputSchema"), ("str", "inputSchema"), ("id", "input_schema")) and toks[k + 1] == ("punct", ":"): o = parent[k] if o < 0 or toks[o] != ("punct", "{") or o in done: continue done.add(o) end = match.get(o, n - 1) name = keyed_string(toks, o, end, "name") if name is None and o >= 2 and toks[o - 1] == ("punct", ":") and toks[o - 2][0] in ("id", "str"): name = toks[o - 2][1] # { tool_name: { description, inputSchema } } if name is None or not re.fullmatch(r"[A-Za-z0-9_.\-/]{1,128}", name): continue desc = keyed_string(toks, o, end, "description") title = keyed_string(toks, o, end, "title") tools.append(make_tool(name, title, desc, param_descriptions(toks, o + 1, end, skip=desc))) return tools def package_tools(tgz_bytes): """Statically extracted, de-duplicated tool definitions from one npm tarball (read in memory).""" tools, seen, files = [], set(), 0 with tarfile.open(fileobj=io.BytesIO(tgz_bytes), mode="r:gz") as tf: for m in tf.getmembers(): low = m.name.lower() if not m.isfile() or not low.endswith(SRC_EXT) or low.endswith((".d.ts", ".d.mts", ".d.cts")) or "/node_modules/" in low: continue if re.search(r"/(tests?|__tests__|examples?|fixtures?)/|\.(test|spec)\.[cm]?[jt]s$", low): continue if m.size > 8 * 1024 * 1024: continue files += 1 src = tf.extractfile(m).read().decode("utf-8", "replace") if "tool" not in src and "Tool" not in src and "inputSchema" not in src: continue for tool in extract(tokenize(src)): key = tool["name"] # first definition of each name wins (src/ and dist/ often both ship) if key not in seen: seen.add(key) tools.append(tool) return tools, files def cmd_scan(a): aud = load_auditor(a.auditor) entries = registry_entries(a.cache) meta = lambda e: e.get("_meta", {}).get("io.modelcontextprotocol.registry/official", {}) # noqa: E731 latest = [e for e in entries if meta(e).get("isLatest")] active = [e for e in latest if meta(e).get("status", "active") == "active"] everything = npm_targets(entries) targets = sample(everything, a.sample) kinds = collections.Counter() for e in active: s = e["server"] types = {p.get("registryType") for p in s.get("packages") or []} - {None} kinds.update(types) kinds["(remote endpoint only)" if s.get("remotes") and not types else "(no package, no remote)" if not types else "(has a package)"] += 1 sens_terms = collections.Counter() per_pkg, findings_by_code, pkgs_by_code, examples = [], collections.Counter(), collections.Counter(), collections.defaultdict(list) level_of = {} missing = no_tools = 0 for server, pkg, ver in targets: p = os.path.join(a.cache, "npm", f"{safe(pkg)}@{safe(ver)}.tgz") if not os.path.exists(p): missing += 1 continue with open(p, "rb") as fh: tools, files = package_tools(fh.read()) if not tools: no_tools += 1 per_pkg.append({"server": server, "package": pkg, "version": ver, "files": files, "tools": 0, "findings": {}}) continue au = aud.Audit() aud.audit_tools(au, f"{pkg}@{ver}", tools) if a.tool and a.tool[0] == pkg: for t in tools: if t["name"] == a.tool[1]: print(json.dumps(t, indent=1, ensure_ascii=False)) for lvl, code, where, msg in au.findings: if f"tools[{a.tool[1]}]" in where: print(f"{lvl.upper():<7}{code:<10}{where}\n {msg}") return 0 codes = collections.Counter(c for _, c, _, _ in au.findings) for lvl, code, where, msg in au.findings: level_of[code] = lvl if code == "SENSITIVE": sens_terms[re.search(r'"(.*)"$', msg).group(1).strip('"`').lower()] += 1 if len(examples[code]) < 200: examples[code].append((where, msg)) findings_by_code.update(codes) pkgs_by_code.update(set(codes)) strings = sum(1 for t in tools for _ in aud.tool_strings({k: v for k, v in t.items()}, "")) per_pkg.append({"server": server, "package": pkg, "version": ver, "files": files, "tools": len(tools), "strings": strings, "findings": dict(codes)}) scanned = [r for r in per_pkg if r["tools"]] print(f"registry entries (latest versions): {len(latest):,}; active {len(active):,}, deprecated {len(latest) - len(active):,}") print("active entries by what they publish (an entry can publish several kinds):") for k, v in sorted(kinds.items(), key=lambda kv: -kv[1]): print(f" {k:<26}{v:>7,}") print(f"npm packages listed: {len(everything):,}; sample: {len(targets)} (seed {SEED}); tarballs read: {len(targets) - missing}; " f"with tools extracted: {len(scanned)}; no tools extracted: {no_tools}") print(f"tools extracted: {sum(r['tools'] for r in scanned):,}; strings scanned: {sum(r['strings'] for r in scanned):,}") any_high = sum(1 for r in scanned if any(level_of.get(c) == 'high' for c in r['findings'])) any_f = sum(1 for r in scanned if r["findings"]) print(f"packages with any finding: {any_f} of {len(scanned)} ({100 * any_f / max(len(scanned), 1):.1f}%); " f"with a high finding: {any_high} ({100 * any_high / max(len(scanned), 1):.1f}%)") print(f"{'code':<10}{'level':<8}{'findings':>9}{'packages':>10}{'share':>8}") for code in sorted(findings_by_code, key=lambda c: (aud.LEVEL_ORDER[level_of[c]], -pkgs_by_code[c])): print(f"{code:<10}{level_of[code]:<8}{findings_by_code[code]:>9}{pkgs_by_code[code]:>10}{100 * pkgs_by_code[code] / len(scanned):>7.1f}%") if sens_terms: print("SENSITIVE findings by the path or term named: " + ", ".join(f"{k} {v}" for k, v in sens_terms.most_common())) for code in sorted(examples): for where, msg in examples[code][: a.show]: print(f" {code:<10}{where}\n {msg}") if a.json: with open(a.json, "w", encoding="utf-8") as fh: json.dump({"packages": per_pkg, "kinds": dict(kinds)}, fh, indent=1, ensure_ascii=False) return 0 def main(argv): ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) sub = ap.add_subparsers(dest="cmd", required=True) f = sub.add_parser("fetch") f.add_argument("--cache", required=True) f.add_argument("--sample", type=int, default=600, help="how many npm packages to read (0 = all)") s = sub.add_parser("scan") s.add_argument("--cache", required=True) s.add_argument("--sample", type=int, default=600) s.add_argument("--json") s.add_argument("--show", type=int, default=0, help="print up to N findings of each code") s.add_argument("--tool", nargs=2, metavar=("PACKAGE", "TOOL"), help="print one extracted definition and its findings") s.add_argument("--auditor", help="path of another copy of coding-agent-security-prompt-injection-mcp-secrets.py") a = ap.parse_args(argv) return cmd_fetch(a) if a.cmd == "fetch" else cmd_scan(a) if __name__ == "__main__": sys.stdout.reconfigure(encoding="utf-8") sys.exit(main(sys.argv[1:]))