From 9f37c57b472d13ca97696b256f59bd4321a4ec19 Mon Sep 17 00:00:00 2001 From: Shivendra-Coherent Date: Tue, 22 Sep 2026 14:17:55 +0530 Subject: [PATCH 1/2] feat(replay): add finding replay / regression (/replay) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the docs/CAPABILITY-GAPS.md gap "No finding-replay/regression to re-verify past bugs on a target". Builds directly on the merged PoC bundler (#144): once /poc has captured a finding, /replay re-sends that exact request and tells you whether the bug is STILL_VULNERABLE, FIXED, CHANGED, or UNREACHABLE — turning a one-shot PoC into a monitorable, retest-ready asset. tools/replay.py — two modes: - run [--marker "signal"]: replay one bundle. - all [root]: sweep every poc-bundle/1 under root (default findings/) into a ranked regression-report.md + regression.json ("retest day"). Verdict engine (pure, exhaustively unit-tested): - Marker-based is the reliable signal (confidence: confirmed) — the string that proves the bug present -> STILL_VULNERABLE, absent -> FIXED. Markers persist as marker.txt so later sweeps reuse them. - Without a marker, a status+SHA-256 integrity check is used and honestly labelled "heuristic"; it never claims FIXED from a hash diff alone. Reuses existing infra: poc_bundler.capture (SSRF-guarded safe_http), parse_raw_request, and sha256. The manifest URL is trusted for scheme/host so an http:// finding isn't replayed over https (avoids a false UNREACHABLE). Safety: PUT/DELETE/PATCH skipped without --confirm-unsafe (replaying a mutating request could re-trigger it); secrets read from env only (like repro.sh) and never written into replay.json or the sweep report; --fail-if-vulnerable exits 3 for scheduled monitoring/CI. Console output reconfigures to UTF-8 so verdict icons don't crash a non-UTF-8 terminal. 25 tests in tests/test_replay.py (all verdict branches, header resolution, bundle I/O, offline replay via monkeypatch, CLI). /replay command + CLAUDE.md command-table and tools-list entries. Co-Authored-By: Claude Opus 4.8 --- CLAUDE.md | 2 + bughunter/tools/replay.py | 414 ++++++++++++++++++++++++++++++++++++++ commands/replay.md | 62 ++++++ tests/test_replay.py | 280 ++++++++++++++++++++++++++ 4 files changed, 758 insertions(+) create mode 100644 bughunter/tools/replay.py create mode 100644 commands/replay.md create mode 100644 tests/test_replay.py diff --git a/CLAUDE.md b/CLAUDE.md index b452741d..1a5520f2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -70,6 +70,7 @@ This repo is a Claude Code plugin for professional bug bounty hunting across Hac | `/poc` | `/poc capture [-H ...] \| from-request req.txt [--response resp.txt]` — reproducible PoC evidence bundle (request/response, curl, HAR, screenshot, report-ready evidence.md); secrets redacted by default | | `/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 | +| `/replay` | `/replay run [--marker "signal"] \| all [root]` — re-verify a PoC bundle: STILL_VULNERABLE / FIXED / CHANGED / UNREACHABLE; SSRF-guarded, unsafe methods gated, batch retest report | ### Agents (9 specialized agents) @@ -124,6 +125,7 @@ This repo is a Claude Code plugin for professional bug bounty hunting across Hac - `tools/poc_bundler.py` — PoC evidence bundler (`capture` live via SSRF-guarded `safe_http`, or `from-request` offline from a saved/Burp request). Emits `request.http`, `response.http`, `repro.sh` (curl), `evidence.har` (HAR 1.2), optional `screenshot.png`, `evidence.md` (report-ready), and `bundle.json` (manifest + response SHA-256) under `findings/-/evidence//`. Secrets redacted by default (repro.sh uses `$ENV` placeholders); refuses PUT/DELETE/PATCH without `--confirm-unsafe`. See `/poc`. - `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 `tools/dashboard.py` (the in-terminal ANSI progress TUI). See `/dashboard`. - `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`. +- `tools/replay.py` — finding replay / regression. Re-sends the exact request from a `poc_bundler` bundle (via SSRF-guarded `safe_http`) and returns STILL_VULNERABLE / FIXED / CHANGED / UNREACHABLE / INDETERMINATE. Marker-based verdicts are `confirmed`; status+SHA-256 diff is a labelled `heuristic` (never claims FIXED alone). `run ` for one, `all [root]` sweeps every bundle into `regression-report.md`. Secrets read from env only; unsafe methods gated by `--confirm-unsafe`; `--fail-if-vulnerable` exits 3 for CI/monitoring. See `/replay`. - Full catalogue: **`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/tools/replay.py b/bughunter/tools/replay.py new file mode 100644 index 00000000..9bcec510 --- /dev/null +++ b/bughunter/tools/replay.py @@ -0,0 +1,414 @@ +#!/usr/bin/env python3 +""" +replay.py — re-verify a past finding against its PoC evidence bundle. + +The gap this closes (docs/CAPABILITY-GAPS.md → "No finding-replay/regression to +re-verify past bugs on a target"): once /poc has captured a finding, nothing +re-checks whether it's *still* live. Retests pay again, and programs increasingly +reward regression checks — but today that's a manual re-send-and-eyeball. + +This replays the exact request stored in a PoC bundle (produced by poc_bundler) +through the repo's SSRF-guarded opener, compares the fresh response against the +recorded evidence, and returns a verdict: + + STILL_VULNERABLE the vuln signal is still present + FIXED the vuln signal is gone + CHANGED response differs from the PoC; a human should look + UNREACHABLE the host/endpoint no longer answers + INDETERMINATE can't judge (missing marker + no baseline, or missing secret) + SKIPPED_UNSAFE a mutating method (PUT/DELETE/PATCH) — needs --confirm-unsafe + ERROR the bundle could not be read + +Two modes: + + replay.py run [--marker "signal"] replay one finding + replay.py all [root] sweep every bundle (retest day) + +How the verdict is decided: + * A **marker** (a string that proves the bug — the leaked email, the reflected + payload, an error string) is the reliable signal: present -> STILL_VULNERABLE, + absent -> FIXED (confidence: confirmed). Pass it with --marker; it's saved as + `marker.txt` in the bundle so later sweeps reuse it automatically. + * Without a marker we fall back to an integrity **heuristic**: byte-identical to + the recorded response (status + SHA-256) -> STILL_VULNERABLE(unchanged), else + CHANGED. This is honestly labelled "heuristic" — it never claims FIXED. + +Secrets: redacted bundles reference secrets via env vars ($AUTHORIZATION, …), +exactly like the bundle's repro.sh. replay reads those from the environment and +never writes secret values or full response bodies into its own artifacts. + +Design: resolve_headers() and decide_verdict() are pure and unit-tested without a +socket; only replay_one() (via poc_bundler.capture) touches the network. +""" +from __future__ import annotations + +import argparse +import glob +import json +import os +import re +import sys +import urllib.error +from dataclasses import dataclass +from datetime import datetime, timezone + +_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) +import poc_bundler as poc # noqa: E402 (reuse capture/parse/hash/constants) + +# Verdicts +STILL_VULNERABLE = "STILL_VULNERABLE" +FIXED = "FIXED" +CHANGED = "CHANGED" +UNREACHABLE = "UNREACHABLE" +INDETERMINATE = "INDETERMINATE" +SKIPPED_UNSAFE = "SKIPPED_UNSAFE" +ERROR = "ERROR" + +# Matches poc_bundler's redacted placeholder: "‹redacted:$AUTHORIZATION›" +_REDACTED_RE = re.compile(r"^‹redacted:\$([A-Z0-9_]+)›$") + +_ICON = { + STILL_VULNERABLE: "🔴", FIXED: "🟢", CHANGED: "🟡", UNREACHABLE: "⚫", + INDETERMINATE: "⚪", SKIPPED_UNSAFE: "⏭", ERROR: "✖", +} + + +def now_iso() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +# --------------------------------------------------------------------------- +# Pure core +# --------------------------------------------------------------------------- + + +@dataclass +class Observed: + reachable: bool + status: int | None = None + sha256: str | None = None + body: str = "" + elapsed_ms: int = 0 + error: str = "" + + +@dataclass +class Verdict: + verdict: str + confidence: str # "confirmed" (marker) | "heuristic" (hash) | "n/a" + detail: str + + +def resolve_headers(headers: list[tuple[str, str]], + env: dict[str, str]) -> tuple[list[tuple[str, str]], list[str]]: + """Turn a recorded (possibly redacted) header list into one we can actually + send: redacted secret placeholders are resolved from `env`. Returns the + resolved headers and the list of env vars that were referenced but missing.""" + resolved: list[tuple[str, str]] = [] + missing: list[str] = [] + for name, value in headers: + m = _REDACTED_RE.match(value) + if m: + var = m.group(1) + if env.get(var): + resolved.append((name, env[var])) + else: + missing.append(var) # drop it; caller decides how to react + else: + resolved.append((name, value)) + return resolved, missing + + +def decide_verdict(baseline_status: int | None, baseline_sha: str | None, + marker: str | None, observed: Observed) -> Verdict: + """The heart of replay — pure, deterministic, exhaustively tested.""" + if not observed.reachable: + return Verdict(UNREACHABLE, "confirmed", observed.error or "host did not respond") + if marker: + if marker in (observed.body or ""): + return Verdict(STILL_VULNERABLE, "confirmed", + f"marker still present in response ({observed.status})") + return Verdict(FIXED, "confirmed", f"marker no longer present ({observed.status})") + # No marker → integrity heuristic. Never claims FIXED from a hash alone. + if baseline_sha is None: + return Verdict(INDETERMINATE, "n/a", + "no marker and no recorded response baseline — pass --marker") + if observed.status == baseline_status and observed.sha256 == baseline_sha: + return Verdict(STILL_VULNERABLE, "heuristic", + f"response byte-identical to recorded PoC ({observed.status})") + return Verdict(CHANGED, "heuristic", + f"response differs from PoC (was {baseline_status}, now {observed.status}) — review") + + +# --------------------------------------------------------------------------- +# Bundle I/O +# --------------------------------------------------------------------------- + + +def load_bundle(bundle_dir: str) -> dict: + """Read a poc-bundle/1 folder into the pieces replay needs. Raises + ValueError/OSError if it isn't a readable bundle.""" + manifest_path = os.path.join(bundle_dir, "bundle.json") + with open(manifest_path, encoding="utf-8") as fh: + manifest = json.load(fh) + if manifest.get("schema") != "poc-bundle/1": + raise ValueError(f"not a poc-bundle/1 manifest: {manifest_path}") + req_path = os.path.join(bundle_dir, "request.http") + with open(req_path, encoding="utf-8", errors="replace") as fh: + exchange = poc.parse_raw_request(fh.read()) + ex_meta = manifest.get("exchange", {}) + # request.http is origin-form and loses the scheme (parse_raw_request would + # guess https). The manifest records the exact original URL — trust it, so we + # replay against the right scheme/host and don't false-report UNREACHABLE. + if ex_meta.get("url"): + exchange.url = ex_meta["url"] + if ex_meta.get("method"): + exchange.method = ex_meta["method"] + marker = load_marker(bundle_dir, manifest) + return { + "dir": bundle_dir, + "manifest": manifest, + "exchange": exchange, + "baseline_status": ex_meta.get("response_status"), + "baseline_sha": ex_meta.get("response_sha256"), + "marker": marker, + } + + +def load_marker(bundle_dir: str, manifest: dict) -> str | None: + """Marker lookup order: manifest field (forward-compat) → marker.txt sidecar.""" + if manifest.get("marker"): + return manifest["marker"] + sidecar = os.path.join(bundle_dir, "marker.txt") + if os.path.isfile(sidecar): + with open(sidecar, encoding="utf-8", errors="replace") as fh: + text = fh.read().strip() + return text or None + return None + + +def save_marker(bundle_dir: str, marker: str) -> None: + with open(os.path.join(bundle_dir, "marker.txt"), "w", + encoding="utf-8", newline="\n") as fh: + fh.write(marker) + + +def find_bundles(root: str) -> list[str]: + """Every poc-bundle/1 directory under `root`, sorted for stable output.""" + out = [] + for manifest in glob.glob(os.path.join(root, "**", "bundle.json"), recursive=True): + try: + with open(manifest, encoding="utf-8") as fh: + if json.load(fh).get("schema") == "poc-bundle/1": + out.append(os.path.dirname(manifest)) + except (OSError, ValueError): + continue + return sorted(out) + + +# --------------------------------------------------------------------------- +# Replay (network) + result assembly +# --------------------------------------------------------------------------- + + +def replay_one(bundle_dir: str, timeout: float, confirm_unsafe: bool, + marker_override: str | None, env: dict[str, str] | None = None) -> dict: + """Replay a single bundle and return a result dict (never raises).""" + env = env if env is not None else dict(os.environ) + result = { + "dir": bundle_dir, "replayed_at": now_iso(), + "finding": {}, "baseline": {}, "observed": {}, + "verdict": ERROR, "confidence": "n/a", "detail": "", + } + try: + bundle = load_bundle(bundle_dir) + except (OSError, ValueError) as e: + result["detail"] = f"could not read bundle: {e}" + return result + + ex = bundle["exchange"] + marker = marker_override if marker_override is not None else bundle["marker"] + result["finding"] = bundle["manifest"].get("finding", {}) + result["baseline"] = {"status": bundle["baseline_status"], "sha256": bundle["baseline_sha"]} + + def finalize(v: Verdict, observed: Observed | None = None): + result["verdict"], result["confidence"], result["detail"] = v.verdict, v.confidence, v.detail + if observed is not None: + result["observed"] = { + "reachable": observed.reachable, "status": observed.status, + "sha256": observed.sha256, "elapsed_ms": observed.elapsed_ms, + "error": observed.error, + } + result["marker_used"] = bool(marker) + return result + + # Guard mutating methods — replaying a DELETE could re-trigger the action. + if ex.method.upper() in poc.UNSAFE_METHODS and not confirm_unsafe: + return finalize(Verdict(SKIPPED_UNSAFE, "n/a", + f"{ex.method} may mutate state; re-run with --confirm-unsafe")) + + headers, missing = resolve_headers(ex.request_headers, env) + if missing: + return finalize(Verdict(INDETERMINATE, "n/a", + "missing secret env var(s): " + ", ".join(sorted(set(missing))) + + " — export them (see repro.sh) and retry")) + + try: + fresh = poc.capture(ex.url, ex.method, headers, ex.request_body or None, timeout) + observed = Observed( + reachable=True, status=fresh.response_status, + sha256=poc.sha256_hex(fresh.response_body_bytes) if fresh.response_body_bytes else None, + body=fresh.response_body, elapsed_ms=fresh.elapsed_ms, + ) + except urllib.error.URLError as e: + observed = Observed(reachable=False, error=str(getattr(e, "reason", e))) + except Exception as e: # never let one bundle break a sweep + observed = Observed(reachable=False, error=f"{type(e).__name__}: {e}") + + return finalize(decide_verdict(bundle["baseline_status"], bundle["baseline_sha"], + marker, observed), observed) + + +# --------------------------------------------------------------------------- +# Rendering (pure) +# --------------------------------------------------------------------------- + + +def render_report(results: list[dict]) -> str: + counts: dict[str, int] = {} + for r in results: + counts[r["verdict"]] = counts.get(r["verdict"], 0) + 1 + order = [STILL_VULNERABLE, CHANGED, INDETERMINATE, FIXED, UNREACHABLE, + SKIPPED_UNSAFE, ERROR] + summary = " ".join(f"{_ICON.get(k, '')} {k}:{counts[k]}" + for k in order if k in counts) + lines = [f"# Regression sweep — {now_iso()}", "", + f"{len(results)} bundle(s). {summary}", "", + "| verdict | conf | finding | target | detail |", "|---|---|---|---|---|"] + rank = {k: i for i, k in enumerate(order)} + for r in sorted(results, key=lambda x: rank.get(x["verdict"], 99)): + f = r.get("finding", {}) + lines.append( + f"| {_ICON.get(r['verdict'], '')} {r['verdict']} | {r['confidence']} | " + f"{f.get('finding_id') or f.get('title') or os.path.basename(r['dir'])} | " + f"{f.get('target', '')} | {r['detail']} |") + return "\n".join(lines) + "\n" + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def _write_replay_json(bundle_dir: str, result: dict) -> None: + payload = {"schema": "replay/1", **result} + with open(os.path.join(bundle_dir, "replay.json"), "w", + encoding="utf-8", newline="\n") as fh: + json.dump(payload, fh, indent=2) + + +def _print_one(result: dict) -> None: + icon = _ICON.get(result["verdict"], "") + print(f"{icon} {result['verdict']} ({result['confidence']})") + f = result.get("finding", {}) + if f.get("finding_id") or f.get("target"): + print(f" finding: {f.get('finding_id', '—')} target: {f.get('target', '—')}") + print(f" {result['detail']}") + obs = result.get("observed") or {} + if obs: + print(f" observed: status={obs.get('status')} " + f"reachable={obs.get('reachable')} {obs.get('elapsed_ms', 0)}ms") + + +def main(argv=None) -> int: + # Verdict icons are non-ASCII; make sure a non-UTF-8 console (e.g. Windows + # cp1252) degrades gracefully instead of crashing on print. + for _stream in (sys.stdout, sys.stderr): + try: + _stream.reconfigure(encoding="utf-8", errors="replace") + except (AttributeError, ValueError): + pass + + ap = argparse.ArgumentParser( + description="Finding replay / regression — re-verify a PoC bundle.") + sub = ap.add_subparsers(dest="cmd", required=True) + + pr = sub.add_parser("run", help="replay one PoC bundle") + pr.add_argument("bundle_dir") + pr.add_argument("--marker", help="signal string that proves the bug (saved for future sweeps)") + pr.add_argument("--timeout", type=float, default=15) + pr.add_argument("--confirm-unsafe", action="store_true", + help="allow replaying PUT/DELETE/PATCH (may mutate state)") + pr.add_argument("--json", dest="as_json", action="store_true") + pr.add_argument("--fail-if-vulnerable", action="store_true", + help="exit 3 if the verdict is STILL_VULNERABLE (for CI/monitoring)") + + pa = sub.add_parser("all", help="sweep every PoC bundle under a root") + pa.add_argument("root", nargs="?", default=os.path.join(poc._REPO, "findings"), + help="root to search (default: findings/)") + pa.add_argument("--timeout", type=float, default=15) + pa.add_argument("--confirm-unsafe", action="store_true") + pa.add_argument("--json", dest="as_json", action="store_true") + pa.add_argument("--fail-if-vulnerable", action="store_true", + help="exit 3 if any verdict is STILL_VULNERABLE") + + args = ap.parse_args(argv) + + if args.cmd == "run": + if args.marker: + try: + save_marker(args.bundle_dir, args.marker) + except OSError: + pass + result = replay_one(args.bundle_dir, args.timeout, args.confirm_unsafe, args.marker) + try: + _write_replay_json(args.bundle_dir, result) + except OSError: + pass + if args.as_json: + print(json.dumps(result, indent=2)) + else: + _print_one(result) + return 3 if (args.fail_if_vulnerable and result["verdict"] == STILL_VULNERABLE) else 0 + + if args.cmd == "all": + bundles = find_bundles(args.root) + if not bundles: + print(f"[!] no PoC bundles found under {args.root}. Create one with /poc.") + return 0 + results = [] + for d in bundles: + r = replay_one(d, args.timeout, args.confirm_unsafe, None) + try: + _write_replay_json(d, r) + except OSError: + pass + results.append(r) + report = render_report(results) + report_path = os.path.join(args.root, "regression-report.md") + try: + with open(report_path, "w", encoding="utf-8", newline="\n") as fh: + fh.write(report) + with open(os.path.join(args.root, "regression.json"), "w", + encoding="utf-8", newline="\n") as fh: + json.dump(results, fh, indent=2) + except OSError: + report_path = None + if args.as_json: + print(json.dumps(results, indent=2)) + else: + print(report) + if report_path: + print(f"[+] report: {report_path}") + vulnerable = any(r["verdict"] == STILL_VULNERABLE for r in results) + return 3 if (args.fail_if_vulnerable and vulnerable) else 0 + + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/commands/replay.md b/commands/replay.md new file mode 100644 index 00000000..9578cd8f --- /dev/null +++ b/commands/replay.md @@ -0,0 +1,62 @@ +--- +description: Re-verify a past finding against its PoC evidence bundle — is it still vulnerable, fixed, or changed? Replays the exact captured request (SSRF-guarded) and compares to the recorded evidence. Usage: /replay run [--marker "signal"] | /replay all [root] +--- + +# /replay + +Turn a `/poc` evidence bundle into a **regression check**. Retests pay again and +programs increasingly reward regression reports — this replays the exact request +stored in a bundle and tells you whether the bug is still live. + +## Usage + +``` +/replay run findings/target.com-idor/evidence/F-12 --marker "victim@corp.com" +/replay all # sweep every bundle under findings/ (retest day) +``` + +Run directly: + +```bash +tools/replay.py run --marker "signal-that-proves-the-bug" +tools/replay.py all [root] # default root: findings/ +``` + +## Verdicts + +| Verdict | Meaning | +|---|---| +| 🔴 `STILL_VULNERABLE` | the bug's signal is still present | +| 🟢 `FIXED` | the signal is gone | +| 🟡 `CHANGED` | response differs from the PoC — a human should look | +| ⚫ `UNREACHABLE` | the host/endpoint no longer answers | +| ⚪ `INDETERMINATE` | can't judge (no marker + no baseline, or a missing secret) | +| ⏭ `SKIPPED_UNSAFE` | a mutating method (PUT/DELETE/PATCH) — needs `--confirm-unsafe` | + +## How the verdict is decided + +- **Marker (reliable, `confidence: confirmed`)** — a string that proves the bug: + the leaked email, the reflected payload, an error message. Present → still + vulnerable; absent → fixed. Pass it with `--marker`; it's saved as `marker.txt` + in the bundle so later `all` sweeps reuse it automatically. +- **Heuristic (`confidence: heuristic`)** — with no marker, replay compares the + fresh response to the recorded one (status + SHA-256). Byte-identical → still + vulnerable (unchanged); different → `CHANGED`. It never claims `FIXED` from a + hash alone, since dynamic content also changes. + +## Safety + +- Replays through the repo's SSRF-guarded opener (`tools/safe_http.py`). +- **`PUT`/`DELETE`/`PATCH` are skipped unless you pass `--confirm-unsafe`** — + replaying a mutating request could re-trigger the action. +- Redacted bundles reference secrets via env vars; replay reads them from the + environment (like `repro.sh`) and **never writes secret values or full response + bodies** into `replay.json` / the sweep report. + +## Output + +- `run` writes `replay.json` into the bundle and prints the verdict. +- `all` writes `regression-report.md` + `regression.json` at the root and prints a + ranked table (most-severe first). +- `--fail-if-vulnerable` exits `3` when anything is still vulnerable — handy for + scheduled monitoring / CI. diff --git a/tests/test_replay.py b/tests/test_replay.py new file mode 100644 index 00000000..d8b3df82 --- /dev/null +++ b/tests/test_replay.py @@ -0,0 +1,280 @@ +"""Tests for tools/replay.py — finding replay / regression. + +The pure core (decide_verdict, resolve_headers, render_report) is tested +directly. replay_one and the CLI are tested offline by monkeypatching +poc_bundler.capture, and bundles are built on disk with poc_bundler's own +offline `from-request` path. `bughunter/tools` is on sys.path via conftest. +""" + +import json +import os + +import pytest + +import poc_bundler as poc +import replay as rp + + +# --------------------------------------------------------------------------- +# Helpers — build a real PoC bundle offline, then replay it. +# --------------------------------------------------------------------------- + + +def make_bundle(tmp_path, *, method="GET", body="", auth=True, + resp_status="200 OK", resp_body='{"email":"victim@acme.com"}'): + tmp_path.mkdir(parents=True, exist_ok=True) + req_lines = [f"{method} /users/1 HTTP/1.1", "Host: api.acme.com"] + if auth: + req_lines.append("Authorization: Bearer SECRET") + req = "\n".join(req_lines) + "\n\n" + body + resp = f"HTTP/1.1 {resp_status}\nContent-Type: application/json\n\n{resp_body}" + reqf, respf = tmp_path / "req.txt", tmp_path / "resp.txt" + reqf.write_text(req, encoding="utf-8") + respf.write_text(resp, encoding="utf-8") + out = tmp_path / "bundle" + poc.main(["from-request", str(reqf), "--response", str(respf), + "--target", "acme.com", "--vuln-class", "idor", + "--finding-id", "F-1", "-o", str(out)]) + return str(out) + + +def fake_capture_returning(body, status=200): + """A poc.capture stand-in that returns a fixed response, no network.""" + def _cap(url, method, headers, reqbody, timeout): + raw = body.encode("utf-8") + return poc.Exchange(method=method, url=url, response_status=status, + response_reason="OK", response_body=body, + response_body_bytes=raw) + return _cap + + +# --------------------------------------------------------------------------- +# decide_verdict — the heart, all branches +# --------------------------------------------------------------------------- + + +def test_verdict_unreachable(): + v = rp.decide_verdict(200, "sha", "marker", rp.Observed(reachable=False, error="dns")) + assert v.verdict == rp.UNREACHABLE + + +def test_verdict_marker_present_is_still_vulnerable(): + obs = rp.Observed(reachable=True, status=200, body='...victim@acme.com...') + v = rp.decide_verdict(200, "sha", "victim@acme.com", obs) + assert v.verdict == rp.STILL_VULNERABLE + assert v.confidence == "confirmed" + + +def test_verdict_marker_absent_is_fixed(): + obs = rp.Observed(reachable=True, status=403, body="access denied") + v = rp.decide_verdict(200, "sha", "victim@acme.com", obs) + assert v.verdict == rp.FIXED + assert v.confidence == "confirmed" + + +def test_verdict_no_marker_identical_is_still_vulnerable_heuristic(): + obs = rp.Observed(reachable=True, status=200, sha256="abc") + v = rp.decide_verdict(200, "abc", None, obs) + assert v.verdict == rp.STILL_VULNERABLE + assert v.confidence == "heuristic" + + +def test_verdict_no_marker_changed(): + obs = rp.Observed(reachable=True, status=404, sha256="different") + v = rp.decide_verdict(200, "abc", None, obs) + assert v.verdict == rp.CHANGED + assert v.confidence == "heuristic" + + +def test_verdict_no_marker_no_baseline_is_indeterminate(): + obs = rp.Observed(reachable=True, status=200, sha256="abc") + v = rp.decide_verdict(None, None, None, obs) + assert v.verdict == rp.INDETERMINATE + + +def test_verdict_never_claims_fixed_without_a_marker(): + """A hash mismatch alone must not be reported as FIXED (could be dynamic content).""" + obs = rp.Observed(reachable=True, status=200, sha256="x") + v = rp.decide_verdict(200, "y", None, obs) + assert v.verdict != rp.FIXED + + +# --------------------------------------------------------------------------- +# resolve_headers +# --------------------------------------------------------------------------- + + +def test_resolve_headers_substitutes_env_secret(): + headers = [("Host", "acme.com"), ("Authorization", "‹redacted:$AUTHORIZATION›")] + resolved, missing = rp.resolve_headers(headers, {"AUTHORIZATION": "Bearer real"}) + assert ("Authorization", "Bearer real") in resolved + assert missing == [] + + +def test_resolve_headers_reports_missing_secret(): + headers = [("Authorization", "‹redacted:$AUTHORIZATION›")] + resolved, missing = rp.resolve_headers(headers, {}) + assert missing == ["AUTHORIZATION"] + assert resolved == [] # unresolved secret is dropped, not sent blank + + +def test_resolve_headers_passthrough_plain(): + headers = [("Accept", "application/json")] + resolved, missing = rp.resolve_headers(headers, {}) + assert resolved == headers and missing == [] + + +# --------------------------------------------------------------------------- +# marker + bundle discovery +# --------------------------------------------------------------------------- + + +def test_marker_sidecar_and_manifest(tmp_path): + d = tmp_path / "b" + d.mkdir() + assert rp.load_marker(str(d), {}) is None + rp.save_marker(str(d), "leaked@x.com") + assert rp.load_marker(str(d), {}) == "leaked@x.com" + # manifest field wins over sidecar + assert rp.load_marker(str(d), {"marker": "from-manifest"}) == "from-manifest" + + +def test_find_bundles_only_matches_poc_schema(tmp_path): + good = make_bundle(tmp_path / "g") + (tmp_path / "decoy").mkdir() + (tmp_path / "decoy" / "bundle.json").write_text('{"schema":"other"}', encoding="utf-8") + found = rp.find_bundles(str(tmp_path)) + assert good in found + assert not any("decoy" in f for f in found) + + +def test_load_bundle_reads_baseline_and_request(tmp_path): + d = make_bundle(tmp_path) + b = rp.load_bundle(d) + assert b["exchange"].method == "GET" + assert b["exchange"].url == "https://api.acme.com/users/1" + assert b["baseline_status"] == 200 + assert b["baseline_sha"] is not None + + +def test_load_bundle_trusts_manifest_url_scheme(tmp_path): + """request.http loses the scheme; the manifest URL is authoritative so an + http:// finding isn't replayed over https (the WRONG_VERSION_NUMBER trap).""" + d = make_bundle(tmp_path) + mpath = os.path.join(d, "bundle.json") + manifest = json.loads(open(mpath, encoding="utf-8").read()) + manifest["exchange"]["url"] = "http://api.acme.com:8080/users/1" + open(mpath, "w", encoding="utf-8").write(json.dumps(manifest)) + assert rp.load_bundle(d)["exchange"].url == "http://api.acme.com:8080/users/1" + + +# --------------------------------------------------------------------------- +# replay_one (offline via monkeypatch) +# --------------------------------------------------------------------------- + + +def test_replay_unsafe_method_is_skipped(tmp_path): + d = make_bundle(tmp_path, method="DELETE", auth=False) + r = rp.replay_one(d, timeout=5, confirm_unsafe=False, marker_override=None) + assert r["verdict"] == rp.SKIPPED_UNSAFE + + +def test_replay_missing_secret_is_indeterminate(tmp_path): + d = make_bundle(tmp_path, auth=True) # request.http has redacted Authorization + r = rp.replay_one(d, timeout=5, confirm_unsafe=False, marker_override=None, env={}) + assert r["verdict"] == rp.INDETERMINATE + assert "AUTHORIZATION" in r["detail"] + + +def test_replay_still_vulnerable_via_marker(tmp_path, monkeypatch): + d = make_bundle(tmp_path, auth=False, resp_body='{"email":"victim@acme.com"}') + monkeypatch.setattr(poc, "capture", fake_capture_returning('{"email":"victim@acme.com"}')) + r = rp.replay_one(d, timeout=5, confirm_unsafe=False, + marker_override="victim@acme.com", env={}) + assert r["verdict"] == rp.STILL_VULNERABLE + assert r["observed"]["reachable"] is True + + +def test_replay_fixed_via_marker(tmp_path, monkeypatch): + d = make_bundle(tmp_path, auth=False) + monkeypatch.setattr(poc, "capture", fake_capture_returning("Access Denied", status=403)) + r = rp.replay_one(d, timeout=5, confirm_unsafe=False, + marker_override="victim@acme.com", env={}) + assert r["verdict"] == rp.FIXED + + +def test_replay_unreachable(tmp_path, monkeypatch): + import urllib.error + d = make_bundle(tmp_path, auth=False) + + def boom(*a, **k): + raise urllib.error.URLError("connection refused") + monkeypatch.setattr(poc, "capture", boom) + r = rp.replay_one(d, timeout=5, confirm_unsafe=False, marker_override=None, env={}) + assert r["verdict"] == rp.UNREACHABLE + assert r["observed"]["reachable"] is False + + +def test_replay_bad_bundle_is_error(tmp_path): + (tmp_path / "empty").mkdir() + r = rp.replay_one(str(tmp_path / "empty"), timeout=5, confirm_unsafe=False, + marker_override=None) + assert r["verdict"] == rp.ERROR + + +def test_replay_writes_no_secret_into_replay_json(tmp_path, monkeypatch): + d = make_bundle(tmp_path, auth=True, resp_body="ok") + monkeypatch.setattr(poc, "capture", fake_capture_returning("ok")) + r = rp.replay_one(d, timeout=5, confirm_unsafe=False, marker_override=None, + env={"AUTHORIZATION": "Bearer TOPSECRET"}) + assert "TOPSECRET" not in json.dumps(r) + + +# --------------------------------------------------------------------------- +# render_report +# --------------------------------------------------------------------------- + + +def test_render_report_counts_and_orders(): + results = [ + {"verdict": rp.FIXED, "confidence": "confirmed", "detail": "gone", "dir": "a", + "finding": {"finding_id": "F-2", "target": "acme.com"}}, + {"verdict": rp.STILL_VULNERABLE, "confidence": "confirmed", "detail": "present", + "dir": "b", "finding": {"finding_id": "F-1", "target": "acme.com"}}, + ] + md = rp.render_report(results) + assert "STILL_VULNERABLE:1" in md and "FIXED:1" in md + # most-severe first: STILL_VULNERABLE row appears before FIXED row + assert md.index("F-1") < md.index("F-2") + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def test_cli_run_writes_replay_json_and_exit_code(tmp_path, monkeypatch): + d = make_bundle(tmp_path, auth=False, resp_body="pwned") + monkeypatch.setattr(poc, "capture", fake_capture_returning("pwned")) + rc = rp.main(["run", d, "--marker", "pwned", "--fail-if-vulnerable"]) + assert rc == 3 # STILL_VULNERABLE + fail flag + saved = json.loads((tmp_path / "bundle" / "replay.json").read_text(encoding="utf-8")) + assert saved["schema"] == "replay/1" + assert saved["verdict"] == rp.STILL_VULNERABLE + # marker persisted for future sweeps + assert (tmp_path / "bundle" / "marker.txt").read_text(encoding="utf-8") == "pwned" + + +def test_cli_all_sweep_writes_report(tmp_path, monkeypatch): + make_bundle(tmp_path / "b1", auth=False, resp_body="x") + monkeypatch.setattr(poc, "capture", fake_capture_returning("x")) + rc = rp.main(["all", str(tmp_path)]) + assert rc == 0 + assert (tmp_path / "regression-report.md").exists() + assert (tmp_path / "regression.json").exists() + + +def test_cli_all_empty_root(tmp_path, capsys): + rc = rp.main(["all", str(tmp_path)]) + assert rc == 0 + assert "no PoC bundles" in capsys.readouterr().out From a7773d8275676435c41748ac743a25b47b229cf5 Mon Sep 17 00:00:00 2001 From: Shivendra-Coherent Date: Mon, 28 Sep 2026 13:36:20 +0530 Subject: [PATCH 2/2] =?UTF-8?q?fix(replay):=20address=20review=20=E2=80=94?= =?UTF-8?q?=20scope=20gate,=20initial-hop=20SSRF=20guard,=20BLOCKED=20verd?= =?UTF-8?q?ict?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Resolves the change requests from @shuvonsec on the /replay PR. All three are about safely re-issuing live traffic. 1. (scope) replay now gates every send through scope_checker.py (Critical Rule #1: read full scope before touching any asset). Pass --scope '*.target.com' (repeatable) / --scope-file ; out-of-scope targets are OUT_OF_SCOPE and never sent. `all` REFUSES to run with no scope unless --no-scope is passed (explicitly asserting every stored target is in scope). `run` warns when no scope is given. 2. (SSRF) the initial hop is now validated. load_bundle trusts the bundle URL, and safe_urlopen only guards redirects — so a tampered bundle could make the FIRST request hit 169.254.169.254 / localhost. initial_host_blocked() checks the initial host with the exact same blocklist safe_http uses on redirects, before any request; a blocked host returns BLOCKED and nothing is sent. 3. (verdict) guard/redirect URLErrors are no longer masked as UNREACHABLE. A new BLOCKED verdict is returned for SSRF-guard / too-many-redirects errors, so a bug that now 302s to an internal host reads as a live signal to investigate, not "endpoint gone / fixed". Tests: +9 in tests/test_replay.py — scope in/out + `all` refusal, the initial-host guard on literals (169.254/localhost/127.0.0.1 blocked; public IP allowed), a tampered-bundle replay that returns BLOCKED without sending, and redirect-to- internal classified BLOCKED vs a real connection error UNREACHABLE. 34 pass. Verified live via the CLI. Docs updated. Co-Authored-By: Claude Opus 4.8 --- bughunter/tools/replay.py | 110 +++++++++++++++++++++++++++++++++--- commands/replay.md | 14 ++++- tests/test_replay.py | 114 ++++++++++++++++++++++++++++++++++++-- 3 files changed, 225 insertions(+), 13 deletions(-) diff --git a/bughunter/tools/replay.py b/bughunter/tools/replay.py index 9bcec510..01ff6066 100644 --- a/bughunter/tools/replay.py +++ b/bughunter/tools/replay.py @@ -37,6 +37,16 @@ exactly like the bundle's repro.sh. replay reads those from the environment and never writes secret values or full response bodies into its own artifacts. +Safety on re-issued traffic: + * Scope gate (Critical Rule #1): pass --scope/--scope-file and every send is + checked with the repo's scope_checker; out-of-scope targets are OUT_OF_SCOPE + and never sent. `all` refuses to run with no scope unless --no-scope is given. + * SSRF: the bundle URL is untrusted, so the INITIAL host is validated with the + same blocklist safe_http uses on redirects (metadata/private/loopback) before + any request — a tampered bundle can't make the first hop hit 169.254.169.254. + * A guard/redirect block is reported as BLOCKED, distinct from UNREACHABLE, so a + bug that now 302s to an internal host is never masked as "fixed". + Design: resolve_headers() and decide_verdict() are pure and unit-tested without a socket; only replay_one() (via poc_bundler.capture) touches the network. """ @@ -51,6 +61,7 @@ import urllib.error from dataclasses import dataclass from datetime import datetime, timezone +from urllib.parse import urlparse _PKG = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # .../bughunter _TOOLS = os.path.join(_PKG, "tools") @@ -58,6 +69,8 @@ if _p not in sys.path: sys.path.insert(0, _p) import poc_bundler as poc # noqa: E402 (reuse capture/parse/hash/constants) +from safe_http import _is_blocked_redirect_target # noqa: E402 (same SSRF blocklist) +from scope_checker import ScopeChecker # noqa: E402 (deterministic scope gate) # Verdicts STILL_VULNERABLE = "STILL_VULNERABLE" @@ -66,6 +79,8 @@ UNREACHABLE = "UNREACHABLE" INDETERMINATE = "INDETERMINATE" SKIPPED_UNSAFE = "SKIPPED_UNSAFE" +OUT_OF_SCOPE = "OUT_OF_SCOPE" # scope guard: target not in the program's scope +BLOCKED = "BLOCKED" # SSRF guard: initial host or a redirect went internal ERROR = "ERROR" # Matches poc_bundler's redacted placeholder: "‹redacted:$AUTHORIZATION›" @@ -73,10 +88,24 @@ _ICON = { STILL_VULNERABLE: "🔴", FIXED: "🟢", CHANGED: "🟡", UNREACHABLE: "⚫", - INDETERMINATE: "⚪", SKIPPED_UNSAFE: "⏭", ERROR: "✖", + INDETERMINATE: "⚪", SKIPPED_UNSAFE: "⏭", OUT_OF_SCOPE: "🚫", BLOCKED: "🛑", + ERROR: "✖", } +def initial_host_blocked(url: str) -> bool: + """True if a URL's *initial* host is a metadata/private/loopback/link-local + target. safe_urlopen only guards redirect hops, and load_bundle trusts the + bundle's URL verbatim — so a tampered bundle could point the FIRST request at + 169.254.169.254 or localhost. We validate the initial host with the exact same + blocklist safe_http uses on redirects, before ever sending.""" + try: + host = urlparse(url).hostname + except ValueError: + return True + return _is_blocked_redirect_target(host) + + def now_iso() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") @@ -94,6 +123,7 @@ class Observed: body: str = "" elapsed_ms: int = 0 error: str = "" + blocked: bool = False # request refused by the SSRF guard (host/redirect internal) @dataclass @@ -126,6 +156,12 @@ def resolve_headers(headers: list[tuple[str, str]], def decide_verdict(baseline_status: int | None, baseline_sha: str | None, marker: str | None, observed: Observed) -> Verdict: """The heart of replay — pure, deterministic, exhaustively tested.""" + if observed.blocked: + # A guard/redirect block is NOT "endpoint gone / fixed" — a bug that now + # 302s to an internal host is a live signal, not an absence. Keep it + # distinct from UNREACHABLE so it never reads as a silent pass. + return Verdict(BLOCKED, "confirmed", + observed.error or "request blocked by SSRF guard — investigate") if not observed.reachable: return Verdict(UNREACHABLE, "confirmed", observed.error or "host did not respond") if marker: @@ -216,7 +252,8 @@ def find_bundles(root: str) -> list[str]: def replay_one(bundle_dir: str, timeout: float, confirm_unsafe: bool, - marker_override: str | None, env: dict[str, str] | None = None) -> dict: + marker_override: str | None, env: dict[str, str] | None = None, + scope: ScopeChecker | None = None) -> dict: """Replay a single bundle and return a result dict (never raises).""" env = env if env is not None else dict(os.environ) result = { @@ -241,16 +278,33 @@ def finalize(v: Verdict, observed: Observed | None = None): result["observed"] = { "reachable": observed.reachable, "status": observed.status, "sha256": observed.sha256, "elapsed_ms": observed.elapsed_ms, - "error": observed.error, + "error": observed.error, "blocked": observed.blocked, } result["marker_used"] = bool(marker) return result + # Scope gate FIRST — Critical Rule #1: never touch an out-of-scope asset. + # No request is sent for a target the program's scope doesn't cover. + if scope is not None and not scope.is_in_scope(ex.url): + host = urlparse(ex.url).hostname or ex.url + return finalize(Verdict(OUT_OF_SCOPE, "n/a", + f"{host} is not in scope — not sent")) + # Guard mutating methods — replaying a DELETE could re-trigger the action. if ex.method.upper() in poc.UNSAFE_METHODS and not confirm_unsafe: return finalize(Verdict(SKIPPED_UNSAFE, "n/a", f"{ex.method} may mutate state; re-run with --confirm-unsafe")) + # SSRF guard on the INITIAL host — the bundle URL is untrusted input. + if initial_host_blocked(ex.url): + host = urlparse(ex.url).hostname or ex.url + return finalize( + decide_verdict(bundle["baseline_status"], bundle["baseline_sha"], marker, + Observed(reachable=False, blocked=True, + error=f"initial host blocked by SSRF guard: {host}")), + Observed(reachable=False, blocked=True, + error=f"initial host blocked by SSRF guard: {host}")) + headers, missing = resolve_headers(ex.request_headers, env) if missing: return finalize(Verdict(INDETERMINATE, "n/a", @@ -265,7 +319,12 @@ def finalize(v: Verdict, observed: Observed | None = None): body=fresh.response_body, elapsed_ms=fresh.elapsed_ms, ) except urllib.error.URLError as e: - observed = Observed(reachable=False, error=str(getattr(e, "reason", e))) + reason = str(getattr(e, "reason", e)) + # safe_urlopen raises URLError for a blocked redirect ("SSRF guard") or + # "too many redirects". Those are guard signals, NOT "endpoint gone" — + # flag them as blocked so a bug that now 302s internal isn't masked. + blocked = ("SSRF guard" in reason) or ("too many redirects" in reason) + observed = Observed(reachable=False, blocked=blocked, error=reason) except Exception as e: # never let one bundle break a sweep observed = Observed(reachable=False, error=f"{type(e).__name__}: {e}") @@ -282,8 +341,8 @@ def render_report(results: list[dict]) -> str: counts: dict[str, int] = {} for r in results: counts[r["verdict"]] = counts.get(r["verdict"], 0) + 1 - order = [STILL_VULNERABLE, CHANGED, INDETERMINATE, FIXED, UNREACHABLE, - SKIPPED_UNSAFE, ERROR] + order = [STILL_VULNERABLE, BLOCKED, CHANGED, INDETERMINATE, FIXED, UNREACHABLE, + OUT_OF_SCOPE, SKIPPED_UNSAFE, ERROR] summary = " ".join(f"{_ICON.get(k, '')} {k}:{counts[k]}" for k in order if k in counts) lines = [f"# Regression sweep — {now_iso()}", "", @@ -324,6 +383,22 @@ def _print_one(result: dict) -> None: f"reachable={obs.get('reachable')} {obs.get('elapsed_ms', 0)}ms") +def build_scope(patterns: list[str] | None, scope_file: str | None) -> ScopeChecker | None: + """Build a ScopeChecker from inline patterns and/or a scope file (one pattern + per line, '#' comments). Returns None when no scope was supplied.""" + pats = list(patterns or []) + if scope_file: + with open(scope_file, encoding="utf-8", errors="replace") as fh: + pats += [ln.strip() for ln in fh if ln.strip() and not ln.startswith("#")] + return ScopeChecker(domains=pats) if pats else None + + +def _add_scope_args(p) -> None: + p.add_argument("--scope", action="append", default=[], + help="in-scope domain pattern, e.g. '*.target.com' (repeatable)") + p.add_argument("--scope-file", help="file of in-scope patterns (one per line)") + + def main(argv=None) -> int: # Verdict icons are non-ASCII; make sure a non-UTF-8 console (e.g. Windows # cp1252) degrades gracefully instead of crashing on print. @@ -346,6 +421,7 @@ def main(argv=None) -> int: pr.add_argument("--json", dest="as_json", action="store_true") pr.add_argument("--fail-if-vulnerable", action="store_true", help="exit 3 if the verdict is STILL_VULNERABLE (for CI/monitoring)") + _add_scope_args(pr) pa = sub.add_parser("all", help="sweep every PoC bundle under a root") pa.add_argument("root", nargs="?", default=os.path.join(poc._REPO, "findings"), @@ -355,16 +431,25 @@ def main(argv=None) -> int: pa.add_argument("--json", dest="as_json", action="store_true") pa.add_argument("--fail-if-vulnerable", action="store_true", help="exit 3 if any verdict is STILL_VULNERABLE") + _add_scope_args(pa) + pa.add_argument("--no-scope", action="store_true", + help="explicitly re-issue traffic to ALL stored targets with no " + "scope check (dangerous — you assert every target is in scope)") args = ap.parse_args(argv) if args.cmd == "run": + scope = build_scope(args.scope, args.scope_file) + if scope is None: + print("[!] no scope provided — replaying only this chosen bundle without a " + "scope check. Pass --scope/--scope-file to gate the send.", file=sys.stderr) if args.marker: try: save_marker(args.bundle_dir, args.marker) except OSError: pass - result = replay_one(args.bundle_dir, args.timeout, args.confirm_unsafe, args.marker) + result = replay_one(args.bundle_dir, args.timeout, args.confirm_unsafe, args.marker, + scope=scope) try: _write_replay_json(args.bundle_dir, result) except OSError: @@ -376,13 +461,22 @@ def main(argv=None) -> int: return 3 if (args.fail_if_vulnerable and result["verdict"] == STILL_VULNERABLE) else 0 if args.cmd == "all": + scope = build_scope(args.scope, args.scope_file) + if scope is None and not args.no_scope: + # Critical Rule #1: never mass re-attack every stored target blind. + print("[!] refusing to sweep-replay every stored target with NO scope check " + "(Critical Rule #1: read full scope before touching any asset).", + file=sys.stderr) + print("[!] pass --scope '*.target.com' / --scope-file to gate, or " + "--no-scope to explicitly override.", file=sys.stderr) + return 2 bundles = find_bundles(args.root) if not bundles: print(f"[!] no PoC bundles found under {args.root}. Create one with /poc.") return 0 results = [] for d in bundles: - r = replay_one(d, args.timeout, args.confirm_unsafe, None) + r = replay_one(d, args.timeout, args.confirm_unsafe, None, scope=scope) try: _write_replay_json(d, r) except OSError: diff --git a/commands/replay.md b/commands/replay.md index 9578cd8f..814ef619 100644 --- a/commands/replay.md +++ b/commands/replay.md @@ -30,6 +30,8 @@ tools/replay.py all [root] # default root: findings/ | 🟢 `FIXED` | the signal is gone | | 🟡 `CHANGED` | response differs from the PoC — a human should look | | ⚫ `UNREACHABLE` | the host/endpoint no longer answers | +| 🛑 `BLOCKED` | the SSRF guard refused the request (initial host or a redirect went internal) — a live signal, **not** "fixed" | +| 🚫 `OUT_OF_SCOPE` | the target isn't in the supplied scope — not sent | | ⚪ `INDETERMINATE` | can't judge (no marker + no baseline, or a missing secret) | | ⏭ `SKIPPED_UNSAFE` | a mutating method (PUT/DELETE/PATCH) — needs `--confirm-unsafe` | @@ -46,7 +48,17 @@ tools/replay.py all [root] # default root: findings/ ## Safety -- Replays through the repo's SSRF-guarded opener (`tools/safe_http.py`). +- **Scope gate (Critical Rule #1).** Pass `--scope '*.target.com'` (repeatable) or + `--scope-file `; every send is checked with `scope_checker.py` and + out-of-scope targets are `OUT_OF_SCOPE` and never sent. `all` **refuses to run** + with no scope unless you pass `--no-scope` to explicitly assert every stored + target is in scope. +- **SSRF on the first hop.** The bundle URL is untrusted, so the *initial* host is + validated with the same blocklist `safe_http` uses on redirects + (metadata/private/loopback) **before any request** — a tampered bundle can't make + the first hop hit `169.254.169.254`. Redirect/guard errors are reported as + `BLOCKED`, distinct from `UNREACHABLE`, so a bug that now 302s internal isn't + masked as fixed. - **`PUT`/`DELETE`/`PATCH` are skipped unless you pass `--confirm-unsafe`** — replaying a mutating request could re-trigger the action. - Redacted bundles reference secrets via env vars; replay reads them from the diff --git a/tests/test_replay.py b/tests/test_replay.py index d8b3df82..bdf38dfe 100644 --- a/tests/test_replay.py +++ b/tests/test_replay.py @@ -8,22 +8,34 @@ import json import os +import urllib.error import pytest import poc_bundler as poc import replay as rp +# The real initial-host SSRF guard, captured before the autouse fixture stubs it. +_REAL_HOST_GUARD = rp.initial_host_blocked + + +@pytest.fixture(autouse=True) +def _offline_ssrf_guard(monkeypatch): + """Keep unit tests offline/fast: stub the initial-host guard to allow by + default (it would otherwise do a DNS lookup). SSRF-specific tests restore the + real guard via _REAL_HOST_GUARD.""" + monkeypatch.setattr(rp, "initial_host_blocked", lambda url: False) + # --------------------------------------------------------------------------- # Helpers — build a real PoC bundle offline, then replay it. # --------------------------------------------------------------------------- -def make_bundle(tmp_path, *, method="GET", body="", auth=True, +def make_bundle(tmp_path, *, method="GET", body="", auth=True, host="api.acme.com", resp_status="200 OK", resp_body='{"email":"victim@acme.com"}'): tmp_path.mkdir(parents=True, exist_ok=True) - req_lines = [f"{method} /users/1 HTTP/1.1", "Host: api.acme.com"] + req_lines = [f"{method} /users/1 HTTP/1.1", f"Host: {host}"] if auth: req_lines.append("Authorization: Bearer SECRET") req = "\n".join(req_lines) + "\n\n" + body @@ -268,13 +280,107 @@ def test_cli_run_writes_replay_json_and_exit_code(tmp_path, monkeypatch): def test_cli_all_sweep_writes_report(tmp_path, monkeypatch): make_bundle(tmp_path / "b1", auth=False, resp_body="x") monkeypatch.setattr(poc, "capture", fake_capture_returning("x")) - rc = rp.main(["all", str(tmp_path)]) + rc = rp.main(["all", str(tmp_path), "--no-scope"]) assert rc == 0 assert (tmp_path / "regression-report.md").exists() assert (tmp_path / "regression.json").exists() def test_cli_all_empty_root(tmp_path, capsys): - rc = rp.main(["all", str(tmp_path)]) + rc = rp.main(["all", str(tmp_path), "--no-scope"]) assert rc == 0 assert "no PoC bundles" in capsys.readouterr().out + + +# --------------------------------------------------------------------------- +# Review fix 1 — scope gate (Critical Rule #1) +# --------------------------------------------------------------------------- + + +def test_replay_out_of_scope_is_not_sent(tmp_path, monkeypatch): + d = make_bundle(tmp_path, auth=False) + sent = {"n": 0} + monkeypatch.setattr(poc, "capture", lambda *a, **k: sent.__setitem__("n", 1)) + scope = rp.ScopeChecker(domains=["other-company.com"]) # acme.com NOT in scope + r = rp.replay_one(d, timeout=3, confirm_unsafe=False, marker_override=None, + env={}, scope=scope) + assert r["verdict"] == rp.OUT_OF_SCOPE + assert sent["n"] == 0 # no request issued to an out-of-scope target + + +def test_replay_in_scope_proceeds(tmp_path, monkeypatch): + d = make_bundle(tmp_path, auth=False, resp_body="marker-here") + monkeypatch.setattr(poc, "capture", fake_capture_returning("marker-here")) + scope = rp.ScopeChecker(domains=["*.acme.com"]) # api.acme.com IS in scope + r = rp.replay_one(d, timeout=3, confirm_unsafe=False, marker_override="marker-here", + env={}, scope=scope) + assert r["verdict"] == rp.STILL_VULNERABLE + + +def test_cli_all_refuses_without_scope(tmp_path, monkeypatch, capsys): + make_bundle(tmp_path / "b", auth=False, resp_body="x") + monkeypatch.setattr(poc, "capture", fake_capture_returning("x")) + rc = rp.main(["all", str(tmp_path)]) # no --scope, no --no-scope + assert rc == 2 + assert "scope" in capsys.readouterr().err.lower() + + +def test_cli_all_with_scope_runs(tmp_path, monkeypatch): + make_bundle(tmp_path / "b", auth=False, resp_body="x") + monkeypatch.setattr(poc, "capture", fake_capture_returning("x")) + assert rp.main(["all", str(tmp_path), "--scope", "*.acme.com"]) == 0 + + +# --------------------------------------------------------------------------- +# Review fix 2 — SSRF on the initial hop +# --------------------------------------------------------------------------- + + +def test_initial_host_guard_literals(): + assert _REAL_HOST_GUARD("http://169.254.169.254/latest/meta-data") is True + assert _REAL_HOST_GUARD("http://localhost:8080/") is True + assert _REAL_HOST_GUARD("http://127.0.0.1/") is True + assert _REAL_HOST_GUARD("https://8.8.8.8/") is False # public IP, no DNS needed + assert _REAL_HOST_GUARD("") is True # empty host → blocked + + +def test_replay_blocks_internal_initial_host(tmp_path, monkeypatch): + monkeypatch.setattr(rp, "initial_host_blocked", _REAL_HOST_GUARD) # real guard + d = make_bundle(tmp_path, host="169.254.169.254", auth=False) # tampered bundle + sent = {"n": 0} + monkeypatch.setattr(poc, "capture", lambda *a, **k: sent.__setitem__("n", 1)) + r = rp.replay_one(d, timeout=3, confirm_unsafe=False, marker_override="x", env={}) + assert r["verdict"] == rp.BLOCKED + assert sent["n"] == 0 # the first request never fired at cloud metadata + + +# --------------------------------------------------------------------------- +# Review fix 3 — guard/redirect errors are BLOCKED, not UNREACHABLE +# --------------------------------------------------------------------------- + + +def test_verdict_blocked_is_distinct_from_unreachable(): + v = rp.decide_verdict(200, "sha", "m", + rp.Observed(reachable=False, blocked=True, error="ssrf")) + assert v.verdict == rp.BLOCKED + + +def _raise(exc): + def _f(*a, **k): + raise exc + return _f + + +def test_redirect_to_internal_is_blocked_not_unreachable(tmp_path, monkeypatch): + d = make_bundle(tmp_path, auth=False) + monkeypatch.setattr(poc, "capture", _raise(urllib.error.URLError( + "blocked redirect to disallowed host (SSRF guard): '169.254.169.254'"))) + r = rp.replay_one(d, timeout=3, confirm_unsafe=False, marker_override=None, env={}) + assert r["verdict"] == rp.BLOCKED # a bug that now 302s internal is NOT masked as fixed + + +def test_real_connection_error_is_unreachable(tmp_path, monkeypatch): + d = make_bundle(tmp_path, auth=False) + monkeypatch.setattr(poc, "capture", _raise(urllib.error.URLError("Connection refused"))) + r = rp.replay_one(d, timeout=3, confirm_unsafe=False, marker_override="x", env={}) + assert r["verdict"] == rp.UNREACHABLE