1"""Detection rules. Every rule is a pure function over a parsed Document.
2
3Rules only *read* configuration. They never start servers, open connections,
4or send data anywhere.
5"""
6from __future__ import annotations
7
8import ipaddress
9import re
10import unicodedata
11from dataclasses import dataclass
12from typing import Any, Callable, Iterator
13from urllib.parse import urlparse
14
15from .model import Document, Finding, Server
16from .secrets import find_secret, redact
17
18
19@dataclass(frozen=True)
20class Rule:
21 id: str
22 name: str
23 title: str
24 severity: str
25 description: str
26 fix: str
27 check: Callable[[Document], Iterator[Finding]]
28 references: tuple[str, ...] = ()
29
30
31RULES: list[Rule] = []
32
33
34def rule(id: str, name: str, title: str, severity: str, description: str, fix: str, references: tuple[str, ...] = ()):
35 def deco(fn: Callable[[Document], Iterator[Finding]]):
36 RULES.append(Rule(id, name, title, severity, description, fix, fn, references))
37 return fn
38
39 return deco
40
41
42def _finding(r_id: str, doc: Document, path: str, message: str, severity: str | None = None,
43 evidence: str = "", fix: str = "", needle: str = "") -> Finding:
44 r = next(x for x in RULES if x.id == r_id)
45 return Finding(
46 rule_id=r.id,
47 title=r.title,
48 severity=severity or r.severity,
49 source=doc.source,
50 path=path,
51 message=message,
52 evidence=evidence,
53 fix=fix or r.fix,
54 line=find_line(doc.text, needle or path.split(".")[-1]),
55 )
56
57
58def find_line(text: str, needle: str) -> int:
59 if not needle:
60 return 0
61 for cand in (needle, needle.strip('"'), needle[:40]):
62 if not cand:
63 continue
64 idx = text.find(cand)
65 if idx == -1:
66
67 import json
68
69 enc = json.dumps(cand)[1:-1]
70 idx = text.find(enc)
71 if idx != -1:
72 return text.count("\n", 0, idx) + 1
73 return 0
74
75
76def _snippet(text: str, start: int, end: int, width: int = 50) -> str:
77 a, b = max(0, start - width), min(len(text), end + width)
78 s = text[a:b].replace("\n", " ")
79 return ("…" if a else "") + s + ("…" if b < len(text) else "")
80
81
82
83
84
85TEXT_KEYS = {"description", "instructions", "systemprompt", "system_prompt", "prompt", "title", "usage"}
86
87
88def _walk(data: Any, path: str = "$") -> Iterator[tuple[str, str, Any]]:
89 """Yield (path, key, value) for every leaf."""
90 if isinstance(data, dict):
91 for k, v in data.items():
92 p = f"{path}.{k}" if path != "$" else str(k)
93 if isinstance(v, (dict, list)):
94 yield from _walk(v, p)
95 else:
96 yield p, str(k), v
97 elif isinstance(data, list):
98 for i, v in enumerate(data):
99 p = f"{path}[{i}]"
100 if isinstance(v, (dict, list)):
101 yield from _walk(v, p)
102 else:
103 yield p, path.rsplit(".", 1)[-1], v
104
105
106def text_fields(doc: Document) -> list[tuple[str, str, str]]:
107 """(path, label, text) for tool descriptions, schema descriptions and other prose fields."""
108 out: list[tuple[str, str, str]] = []
109 seen: set[str] = set()
110 for t in doc.tools:
111 out.append((t.path + ".description", f"tool '{t.name}'", t.description))
112 seen.add(t.description)
113 for p, k, v in _walk(t.input_schema, t.path + ".inputSchema"):
114 if isinstance(v, str) and k.lower() in TEXT_KEYS | {"default", "enum"} and v not in seen:
115 out.append((p, f"tool '{t.name}' parameter", v))
116 seen.add(v)
117 for p, k, v in _walk(doc.data):
118 if isinstance(v, str) and k.lower() in TEXT_KEYS and v not in seen:
119 out.append((p, k, v))
120 seen.add(v)
121 return out
122
123
124
125
126
127INJECTION_PATTERNS: list[tuple[str, str, str]] = [
128
129 ("critical", r"\b(ignore|disregard|forget|override)\b.{0,30}\b(all\s+)?(previous|prior|above|earlier|system|other)\b.{0,20}\b(instructions?|prompts?|rules?|messages?)", "instruction override"),
130 ("critical", r"(忽略|无视|忘记|忽视|覆盖)(掉)?(之前|以上|上面|先前|此前|所有|其他|系统)的?(所有)?(指令|指示|提示词?|规则|要求|设定)", "instruction override (zh)"),
131 ("critical", r"\b(do\s*n[o']?t|never)\b.{0,25}\b(tell|inform|notify|mention|reveal|show|alert)\b.{0,60}\b(the\s+)?user", "concealment from user"),
132 ("critical", r"(不要|不得|切勿|别|无需|不必)(向|对|让)?(用户|使用者).{0,6}(透露|告知|告诉|提及|提示|知道|看到|展示)", "concealment from user (zh)"),
133 ("critical", r"(不要|不得|切勿|别)(告诉|告知|通知|提醒|提示)(用户|使用者)", "concealment from user (zh)"),
134 ("high", r"\bwithout\s+(the\s+)?user'?s?\s+(knowledge|consent|confirmation|approval|noticing)|\bwithout\s+(telling|asking|informing|notifying)\b", "act without user consent"),
135 ("high", r"(静默|悄悄|偷偷|秘密|暗中|后台)(地)?(执行|发送|上传|读取|调用|抄送|转发|运行)", "covert action (zh)"),
136 ("high", r"\b(silently|secretly|covertly|quietly)\b.{0,30}\b(send|upload|read|call|run|execute|forward|copy|include|exfiltrate)", "covert action"),
137 ("high", r"<\s*(important|system|instructions?|admin|secret|hidden)\s*>|\[\s*(system|admin|instructions?)\s*\]|<\|im_start\|>|\[INST\]", "fake system/priority tag"),
138 ("high", r"\bbefore\s+(using|calling|invoking|running)\s+(this|any|the\s+\w+)\s+tool\b.{0,60}\b(must|first|always|read|send|call)", "pre-condition hijack"),
139 ("high", r"(在|调用|使用)(本|此|任何|其他)?工具(之前|前).{0,20}(必须|务必|先|首先)", "pre-condition hijack (zh)"),
140 ("medium", r"\b(you\s+(must|should)\s+(always|first|now))\b|\bfrom\s+now\s+on\b|\byou\s+are\s+now\b|\bact\s+as\b", "behavioural override"),
141 ("medium", r"(从现在(开始|起)|你现在是|你必须(始终|总是|首先|先)|请扮演)", "behavioural override (zh)"),
142 ("medium", r"<!--.*?-->", "hidden HTML comment"),
143 ("medium", r"\b(sidenote|side_note|hidden_param)\b", "covert side-channel parameter"),
144]
145_INJ = [(sev, re.compile(rx, re.I | re.S), why) for sev, rx, why in INJECTION_PATTERNS]
146
147
148@rule(
149 "ASC001", "tool-poisoning", "Prompt injection / tool poisoning in tool or agent text", "critical",
150 "Tool descriptions and other prose fields are fed to the model verbatim. Imperative text aimed at the model "
151 "(override instructions, hide actions from the user, fake <IMPORTANT>/<system> tags) is the classic MCP tool-poisoning vector. "
152 "Patterns cover English and Chinese.",
153 "Do not install/enable this server until the description is reviewed. Descriptions should only describe what the tool does "
154 "for a human; remove imperative instructions to the model, hidden tags and comments. Pin the server version so a later update "
155 "cannot silently change the text (rug pull).",
156 ("https://invariantlabs.ai/blog/mcp-security-notification-tool-poisoning-attacks",
157 "https://genai.owasp.org/llmrisk/llm01-prompt-injection/"),
158)
159def check_injection(doc: Document) -> Iterator[Finding]:
160 for path, label, text in text_fields(doc):
161 hits = []
162 for sev, rx, why in _INJ:
163 m = rx.search(text)
164 if m:
165 hits.append((sev, why, m))
166 if not hits:
167 continue
168 hits.sort(key=lambda h: ["info", "low", "medium", "high", "critical"].index(h[0]), reverse=True)
169 sev, why, m = hits[0]
170 reasons = ", ".join(sorted({h[1] for h in hits}))
171 yield _finding("ASC001", doc, path, f"{label}: suspicious model-directed instructions ({reasons})",
172 severity=sev, evidence=_snippet(text, m.start(), m.end()), needle=m.group(0))
173
174
175
176
177
178_BIDI = set("\u202a\u202b\u202c\u202d\u202e\u2066\u2067\u2068\u2069\u200e\u200f")
179_ZW = set("\u200b\u200c\u200d\u2060\ufeff\u180e")
180
181
182def _hidden_chars(s: str) -> list[str]:
183 found = []
184 for ch in s:
185 o = ord(ch)
186 if ch in _BIDI:
187 found.append(f"U+{o:04X} bidi control")
188 elif ch in _ZW:
189 found.append(f"U+{o:04X} zero-width")
190 elif 0xE0000 <= o <= 0xE007F:
191 found.append(f"U+{o:05X} tag character (ASCII smuggling)")
192 elif 0xE000 <= o <= 0xF8FF:
193 found.append(f"U+{o:04X} private use")
194 elif unicodedata.category(ch) == "Cf" and ch not in "\u00ad":
195 found.append(f"U+{o:04X} format char")
196 return found
197
198
199@rule(
200 "ASC002", "hidden-unicode", "Invisible / bidirectional Unicode characters", "high",
201 "Zero-width, bidi-override and Unicode 'tag' characters render invisibly to humans but are read by the model. "
202 "They are used to smuggle instructions past reviewers.",
203 "Strip non-printing characters (e.g. normalise with a whitelist) and re-review the visible text. Treat the source as untrusted.",
204 ("https://embracethered.com/blog/posts/2024/hiding-and-finding-text-with-unicode-tags/",),
205)
206def check_hidden_unicode(doc: Document) -> Iterator[Finding]:
207 for path, key, v in list(_walk(doc.data)):
208 if isinstance(v, str):
209 hc = _hidden_chars(v)
210 if hc:
211 smuggled = "".join(chr(ord(c) - 0xE0000) for c in v if 0xE0020 <= ord(c) <= 0xE007E)
212 ev = f"{len(hc)} hidden char(s): {', '.join(sorted(set(hc))[:4])}"
213 if smuggled:
214 ev += f"; decoded tag text: {smuggled[:80]!r}"
215 yield _finding("ASC002", doc, path, f"field '{key}' contains invisible Unicode", evidence=ev,
216 severity="critical" if smuggled else "high", needle=key)
217
218
219
220
221
222SENSITIVE_PATHS = re.compile(
223 r"(~|\$HOME|%USERPROFILE%)?/?\.(ssh|aws|gnupg|kube|docker)/?[\w\-/.]*|id_(rsa|ed25519|ecdsa)\b|\.env\b|/etc/(passwd|shadow)|"
224 r"\.npmrc|\.pypirc|\.git-credentials|\.netrc|credentials\.json|mcp\.json|claude_desktop_config\.json|"
225 r"Login Data|Cookies\.sqlite|keychain|wallet\.dat|私钥|助记词",
226 re.I,
227)
228EXFIL = re.compile(
229 r"\b(send|upload|post|forward|transmit|exfiltrate|copy|cc|bcc|report)\b.{0,40}(https?://|\b[\w.+-]+@[\w-]+\.[\w.]+|webhook)|"
230 r"(发送|上传|抄送|转发|回传|同步).{0,12}(到|至|给).{0,30}(https?://|[\w.+-]+@[\w-]+\.[\w.]+|服务器|邮箱)",
231 re.I,
232)
233
234
235@rule(
236 "ASC003", "sensitive-data-instruction", "Tool text references credentials or outbound exfiltration", "high",
237 "A tool description that mentions SSH keys, cloud credentials, other agents' config files, or tells the model to "
238 "send content to a URL / e-mail address is a strong indicator of a data-exfiltration attempt.",
239 "Remove the server unless the reference is clearly legitimate (e.g. an SSH-management tool). Never let a tool's text decide "
240 "which local files the agent reads; restrict the server's filesystem access and outbound network instead.",
241)
242def check_sensitive_text(doc: Document) -> Iterator[Finding]:
243 for path, label, text in text_fields(doc):
244 m = SENSITIVE_PATHS.search(text)
245 e = EXFIL.search(text)
246 if m and e:
247 yield _finding("ASC003", doc, path, f"{label}: references sensitive files AND an outbound destination",
248 severity="critical", evidence=_snippet(text, m.start(), m.end()) + " | " + _snippet(text, e.start(), e.end(), 20),
249 needle=m.group(0))
250 elif e:
251 yield _finding("ASC003", doc, path, f"{label}: instructs sending data to an external destination",
252 severity="high", evidence=_snippet(text, e.start(), e.end()), needle=e.group(0))
253 elif m:
254 yield _finding("ASC003", doc, path, f"{label}: references sensitive local files ({m.group(0)})",
255 severity="medium", evidence=_snippet(text, m.start(), m.end()), needle=m.group(0))
256
257
258
259
260
261@rule(
262 "ASC004", "tool-shadowing", "Tool shadowing / cross-tool manipulation", "high",
263 "A tool whose description talks about how *other* tools must behave, or two servers exposing the same tool name, "
264 "lets one server hijack calls meant for another (cross-origin escalation).",
265 "Remove the offending server or rename/namespace the colliding tools. Tools should only describe themselves.",
266 ("https://invariantlabs.ai/blog/mcp-security-notification-tool-poisoning-attacks",),
267)
268def check_shadowing(doc: Document) -> Iterator[Finding]:
269 names: dict[str, list] = {}
270 for t in doc.tools:
271 names.setdefault(t.name.lower(), []).append(t)
272 for n, ts in names.items():
273 servers = {t.server for t in ts}
274 if len(ts) > 1 and len(servers) > 1:
275 yield _finding("ASC004", doc, ts[1].path, f"tool name '{n}' is exposed by multiple servers: {sorted(servers)}",
276 severity="medium", needle=ts[1].name)
277 generic = re.compile(r"\b(when|whenever|if)\b.{0,30}\b(using|calling|invoking)\b.{0,20}\b(the\s+)?[`'\"]?(\w+)[`'\"]?\s+tool\b|"
278 r"\binstead\s+of\s+(the\s+)?[`'\"]?\w+[`'\"]?\s+tool\b|(调用|使用)\s*[`'\"]?\w+[`'\"]?\s*工具时", re.I)
279 for t in doc.tools:
280 others = [o.name for o in doc.tools if o.name != t.name and o.server != t.server and len(o.name) > 3]
281 hit = next((o for o in others if re.search(rf"\b{re.escape(o)}\b", t.description)), None)
282 m = generic.search(t.description)
283 if hit or m:
284 ev = f"mentions tool '{hit}' of another server" if hit else _snippet(t.description, m.start(), m.end())
285 yield _finding("ASC004", doc, t.path + ".description", f"tool '{t.name}' tries to influence other tools",
286 evidence=ev, needle=hit or (m.group(0) if m else t.name))
287
288
289
290
291
292@rule(
293 "ASC010", "hardcoded-secret", "Hard-coded secret / credential in config", "high",
294 "Agent and MCP config files are frequently committed, synced, shared in screenshots, or readable by every installed MCP "
295 "server and by the agent itself (and therefore by any prompt injection).",
296 "Rotate the credential now, then reference it from the environment or a secret manager instead of a literal, e.g. "
297 "VS Code: \"${input:token}\" / \"${env:TOKEN}\"; Claude/Cursor: set it in your shell profile or OS keychain and keep only the "
298 "variable name in the file. Add the file to .gitignore.",
299)
300def check_secrets(doc: Document) -> Iterator[Finding]:
301 for path, key, v in _walk(doc.data):
302 if not isinstance(v, str):
303 continue
304 hit = find_secret(v, key)
305 if hit:
306 sev = "critical" if hit.label.startswith(("Private key", "AWS", "Stripe", "Alibaba", "Tencent", "Volcengine")) else "high"
307 yield _finding("ASC010", doc, path, (hit.label if hit.label.startswith("High-entropy") else f"{hit.label} stored as a literal in '{key}'"), severity=sev,
308 evidence=redact(hit.match), needle=hit.match[:12])
309
310
311
312
313
314DANGEROUS_CMD: list[tuple[str, str, str]] = [
315 ("critical", r"\b(curl|wget|iwr|irm|Invoke-WebRequest|Invoke-RestMethod)\b[^|;&]*\|\s*(sudo\s+)?(sh|bash|zsh|python3?|node|iex|powershell|pwsh)\b", "downloads and pipes remote code into an interpreter"),
316 ("critical", r"\b(iex|Invoke-Expression)\b|FromBase64String|\s-(e|enc|EncodedCommand)\s+[A-Za-z0-9+/=]{16,}", "PowerShell expression / encoded command"),
317 ("high", r"base64\s+(-d|--decode)|\beval\s|\bexec\(", "decodes or evals dynamic code"),
318 ("high", r"\bsudo\b|\brunas\b|\bdoas\b", "runs with elevated privileges"),
319 ("critical", r"--privileged\b|/var/run/docker\.sock|--pid[= ]host|--cap-add[= ](SYS_ADMIN|ALL)", "container escape-equivalent docker flags"),
320 ("high", r"(-v|--volume|--mount)[= ]\s*(type=bind,\s*source=)?(/|~|\$HOME|/home|/Users|C:\\\\?)(:|,|\s|$)", "mounts host root/home into the container"),
321 ("medium", r"--net(work)?[= ]host\b", "shares host network namespace"),
322 ("high", r"\brm\s+-[a-z]*r[a-z]*f|\bmkfs\b|\bdd\s+if=", "destructive filesystem command"),
323 ("medium", r"\b(nc|ncat|netcat)\b.*\s-e\s|/dev/tcp/", "raw socket / reverse-shell style construct"),
324]
325_DC = [(s, re.compile(rx, re.I), w) for s, rx, w in DANGEROUS_CMD]
326SHELLS = {"sh", "bash", "zsh", "dash", "fish", "cmd", "cmd.exe", "powershell", "powershell.exe", "pwsh"}
327
328
329@rule(
330 "ASC020", "dangerous-command", "Unsafe command used to launch an MCP server", "high",
331 "stdio MCP servers run as local processes with your user's privileges. Shell one-liners that download and execute code, "
332 "elevated privileges, or privileged containers turn any compromise of the package or URL into full machine compromise.",
333 "Install the server from a pinned, reviewed package and call its binary directly (no `sh -c`, no `curl | sh`, no sudo). "
334 "If you containerise it, drop --privileged/docker.sock and mount only the project directory read-only.",
335)
336def check_dangerous_command(doc: Document) -> Iterator[Finding]:
337 for s in doc.servers:
338 line = s.cmdline
339 if not line:
340 continue
341 reported = False
342 for sev, rx, why in _DC:
343 m = rx.search(line)
344 if m:
345 reported = True
346 yield _finding("ASC020", doc, s.path, f"server '{s.name}' {why}", severity=sev,
347 evidence=_snippet(line, m.start(), m.end(), 40), needle=s.name)
348 base = s.command.replace("\\", "/").rsplit("/", 1)[-1].lower()
349 if not reported and base in SHELLS and any(a.lower() in ("-c", "/c", "-command", "/k") for a in s.args):
350 yield _finding("ASC020", doc, s.path, f"server '{s.name}' is launched through a shell wrapper ({base} -c)",
351 severity="medium", evidence=line[:160], needle=s.name,
352 fix="Call the server executable directly with an argument list; shell wrappers hide what actually runs and enable injection via env/args.")
353
354
355
356
357
358def _first_pkg(args: list[str], skip_values: set[str]) -> str | None:
359 it = iter(range(len(args)))
360 for i in it:
361 a = args[i]
362 if a in skip_values:
363 next(it, None)
364 continue
365 if a.startswith("-"):
366 continue
367 return a
368 return None
369
370
371@rule(
372 "ASC021", "unpinned-package", "MCP server package/image is not version-pinned", "medium",
373 "`npx -y pkg`, `uvx pkg`, `pkg@latest` or `image:latest` fetch whatever was published most recently at every launch. "
374 "A compromised or malicious update (a 'rug pull') runs automatically with no review.",
375 "Pin an exact version (npx -y pkg@1.2.3, uvx pkg==1.2.3, docker image@sha256:...) and review changes before bumping.",
376)
377def check_unpinned(doc: Document) -> Iterator[Finding]:
378 for s in doc.servers:
379 base = s.command.replace("\\", "/").rsplit("/", 1)[-1].lower().removesuffix(".cmd").removesuffix(".exe")
380 args = s.args
381 pkg = None
382 pinned = True
383 if base in ("npx", "bunx", "pnpx") or (base in ("pnpm", "yarn", "npm") and args[:1] in (["dlx"], ["exec"])):
384 pkg = _first_pkg([a for a in args if a not in ("dlx", "exec")], {"-p", "--package"})
385 if pkg:
386 name_part = pkg[1:] if pkg.startswith("@") else pkg
387 pinned = "@" in name_part and not name_part.endswith(("@latest", "@next", "@*"))
388 elif base in ("uvx", "pipx") or (base == "uv" and args[:2] in (["tool", "run"],)):
389 a2 = args[2:] if base == "uv" else (args[1:] if base == "pipx" and args[:1] == ["run"] else args)
390 pkg = _first_pkg(a2, {"--from", "--python", "--with", "-p"})
391 if "--from" in a2:
392 idx = a2.index("--from")
393 pkg = a2[idx + 1] if idx + 1 < len(a2) else pkg
394 if pkg:
395 pinned = bool(re.search(r"(==|@)[\w.]+", pkg)) and not pkg.endswith("@latest")
396 elif base in ("docker", "podman") and args[:1] == ["run"]:
397 pkg = _first_pkg(args[1:], {"-e", "--env", "-v", "--volume", "--name", "--network", "--net", "-p", "--mount", "-w", "--env-file", "-u", "--user"})
398 if pkg:
399 pinned = "@sha256:" in pkg or (":" in pkg.rsplit("/", 1)[-1] and not pkg.endswith(":latest"))
400 if pkg and not pinned:
401 yield _finding("ASC021", doc, s.path, f"server '{s.name}' runs unpinned '{pkg}' via {base}",
402 evidence=s.cmdline[:160], needle=pkg)
403 for a in args:
404 if re.match(r"^(git\+)?https://github\.com/[^@#]+(\.git)?$", a):
405 yield _finding("ASC021", doc, s.path, f"server '{s.name}' installs from a git URL without a pinned commit",
406 severity="medium", evidence=a, needle=a)
407
408
409
410
411
412@rule(
413 "ASC022", "tls-disabled", "TLS certificate verification disabled", "medium",
414 "Disabling certificate checks lets anyone on the network path impersonate the MCP endpoint or package registry and "
415 "inject tool descriptions or responses.",
416 "Remove the flag/variable; if you use a corporate proxy, add its CA to the trust store (NODE_EXTRA_CA_CERTS / SSL_CERT_FILE) instead.",
417)
418def check_tls(doc: Document) -> Iterator[Finding]:
419 bad_env = {"NODE_TLS_REJECT_UNAUTHORIZED": "0", "PYTHONHTTPSVERIFY": "0", "GIT_SSL_NO_VERIFY": "1", "CURL_INSECURE": "1"}
420 for s in doc.servers:
421 for k, v in s.env.items():
422 if k.upper() in bad_env and v.strip().lower() in (bad_env[k.upper()], "true" if bad_env[k.upper()] == "1" else "false"):
423 yield _finding("ASC022", doc, f"{s.path}.env.{k}", f"server '{s.name}' sets {k}={v}", evidence=f"{k}={v}", needle=k)
424 for a in s.args:
425 if a in ("--insecure", "-k", "--no-verify-ssl", "--skip-ssl-verify", "--no-check-certificate", "--ssl-no-verify"):
426 yield _finding("ASC022", doc, s.path, f"server '{s.name}' passes {a}", evidence=a, needle=a)
427
428
429
430
431
432BROAD_PATHS = {"/": "critical", "/*": "critical", "c:\\": "critical", "c:/": "critical", "c:": "critical",
433 "~": "high", "~/": "high", "$home": "high", "${home}": "high", "%userprofile%": "high",
434 "/home": "high", "/users": "high", "/etc": "high", "/root": "high", "c:\\users": "high"}
435
436
437@rule(
438 "ASC030", "overbroad-filesystem", "File-system access scope is too broad", "high",
439 "Granting a filesystem/terminal server your whole disk or home directory gives every prompt injection access to SSH keys, "
440 "browser profiles, cloud credentials and other agents' configs.",
441 "Scope the server to the specific project directories it needs (preferably read-only), e.g. "
442 "`npx -y @modelcontextprotocol/server-filesystem@<ver> /path/to/project`.",
443)
444def check_fs_scope(doc: Document) -> Iterator[Finding]:
445 for s in doc.servers:
446 for a in s.args:
447 norm = a.strip().rstrip("/").lower() or "/"
448 if a.strip() in ("/", "/*"):
449 norm = a.strip()
450 if norm in BROAD_PATHS or re.fullmatch(r"(/home/[^/]+|/users/[^/]+|c:\\users\\[^\\]+)", norm):
451 sev = BROAD_PATHS.get(norm, "high")
452 yield _finding("ASC030", doc, s.path, f"server '{s.name}' is given access to '{a}'", severity=sev,
453 evidence=s.cmdline[:160], needle=s.name)
454 if any(k in s.raw for k in ("roots", "allowedDirectories")):
455 for r in s.raw.get("roots") or s.raw.get("allowedDirectories") or []:
456 rr = str(r.get("uri", r) if isinstance(r, dict) else r).replace("file://", "")
457 if rr.rstrip("/").lower() in ("", "~") or rr == "/":
458 yield _finding("ASC030", doc, s.path, f"server '{s.name}' root '{rr}' is too broad", severity="critical", needle=s.name)
459
460
461
462
463
464EXEC_TOOL = re.compile(r"(exec|shell|bash|terminal|command|run|write|delete|remove|rm|edit|push|deploy|eval|kill|sql|query)", re.I)
465
466
467@rule(
468 "ASC031", "auto-approve", "Human approval or sandbox disabled for agent tool calls", "high",
469 "Auto-approving tool calls (YOLO / bypassPermissions / alwaysAllow / approval_policy=never) removes the last human check "
470 "between a prompt injection and a real side effect (shell commands, file writes, outbound requests).",
471 "Keep confirmation on for exec/write/network tools. Allow-list only narrow, read-only commands (e.g. `Bash(git status)` rather "
472 "than `Bash(*)`), keep the sandbox on (Codex: sandbox_mode = \"workspace-write\"), and never trust a whole MCP server.",
473)
474def check_posture(doc: Document) -> Iterator[Finding]:
475 data = doc.data if isinstance(doc.data, dict) else {}
476
477 perms = data.get("permissions") if isinstance(data.get("permissions"), dict) else None
478 if perms:
479 if perms.get("defaultMode") == "bypassPermissions":
480 yield _finding("ASC031", doc, "permissions.defaultMode", "Claude Code permission prompts are bypassed (bypassPermissions)",
481 severity="critical", evidence="defaultMode: bypassPermissions", needle="bypassPermissions")
482 for i, a in enumerate(perms.get("allow") or []):
483 a = str(a)
484 if re.fullmatch(r"Bash(\((\*|\*:\*|:\*)\))?", a) or a in ("*", "mcp__*"):
485 yield _finding("ASC031", doc, f"permissions.allow[{i}]", f"unrestricted allow rule '{a}'",
486 severity="high", evidence=a, needle=a)
487 elif re.fullmatch(r"Bash\((curl|wget|python3?|node|sh|bash|npx|rm|sudo)[ :]?\*?\)", a):
488 yield _finding("ASC031", doc, f"permissions.allow[{i}]", f"allow rule '{a}' permits arbitrary code/network",
489 severity="medium", evidence=a, needle=a)
490 if data.get("enableAllProjectMcpServers") is True:
491 yield _finding("ASC031", doc, "enableAllProjectMcpServers",
492 "every MCP server declared in any opened repo's .mcp.json is auto-enabled", severity="medium",
493 needle="enableAllProjectMcpServers")
494
495 scopes = [("", data)] + [(f"profiles.{k}.", v) for k, v in (data.get("profiles") or {}).items() if isinstance(v, dict)]
496 for prefix, sc in scopes:
497 ap, sb = sc.get("approval_policy"), sc.get("sandbox_mode")
498 if ap == "never" and sb == "danger-full-access":
499 yield _finding("ASC031", doc, f"{prefix}approval_policy", "Codex runs with no approvals AND no sandbox",
500 severity="critical", evidence="approval_policy=never, sandbox_mode=danger-full-access", needle="danger-full-access")
501 elif sb == "danger-full-access":
502 yield _finding("ASC031", doc, f"{prefix}sandbox_mode", "Codex sandbox disabled (danger-full-access)",
503 severity="high", evidence="sandbox_mode=danger-full-access", needle="danger-full-access")
504 elif ap == "never":
505 yield _finding("ASC031", doc, f"{prefix}approval_policy", "Codex never asks for approval", severity="medium",
506 evidence="approval_policy=never", needle="approval_policy")
507
508 for path, key, v in _walk(data):
509 k = key.lower()
510 if v is True and (k in ("yolo", "autoaccept", "dangerouslyskippermissions", "chat.tools.autoapprove",
511 "chat.tools.global.autoapprove", "autoapproveall", "alwaysallowall") or k.endswith("autoapprove")):
512 yield _finding("ASC031", doc, path, f"auto-approval switch '{key}' is enabled", severity="high", evidence=f"{key}: true", needle=key)
513
514 for s in doc.servers:
515 if s.trust:
516 yield _finding("ASC031", doc, f"{s.path}.trust", f"server '{s.name}' is fully trusted (all confirmations skipped)",
517 severity="high", evidence="trust: true", needle="trust")
518 if s.auto_approve:
519 risky = [t for t in s.auto_approve if t == "*" or EXEC_TOOL.search(t)]
520 yield _finding("ASC031", doc, f"{s.path}.alwaysAllow", f"server '{s.name}' auto-approves tools: {s.auto_approve[:6]}",
521 severity="high" if risky else "low", evidence=f"risky: {risky}" if risky else "",
522 needle=s.auto_approve[0])
523
524
525
526
527
528TUNNEL_HOSTS = re.compile(
529 r"(ngrok(-free)?\.(io|app|dev)|trycloudflare\.com|loca\.lt|localtunnel\.me|serveo\.net|localhost\.run|pinggy\.(io|link)|"
530 r"webhook\.site|requestbin|pipedream\.net|interact\.sh|oast\.(fun|me|live|site|online|pro)|burpcollaborator\.net|"
531 r"beeceptor\.com|hookbin\.com|requestcatcher\.com|ptpb|transfer\.sh|pastebin\.com)$",
532 re.I,
533)
534
535
536_LOCAL_NETS = ("10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "fc00::/7")
537
538
539def _urls(s: Server) -> list[tuple[str, str]]:
540 out = [("url", s.url)] if s.url else []
541 for a in s.args:
542 for m in re.finditer(r"\b(?:https?|wss?)://[^\s\"'<>]+", a):
543 out.append(("args", m.group(0)))
544 return out
545
546
547@rule(
548 "ASC040", "suspicious-endpoint", "Insecure or suspicious remote MCP endpoint", "high",
549 "Remote MCP servers receive your prompts/tool arguments and control the tool descriptions your agent reads. Plain HTTP, raw "
550 "IPs, tunnel/request-catcher domains and punycode hosts are typical of throw-away or attacker-controlled endpoints.",
551 "Use HTTPS endpoints operated by the vendor you intend to trust, authenticate with OAuth / scoped tokens, and avoid tunnel or "
552 "request-bin domains in shared configs.",
553)
554def check_endpoints(doc: Document) -> Iterator[Finding]:
555 for s in doc.servers:
556 for where, u in _urls(s):
557 try:
558 p = urlparse(u)
559 except ValueError:
560 continue
561 host = (p.hostname or "").lower()
562 if not host:
563 continue
564 local = host in ("localhost", "127.0.0.1", "::1", "0.0.0.0") or host.endswith(".localhost")
565 ip = None
566 try:
567 ip = ipaddress.ip_address(host)
568 except ValueError:
569 pass
570 if ip and (ip.is_loopback or any(ip in ipaddress.ip_network(n) for n in _LOCAL_NETS if ip.version == ipaddress.ip_network(n).version)):
571 local = True
572 if ip and str(ip) in ("169.254.169.254", "100.100.100.200") :
573 yield _finding("ASC040", doc, s.path, f"server '{s.name}' points at a cloud metadata address {host}",
574 severity="critical", evidence=u, needle=u[:30])
575 continue
576 if TUNNEL_HOSTS.search(host):
577 yield _finding("ASC040", doc, s.path, f"server '{s.name}' uses a tunnel/request-capture host '{host}'",
578 severity="high", evidence=u, needle=host)
579 if p.scheme in ("http", "ws") and not local:
580 yield _finding("ASC040", doc, s.path, f"server '{s.name}' uses unencrypted {p.scheme}:// to a non-local host",
581 severity="high", evidence=u, needle=host)
582 elif ip and not local:
583 yield _finding("ASC040", doc, s.path, f"server '{s.name}' uses a raw public IP address {host}",
584 severity="medium", evidence=u, needle=host)
585 if "xn--" in host:
586 yield _finding("ASC040", doc, s.path, f"server '{s.name}' uses a punycode (possibly look-alike) host '{host}'",
587 severity="medium", evidence=u, needle=host)
588
589
590
591
592
593@rule(
594 "ASC041", "exposed-bind", "Local MCP server exposed on all network interfaces", "medium",
595 "A local server bound to 0.0.0.0 / :: is reachable from the LAN (or the internet on a cloud VM). Many MCP servers have no auth.",
596 "Bind to 127.0.0.1 and, for remote use, put the server behind authentication (OAuth / token) and TLS.",
597)
598def check_bind(doc: Document) -> Iterator[Finding]:
599 for s in doc.servers:
600 line = s.cmdline + " " + " ".join(f"{k}={v}" for k, v in s.env.items())
601 m = re.search(r"(--host[= ]|--bind[= ]|HOST=|-H\s)(0\.0\.0\.0|::|\[::\])\b|-p\s+0\.0\.0\.0:", line)
602 if m:
603 yield _finding("ASC041", doc, s.path, f"server '{s.name}' binds to all interfaces", evidence=m.group(0), needle="0.0.0.0")
604
605
606
607
608
609UNTRUSTED = re.compile(r"(fetch|browser|puppeteer|playwright|web|search|crawl|scrape|firecrawl|tavily|brave|email|gmail|mail|slack|"
610 r"discord|telegram|github|gitlab|jira|notion|rss|youtube|reddit|twitter|(?<![a-z])x-mcp|wechat|feishu|lark|dingtalk)", re.I)
611POWERFUL = re.compile(r"(shell|terminal|exec|command|desktop-commander|filesystem|server-filesystem|docker|kubernetes|k8s|ssh|"
612 r"postgres|mysql|sqlite|database|aws|aliyun|gcloud|azure|code-runner|python-repl|computer-use)", re.I)
613
614
615@rule(
616 "ASC050", "toxic-combination", "Untrusted-input + powerful-action servers in the same agent", "medium",
617 "When one agent can both read attacker-controllable content (web, e-mail, issues, chat) and run commands / touch files / "
618 "databases, a single injected document can drive the powerful tool. Risk is higher when calls are auto-approved.",
619 "Split these capabilities into separate agent profiles/workspaces, keep confirmation on for the powerful tools, and scope "
620 "credentials to read-only where possible.",
621 ("https://simonwillison.net/2025/Jun/16/the-lethal-trifecta/",),
622)
623def check_toxic(doc: Document) -> Iterator[Finding]:
624 active = [s for s in doc.servers if not s.disabled]
625 def ident(s: Server) -> str:
626 return " ".join([s.name, *s.args, s.url])
627 src = [s for s in active if UNTRUSTED.search(ident(s))]
628 pwr = [s for s in active if POWERFUL.search(ident(s))]
629 pairs = [(a, b) for a in src for b in pwr if a is not b]
630 if not pairs:
631 return
632 auto = any(s.trust or s.auto_approve for s in active) or any(
633 f.rule_id == "ASC031" for f in check_posture(doc))
634 a, b = pairs[0]
635 yield _finding("ASC050", doc, b.path,
636 f"'{a.name}' ingests untrusted content and '{b.name}' can act on the system" + (" — and tool calls are auto-approved" if auto else ""),
637 severity="high" if auto else "low",
638 evidence=f"untrusted: {sorted({x.name for x in src})}; powerful: {sorted({x.name for x in pwr})}", needle=b.name)
639
640
641def run_rules(doc: Document, only: set[str] | None = None, ignore: set[str] | None = None) -> list[Finding]:
642 out: list[Finding] = []
643 for r in RULES:
644 if only and r.id not in only and r.name not in only:
645 continue
646 if ignore and (r.id in ignore or r.name in ignore):
647 continue
648 out.extend(r.check(doc))
649
650 seen, uniq = set(), []
651 for f in out:
652 key = (f.rule_id, f.source, f.path, f.message)
653 if key not in seen:
654 seen.add(key)
655 uniq.append(f)
656 return uniq
