Add MADBugs-style RCE evidence checks

Author: Hinotoi-agent

README.md

@@ -14,6 +14,7 @@ It is intentionally **not** a full static analyzer. It answers the practical pre
 - The report has a PoC/repro indicator.
 - Dangerous capability terms exist in the repo.
 - Agent/tool context is detected so intended functionality is not mislabeled as RCE.
+- Critical RCE reports include MADBugs-style proof context: affected/tested version, attack surface, root cause, exploit chain, safe PoC result/cleanup, and fix guidance.
 - Optional simple GitHub issue/PR duplicate search via `gh`.
 
 ## Install locally
@@ -66,6 +67,8 @@ Every report should answer:
 5. What asset or user is harmed?
 6. What proof shows this works on current HEAD?
 
+For Critical/High RCE claims, `finding-vetter` also looks for the compact evidence pattern that showed up repeatedly in MADBugs writeups: affected/tested version, default attack surface, root-cause code path, attacker-input-to-impact chain, safe PoC side effect with cleanup, and concise fix guidance.
+
 ## Agent/tool API rule
 
 For agent and AI repos, command execution or file access may be intended functionality.

examples/strong-agent-rce.md

@@ -2,6 +2,10 @@
 
 Severity: Critical
 
+Affected: sample-agent-repo current HEAD.
+Tested on: bundled example repo in default configuration.
+Attack surface: HTTP API exposed by the server routes.
+
 A remote unauthenticated attacker can POST to `/api/tools/run` in the default server configuration.
 
 Affected files: `server/tools.py`, `server/routes.py`
@@ -14,12 +18,19 @@ Boundary crossed: unauthenticated HTTP caller can invoke an agent-only command t
 
 Trusted component: `run_command()` passes request JSON into subprocess.
 
+Root cause: `run_tool_route()` forwards attacker-controlled JSON into `run_command()`, which calls `subprocess.run(..., shell=True)` without authentication, authorization, or approval.
+
+Exploit chain: HTTP request -> JSON `cmd` -> `run_tool_route()` -> `run_command()` -> shell.
+
 Impact: host command execution as the server user.
 
 PoC:
 
 ```bash
 curl -s http://127.0.0.1:8000/api/tools/run -d '{"cmd":"id > /tmp/finding-vetter-poc"}'
+test -f /tmp/finding-vetter-poc && rm /tmp/finding-vetter-poc
 ```
 
-Proof: `/tmp/finding-vetter-poc` is created and then removed.
+Expected result: `/tmp/finding-vetter-poc` is created and then removed, demonstrating a safe side effect.
+
+Fix: require authentication and explicit user approval before invoking the tool, and replace shell execution with an allowlisted argv runner using `shell=False`.

src/finding_vetter/core.py

@@ -37,6 +37,12 @@ class ParsedReport:
     has_repro: bool = False
     attacker_mentions: list[str] = field(default_factory=list)
     impact_terms: list[str] = field(default_factory=list)
+    has_affected_or_tested_version: bool = False
+    has_attack_surface: bool = False
+    has_root_cause: bool = False
+    has_exploit_chain: bool = False
+    has_safe_side_effect: bool = False
+    has_fix_guidance: bool = False
 
 
 @dataclass
@@ -85,6 +91,30 @@ def parse_report(path: Path) -> ParsedReport:
     report.attacker_mentions = sorted(set(re.findall(r"(?i)\b(unauthenticated|remote|authenticated|low-privilege|workspace member|malicious website|prompt injection|admin|operator|local user)\b", text)))
     lower = text.lower()
     report.impact_terms = [t for t in RCE_TERMS + FILE_TERMS + SSRF_TERMS + AUTH_TERMS if t in lower]
+    report.has_affected_or_tested_version = bool(re.search(
+        r"(?im)^\s*(?:affected(?!\s+files?)|tested on|confirmed against|version|commit|current head|default configuration)\b",
+        text,
+    ))
+    report.has_attack_surface = bool(re.search(
+        r"(?i)\b(attack surface|entrypoint|entry point|reachable|default configuration|port \d+|victim action)\b",
+        text,
+    ))
+    report.has_root_cause = bool(re.search(
+        r"(?i)\b(root cause|bug is|vulnerability is|without checking|without auth|bounds check|validation|sanitize|source-to-sink|copies attacker-controlled)\b",
+        text,
+    ))
+    report.has_exploit_chain = bool(re.search(
+        r"(?i)\b(exploit chain|trigger|primitive|source[- ]to[- ]sink|->|leads to|reaches the sink)\b",
+        text,
+    ))
+    report.has_safe_side_effect = bool(re.search(
+        r"(?i)\b(expected result|actual result|safe side effect|created and then removed|cleanup|rm /tmp|test -f|id > /tmp|touch /tmp)\b",
+        text,
+    ))
+    report.has_fix_guidance = bool(re.search(
+        r"(?i)\b(fix|patch|mitigation|remediation|allowlist|denylist|bounds check|validate|sanitize|shell=false|authentication)\b",
+        text,
+    ))
     return report
 
 
