diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 345be425..4be76c5b 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -63,3 +63,28 @@ jobs: exit 1 fi echo "All invoked tool paths resolve." + + bugbench: + name: bugbench (detection-quality gate) + runs-on: ubuntu-latest + # NON-BLOCKING while the corpus is small: this reports detection quality on + # every PR but must not red-gate unrelated PRs if one detector regresses or a + # newly-added hard case fails. Flip to blocking (remove continue-on-error and + # make it a required check) once the corpus is broad enough to trust as a gate. + # NOTE: until then, adding a currently-failing case will show this job red but + # will NOT block merges. + continue-on-error: true + steps: + - uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + # Runs the toolkit's detectors against a ground-truth corpus and reports + # precision / recall / false-positive rate. Threshold is strict so quality + # regressions are visible; the job is continue-on-error so it informs + # without blocking while the corpus matures. + - name: Run BugBench detection-quality gate + run: python bughunter/tools/bench.py run --min-precision 1.0 --max-fp-rate 0.0 diff --git a/CLAUDE.md b/CLAUDE.md index 7cf25d27..1234dc39 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -71,6 +71,7 @@ This repo is a Claude Code plugin for professional bug bounty hunting across Hac | `/dashboard` | `/dashboard [--port 8777] \| --export dashboard.html` — local web Hunt Dashboard: leads, recon surface, findings, reports, galleries, memory flywheel, in one live view | | `/verify` | `/verify check [--marker "signal"] \| sweep [root]` — report gate: independently re-derives a finding from its PoC bundle; PROVEN → reportable, REFUTED/UNPROVEN/INCONCLUSIVE → suppressed; writes verification.json; `--require-proof` gates CI | | `/oracle` | `/oracle train \| rank \| score --class idor --url \| eval` — self-learning prioritizer: trains on your journal outcomes to predict P(lead becomes a real finding), with per-feature explanations + leave-one-out CV vs the static scorer | +| `/bench` | `/bench run [--min-precision 1.0 --max-fp-rate 0.0] \| list` — BugBench: measure detector precision/recall/false-positive rate vs a ground-truth corpus (known-vuln + known-safe); `--min-precision`/`--max-fp-rate` gate CI against quality regressions | ### Agents (9 specialized agents) @@ -126,6 +127,7 @@ This repo is a Claude Code plugin for professional bug bounty hunting across Hac - `bughunter/tools/hunt_dashboard.py` — local web Hunt Dashboard (`serve`/`export`); reads `memory/leads/`, `recon/`, `findings/`, `reports/`, galleries, and `hunt-memory/` into one live view. Pure stdlib, binds to 127.0.0.1, HTML-escapes all recon-derived data, path-traversal-guarded `/file` endpoint. Distinct from `bughunter/tools/dashboard.py` (the in-terminal ANSI progress TUI). See `/dashboard`. - `bughunter/tools/verifier.py` — the report gate (prove-or-suppress). Independently re-derives a finding from its `poc_bundler` bundle (re-sends via SSRF-guarded `safe_http` with an initial-host guard, checks a confirmation marker in the fresh response headers+body) and decides reportability: PROVEN only if the marker was in the original response AND is still present now; else REFUTED/UNPROVEN/INCONCLUSIVE → suppressed. Records the re-fetched response's SHA + status. `decide()` is a pure, total policy function; only PROVEN ships. Writes `verification.json`; `check`/`sweep`; `--require-proof` exits non-zero for gating; unsafe methods never re-fired without `--confirm-unsafe`; secrets read from env only. Reuses only merged code (poc_bundler). See `/verify`. - `bughunter/tools/oracle.py` — self-learning lead/finding prioritizer. Trains a smoothed Naive-Bayes log-odds model on `journal.jsonl` outcomes (confirmed/rejected + payout/severity/vuln-class/endpoint/tech features) to score P(lead becomes a real finding), replacing the static regex priority with a data-driven, per-feature-explainable rank. Cold-start prior blends toward the learned model as evidence grows; `eval` runs leave-one-out CV against the prior + majority baselines to prove lift. `train` / `eval` / `rank ` (reuses `lead_board`) / `score`. Pure-stdlib, deterministic, read-only. See `/oracle`. +- `bughunter/tools/bench.py` — BugBench detection-quality harness. Runs the toolkit's detectors against a ground-truth corpus (`bughunter/bench/cases/*.json`, known-vuln + known-safe, recorded inputs = deterministic) and reports precision / recall / false-positive rate per detector. Detectors register a one-function adapter via `@detector`; cases are contributable JSON. `run` / `list`; `--min-precision`/`--max-fp-rate` exit non-zero to gate CI against quality regressions. Pure-stdlib. See `/bench`. - Full catalogue: **`bughunter/tools/README.md`** (~50 tools). `hunt.py` auto-ingests leads after recon (`--graphql` / `--cve-hunt` / `--skip-leads` flags). ### External tool references diff --git a/bughunter/bench/cases/cors-fixed-allowlist-safe.json b/bughunter/bench/cases/cors-fixed-allowlist-safe.json new file mode 100644 index 00000000..c07aaf93 --- /dev/null +++ b/bughunter/bench/cases/cors-fixed-allowlist-safe.json @@ -0,0 +1,15 @@ +{ + "id": "cors-fixed-allowlist-safe", + "detector": "cors", + "note": "Server returns a fixed, trusted allow-list origin and ignores the attacker origin — correct CORS. A detector that flags this is a false positive.", + "input": { + "url": "https://api.example.com/me", + "sent_origin": "https://evil.example", + "acao": "https://app.example.com", + "acac": "true" + }, + "expected": { + "vulnerable": false, + "class": "cors" + } +} diff --git a/bughunter/bench/cases/cors-null-origin-credentialed.json b/bughunter/bench/cases/cors-null-origin-credentialed.json new file mode 100644 index 00000000..f76045c5 --- /dev/null +++ b/bughunter/bench/cases/cors-null-origin-credentialed.json @@ -0,0 +1,16 @@ +{ + "id": "cors-null-origin-credentialed", + "detector": "cors", + "note": "Server reflects the `null` origin AND allows credentials — reachable from a sandboxed iframe / data: URI, cookies sent cross-origin. A real HIGH bug.", + "input": { + "url": "https://api.example.com/me", + "sent_origin": "null", + "acao": "null", + "acac": "true" + }, + "expected": { + "vulnerable": true, + "class": "cors", + "min_severity": "high" + } +} diff --git a/bughunter/bench/cases/cors-reflect-credentialed.json b/bughunter/bench/cases/cors-reflect-credentialed.json new file mode 100644 index 00000000..c2ecfdf9 --- /dev/null +++ b/bughunter/bench/cases/cors-reflect-credentialed.json @@ -0,0 +1,16 @@ +{ + "id": "cors-reflect-credentialed", + "detector": "cors", + "note": "Server reflects an arbitrary attacker origin AND allows credentials — a real, critical CORS bug (authenticated cross-origin read).", + "input": { + "url": "https://api.example.com/me", + "sent_origin": "https://evil.example", + "acao": "https://evil.example", + "acac": "true" + }, + "expected": { + "vulnerable": true, + "class": "cors", + "min_severity": "high" + } +} diff --git a/bughunter/bench/cases/cors-wildcard-with-credentials.json b/bughunter/bench/cases/cors-wildcard-with-credentials.json new file mode 100644 index 00000000..f8bedc12 --- /dev/null +++ b/bughunter/bench/cases/cors-wildcard-with-credentials.json @@ -0,0 +1,16 @@ +{ + "id": "cors-wildcard-with-credentials", + "detector": "cors", + "note": "ACAO:* together with ACAC:true is invalid per spec (browsers reject it) but signals a broken CORS layer — flagged MEDIUM, not treated as a clean public wildcard.", + "input": { + "url": "https://api.example.com/data", + "sent_origin": "https://evil.example", + "acao": "*", + "acac": "true" + }, + "expected": { + "vulnerable": true, + "class": "cors", + "min_severity": "medium" + } +} diff --git a/bughunter/tools/bench.py b/bughunter/tools/bench.py new file mode 100644 index 00000000..9cfcb953 --- /dev/null +++ b/bughunter/tools/bench.py @@ -0,0 +1,435 @@ +#!/usr/bin/env python3 +""" +bench.py — BugBench: measure the toolkit's detection quality. + +Reputation in security tooling is trust, and trust is *measured* false positives +and reproducibility. This harness runs the toolkit's detectors against a +ground-truth corpus of known-vulnerable AND known-safe cases and reports +precision / recall / false-positive rate — so the project can make a claim few +AI security tools can: "here is our measured accuracy, reproduce it yourself." + +Determinism is the whole point: a benchmark case carries a *recorded* input (the +exact response/data a detector would see), never a live URL. CI, and anyone, get +the identical score every time. + +This file is being built step by step. + STEP 1: the data contract — what a benchmark case is, plus load + validation. + STEP 2: the metrics — precision / recall / false-positive rate from a + confusion matrix. FP-rate is the headline: for a security tool, trust + is "almost never cries wolf", and accuracy alone hides that under + class imbalance. + STEP 3: the runner — a detector registry where each detector registers a tiny + adapter (recorded input -> DetectorResult). Adding a detector is a + one-function change; the harness core never grows an if/elif tree. + STEP 4: the scoreboard + `run` CLI + a CI gate (--min-precision / --max-fp-rate) + so a change that regresses detection quality fails the build. +""" +from __future__ import annotations + +import argparse +import glob +import json +import os +import sys +from dataclasses import dataclass, field + +_PKG = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # .../bughunter +_TOOLS = os.path.join(_PKG, "tools") +for _p in (_PKG, _TOOLS): + if _p not in sys.path: + sys.path.insert(0, _p) +_CASES_DIR = os.path.join(_PKG, "bench", "cases") + + +class BenchError(Exception): + """Raised when a benchmark case is malformed — a bad corpus must fail loud, + never silently skew the score.""" + + +@dataclass +class Case: + """One ground-truth benchmark case. + + id unique identifier, names the case in reports + detector which classifier judges this case (wired in a later step) + input recorded data the detector sees (kept opaque here; detector-specific) + expected ground truth: {"vulnerable": bool, "class": str?, "min_severity": str?} + note human-readable provenance / description + """ + id: str + detector: str + expected: dict + input: dict = field(default_factory=dict) + note: str = "" + + @property + def is_vulnerable(self) -> bool: + return bool(self.expected.get("vulnerable")) + + +def _validate_raw(raw: dict, source: str) -> Case: + """Turn one parsed JSON object into a validated Case, or raise BenchError.""" + if not isinstance(raw, dict): + raise BenchError(f"{source}: case must be a JSON object") + for key in ("id", "detector", "expected"): + if key not in raw: + raise BenchError(f"{source}: missing required field '{key}'") + if not isinstance(raw["id"], str) or not raw["id"].strip(): + raise BenchError(f"{source}: 'id' must be a non-empty string") + if not isinstance(raw["detector"], str) or not raw["detector"].strip(): + raise BenchError(f"{source}: 'detector' must be a non-empty string") + expected = raw["expected"] + if not isinstance(expected, dict) or "vulnerable" not in expected: + raise BenchError(f"{source}: 'expected' must be an object with a 'vulnerable' field") + if not isinstance(expected["vulnerable"], bool): + raise BenchError(f"{source}: 'expected.vulnerable' must be true or false") + inp = raw.get("input", {}) + if not isinstance(inp, dict): + raise BenchError(f"{source}: 'input' must be an object") + return Case(id=raw["id"].strip(), detector=raw["detector"].strip(), + expected=expected, input=inp, note=raw.get("note", "")) + + +def load_cases(cases_dir: str | None = None) -> list[Case]: + """Load and validate every *.json case under `cases_dir`. Raises BenchError + on a malformed case or a duplicate id — a broken corpus must never quietly + produce a wrong score.""" + cases_dir = cases_dir or _CASES_DIR + cases: list[Case] = [] + seen: set[str] = set() + for path in sorted(glob.glob(os.path.join(cases_dir, "*.json"))): + with open(path, encoding="utf-8") as fh: + try: + raw = json.load(fh) + except ValueError as e: + raise BenchError(f"{os.path.basename(path)}: invalid JSON — {e}") from e + records = raw if isinstance(raw, list) else [raw] + for rec in records: + case = _validate_raw(rec, os.path.basename(path)) + if case.id in seen: + raise BenchError(f"duplicate case id: {case.id!r}") + seen.add(case.id) + cases.append(case) + return cases + + +# --------------------------------------------------------------------------- +# STEP 2 — metrics. Pure: a confusion matrix in, quality numbers out. +# --------------------------------------------------------------------------- + + +def _safe_div(a: int, b: int) -> float: + """Divide, treating 0/0 as 0.0 — a metric with no applicable cases is not a + failure, it's simply undefined; reporting 0 keeps the scoreboard readable.""" + return a / b if b else 0.0 + + +@dataclass +class Metrics: + """Detection quality from a confusion matrix. tp/fp/tn/fn are the raw counts; + the rest are derived. Rounding happens only at display time (as_dict).""" + tp: int = 0 + fp: int = 0 + tn: int = 0 + fn: int = 0 + + @property + def total(self) -> int: + return self.tp + self.fp + self.tn + self.fn + + @property + def precision(self) -> float: + # of everything we flagged, how much was real + return _safe_div(self.tp, self.tp + self.fp) + + @property + def recall(self) -> float: + # of the real bugs present, how many we caught + return _safe_div(self.tp, self.tp + self.fn) + + @property + def fp_rate(self) -> float: + # of the safe cases, how many we wrongly flagged — the trust metric + return _safe_div(self.fp, self.fp + self.tn) + + @property + def f1(self) -> float: + p, r = self.precision, self.recall + return _safe_div(2 * p * r, p + r) if (p + r) else 0.0 + + @property + def accuracy(self) -> float: + # included for completeness — but it is misleading under class imbalance + # (all-safe corpus + a do-nothing detector still scores high), so the + # scoreboard leads with precision / fp_rate, not this. + return _safe_div(self.tp + self.tn, self.total) + + def as_dict(self) -> dict: + return { + "n": self.total, "tp": self.tp, "fp": self.fp, "tn": self.tn, "fn": self.fn, + "precision": round(self.precision, 4), + "recall": round(self.recall, 4), + "fp_rate": round(self.fp_rate, 4), + "f1": round(self.f1, 4), + "accuracy": round(self.accuracy, 4), + } + + def summary_line(self) -> str: + return (f"precision {self.precision:.0%} recall {self.recall:.0%} " + f"FP-rate {self.fp_rate:.0%} F1 {self.f1:.2f} (n={self.total})") + + +def score(pairs: list[tuple[bool, bool]]) -> Metrics: + """Build a Metrics from (predicted_vulnerable, actually_vulnerable) pairs. + + Pure and decoupled from how predictions were produced — the runner (a later + step) turns cases + detector output into these pairs, and this stays trivial + to unit-test with synthetic data.""" + m = Metrics() + for predicted, actual in pairs: + if predicted and actual: + m.tp += 1 + elif predicted and not actual: + m.fp += 1 + elif not predicted and actual: + m.fn += 1 + else: + m.tn += 1 + return m + + +def score_cases(cases: list[Case], predictions: dict[str, bool]) -> Metrics: + """Convenience: score a run given cases and a {case_id: predicted} map. + A case with no prediction is treated as 'not flagged' (a miss on a vuln + case, a correct silence on a safe one) — never dropped, so the denominator + always reflects the full corpus.""" + return score([(bool(predictions.get(c.id, False)), c.is_vulnerable) for c in cases]) + + +# --------------------------------------------------------------------------- +# STEP 3 — the runner. A registry of detector adapters; the harness stays +# generic (open for new detectors, closed for modification). +# --------------------------------------------------------------------------- + + +@dataclass +class DetectorResult: + """What a detector concluded about one case. Reduced to `.vulnerable` at the + metric boundary; `severity`/`detail` are kept for the human scoreboard.""" + vulnerable: bool + severity: str | None = None + detail: str = "" + + +# name -> callable(input: dict) -> DetectorResult +DETECTORS: dict[str, "callable"] = {} + + +def detector(name: str): + """Register a detector adapter under `name` (matches Case.detector).""" + def deco(fn): + DETECTORS[name] = fn + return fn + return deco + + +@detector("cors") +def _detect_cors(inp: dict) -> DetectorResult: + """Adapter for the CORS classifier. Feeds the case's recorded ACAO/ACAC to + the real, pure cors_scanner.classify — no network. Lazy-imported so bench's + core stays dependency-free.""" + from cors_scanner import classify, CorsTest # lazy: keep core import-light + test = CorsTest(origin=inp.get("sent_origin", ""), label="bench", + weakness="reflect-any") + finding = classify(inp.get("url", ""), test, inp.get("acao"), inp.get("acac")) + if finding is None: + return DetectorResult(False, None, "no CORS finding") + # A public wildcard (INFO) is a misconfig note, not a reportable bug — + # counting it as a detection would manufacture false positives. + is_vuln = finding.severity != "INFO" + return DetectorResult(is_vuln, finding.severity, finding.title) + + +def run_case(case: Case) -> DetectorResult: + """Run the case's registered detector on its recorded input.""" + fn = DETECTORS.get(case.detector) + if fn is None: + raise BenchError(f"no detector registered for {case.detector!r} " + f"(case {case.id!r}); registered: {sorted(DETECTORS)}") + return fn(case.input) + + +def run_cases(cases: list[Case]) -> dict[str, DetectorResult]: + """Run every case, returning {case_id: DetectorResult}.""" + return {c.id: run_case(c) for c in cases} + + +# --------------------------------------------------------------------------- +# STEP 4 — scoreboard + CI gate. +# --------------------------------------------------------------------------- + + +# Severity ordering for min_severity enforcement. cors emits MEDIUM_LOW for "low". +SEVERITY_RANK = {"critical": 4, "high": 3, "medium": 2, "medium_low": 1, "low": 1, + "informational": 0, "info": 0, "none": 0, "": 0} + + +def _sev_rank(sev: str | None) -> int: + return SEVERITY_RANK.get((sev or "").strip().lower(), 0) + + +def meets_min_severity(case: Case, result: DetectorResult) -> bool: + """A detection must reach the case's required severity to count as a catch. + Without this, a detector that downgrades CRITICAL->LOW would still 'pass' — + the exact gap this enforces.""" + min_sev = case.expected.get("min_severity") + if not min_sev: + return True + return _sev_rank(result.severity) >= _sev_rank(min_sev) + + +def predicted_positive(case: Case, result: DetectorResult) -> bool: + """Does the detector's result count as correctly flagging THIS case? For a + vulnerable case with a min_severity, a below-severity detection is a miss.""" + if not result.vulnerable: + return False + if case.is_vulnerable and not meets_min_severity(case, result): + return False # flagged, but below required severity → not a valid catch + return True + + +def evaluate(cases: list[Case]) -> dict: + """Run every case and compute overall + per-detector metrics. Returns a + plain dict so it serializes straight to JSON for CI.""" + results = run_cases(cases) + # min_severity-aware prediction: a downgrade below the required severity is + # NOT counted as catching the bug (see meets_min_severity / the reviewer note). + preds = {c.id: predicted_positive(c, results[c.id]) for c in cases} + overall = score_cases(cases, preds) + detectors = sorted({c.detector for c in cases}) + per = {det: score_cases([c for c in cases if c.detector == det], preds) + for det in detectors} + # Actionable failure rows: what did we get wrong, and how. + failures = [] + for c in cases: + r = results[c.id] + pred = preds.get(c.id, False) + if pred and not c.is_vulnerable: + kind = "false_positive" # cried wolf + elif not pred and c.is_vulnerable: + # distinguish a severity downgrade from missing the bug entirely + kind = ("severity_downgrade" if r.vulnerable + and not meets_min_severity(c, r) else "miss") + else: + continue + failures.append({"id": c.id, "detector": c.detector, "kind": kind, + "detail": r.detail, "severity": r.severity}) + return {"overall": overall, "per_detector": per, + "failures": failures, "results": results, "predictions": preds} + + +def render_report(cases: list[Case]) -> str: + """A Markdown scoreboard. Leads with precision / FP-rate (the trust metrics); + accuracy is intentionally not the headline.""" + ev = evaluate(cases) + o: Metrics = ev["overall"] + lines = [ + "# BugBench — detection quality", "", + f"**{o.precision:.0%} precision · {o.fp_rate:.0%} false-positive rate · " + f"{o.recall:.0%} recall · F1 {o.f1:.2f}** (n={o.total})", "", + "| detector | cases | precision | recall | FP-rate | F1 |", + "|---|---|---|---|---|---|", + ] + for det, m in ev["per_detector"].items(): + lines.append(f"| {det} | {m.total} | {m.precision:.0%} | {m.recall:.0%} " + f"| {m.fp_rate:.0%} | {m.f1:.2f} |") + lines += ["", f"| **overall** | **{o.total}** | **{o.precision:.0%}** | " + f"**{o.recall:.0%}** | **{o.fp_rate:.0%}** | **{o.f1:.2f}** |"] + if ev["failures"]: + lines += ["", "## Failures", "", "| case | detector | kind | detail |", + "|---|---|---|---|"] + for f in ev["failures"]: + lines.append(f"| {f['id']} | {f['detector']} | {f['kind']} | {f['detail']} |") + else: + lines += ["", "_No false positives, no misses on the current corpus._"] + return "\n".join(lines) + "\n" + + +def gate(overall: Metrics, min_precision: float | None, + max_fp_rate: float | None) -> list[str]: + """Return a list of threshold violations (empty = passed). Kept pure so the + CI decision is testable without spawning a process.""" + violations = [] + if min_precision is not None and overall.precision < min_precision: + violations.append(f"precision {overall.precision:.2%} < required {min_precision:.2%}") + if max_fp_rate is not None and overall.fp_rate > max_fp_rate: + violations.append(f"FP-rate {overall.fp_rate:.2%} > allowed {max_fp_rate:.2%}") + return violations + + +def main(argv=None) -> int: + for _stream in (sys.stdout, sys.stderr): # non-ASCII marks must not crash a cp1252 console + try: + _stream.reconfigure(encoding="utf-8", errors="replace") + except (AttributeError, ValueError): + pass + + ap = argparse.ArgumentParser(description="BugBench — detection-quality benchmark.") + ap.add_argument("--cases-dir", default=_CASES_DIR, help="corpus directory") + sub = ap.add_subparsers(dest="cmd", required=True) + sub.add_parser("list", help="list loaded benchmark cases") + + prun = sub.add_parser("run", help="run detectors over the corpus and score them") + prun.add_argument("--json", dest="as_json", action="store_true") + prun.add_argument("--report", help="write the Markdown scoreboard to this path") + prun.add_argument("--min-precision", type=float, + help="CI gate: fail (exit 1) if precision falls below this") + prun.add_argument("--max-fp-rate", type=float, + help="CI gate: fail (exit 1) if false-positive rate exceeds this") + + args = ap.parse_args(argv) + + try: + cases = load_cases(args.cases_dir) + except BenchError as e: + print(f"[!] corpus error: {e}", file=sys.stderr) + return 2 + + if args.cmd == "list": + vuln = sum(1 for c in cases if c.is_vulnerable) + print(f"{len(cases)} case(s): {vuln} vulnerable, {len(cases) - vuln} safe") + for c in cases: + tag = "VULN" if c.is_vulnerable else "safe" + print(f" [{tag:>4}] {c.id:<32} detector={c.detector}") + return 0 + + if args.cmd == "run": + try: + ev = evaluate(cases) + except BenchError as e: + print(f"[!] {e}", file=sys.stderr) + return 2 + overall: Metrics = ev["overall"] + report = render_report(cases) + if args.report: + with open(args.report, "w", encoding="utf-8", newline="\n") as fh: + fh.write(report) + if args.as_json: + print(json.dumps({"overall": overall.as_dict(), + "per_detector": {d: m.as_dict() for d, m in ev["per_detector"].items()}, + "failures": ev["failures"]}, indent=2)) + else: + print(report) + violations = gate(overall, args.min_precision, args.max_fp_rate) + if violations: + for v in violations: + print(f"[✗] BugBench gate failed: {v}", file=sys.stderr) + return 1 + if args.min_precision is not None or args.max_fp_rate is not None: + print("[✓] BugBench gate passed.") + return 0 + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/commands/bench.md b/commands/bench.md new file mode 100644 index 00000000..900721c2 --- /dev/null +++ b/commands/bench.md @@ -0,0 +1,77 @@ +--- +description: BugBench — measure the toolkit's detection quality (precision / recall / false-positive rate) against a ground-truth corpus of known-vulnerable AND known-safe cases. Usage: /bench run [--min-precision 1.0 --max-fp-rate 0.0] | /bench list +--- + +# /bench + +Reputation in security tooling is trust, and trust is *measured*. BugBench runs +the toolkit's detectors against a ground-truth corpus — cases that are **known +vulnerable** and cases that are **known safe** — and reports precision, recall, +and **false-positive rate**, so the project can make a claim few AI security +tools can: *"here is our measured accuracy — reproduce it yourself."* + +Every case carries a **recorded input** (the exact response a detector would +see), never a live URL, so the score is identical on every machine and in CI. + +## Usage + +``` +/bench list # show the corpus +/bench run # score every detector +/bench run --json # machine-readable +/bench run --report bench-score.md # write the Markdown scoreboard +/bench run --min-precision 1.0 --max-fp-rate 0.0 # CI gate (exit 1 on regression) +``` + +Run directly: + +```bash +bughunter/tools/bench.py run --min-precision 1.0 --max-fp-rate 0.0 +``` + +## What it reports + +- **precision** — of everything flagged, how much was a real bug (noise ↓). +- **recall** — of the real bugs present, how many were caught. +- **false-positive rate** — of the safe cases, how many were wrongly flagged. + *This is the headline trust metric.* Accuracy is deliberately **not** the + headline: on a mostly-safe corpus a do-nothing detector scores high accuracy + while catching zero bugs. +- **failures** — the exact cases that were false positives, misses, or + **severity downgrades**, so a regression is actionable, not just a number. + +Scoring enforces each case's `min_severity`: a detector that flags a CRITICAL +case at LOW is counted as a **miss** (`severity_downgrade`), not a catch — so the +benchmark can't be fooled by a downgrade. `class` is descriptive metadata. + +## The CI gate + +`--min-precision` / `--max-fp-rate` make `/bench run` **exit non-zero** when +quality drops below the bar. In CI the `bugbench` job runs this on every PR but is +currently **non-blocking** (`continue-on-error`) while the corpus is small — it +*reports* quality without red-gating unrelated PRs. Flip it to a required check +once the corpus is broad enough to trust as a hard gate. (Until then, note that a +currently-failing case shows the job red but does not block merges.) + +## Adding a case + +Drop a JSON file in `bughunter/bench/cases/`: + +```json +{ + "id": "cors-reflect-credentialed", + "detector": "cors", + "note": "reflects arbitrary origin with credentials", + "input": { "url": "...", "sent_origin": "https://evil.example", + "acao": "https://evil.example", "acac": "true" }, + "expected": { "vulnerable": true, "class": "cors", "min_severity": "high" } +} +``` + +Cases are data — anyone can contribute one without touching Python. Add safe +cases too: they're what let BugBench measure false positives. + +## Adding a detector + +Register a one-function adapter in `bughunter/tools/bench.py` (recorded `input` → +`DetectorResult`) with `@detector("name")`. The harness core never changes. diff --git a/tests/test_bench.py b/tests/test_bench.py new file mode 100644 index 00000000..c545f7a9 --- /dev/null +++ b/tests/test_bench.py @@ -0,0 +1,336 @@ +"""Tests for bughunter/tools/bench.py — BugBench detection-quality harness. + +STEP 1 scope: the data contract — loading + strict validation of benchmark +cases. (Detector running and metrics are added in later steps, with their own +tests.) `bughunter/tools` is on sys.path via conftest. +""" + +import json + +import pytest + +import bench + + +def _write_case(tmp_path, name, obj): + p = tmp_path / name + p.write_text(json.dumps(obj), encoding="utf-8") + return p + + +def _valid_case(**over): + base = {"id": "c1", "detector": "cors", + "expected": {"vulnerable": True}, "input": {}} + base.update(over) + return base + + +# --------------------------------------------------------------------------- +# Happy path +# --------------------------------------------------------------------------- + + +def test_loads_the_shipped_corpus(): + """The corpus that ships with the repo must always be valid — a broken + case file would poison every score.""" + cases = bench.load_cases() + assert len(cases) >= 2 + assert any(c.is_vulnerable for c in cases) + assert any(not c.is_vulnerable for c in cases) # safe cases exist (for FP measurement) + + +def test_loads_valid_cases_from_dir(tmp_path): + _write_case(tmp_path, "a.json", _valid_case(id="a")) + _write_case(tmp_path, "b.json", _valid_case(id="b", expected={"vulnerable": False})) + cases = bench.load_cases(str(tmp_path)) + assert {c.id for c in cases} == {"a", "b"} + assert cases[0].is_vulnerable is True + + +def test_a_file_may_hold_a_list_of_cases(tmp_path): + _write_case(tmp_path, "many.json", + [_valid_case(id="x"), _valid_case(id="y", expected={"vulnerable": False})]) + assert len(bench.load_cases(str(tmp_path))) == 2 + + +# --------------------------------------------------------------------------- +# Validation — a bad corpus must fail loud, never skew the score silently +# --------------------------------------------------------------------------- + + +def test_missing_required_field_raises(tmp_path): + _write_case(tmp_path, "bad.json", {"detector": "cors", "expected": {"vulnerable": True}}) + with pytest.raises(bench.BenchError, match="missing required field 'id'"): + bench.load_cases(str(tmp_path)) + + +def test_expected_must_have_bool_vulnerable(tmp_path): + _write_case(tmp_path, "bad.json", _valid_case(expected={"vulnerable": "yes"})) + with pytest.raises(bench.BenchError, match="vulnerable"): + bench.load_cases(str(tmp_path)) + + +def test_duplicate_ids_raise(tmp_path): + _write_case(tmp_path, "a.json", _valid_case(id="dup")) + _write_case(tmp_path, "b.json", _valid_case(id="dup")) + with pytest.raises(bench.BenchError, match="duplicate case id"): + bench.load_cases(str(tmp_path)) + + +def test_invalid_json_raises(tmp_path): + (tmp_path / "broken.json").write_text("{ not json", encoding="utf-8") + with pytest.raises(bench.BenchError, match="invalid JSON"): + bench.load_cases(str(tmp_path)) + + +def test_input_must_be_object(tmp_path): + _write_case(tmp_path, "bad.json", _valid_case(input=["not", "an", "object"])) + with pytest.raises(bench.BenchError, match="'input' must be an object"): + bench.load_cases(str(tmp_path)) + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def test_cli_list(capsys): + rc = bench.main(["list"]) + assert rc == 0 + out = capsys.readouterr().out + assert "vulnerable" in out + assert "cors-reflect-credentialed" in out + + +# --------------------------------------------------------------------------- +# STEP 2 — metrics +# --------------------------------------------------------------------------- + + +def test_confusion_matrix_counts(): + # (predicted, actual): TP, FP, FN, TN + m = bench.score([(True, True), (True, False), (False, True), (False, False)]) + assert (m.tp, m.fp, m.fn, m.tn) == (1, 1, 1, 1) + assert m.total == 4 + + +def test_perfect_detector(): + m = bench.score([(True, True), (True, True), (False, False)]) + assert m.precision == 1.0 + assert m.recall == 1.0 + assert m.fp_rate == 0.0 + assert m.f1 == 1.0 + + +def test_false_positives_drive_precision_and_fp_rate(): + # 1 real bug caught, but 3 safe cases wrongly flagged + m = bench.score([(True, True), (True, False), (True, False), (True, False)]) + assert m.precision == 0.25 # 1 of 4 flags was real + assert m.fp_rate == 1.0 # every safe case was flagged — max noise + assert m.recall == 1.0 # it did catch the one real bug + + +def test_accuracy_is_misleading_under_imbalance(): + """The core lesson: a do-nothing detector on a mostly-safe corpus scores + high accuracy while catching zero bugs — which is why the scoreboard leads + with precision / fp_rate, not accuracy.""" + pairs = [(False, False)] * 19 + [(False, True)] # 19 safe, 1 missed bug + m = bench.score(pairs) + assert m.accuracy == 0.95 # looks great... + assert m.recall == 0.0 # ...but it caught nothing + assert m.fp_rate == 0.0 + + +def test_empty_does_not_crash(): + m = bench.score([]) + assert m.total == 0 + assert m.precision == 0.0 and m.recall == 0.0 and m.fp_rate == 0.0 + + +def test_score_cases_missing_prediction_counts_as_not_flagged(): + cases = [ + bench.Case(id="v", detector="d", expected={"vulnerable": True}), # a real bug + bench.Case(id="s", detector="d", expected={"vulnerable": False}), # safe + ] + # no prediction for either → vuln case becomes a miss (FN), safe stays TN + m = bench.score_cases(cases, {}) + assert (m.fn, m.tn, m.tp, m.fp) == (1, 1, 0, 0) + # now flag only the real bug → perfect + m2 = bench.score_cases(cases, {"v": True}) + assert m2.precision == 1.0 and m2.recall == 1.0 and m2.fp_rate == 0.0 + + +def test_as_dict_and_summary_line(): + m = bench.score([(True, True), (True, False)]) + d = m.as_dict() + assert d["tp"] == 1 and d["fp"] == 1 and d["precision"] == 0.5 + assert "FP-rate" in m.summary_line() + + +# --------------------------------------------------------------------------- +# STEP 3 — the runner (real cors classifier, offline) +# --------------------------------------------------------------------------- + + +def test_cors_adapter_flags_credentialed_reflection(): + inp = {"url": "https://api.x/me", "sent_origin": "https://evil.example", + "acao": "https://evil.example", "acac": "true"} + res = bench._detect_cors(inp) + assert res.vulnerable is True + assert res.severity == "CRITICAL" + + +def test_cors_adapter_clears_fixed_allowlist(): + inp = {"url": "https://api.x/me", "sent_origin": "https://evil.example", + "acao": "https://app.x", "acac": "true"} + res = bench._detect_cors(inp) + assert res.vulnerable is False + + +def test_cors_adapter_public_wildcard_is_not_a_bug(): + """INFO-level public wildcard must NOT count as a detection (else it's a + manufactured false positive).""" + inp = {"url": "https://api.x/data", "sent_origin": "https://evil.example", + "acao": "*", "acac": "false"} + res = bench._detect_cors(inp) + assert res.vulnerable is False + assert res.severity == "INFO" + + +def test_run_case_uses_registered_detector(): + case = bench.Case(id="c", detector="cors", expected={"vulnerable": True}, + input={"url": "https://x/me", "sent_origin": "https://evil.example", + "acao": "https://evil.example", "acac": "true"}) + assert bench.run_case(case).vulnerable is True + + +def test_unknown_detector_raises(): + case = bench.Case(id="c", detector="nope", expected={"vulnerable": True}) + with pytest.raises(bench.BenchError, match="no detector registered"): + bench.run_case(case) + + +def test_run_the_shipped_corpus_end_to_end(): + """Run every real detector on the shipped cases and score it — the two cors + cases should come out perfect (1 caught, 0 false alarms).""" + cases = bench.load_cases() + results = bench.run_cases(cases) + predictions = {cid: r.vulnerable for cid, r in results.items()} + m = bench.score_cases(cases, predictions) + assert m.fp == 0 # never cries wolf on the safe case + assert m.fn == 0 # catches the real bug + assert m.precision == 1.0 + + +# --------------------------------------------------------------------------- +# STEP 4 — scoreboard + gate +# --------------------------------------------------------------------------- + + +def test_evaluate_overall_and_per_detector(): + ev = bench.evaluate(bench.load_cases()) + assert ev["overall"].precision == 1.0 + assert "cors" in ev["per_detector"] + assert ev["failures"] == [] + + +def test_render_report_is_markdown_with_headline(): + md = bench.render_report(bench.load_cases()) + assert "detection quality" in md + assert "FP-rate" in md or "false-positive" in md + assert "| cors |" in md + assert "No false positives" in md + + +def test_gate_passes_and_fails(): + good = bench.Metrics(tp=5, fp=0, tn=5, fn=0) # precision 1.0, fp_rate 0 + bad = bench.Metrics(tp=1, fp=4, tn=1, fn=0) # precision .2, fp_rate .8 + assert bench.gate(good, min_precision=0.9, max_fp_rate=0.0) == [] + viol = bench.gate(bad, min_precision=0.9, max_fp_rate=0.0) + assert len(viol) == 2 # both thresholds violated + + +def _register_dummy(): + @bench.detector("dummy_always_flags") + def _d(inp): + return bench.DetectorResult(True, "HIGH", "always flags") + + +def test_cli_run_gate_blocks_a_false_positive(tmp_path, capsys): + """A detector that flags a SAFE case must fail the --max-fp-rate 0 gate — + this is the CI mechanism that blocks quality regressions.""" + _register_dummy() + (tmp_path / "safe.json").write_text(json.dumps({ + "id": "dummy-safe", "detector": "dummy_always_flags", + "expected": {"vulnerable": False}, "input": {}}), encoding="utf-8") + rc = bench.main(["--cases-dir", str(tmp_path), "run", "--max-fp-rate", "0.0"]) + assert rc == 1 + assert "gate failed" in capsys.readouterr().err + + +def test_cli_run_shipped_corpus_passes_strict_gate(capsys): + rc = bench.main(["run", "--min-precision", "1.0", "--max-fp-rate", "0.0"]) + assert rc == 0 + assert "gate passed" in capsys.readouterr().out + + +def test_cli_run_writes_report_file(tmp_path): + out = tmp_path / "score.md" + rc = bench.main(["run", "--report", str(out)]) + assert rc == 0 + assert "detection quality" in out.read_text(encoding="utf-8") + + +# --------------------------------------------------------------------------- +# min_severity enforcement (review fix): a severity downgrade is NOT a catch +# --------------------------------------------------------------------------- + + +def test_meets_min_severity_ranking(): + hi = bench.Case(id="c", detector="cors", expected={"vulnerable": True, "min_severity": "high"}) + assert bench.meets_min_severity(hi, bench.DetectorResult(True, "CRITICAL")) + assert bench.meets_min_severity(hi, bench.DetectorResult(True, "HIGH")) + assert not bench.meets_min_severity(hi, bench.DetectorResult(True, "MEDIUM")) + assert not bench.meets_min_severity(hi, bench.DetectorResult(True, "MEDIUM_LOW")) + # no min_severity declared → any severity is acceptable + no_min = bench.Case(id="c", detector="cors", expected={"vulnerable": True}) + assert bench.meets_min_severity(no_min, bench.DetectorResult(True, "LOW")) + + +def test_severity_downgrade_counts_as_a_miss(monkeypatch): + """A detector that flags a CRITICAL case at LOW must NOT pass — it's a miss, + not a catch. Guards against the 'min_severity silently ignored' gap.""" + case = bench.Case(id="dg", detector="stub_low", + expected={"vulnerable": True, "min_severity": "high"}) + + @bench.detector("stub_low") + def _stub(inp): + return bench.DetectorResult(True, "LOW", "flagged but downgraded") + + ev = bench.evaluate([case]) + assert ev["overall"].fn == 1 # counted as missed, not caught + assert ev["overall"].tp == 0 + assert ev["failures"][0]["kind"] == "severity_downgrade" + + +def test_at_or_above_min_severity_is_a_catch(): + case = bench.Case(id="ok", detector="stub_hi", + expected={"vulnerable": True, "min_severity": "high"}) + + @bench.detector("stub_hi") + def _stub(inp): + return bench.DetectorResult(True, "CRITICAL", "flagged critical") + + ev = bench.evaluate([case]) + assert ev["overall"].tp == 1 and ev["overall"].fn == 0 + + +def test_expanded_shipped_corpus_scores_clean(): + """The shipped corpus (incl. null-origin + wildcard+creds, each with a + min_severity) must still be caught at the required severity, 0 FP.""" + cases = bench.load_cases() + assert len(cases) >= 4 + ev = bench.evaluate(cases) + assert ev["overall"].fp == 0 + assert ev["overall"].fn == 0 + assert ev["overall"].precision == 1.0