diff --git a/CLAUDE.md b/CLAUDE.md index b452741d..2519acce 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 | +| `/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 | ### 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/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`. - 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/oracle.py b/bughunter/tools/oracle.py new file mode 100644 index 00000000..0ed80e57 --- /dev/null +++ b/bughunter/tools/oracle.py @@ -0,0 +1,555 @@ +#!/usr/bin/env python3 +""" +oracle.py — a self-learning lead & finding prioritizer. + +The problem: the hunt-memory flywheel records every finding's *outcome* +(confirmed / rejected / false_positive, payout, severity, vuln_class, tech +stack, endpoint) in journal.jsonl — but nothing learns from it. The lead board +still ranks purely by hardcoded regex priority (high/med/low). So the single +richest asset a hunter builds up — their own track record of what actually pays +on which surfaces — never feeds back into deciding what to hunt next +(docs/TODOS.md TODO-6: the memory→hunt loop "never spins up in practice"). + +Oracle closes that loop. It trains on the hunter's own journal and scores any +lead by P(this turns into a real, valuable finding), replacing the static +regex rank with one that gets sharper every hunt. + +Why Naive Bayes and not a heavy ML stack: + Bug-bounty history is *small* (tens–hundreds of labelled findings). A big + model overfits that and would drag in a dependency this repo deliberately + avoids. A smoothed Naive-Bayes log-odds model is the right tool for the data + regime: it works with little data, produces calibrated-ish probabilities, + is fully explainable (per-feature log-odds you can read off), pure-stdlib, + and deterministic. A cold-start PRIOR (mirroring the current high-value skill + intuition) is blended in and decays as real evidence accumulates, so Oracle + is useful on day one and data-driven by day thirty. + +Commands: + oracle.py train [--memory-dir DIR] [--out model.json] learn from journal, save + report + oracle.py eval [--memory-dir DIR] leave-one-out CV vs baselines + oracle.py rank [--model model.json] score & rank that target's leads + oracle.py score --class idor [--url U] [--tech nginx] score one hypothetical lead + +Everything except file/lead I/O is pure and unit-tested. +""" +from __future__ import annotations + +import argparse +import json +import math +import os +import re +import sys +from dataclasses import dataclass, field +from datetime import datetime, timezone +from urllib.parse import parse_qs, urlparse + +_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 lead_board # noqa: E402 (reuse load_ledger + high-value skills) + +# Outcome → label. Ambiguous results are left out of training entirely. +POSITIVE_RESULTS = {"confirmed"} +NEGATIVE_RESULTS = {"rejected", "false_positive"} + +# When history is thin, blend the learned model with a heuristic prior. The +# learned weight is n/(n+SHRINKAGE_K): ~0 with no data, ~0.8 at 80 samples. +SHRINKAGE_K = 20 +MIN_TRAIN = 6 # below this, output is prior-dominated and flagged as such + +# Endpoint keywords worth a categorical feature (high-signal path/param hints). +_EP_KEYWORDS = ("admin", "api", "graphql", "upload", "login", "oauth", "sso", + "user", "account", "internal", "debug", "token", "file", + "redirect", "password", "reset", "payment", "invoice") + +# Vuln classes that historically pay well — the cold-start prior leans on these +# (normalized, i.e. lead_board's "hunt-idor" -> "idor"). +_HIGH_VALUE_CLASSES = {re.sub(r"^hunt-", "", s) for s in lead_board.HIGH_VALUE} +_SEV_PRIOR = {"critical": 0.22, "high": 0.15, "medium": 0.05, "low": -0.05, + "informational": -0.15, "none": -0.15} + + +def now_iso() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +# --------------------------------------------------------------------------- +# Feature engineering (pure) +# --------------------------------------------------------------------------- + + +def norm_class(value: str | None) -> str: + return re.sub(r"^hunt-", "", (value or "").strip().lower()) + + +def _tech_tokens(tech) -> set[str]: + if not tech: + return set() + if isinstance(tech, (list, tuple, set)): + parts = tech + else: + parts = re.split(r"[,\s/|;]+", str(tech)) + return {f"tech:{re.sub(r'[^a-z0-9.]+', '', p.lower())}" + for p in parts if p and p.strip()} - {"tech:"} + + +def endpoint_tokens(url: str | None) -> set[str]: + """Categorical features describing an endpoint's shape.""" + toks: set[str] = set() + if not url: + return toks + p = urlparse(url if "://" in url else "//" + url) + query = p.query.lower() + if query: + toks.add("ep:has_params") + if re.search(r"=\d+(?:&|$)", query): + toks.add("ep:numeric_id") + path = (p.path or "").lower() + depth = len([s for s in path.split("/") if s]) + toks.add(f"ep:depth_{min(depth, 4)}") + # Host matters too: api./admin./internal. subdomains are strong signals. + hay = (p.netloc or "").lower() + " " + path + " " + query + for kw in _EP_KEYWORDS: + if kw in hay: + toks.add(f"ep:kw_{kw}") + return toks + + +def _tag_tokens(tags) -> set[str]: + if not tags: + return set() + if isinstance(tags, str): + tags = re.split(r"[,\s]+", tags) + return {f"tag:{re.sub(r'[^a-z0-9]+', '', t.lower())}" for t in tags if t} - {"tag:"} + + +def features_from_journal(entry: dict, tech_index: dict[str, set[str]]) -> set[str]: + f: set[str] = set() + cls = norm_class(entry.get("vuln_class")) + if cls: + f.add(f"class:{cls}") + sev = (entry.get("severity") or "").lower() + if sev: + f.add(f"sev:{sev}") + f |= endpoint_tokens(entry.get("endpoint")) + f |= _tag_tokens(entry.get("tags")) + f |= _tech_tokens(entry.get("tech_stack")) + f |= tech_index.get(entry.get("target", ""), set()) + return f + + +def features_from_lead(lead: dict, tech_index: dict[str, set[str]]) -> set[str]: + """Same feature space as journal, derived from a lead_board lead.""" + f: set[str] = set() + cls = norm_class(lead.get("skill")) + if cls: + f.add(f"class:{cls}") + f |= endpoint_tokens(lead.get("evidence")) + f |= tech_index.get(lead.get("target", ""), set()) + return f + + +def label_from_journal(entry: dict) -> int | None: + r = (entry.get("result") or "").lower() + if r in POSITIVE_RESULTS: + return 1 + if r in NEGATIVE_RESULTS: + return 0 + return None + + +# --------------------------------------------------------------------------- +# The model (pure) — smoothed Naive-Bayes log-odds + heuristic prior blend. +# --------------------------------------------------------------------------- + + +def _payout_num(value) -> float: + if value is None: + return 0.0 + if isinstance(value, (int, float)): + return float(value) + m = re.search(r"[\d,]+(?:\.\d+)?", str(value)) + return float(m.group(0).replace(",", "")) if m else 0.0 + + +def prior_score(features: set[str]) -> float: + """Heuristic cold-start probability from class/severity/endpoint shape — + the same intuition the regex prioritizer encodes, expressed as a probability + so it can be blended with the learned model.""" + z = 0.0 + for f in features: + if f.startswith("class:") and f[6:] in _HIGH_VALUE_CLASSES: + z += 0.6 + elif f.startswith("sev:"): + z += _SEV_PRIOR.get(f[4:], 0.0) * 4 + elif f in ("ep:numeric_id", "ep:kw_admin", "ep:kw_graphql", + "ep:kw_upload", "ep:kw_oauth", "ep:kw_internal"): + z += 0.25 + return 1.0 / (1.0 + math.exp(-z)) + + +@dataclass +class OracleModel: + alpha: float = 1.0 + n_pos: int = 0 + n_neg: int = 0 + feat_pos: dict[str, int] = field(default_factory=dict) + feat_neg: dict[str, int] = field(default_factory=dict) + mean_payout: dict[str, float] = field(default_factory=dict) # class -> mean payout (positives) + trained_at: str = "" + + @property + def n(self) -> int: + return self.n_pos + self.n_neg + + def train(self, samples: list[tuple[set[str], int]], + payouts: list[tuple[str, float]] | None = None) -> "OracleModel": + for feats, label in samples: + if label == 1: + self.n_pos += 1 + dst = self.feat_pos + else: + self.n_neg += 1 + dst = self.feat_neg + for f in feats: + dst[f] = dst.get(f, 0) + 1 + if payouts: + acc: dict[str, list[float]] = {} + for cls, amt in payouts: + acc.setdefault(cls, []).append(amt) + self.mean_payout = {c: sum(v) / len(v) for c, v in acc.items() if v} + self.trained_at = now_iso() + return self + + def _prior_logodds(self) -> float: + return math.log((self.n_pos + self.alpha) / (self.n_neg + self.alpha)) + + def contributions(self, features: set[str]) -> dict[str, float]: + """Per-feature log-odds contribution — the model's explanation.""" + out = {} + for f in features: + p1 = (self.feat_pos.get(f, 0) + self.alpha) / (self.n_pos + 2 * self.alpha) + p0 = (self.feat_neg.get(f, 0) + self.alpha) / (self.n_neg + 2 * self.alpha) + out[f] = math.log(p1 / p0) + return out + + def _learned_score(self, features: set[str]) -> float: + z = self._prior_logodds() + sum(self.contributions(features).values()) + z = max(-30.0, min(30.0, z)) + return 1.0 / (1.0 + math.exp(-z)) + + def score(self, features: set[str]) -> float: + """Blend the learned score with the cold-start prior by evidence weight.""" + learned = self._learned_score(features) + w = self.n / (self.n + SHRINKAGE_K) # 0 with no data → ~1 with lots + return w * learned + (1 - w) * prior_score(features) + + def confidence(self) -> str: + if self.n >= MIN_TRAIN * 5: + return "learned" + if self.n >= MIN_TRAIN: + return "blended" + return "prior" + + def explain(self, features: set[str], k: int = 5) -> list[dict]: + contribs = self.contributions(features) + rows = sorted(contribs.items(), key=lambda kv: abs(kv[1]), reverse=True)[:k] + return [{"feature": f, "logodds": round(c, 3), + "pos": self.feat_pos.get(f, 0), "neg": self.feat_neg.get(f, 0)} + for f, c in rows] + + def expected_value(self, features: set[str]) -> float: + cls = next((f[6:] for f in features if f.startswith("class:")), "") + return self.score(features) * self.mean_payout.get(cls, 0.0) + + def to_dict(self) -> dict: + return {"schema": "oracle-model/1", "alpha": self.alpha, + "n_pos": self.n_pos, "n_neg": self.n_neg, + "feat_pos": self.feat_pos, "feat_neg": self.feat_neg, + "mean_payout": self.mean_payout, "trained_at": self.trained_at} + + @classmethod + def from_dict(cls, d: dict) -> "OracleModel": + m = cls(alpha=d.get("alpha", 1.0)) + m.n_pos, m.n_neg = d.get("n_pos", 0), d.get("n_neg", 0) + m.feat_pos = dict(d.get("feat_pos", {})) + m.feat_neg = dict(d.get("feat_neg", {})) + m.mean_payout = dict(d.get("mean_payout", {})) + m.trained_at = d.get("trained_at", "") + return m + + +# --------------------------------------------------------------------------- +# Data loading +# --------------------------------------------------------------------------- + + +def _read_jsonl(path: str) -> list[dict]: + out = [] + if not os.path.isfile(path): + return out + with open(path, encoding="utf-8", errors="replace") as fh: + for line in fh: + line = line.strip() + if line: + try: + obj = json.loads(line) + if isinstance(obj, dict): + out.append(obj) + except ValueError: + pass + return out + + +def build_tech_index(patterns: list[dict]) -> dict[str, set[str]]: + idx: dict[str, set[str]] = {} + for p in patterns: + tgt = p.get("target") + if tgt: + idx.setdefault(tgt, set()).update(_tech_tokens(p.get("tech_stack"))) + return idx + + +def build_training_set(journal: list[dict], tech_index: dict[str, set[str]] + ) -> tuple[list[tuple[set[str], int]], list[tuple[str, float]]]: + samples, payouts = [], [] + for e in journal: + label = label_from_journal(e) + if label is None: + continue + feats = features_from_journal(e, tech_index) + if not feats: + continue + samples.append((feats, label)) + if label == 1: + payouts.append((norm_class(e.get("vuln_class")), _payout_num(e.get("payout")))) + return samples, payouts + + +def load_memory(memory_dir: str) -> tuple[list[dict], list[dict]]: + journal = _read_jsonl(os.path.join(memory_dir, "journal.jsonl")) + patterns = _read_jsonl(os.path.join(memory_dir, "patterns.jsonl")) + return journal, patterns + + +# --------------------------------------------------------------------------- +# Evaluation (pure) — honest metrics + baselines to prove Oracle earns its keep. +# --------------------------------------------------------------------------- + + +def _metrics(pairs: list[tuple[int, int]]) -> dict: + """pairs = list of (predicted, actual) in {0,1}.""" + tp = sum(1 for p, a in pairs if p == 1 and a == 1) + fp = sum(1 for p, a in pairs if p == 1 and a == 0) + tn = sum(1 for p, a in pairs if p == 0 and a == 0) + fn = sum(1 for p, a in pairs if p == 0 and a == 1) + n = len(pairs) or 1 + prec = tp / (tp + fp) if (tp + fp) else 0.0 + rec = tp / (tp + fn) if (tp + fn) else 0.0 + f1 = 2 * prec * rec / (prec + rec) if (prec + rec) else 0.0 + return {"n": len(pairs), "accuracy": round((tp + tn) / n, 3), + "precision": round(prec, 3), "recall": round(rec, 3), + "f1": round(f1, 3), "tp": tp, "fp": fp, "tn": tn, "fn": fn} + + +def loocv(samples: list[tuple[set[str], int]], threshold: float = 0.5) -> dict: + """Leave-one-out cross-validation — the honest way to score a small model. + + The primary "oracle" number scores each held-out fold with the DEPLOYED + scorer — `model.score()`, the same blend of learned + cold-start prior that + `rank`/`score` actually serve (weight w = n/(n+SHRINKAGE_K)). This is what a + user with that much history really gets, so the reported lift never overstates + reality. `oracle_learned_only` (pure `_learned_score`, w=1) is reported + separately as a learned-only *upper bound* it converges to as data grows. + Baselines: the heuristic prior alone, and majority class.""" + if len(samples) < 3: + return {"error": "need >=3 labelled samples", "n": len(samples)} + deployed_pairs, learned_pairs, prior_pairs = [], [], [] + n_pos = sum(1 for _, y in samples if y == 1) + majority = 1 if n_pos * 2 >= len(samples) else 0 + for i in range(len(samples)): + train = samples[:i] + samples[i + 1:] + feats, actual = samples[i] + m = OracleModel().train(train) + deployed_pairs.append((1 if m.score(feats) >= threshold else 0, actual)) # what users get + learned_pairs.append((1 if m._learned_score(feats) >= threshold else 0, actual)) # upper bound + prior_pairs.append((1 if prior_score(feats) >= threshold else 0, actual)) + majority_pairs = [(majority, a) for _, a in samples] + return { + "oracle": _metrics(deployed_pairs), # deployed (blended) — the honest number + "oracle_learned_only": _metrics(learned_pairs), # learned-only upper bound + "baseline_prior": _metrics(prior_pairs), + "baseline_majority": _metrics(majority_pairs), + "positives": n_pos, "negatives": len(samples) - n_pos, + } + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def default_memory_dir() -> str: + return os.environ.get("BBHUNT_MEMORY_DIR", "hunt-memory") + + +def _fit(memory_dir: str) -> tuple[OracleModel, list, dict]: + journal, patterns = load_memory(memory_dir) + tech_index = build_tech_index(patterns) + samples, payouts = build_training_set(journal, tech_index) + model = OracleModel().train(samples, payouts) + return model, samples, tech_index + + +def cmd_train(args) -> int: + model, samples, _ = _fit(args.memory_dir) + out = args.out or os.path.join(args.memory_dir, "oracle_model.json") + if not samples: + print(f"[!] no labelled findings in {args.memory_dir}/journal.jsonl " + "(need results of 'confirmed' / 'rejected' / 'false_positive').") + print("[*] Oracle will run on its cold-start prior until you log outcomes.") + os.makedirs(args.memory_dir, exist_ok=True) + with open(out, "w", encoding="utf-8", newline="\n") as fh: + json.dump(model.to_dict(), fh, indent=2) + print(f"[+] trained on {model.n} labelled findings " + f"({model.n_pos} confirmed / {model.n_neg} rejected) — confidence: {model.confidence()}") + print(f"[+] model saved: {out}") + ev = loocv(samples) + if "error" not in ev: + _print_eval(ev) + return 0 + + +def _print_eval(ev: dict) -> None: + o, p, mj = ev["oracle"], ev["baseline_prior"], ev["baseline_majority"] + ub = ev.get("oracle_learned_only", o) + print(f"\n leave-one-out CV ({ev['positives']} confirmed / {ev['negatives']} rejected):") + print(f" {'model':<26}{'acc':>6}{'prec':>7}{'rec':>7}{'f1':>7}") + # The deployed (blended) scorer is the honest headline — it's what rank/score serve. + print(f" {'Oracle (deployed)':<26}{o['accuracy']:>6}{o['precision']:>7}{o['recall']:>7}{o['f1']:>7}") + print(f" {'Oracle (learned-only, UB)':<26}{ub['accuracy']:>6}{ub['precision']:>7}{ub['recall']:>7}{ub['f1']:>7}") + print(f" {'prior heuristic':<26}{p['accuracy']:>6}{p['precision']:>7}{p['recall']:>7}{p['f1']:>7}") + print(f" {'majority class':<26}{mj['accuracy']:>6}{'—':>7}{'—':>7}{'—':>7}") + lift = round(o["accuracy"] - p["accuracy"], 3) + print(f" → Oracle (deployed) vs prior heuristic: {'+' if lift >= 0 else ''}{lift} accuracy") + print(" (learned-only UB = upper bound the deployed blend converges to as history grows)") + + +def cmd_eval(args) -> int: + _, samples, _ = _fit(args.memory_dir) + ev = loocv(samples) + if "error" in ev: + print(f"[!] {ev['error']} (have {ev['n']}). Log more outcomes with /remember.") + return 0 + _print_eval(ev) + return 0 + + +def _load_model(args) -> OracleModel: + path = getattr(args, "model", None) or os.path.join(args.memory_dir, "oracle_model.json") + if os.path.isfile(path): + with open(path, encoding="utf-8") as fh: + return OracleModel.from_dict(json.load(fh)) + # fall back to training on the fly + return _fit(args.memory_dir)[0] + + +def cmd_rank(args) -> int: + model = _load_model(args) + _, patterns = load_memory(args.memory_dir) + tech_index = build_tech_index(patterns) + leads = [l for l in lead_board.load_ledger(args.target) if l.get("status") == "new"] + if not leads: + print(f"[!] no untouched leads for {args.target}. Run lead_board.py ingest first.") + return 0 + scored = [] + for l in leads: + feats = features_from_lead(l, tech_index) + scored.append((model.score(feats), model.expected_value(feats), l, feats)) + scored.sort(key=lambda x: (x[0], x[1]), reverse=True) + + if args.as_json: + print(json.dumps([{"score": round(s, 3), "ev": round(ev, 1), + "id": l.get("id"), "skill": l.get("skill"), + "evidence": l.get("evidence"), + "explain": model.explain(f)} for s, ev, l, f in scored], indent=2)) + return 0 + + print(f"\n=== ORACLE RANK: {args.target} — {len(scored)} leads " + f"(model confidence: {model.confidence()}) ===") + for s, ev, l, feats in scored[:args.top]: + evs = f" ~${ev:,.0f}" if ev else "" + print(f" {s*100:5.1f}%{evs} {l.get('skill',''):<18} {str(l.get('evidence',''))[:56]}") + top = model.explain(feats, k=3) + if top and model.confidence() != "prior": + why = ", ".join(f"{r['feature']}({'+' if r['logodds']>=0 else ''}{r['logodds']})" + for r in top) + print(f" why: {why}") + return 0 + + +def cmd_score(args) -> int: + model = _load_model(args) + feats = {f"class:{norm_class(args.vuln_class)}"} if args.vuln_class else set() + feats |= endpoint_tokens(args.url) + for t in (args.tech or []): + feats |= _tech_tokens(t) + if args.severity: + feats.add(f"sev:{args.severity.lower()}") + score = model.score(feats) + result = {"score": round(score, 4), "confidence": model.confidence(), + "expected_value": round(model.expected_value(feats), 2), + "features": sorted(feats), "explain": model.explain(feats)} + if args.as_json: + print(json.dumps(result, indent=2)) + else: + print(f"P(worth pursuing) = {score*100:.1f}% ({model.confidence()})") + if result["expected_value"]: + print(f"expected value = ~${result['expected_value']:,.0f}") + for r in result["explain"]: + print(f" {r['feature']:<22} {'+' if r['logodds']>=0 else ''}{r['logodds']:>6}" + f" (seen {r['pos']}✓ / {r['neg']}✗)") + return 0 + + +def main(argv=None) -> int: + for _stream in (sys.stdout, sys.stderr): + try: + _stream.reconfigure(encoding="utf-8", errors="replace") + except (AttributeError, ValueError): + pass + + ap = argparse.ArgumentParser(description="Oracle — self-learning lead/finding prioritizer.") + ap.add_argument("--memory-dir", default=default_memory_dir(), + help="hunt-memory dir with journal.jsonl / patterns.jsonl") + sub = ap.add_subparsers(dest="cmd", required=True) + + pt = sub.add_parser("train", help="learn from journal history, save + report") + pt.add_argument("--out", help="model output path (default: /oracle_model.json)") + + sub.add_parser("eval", help="leave-one-out CV vs baselines") + + prk = sub.add_parser("rank", help="score & rank a target's untouched leads") + prk.add_argument("target") + prk.add_argument("--model", help="model path (default: /oracle_model.json)") + prk.add_argument("--top", type=int, default=20) + prk.add_argument("--json", dest="as_json", action="store_true") + + psc = sub.add_parser("score", help="score one hypothetical lead") + psc.add_argument("--class", dest="vuln_class", help="vuln class / skill, e.g. idor") + psc.add_argument("--url", help="candidate endpoint") + psc.add_argument("--tech", action="append", help="tech token(s), repeatable") + psc.add_argument("--severity", help="critical/high/medium/low") + psc.add_argument("--model", help="model path") + psc.add_argument("--json", dest="as_json", action="store_true") + + args = ap.parse_args(argv) + return {"train": cmd_train, "eval": cmd_eval, + "rank": cmd_rank, "score": cmd_score}[args.cmd](args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/commands/oracle.md b/commands/oracle.md new file mode 100644 index 00000000..c55ae056 --- /dev/null +++ b/commands/oracle.md @@ -0,0 +1,60 @@ +--- +description: Self-learning lead/finding prioritizer — trains on your own hunt-memory outcomes to predict which leads are worth pursuing, replacing the static regex priority. Usage: /oracle train | /oracle rank | /oracle score --class idor --url | /oracle eval +--- + +# /oracle + +Turn the hunt-memory flywheel into a decision engine. Every finding's outcome +(confirmed / rejected / false_positive, payout, severity, vuln class, endpoint, +tech stack) already lives in `journal.jsonl` — Oracle learns from it and scores +any lead by **P(this becomes a real, valuable finding)**, so you work the leads +your own track record says pay, not the ones a hardcoded regex guesses. + +## Usage + +``` +/oracle train # learn from history, save model, print eval +/oracle eval # leave-one-out CV vs baselines (no save) +/oracle rank target.com # score & rank that target's untouched leads +/oracle score --class idor --url https://api.t.com/v1/o?id=9 --severity high +``` + +Run directly: + +```bash +tools/oracle.py train +tools/oracle.py rank target.com --top 20 +tools/oracle.py score --class open-redirect --url https://t.com/go?next=/x --json +``` + +## What it learns + +Features per finding: vuln class, severity, endpoint shape (`numeric_id`, +`api`/`admin`/`graphql`/`upload`/`oauth` keywords, path depth, host), tags, and +the target's tech stack (from `patterns.jsonl`). Label: `confirmed` → positive, +`rejected`/`false_positive` → negative; ambiguous results are excluded. + +## The model (and why this one) + +Bug-bounty history is small, so Oracle uses a **smoothed Naive-Bayes log-odds +model** — the right tool for little data: calibrated, fully explainable +(every score comes with the per-feature log-odds that produced it), pure-stdlib, +deterministic. A **cold-start prior** (the same high-value-class intuition the +regex scorer encodes) is blended in and **decays as real evidence accumulates** +(`prior` → `blended` → `learned`), so Oracle is useful on day one and +data-driven by day thirty. + +## Why it beats the static prioritizer + +The regex priority is the same for every program. Oracle learns **program-specific +reality** — e.g. that IDOR is always a duplicate on one program (down-rank it) or +that open-redirect chains to ATO and pays on another (up-rank it). `oracle eval` +proves this on *your* data with leave-one-out CV against two baselines (the prior +heuristic and majority class), so you can see the lift before you trust it. + +## Output + +`rank` prints each lead's probability, expected value (`P × mean historical +payout for that class`), and the top features driving the score. `score` explains +a single hypothetical lead. `train` saves `oracle_model.json` into the memory dir. +Read-only — Oracle never mutates the lead board. diff --git a/tests/test_oracle.py b/tests/test_oracle.py new file mode 100644 index 00000000..2fa8bd03 --- /dev/null +++ b/tests/test_oracle.py @@ -0,0 +1,291 @@ +"""Tests for tools/oracle.py — the self-learning prioritizer. + +The whole model/feature/eval layer is pure and tested here without touching the +network or the real hunt-memory. `bughunter/tools` is on sys.path via conftest. +""" + +import json +import os + +import pytest + +import oracle as oc + + +# --------------------------------------------------------------------------- +# Feature engineering +# --------------------------------------------------------------------------- + + +def test_norm_class_strips_hunt_prefix(): + assert oc.norm_class("hunt-idor") == "idor" + assert oc.norm_class("IDOR") == "idor" + assert oc.norm_class(None) == "" + + +def test_endpoint_tokens(): + toks = oc.endpoint_tokens("https://api.acme.com/v1/users/1?id=5&admin=1") + assert "ep:has_params" in toks + assert "ep:numeric_id" in toks + assert "ep:kw_api" in toks + assert "ep:kw_user" in toks + assert any(t.startswith("ep:depth_") for t in toks) + + +def test_endpoint_tokens_empty(): + assert oc.endpoint_tokens("") == set() + assert oc.endpoint_tokens(None) == set() + + +def test_tech_tokens_handles_list_and_string(): + assert oc._tech_tokens("nginx, PHP") == {"tech:nginx", "tech:php"} + assert oc._tech_tokens(["Next.js", "node"]) == {"tech:next.js", "tech:node"} + assert oc._tech_tokens(None) == set() + + +def test_features_from_lead_maps_skill_to_class(): + lead = {"skill": "hunt-idor", "evidence": "https://acme.com/api/order?id=3", + "target": "acme.com"} + feats = oc.features_from_lead(lead, {"acme.com": {"tech:nginx"}}) + assert "class:idor" in feats + assert "ep:numeric_id" in feats + assert "tech:nginx" in feats + + +def test_label_from_journal(): + assert oc.label_from_journal({"result": "confirmed"}) == 1 + assert oc.label_from_journal({"result": "rejected"}) == 0 + assert oc.label_from_journal({"result": "false_positive"}) == 0 + assert oc.label_from_journal({"result": "partial"}) is None # ambiguous → excluded + assert oc.label_from_journal({"result": "informational"}) is None + + +# --------------------------------------------------------------------------- +# The model learns a real signal +# --------------------------------------------------------------------------- + + +def _signal_samples(n=40): + """idor+numeric_id => confirmed; xss without signal => rejected.""" + samples = [] + for i in range(n): + if i % 2 == 0: + samples.append(({"class:idor", "ep:numeric_id", "sev:high"}, 1)) + else: + samples.append(({"class:xss", "ep:has_params", "sev:low"}, 0)) + return samples + + +def test_model_learns_positive_signal(): + m = oc.OracleModel().train(_signal_samples()) + good = m.score({"class:idor", "ep:numeric_id", "sev:high"}) + bad = m.score({"class:xss", "ep:has_params", "sev:low"}) + assert good > 0.7 + assert bad < 0.3 + assert good > bad + + +def test_scores_are_probabilities(): + m = oc.OracleModel().train(_signal_samples()) + for feats in ({"class:idor", "ep:numeric_id"}, {"class:xss"}, set(), {"unknown:feat"}): + s = m.score(feats) + assert 0.0 <= s <= 1.0 + + +def test_explanation_points_at_the_learned_feature(): + m = oc.OracleModel().train(_signal_samples()) + top = m.explain({"class:idor", "ep:numeric_id", "sev:high"}, k=3) + # the strongest positive contributor should be one of the idor-signal features + best = max(top, key=lambda r: r["logodds"]) + assert best["feature"] in {"class:idor", "ep:numeric_id", "sev:high"} + assert best["logodds"] > 0 + + +def test_expected_value_uses_mean_payout(): + samples = [({"class:idor", "ep:numeric_id"}, 1)] * 10 + [({"class:xss"}, 0)] * 10 + payouts = [("idor", 500.0)] * 10 + m = oc.OracleModel().train(samples, payouts) + ev = m.expected_value({"class:idor", "ep:numeric_id"}) + assert 0 < ev <= 500.0 + + +def test_serialization_round_trip(): + m = oc.OracleModel().train(_signal_samples()) + m2 = oc.OracleModel.from_dict(json.loads(json.dumps(m.to_dict()))) + feats = {"class:idor", "ep:numeric_id", "sev:high"} + assert abs(m.score(feats) - m2.score(feats)) < 1e-9 + + +# --------------------------------------------------------------------------- +# Cold-start prior + blending +# --------------------------------------------------------------------------- + + +def test_prior_favors_high_value_class(): + assert oc.prior_score({"class:idor"}) > oc.prior_score({"class:misc"}) + assert oc.prior_score({"sev:critical"}) > oc.prior_score({"sev:low"}) + + +def test_empty_model_falls_back_to_prior(): + m = oc.OracleModel() # untrained + assert m.confidence() == "prior" + # with no learned data, score should equal the prior + feats = {"class:idor", "ep:numeric_id"} + assert abs(m.score(feats) - oc.prior_score(feats)) < 1e-9 + + +def test_confidence_grows_with_data(): + assert oc.OracleModel().train(_signal_samples(4)).confidence() in ("prior", "blended") + assert oc.OracleModel().train(_signal_samples(40)).confidence() in ("blended", "learned") + assert oc.OracleModel().train(_signal_samples(60)).confidence() == "learned" + + +# --------------------------------------------------------------------------- +# Evaluation: Oracle must beat the baselines on a learnable signal +# --------------------------------------------------------------------------- + + +def test_loocv_oracle_beats_majority_on_signal(): + ev = oc.loocv(_signal_samples(40)) + assert ev["oracle"]["accuracy"] >= 0.9 + assert ev["oracle"]["accuracy"] >= ev["baseline_majority"]["accuracy"] + + +def test_loocv_reports_deployed_and_learned_upper_bound(): + """The headline 'oracle' number must be the DEPLOYED (blended) scorer users + actually get, with a separate learned-only upper bound — so the reported lift + never overstates production behaviour.""" + ev = oc.loocv(_signal_samples(40)) + assert "oracle" in ev and "oracle_learned_only" in ev + # both are valid metric dicts + for key in ("oracle", "oracle_learned_only"): + assert 0.0 <= ev[key]["accuracy"] <= 1.0 + # the learned-only bound is never worse than the blended deployed score on a + # clean signal (the blend only pulls toward the prior) + assert ev["oracle_learned_only"]["accuracy"] >= ev["oracle"]["accuracy"] - 1e-9 + + +def test_loocv_deployed_matches_model_score_not_learned_only(monkeypatch): + """Regression guard for the review fix: LOOCV must call the deployed score(), + not _learned_score(). We tag which path was used and assert it's score().""" + calls = {"score": 0, "learned": 0} + real_score = oc.OracleModel.score + real_learned = oc.OracleModel._learned_score + + def spy_score(self, feats): + calls["score"] += 1 + return real_score(self, feats) + + def spy_learned(self, feats): + calls["learned"] += 1 + return real_learned(self, feats) + + monkeypatch.setattr(oc.OracleModel, "score", spy_score) + monkeypatch.setattr(oc.OracleModel, "_learned_score", spy_learned) + oc.loocv(_signal_samples(6)) + assert calls["score"] > 0 # the deployed scorer is exercised per fold + + +def test_loocv_needs_min_samples(): + assert "error" in oc.loocv([({"a"}, 1), ({"b"}, 0)]) + + +# --------------------------------------------------------------------------- +# Data loading + training set +# --------------------------------------------------------------------------- + + +def _write_memory(tmp_path, journal_rows, pattern_rows=()): + d = tmp_path / "hunt-memory" + d.mkdir() + with open(d / "journal.jsonl", "w", encoding="utf-8") as fh: + for r in journal_rows: + fh.write(json.dumps(r) + "\n") + if pattern_rows: + with open(d / "patterns.jsonl", "w", encoding="utf-8") as fh: + for r in pattern_rows: + fh.write(json.dumps(r) + "\n") + return str(d) + + +def test_build_tech_index(): + patterns = [{"target": "acme.com", "tech_stack": "nginx, php"}, + {"target": "acme.com", "tech_stack": ["laravel"]}] + idx = oc.build_tech_index(patterns) + assert idx["acme.com"] == {"tech:nginx", "tech:php", "tech:laravel"} + + +def test_build_training_set_skips_ambiguous_and_featureless(): + journal = [ + {"vuln_class": "idor", "endpoint": "https://a/x?id=1", "result": "confirmed", "payout": 300}, + {"vuln_class": "xss", "endpoint": "https://a/y?q=1", "result": "rejected"}, + {"vuln_class": "ssrf", "endpoint": "https://a/z", "result": "partial"}, # ambiguous + ] + samples, payouts = oc.build_training_set(journal, {}) + assert len(samples) == 2 + assert ("idor", 300.0) in payouts + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def _big_journal(n=30): + rows = [] + for i in range(n): + if i % 2 == 0: + rows.append({"vuln_class": "idor", "endpoint": "https://a/api/o?id=2", + "result": "confirmed", "severity": "high", "payout": 400, + "target": "a", "action": "hunt", "ts": "t", "schema_version": 2}) + else: + rows.append({"vuln_class": "xss", "endpoint": "https://a/s?q=1", + "result": "rejected", "severity": "low", + "target": "a", "action": "hunt", "ts": "t", "schema_version": 2}) + return rows + + +def test_cli_train_saves_model(tmp_path): + mem = _write_memory(tmp_path, _big_journal()) + rc = oc.main(["--memory-dir", mem, "train"]) + assert rc == 0 + model = json.loads((tmp_path / "hunt-memory" / "oracle_model.json").read_text(encoding="utf-8")) + assert model["schema"] == "oracle-model/1" + assert model["n_pos"] > 0 and model["n_neg"] > 0 + + +def test_cli_train_no_data_is_graceful(tmp_path, capsys): + mem = _write_memory(tmp_path, []) + rc = oc.main(["--memory-dir", mem, "train"]) + assert rc == 0 + assert "cold-start" in capsys.readouterr().out + + +def test_cli_score_json(tmp_path, capsys): + mem = _write_memory(tmp_path, _big_journal()) + oc.main(["--memory-dir", mem, "train"]) + capsys.readouterr() # discard training output before capturing JSON + rc = oc.main(["--memory-dir", mem, "score", "--class", "idor", + "--url", "https://a/api/o?id=9", "--json"]) + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert 0.0 <= out["score"] <= 1.0 + assert out["confidence"] in ("prior", "blended", "learned") + + +def test_cli_rank_uses_lead_board(tmp_path, monkeypatch, capsys): + mem = _write_memory(tmp_path, _big_journal()) + oc.main(["--memory-dir", mem, "train"]) + leads = [ + {"id": "lb-1", "skill": "hunt-idor", "evidence": "https://a/api/o?id=1", + "target": "a", "status": "new"}, + {"id": "lb-2", "skill": "hunt-xss", "evidence": "https://a/s?q=x", + "target": "a", "status": "new"}, + ] + monkeypatch.setattr(oc.lead_board, "load_ledger", lambda t: leads) + capsys.readouterr() # discard training output before capturing JSON + rc = oc.main(["--memory-dir", mem, "rank", "a", "--json"]) + assert rc == 0 + ranked = json.loads(capsys.readouterr().out) + assert ranked[0]["skill"] == "hunt-idor" # learned high-value lead ranks first + assert ranked[0]["score"] >= ranked[1]["score"]