@@ -161,9 +191,59 @@ def collect_repo_evidence(repo: Path, report: ParsedReport) -> tuple[list[str],
     else:
         missing.append("No clear attacker position found, e.g. unauthenticated, low-privilege, malicious website, prompt injection.")
 
+    if _claims_high_impact_rce(report):
+        _add_madbugs_style_evidence(report, confirmed, missing)
+
     return confirmed, missing, agent_context, dangerous_hits
 
 
+def _claims_high_impact_rce(report: ParsedReport) -> bool:
+    lower = report.text.lower()
+    return ((report.severity or "").lower() in {"critical", "high"} or "critical" in lower) and any(
+        term in lower for term in RCE_TERMS
+    )
+
+
+def _add_madbugs_style_evidence(report: ParsedReport, confirmed: list[str], missing: list[str]) -> None:
+    checks = [
+        (
+            report.has_affected_or_tested_version,
+            "MADBugs-style affected/tested version or commit context is present.",
+            "Critical RCE needs affected/tested version or current-HEAD/default-config context.",
+        ),
+        (
+            report.has_attack_surface,
+            "Attack surface or reachable entrypoint context is present.",
+            "Critical RCE needs attack-surface/default-reachability context.",
+        ),
+        (
+            report.has_root_cause,
+            "Root-cause explanation is present.",
+            "Critical RCE needs root-cause/source-to-sink explanation, not only a dangerous sink.",
+        ),
+        (
+            report.has_exploit_chain,
+            "Exploit chain or trigger path is present.",
+            "Critical RCE needs a trigger/exploit chain from attacker input to impact.",
+        ),
+        (
+            report.has_safe_side_effect,
+            "PoC describes an expected result, cleanup, or safe side effect.",
+            "Critical RCE PoC should use a safe side effect and state expected result/cleanup.",
+        ),
+        (
+            report.has_fix_guidance,
+            "Fix, patch, or mitigation guidance is present.",
+            "Critical RCE report should include concise fix/mitigation guidance.",
+        ),
+    ]
+    for ok, good, bad in checks:
+        if ok:
+            confirmed.append(good)
+        else:
+            missing.append(bad)
+
+
 def generate_questions(report: ParsedReport, agent_context: bool, dangerous_hits: dict[str, list[str]]) -> list[str]:
     lower = report.text.lower()
     questions: list[str] = [
@@ -183,7 +263,9 @@ def generate_questions(report: ParsedReport, agent_context: bool, dangerous_hits
         questions += [
             "Who can reach the command/code execution sink in default configuration?",
             "Is there authentication, authorization, user approval, or sandboxing before execution?",
-            "What safe command proves impact, and is cleanup documented?",
+            "What affected/tested version or commit proves this is current?",
+            "What is the root-cause source-to-sink chain from attacker input to execution?",
+            "What safe side effect proves impact, and is cleanup documented?",
         ]
     if any(term in lower for term in FILE_TERMS) or "file read/write" in dangerous_hits:
         questions += [
@@ -208,6 +290,7 @@ def decide_verdict(report: ParsedReport, missing: list[str], agent_context: bool
     missing_symbols = [m for m in missing if "symbol/string not found" in m or "endpoint/path string not found" in m]
     no_repro = any("No obvious PoC" in m for m in missing)
     no_attacker = any("No clear attacker" in m for m in missing)
+    missing_madbugs_rce_context = [m for m in missing if m.startswith("Critical RCE needs") or m.startswith("Critical RCE PoC")]
     lower = report.text.lower()
     claimed_critical_rce = (report.severity or "").lower() == "critical" or "critical" in lower and any(t in lower for t in RCE_TERMS)
 
@@ -217,6 +300,8 @@ def decide_verdict(report: ParsedReport, missing: list[str], agent_context: bool
         return "WEAK", "This appears to involve agent/tool functionality, but the report does not prove unauthorized boundary crossing.", "Rewrite as a potential boundary issue, then prove unauthorized invocation, approval bypass, sandbox escape, cross-user impact, or secret exposure."
     if no_attacker:
         return "WEAK", "The report does not define who the attacker is or how they reach the issue.", "Add a precise attacker model before claiming severity."
+    if claimed_critical_rce and missing_madbugs_rce_context:
+        return "NEEDS_WORK", "Critical RCE is plausible, but missing MADBugs-style context: affected/tested version, root cause, exploit chain, safe PoC evidence, or fix guidance.", "Add the affected/tested version, default attack surface, source-to-sink root cause, safe side-effect PoC with cleanup, and concise fix guidance."
     if no_repro:
         return "NEEDS_WORK", "The claim may be plausible, but it lacks a minimal PoC/repro.", "Add a safe repro against current HEAD and document expected vs actual result."
     if claimed_critical_rce and "auth" not in lower and "unauth" not in lower and "approval" not in lower:
@@ -287,6 +372,12 @@ def vet(repo: Path, report_path: Path, owner_repo: str | None = None) -> VetResu
             "has_repro": report.has_repro,
             "attacker_mentions": report.attacker_mentions,
             "impact_terms": report.impact_terms,
+            "has_affected_or_tested_version": report.has_affected_or_tested_version,
+            "has_attack_surface": report.has_attack_surface,
+            "has_root_cause": report.has_root_cause,
+            "has_exploit_chain": report.has_exploit_chain,
+            "has_safe_side_effect": report.has_safe_side_effect,
+            "has_fix_guidance": report.has_fix_guidance,
         },
     )
 

tests/test_madbugs_style.py

@@ -0,0 +1,71 @@
+from pathlib import Path
+
+from finding_vetter.core import vet
+
+
+def _write_rce_repo(tmp_path: Path) -> Path:
+    repo = tmp_path / "repo"
+    (repo / "server").mkdir(parents=True)
+    (repo / "server" / "handler.py").write_text(
+        "import subprocess\n\n"
+        "def handle(request):\n"
+        "    cmd = request.json['cmd']\n"
+        "    return subprocess.run(cmd, shell=True)\n",
+        encoding="utf-8",
+    )
+    return repo
+
+
+def test_critical_rce_needs_madbugs_style_repro_context(tmp_path: Path):
+    repo = _write_rce_repo(tmp_path)
+    report = tmp_path / "thin-rce.md"
+    report.write_text(
+        "# Critical RCE\n\n"
+        "Severity: Critical\n\n"
+        "Affected files: `server/handler.py`\n\n"
+        "Attacker: remote unauthenticated user.\n\n"
+        "Entrypoint: `POST /run`\n\n"
+        "Boundary crossed: unauthenticated HTTP caller reaches subprocess.\n\n"
+        "PoC: curl http://target/run -d '{\"cmd\":\"id\"}'\n",
+        encoding="utf-8",
+    )
+
+    result = vet(repo, report)
+
+    assert result.verdict == "NEEDS_WORK"
+    assert any("affected/tested version" in item.lower() for item in result.missing)
+    assert any("root-cause" in item.lower() for item in result.missing)
+    assert any("safe side effect" in q.lower() for q in result.questions)
+
+
+def test_madbugs_style_rce_report_passes_with_versions_chain_and_fix(tmp_path: Path):
+    repo = _write_rce_repo(tmp_path)
+    report = tmp_path / "grounded-rce.md"
+    report.write_text(
+        "# Unauthenticated command execution in handler\n\n"
+        "Severity: Critical\n\n"
+        "Affected: demo-agent <= 1.2.3.\n"
+        "Tested on: commit abc123 in default configuration.\n"
+        "Attack surface: HTTP API exposed on port 8000.\n\n"
+        "Affected files: `server/handler.py`\n\n"
+        "Attacker: remote unauthenticated user.\n"
+        "Entrypoint: `POST /run`\n"
+        "Boundary crossed: unauthenticated HTTP caller can invoke a server-side shell command.\n\n"
+        "Root cause: `handle()` copies attacker-controlled JSON into `subprocess.run(..., shell=True)` without auth.\n"
+        "Exploit chain: HTTP request -> JSON `cmd` -> `handle()` -> shell.\n"
+        "Impact: host command execution as the server user.\n\n"
+        "PoC:\n```bash\n"
+        "curl http://127.0.0.1:8000/run -d '{\"cmd\":\"id > /tmp/fv-safe-poc\"}'\n"
+        "test -f /tmp/fv-safe-poc && rm /tmp/fv-safe-poc\n"
+        "```\n\n"
+        "Expected result: `/tmp/fv-safe-poc` is created, proving a safe side effect.\n"
+        "Fix: require authentication and pass an allowlisted argv array with `shell=False`.\n",
+        encoding="utf-8",
+    )
+
+    result = vet(repo, report)
+
+    assert result.verdict == "PASS"
+    assert any("affected/tested version" in item.lower() for item in result.confirmed)
+    assert any("root-cause" in item.lower() for item in result.confirmed)
+    assert not any("safe side effect" in item.lower() for item in result.missing)