From abf28dfdb1769d0ecb7f1aa0d880e9d41cf385a0 Mon Sep 17 00:00:00 2001 From: henleda Date: Mon, 27 Jul 2026 11:38:17 -0500 Subject: [PATCH] feat(refiner): the refine loop now gates on blast radius too (G3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `refiner.py` retried until the band-aid blocked the exploit. A refinement that widened the policy far enough to catch the exploit had nothing telling it it went too far, so the loop would happily accept a rule that denies production. With a traffic sample, "works" now means two things: the policy blocks the exploit AND stays under the blast-radius threshold. An over-broad candidate is fed back as `over_block` — a diagnosis the refine agent already understands ("blocked the LEGIT request too. Narrow the rule") — with the measured rate and the top blocked paths attached, so it knows what it over-blocked rather than guessing. Measured where the policy is already attached. The refiner has just confirmed the candidate is enforcing, so `simulate.measure_attached` replays through that LB instead of paying a second attach and propagation wait on another one: slower-but-correct per the decision, but one round trip per attempt rather than two. `_score` is now shared by `simulate_policies` and `measure_attached`, so the standalone run and the refine gate cannot drift on what "too broad" means. The sample comes from `VPCOPILOT_SIM_LOGS` or an explicit `records=`. With neither, the gate is skipped and the loop behaves exactly as it did before — pinned by a test that fails if anything is replayed without a sample. An unreadable sample logs a warning and refines without the gate rather than failing a working loop. Tests cover both directions: an over-broad candidate is rejected and refined (and the agent is handed the blast radius), a surgical one passes on the first attempt, an unfixable over-block gives up honestly with a reason, the threshold is configurable, and the no-sample path replays nothing. 250 pass, coverage 73%. Co-Authored-By: Claude Opus 5 (1M context) --- .env.example | 1 + ROADMAP.md | 19 +++- docs/USAGE.md | 5 + src/vpcopilot/console/static/index.html | 1 + src/vpcopilot/refiner.py | 77 ++++++++++++-- src/vpcopilot/simulate.py | 59 +++++++---- tests/test_refiner.py | 132 ++++++++++++++++++++++++ 7 files changed, 262 insertions(+), 32 deletions(-) diff --git a/.env.example b/.env.example index 84c7255..40567da 100644 --- a/.env.example +++ b/.env.example @@ -24,6 +24,7 @@ GITHUB_TOKEN= # repo-scoped PAT for opening PRs (or use `gh auth login`) # VPCOPILOT_REQUIRE_PROBE=1 # fail-closed: don't validate a finding that has no derived probe # VPCOPILOT_ACTOR=jane.doe # audit-log attribution — default: the OS user; set it in CI # VPCOPILOT_SIM_THRESHOLD=0.01 # would-block rate that flags a policy in `vpcopilot simulate` +# VPCOPILOT_SIM_LOGS=traffic.jsonl # traffic sample the refiner also checks each candidate against # --- Validation auth (auth-protected targets: let the probe log in so it can demonstrate the exploit) --- # VPCOPILOT_PROBE_USER= # username the validation probe logs in with (cookie or token app) # VPCOPILOT_PROBE_PASS= # its password diff --git a/ROADMAP.md b/ROADMAP.md index ae7a3cb..b54528c 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -150,12 +150,21 @@ Live validation on the LB stays. This runs before it, not instead of it. convenience: it needs its own test file, a redaction count in `simulation.json`, and a `caveats` line in `export.build_manifest` stating what was stripped and what was not. -- [ ] **G3** Simulation feeds the refiner. (S, P2) Depends on G2 (landed). - `refiner.py` retries until the band-aid blocks the exploit. Add the second half: retry - until it blocks the exploit **and** stays under the false-positive threshold. Today a - refinement widening the policy to catch the exploit has nothing telling it it went too far. +- [x] **G3** Simulation feeds the refiner. (S, P2) — **DONE:** the refine loop now accepts a + candidate only when it blocks the exploit **and** stays under the blast-radius threshold. An + over-broad candidate is fed back as the `over_block` diagnosis the refine agent already + understands, with the top blocked paths attached so it knows *what* it over-blocked. - Acceptance: a refinement pass increasing would-block rate past the threshold gets - rejected and the refiner tries again or gives up honestly, matching existing behavior. + rejected and the refiner tries again or gives up honestly, matching existing behavior ✅ + - **Measured where it is already attached.** The refiner has just confirmed the candidate is + enforcing, so `simulate.measure_attached` replays through that LB rather than paying a second + attach and propagation wait elsewhere — slower-but-correct per the decision of 2026-07-27, but + only one round trip per attempt rather than two. + - `_score` is shared by `simulate_policies` and `measure_attached`, so the standalone run and the + refiner cannot drift on what "too broad" means. + - **Sample comes from `VPCOPILOT_SIM_LOGS`** (or an explicit `records=`). With neither, the second + gate is skipped and the loop behaves exactly as before — pinned by a test that fails if anything + is replayed without a sample. An unreadable sample logs a warning and refines without the gate. - [ ] **G4** Published model scorecard. (M, P1) `D3` proved a config-only swap on `gpt-4o`. Turn that into a committed table across four diff --git a/docs/USAGE.md b/docs/USAGE.md index 862c9a1..9aa53e4 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -81,6 +81,11 @@ vpcopilot report --open # standalone shareable HTML dashboard of the results vpcopilot retire --finding # C2: when the cure PR merges, detach the band-aid + mark retired vpcopilot retire --all # retire every mitigated finding whose cure PR merged (--force to skip the check) ``` +**Refine with a blast-radius gate.** Set `VPCOPILOT_SIM_LOGS` to a traffic sample and the +refiner stops at the first policy that blocks the exploit *and* stays under the threshold, instead +of the first that merely blocks it — a refinement that widened the rule too far is fed back as +`over_block` and retried. With the variable unset the refine loop behaves exactly as before. + `export` writes `/audit-bundle.zip` (`--all` → `/audit-bundle-all.zip` with a top-level `index.json`). Inside: `manifest.json` (bundle identity, the run manifest, a SHA-256 per member, and an explicit `caveats` list), `audit.csv` + `audit-events.json` (one normalized row per change, joined diff --git a/src/vpcopilot/console/static/index.html b/src/vpcopilot/console/static/index.html index b420a53..7428399 100644 --- a/src/vpcopilot/console/static/index.html +++ b/src/vpcopilot/console/static/index.html @@ -502,6 +502,7 @@ if(res.attempts>1) bits.push(`self-healed in ${res.attempts} attempts`); else if(res.attempts) bits.push(`${res.attempts} attempt`); if(res.kept) bits.push("· KEPT LIVE"); else if(res.rolled_back) bits.push("(rolled back)"); + if(res.blast_radius) bits.push(`· blast radius ${esc(res.blast_radius.would_block)}/${esc(res.blast_radius.evaluated)} recorded (${(res.blast_radius.block_rate*100).toFixed(1)}%)`); if(res.unfixable) bits.push('unfixable — ship the code fix'); const _before=((res.before_after||{}).before)||{}, _bst=_before.exploit_status; if(_before.auth_failed) diff --git a/src/vpcopilot/refiner.py b/src/vpcopilot/refiner.py index 80da199..4bb9408 100644 --- a/src/vpcopilot/refiner.py +++ b/src/vpcopilot/refiner.py @@ -38,14 +38,51 @@ def _load_finding(out_dir: str, finding_id: str | None) -> Finding | None: return None +def _sim_threshold_default() -> float: + from .simulate import DEFAULT_THRESHOLD + try: + return float(os.environ.get("VPCOPILOT_SIM_THRESHOLD", DEFAULT_THRESHOLD)) + except ValueError: + return DEFAULT_THRESHOLD + + +def _resolve_records(records, log: Callable): + """The sample the second gate measures against: an explicit list, else `VPCOPILOT_SIM_LOGS`. + With neither there is no gate — fail-soft, because a missing or unreadable sample must never + turn a working refine loop into a failure.""" + if records: + return records + path = (os.environ.get("VPCOPILOT_SIM_LOGS") or "").strip() + if not path: + return None + try: + from .traffic import load + recs, _ = load(path) + log(f"blast-radius gate: {len(recs)} record(s) from {path}") + return recs or None + except Exception as e: # noqa: BLE001 + log(f" ⚠ could not read the traffic sample at {path}: {e} — refining without the gate") + return None + + def refine_apply_service_policy(artifact_path: str, lb: str, target_url: str, *, finding_id: str | None = None, name: str | None = None, max_refine: int | None = None, keep: bool = False, allow_protected: bool = False, config_path: str | None = None, retries: int = 6, wait_seconds: int = 8, + records: list | None = None, sim_threshold: float | None = None, out_dir: str = "out", log: Callable = print) -> dict: """Create/attach a service policy, validate it live, and refine-until-it-works (or give up - honestly). Returns passed + attempts + before/after; persists the WORKING spec to the artifact.""" + honestly). Returns passed + attempts + before/after; persists the WORKING spec to the artifact. + + G3 — with a traffic sample, "works" means two things, not one: the policy blocks the exploit + **and** stays under the blast-radius threshold. Without that second gate a refinement that + widens the policy far enough to catch the exploit has nothing telling it it went too far, and + the loop happily accepts a rule that would deny production. An over-broad result is fed back as + the `over_block` diagnosis the refine agent already understands. + + `records` (or `VPCOPILOT_SIM_LOGS`) supplies the sample. With neither, the second gate is + skipped entirely and this behaves exactly as it did before G3.""" xc = XC() if lb in _protected_lbs() and not allow_protected: raise RuntimeError(f"refusing to mutate protected LB '{lb}'. Pass allow_protected=True to override.") @@ -80,6 +117,9 @@ def attach(): def detach(): _put_lb(copy.deepcopy(orig_spec)) + records = _resolve_records(records, log) + threshold = sim_threshold if sim_threshold is not None else _sim_threshold_default() + h = Harness(config_path) if finding else None exploit = (probe or {}).get("exploit") before = _run_validation(target_url, finding_id, out_dir, probe_negative_pay, log) @@ -142,7 +182,17 @@ def detach(): if result["exploit_blocked"] and result["legit_ok"]: break - if result["exploit_blocked"] and result["legit_ok"]: + blast = None + if result["exploit_blocked"] and result["legit_ok"] and records: + # The policy is attached and confirmed enforcing right now — measure it here rather + # than paying a second attach + propagation wait on another LB. + from .simulate import measure_attached + blast = measure_attached(records, target_url, policy_name=policy_name, + finding_id=finding_id or "", threshold=threshold, log=log) + log(f" blast radius: would block {blast['would_block']}/{blast['evaluated']} " + f"recorded request(s) ({blast['block_rate']:.1%})") + + if result["exploit_blocked"] and result["legit_ok"] and not (blast and blast["blocked_promotion"]): log(f"validation PASS on attempt {attempt} ✓") Path(artifact_path).write_text(json.dumps({"metadata": {"name": policy_name}, "spec": spec}, indent=2)) rolled = False @@ -153,18 +203,27 @@ def detach(): rolled = True audit.record(out_dir, "refine_apply", finding_id=finding_id, namespace=xc.ns, control="service_policy", policy=policy_name, lb=lb, passed=True, attempts=attempt, rolled_back=rolled, - before_after={"before": before, "after": result}) - return {"mode": "refine_apply", "control": "service_policy", "policy": policy_name, - "passed": True, "attempts": attempt, "rolled_back": rolled, "kept": keep, - "before_after": {"before": before, "after": result}} + before_after={"before": before, "after": result}, blast_radius=blast) + out = {"mode": "refine_apply", "control": "service_policy", "policy": policy_name, + "passed": True, "attempts": attempt, "rolled_back": rolled, "kept": keep, + "before_after": {"before": before, "after": result}} + if blast: + out["blast_radius"] = blast + return out # failed this attempt — detach, diagnose, refine detach() - diagnosis = "exploit_not_blocked" if not result["exploit_blocked"] else "over_block" - log(f"attempt {attempt}: FAIL ({diagnosis}, exploit_status={result['exploit_status']})") + agent_result = result + if blast and blast["blocked_promotion"]: + diagnosis = "over_block" + agent_result = {**result, "blast_radius": blast} + log(f"attempt {attempt}: FAIL (over_block — {blast['reason']})") + else: + diagnosis = "exploit_not_blocked" if not result["exploit_blocked"] else "over_block" + log(f"attempt {attempt}: FAIL ({diagnosis}, exploit_status={result['exploit_status']})") if attempt == max_refine or not (h and finding): break - refined = refine_agent.run(h, finding, "service_policy", spec, probe, result, diagnosis) + refined = refine_agent.run(h, finding, "service_policy", spec, probe, agent_result, diagnosis) log(f" refined: {refined.rationale}" + (" [UNFIXABLE]" if refined.unfixable else "")) if refined.unfixable: audit.record(out_dir, "refine_apply", finding_id=finding_id, namespace=xc.ns, control="service_policy", policy=policy_name, lb=lb, diff --git a/src/vpcopilot/simulate.py b/src/vpcopilot/simulate.py index 6cc1974..fd9bf6d 100644 --- a/src/vpcopilot/simulate.py +++ b/src/vpcopilot/simulate.py @@ -140,6 +140,44 @@ def _replay(records: list[RequestRecord], url: str, return len(records) - errored, blocked, errored +def _score(sim: PolicySimulation, total: int, blocked: list, errored: int, threshold: float) -> PolicySimulation: + """Fill a verdict from a replay. Shared by the standalone run and the refiner's per-attempt + check so the two can never drift on what "too broad" means.""" + sim.errored = errored + sim.evaluated = total + sim.would_block = len(blocked) + sim.block_rate = round(len(blocked) / total, 4) if total else 0.0 + sim.threshold = threshold + sim.samples = [b.model_dump(exclude={"headers"}) for b in blocked[:MAX_SAMPLES]] + sim.top_paths = [list(x) for x in Counter(b.path for b in blocked).most_common(5)] + sim.top_user_agents = [list(x) for x in + Counter(b.user_agent for b in blocked if b.user_agent).most_common(5)] + if not total: + sim.reason = ("nothing measured — every replayed request failed in transit, so this is " + "zero evidence, not a clean result") + elif sim.block_rate > threshold: + sim.blocked_promotion = True + sim.reason = (f"would block {sim.would_block}/{total} recorded requests " + f"({sim.block_rate:.1%}), over the {threshold:.1%} threshold") + return sim + + +def measure_attached(records: list[RequestRecord], url: str, *, policy_name: str = "", + finding_id: str = "", threshold: float = DEFAULT_THRESHOLD, + max_records: int | None = None, log: Callable = print) -> dict: + """G3: blast radius of the policy that is ALREADY attached and validating. + + The refiner has just attached a candidate and confirmed it blocks the exploit, so there is no + second attach, no second propagation wait, and no second LB — we measure the exact policy that + just passed, where it passed. Returns a plain dict so it drops straight into the refine agent's + result payload, the audit record and the apply result.""" + sample = records[:max_records] if max_records else records + total, blocked, errored = _replay(sample, url, log) + sim = _score(PolicySimulation(policy_name=policy_name, finding_id=finding_id), total, blocked, + errored, threshold) + return sim.model_dump(exclude={"samples", "top_user_agents", "enforcement_confirmed", "error"}) + + def simulate_policies(candidates: list[dict], records: list[RequestRecord], *, lb: str, url: str, out_dir: str = "out", threshold: float = DEFAULT_THRESHOLD, max_records: int | None = None, source: str = "", window: str = "", @@ -186,26 +224,11 @@ def simulate_policies(candidates: list[dict], records: list[RequestRecord], *, l wait_seconds=wait_seconds, retries=retries, sleep=ctx.sleep) log(f" replaying {len(sample)} record(s)") total, blocked, errored = _replay(sample, url, log) - sim.errored = errored if errored: log(f" {errored} record(s) failed in transit — excluded from the rate") - sim.evaluated = total - sim.would_block = len(blocked) - sim.block_rate = round(len(blocked) / total, 4) if total else 0.0 - sim.samples = [b.model_dump(exclude={"headers"}) for b in blocked[:MAX_SAMPLES]] - sim.top_paths = [list(x) for x in Counter(b.path for b in blocked).most_common(5)] - sim.top_user_agents = [list(x) for x in - Counter(b.user_agent for b in blocked if b.user_agent).most_common(5)] - if not total: - sim.reason = ("nothing measured — every replayed request failed in transit, so this " - "is zero evidence, not a clean result") - if total and sim.block_rate > threshold: - sim.blocked_promotion = True - sim.reason = (f"would block {sim.would_block}/{total} recorded requests " - f"({sim.block_rate:.1%}), over the {threshold:.1%} threshold") - log(f" ⚠ {name}: {sim.reason}") - else: - log(f" {name}: would block {sim.would_block}/{total}") + _score(sim, total, blocked, errored, threshold) + log(f" ⚠ {name}: {sim.reason}" if sim.blocked_promotion + else f" {name}: would block {sim.would_block}/{total}") except Exception as e: # noqa: BLE001 — a replay failure is reported, never raised past restore sim.error = str(e) log(f" ⚠ {name}: simulation failed — {e}") diff --git a/tests/test_refiner.py b/tests/test_refiner.py index 64f3491..5a11678 100644 --- a/tests/test_refiner.py +++ b/tests/test_refiner.py @@ -147,3 +147,135 @@ def put_service_policy(self, n, b): pass max_refine=5, out_dir=str(tmp_path), log=lambda m: None) assert res["passed"] is False and res["unfixable"] is True assert res["attempts"] < 5 and res.get("recommend") == "ship the code fix" + + +# ---- G3: the refiner's second gate — block the exploit AND stay under the blast-radius threshold ---- +def _g3_env(monkeypatch, tmp_path, art_spec): + """Shared harness: a lint-clean policy, a finding + probe, and a fake XC/validation so the loop + reaches the point where blast radius is measured.""" + import json as _json + + from vpcopilot import refiner + art = tmp_path / "service_policy.deny-x.json" + art.write_text(_json.dumps({"metadata": {"name": "deny-x"}, "spec": art_spec})) + (tmp_path / "findings.json").write_text(_json.dumps([{ + "id": "f1", "title": "t", "vuln_class": "other", "severity": "high", + "file": "a.py", "description": "d", "exploit_sketch": "e"}])) + (tmp_path / "probes.json").write_text(_json.dumps([{ + "finding_id": "f1", "exploit": {"method": "POST", "path": "/x"}, + "legit": {"method": "GET", "path": "/y"}}])) + + class FakeXC: + ns = "test-ns" + def get_lb(self, lb): + return {"metadata": {"name": lb, "namespace": self.ns}, "spec": {"no_service_policies": {}}} + def put_lb(self, lb, obj): pass + def service_policy_exists(self, n): return False + def create_service_policy(self, b): pass + def put_service_policy(self, n, b): pass + + monkeypatch.setattr(refiner, "XC", FakeXC) + monkeypatch.setattr(refiner, "normalize_service_policy_spec", lambda s: s) + monkeypatch.setattr(refiner, "_protected_lbs", lambda: set()) + monkeypatch.setattr(refiner, "Harness", lambda cfg=None: object()) + monkeypatch.setattr(refiner.time, "sleep", lambda s: None) + # validation always passes: the exploit is blocked and the legit request survives + monkeypatch.setattr(refiner, "_run_validation", + lambda *a, **k: {"exploit_status": 403, "exploit_blocked": True, "legit_ok": True}) + return refiner, art + + +def _records(*paths): + from vpcopilot.schemas import RequestRecord + return [RequestRecord(path=p, method="GET") for p in paths] + + +CLEAN = {"rule_list": {"rules": [ + {"spec": {"action": "DENY", "path": {"prefix_values": ["/x"]}, "http_method": {"methods": ["POST"]}}}, + {"spec": {"action": "ALLOW", "path": {"prefix_values": ["/"]}}}]}} + + +def test_a_policy_that_blocks_the_exploit_but_overblocks_is_rejected_and_refined(monkeypatch, tmp_path): + """The gap G3 closes: today a refinement that widens the policy enough to catch the exploit has + nothing telling it it went too far.""" + from vpcopilot import simulate + from vpcopilot.schemas import RefinedPolicy + refiner, art = _g3_env(monkeypatch, tmp_path, CLEAN) + + wide = {"rule_list": {"rules": [ + {"spec": {"action": "DENY", "path": {"prefix_values": ["/x"]}, "narrowed": 1}}, + {"spec": {"action": "ALLOW", "path": {"prefix_values": ["/"]}}}]}} + seen = {} + monkeypatch.setattr(refiner.refine_agent, "run", + lambda h, f, c, spec, probe, result, diagnosis: ( + seen.update(diagnosis=diagnosis, result=result), + RefinedPolicy(spec=wide, rationale="narrowed the prefix"))[1]) + # attempt 1 blocks everything; the refined spec blocks only the exploit path + calls = {"n": 0} + + def fire(rec, url): + calls["n"] += 1 + return (403, "") if seen.get("diagnosis") is None else ((403, "") if rec.path == "/x" else (200, "ok")) + monkeypatch.setattr(simulate, "_fire", fire) + + res = refiner.refine_apply_service_policy( + str(art), "lab", "http://x", finding_id="f1", max_refine=3, out_dir=str(tmp_path), + records=_records("/x", *[f"/p{i}" for i in range(199)]), log=lambda m: None) + + assert seen["diagnosis"] == "over_block" # mapped to a diagnosis it knows + assert seen["result"]["blast_radius"]["would_block"] == 200 # the agent is told WHAT it blocked + assert res["passed"] is True and res["attempts"] == 2 + assert res["blast_radius"]["would_block"] == 1 # the accepted policy is surgical + + +def test_a_surgical_policy_passes_on_the_first_attempt(monkeypatch, tmp_path): + from vpcopilot import simulate + refiner, art = _g3_env(monkeypatch, tmp_path, CLEAN) + monkeypatch.setattr(simulate, "_fire", + lambda rec, url: (403, "") if rec.path == "/x" else (200, "ok")) + res = refiner.refine_apply_service_policy( + str(art), "lab", "http://x", finding_id="f1", max_refine=3, out_dir=str(tmp_path), + records=_records("/x", *[f"/p{i}" for i in range(199)]), log=lambda m: None) + assert res["passed"] is True and res["attempts"] == 1 + assert res["blast_radius"]["would_block"] == 1 # 1/200 = 0.5%, under the 1% default + assert res["blast_radius"]["blocked_promotion"] is False + + +def test_it_gives_up_honestly_when_it_can_never_get_under_the_threshold(monkeypatch, tmp_path): + """Matching existing behaviour: an unfixable over-block ends as passed=False with a reason, + not as a silently-accepted policy.""" + from vpcopilot import simulate + from vpcopilot.schemas import RefinedPolicy + refiner, art = _g3_env(monkeypatch, tmp_path, CLEAN) + monkeypatch.setattr(refiner.refine_agent, "run", + lambda *a, **k: RefinedPolicy(spec=CLEAN, rationale="still too wide")) + monkeypatch.setattr(simulate, "_fire", lambda rec, url: (403, "")) # blocks everything, always + res = refiner.refine_apply_service_policy( + str(art), "lab", "http://x", finding_id="f1", max_refine=2, out_dir=str(tmp_path), + records=_records("/x", "/a", "/b"), log=lambda m: None) + assert res["passed"] is False and "over_block" in res["reason"] + + +def test_the_threshold_is_configurable(monkeypatch, tmp_path): + from vpcopilot import simulate + refiner, art = _g3_env(monkeypatch, tmp_path, CLEAN) + monkeypatch.setattr(simulate, "_fire", lambda rec, url: (403, "") if rec.path in ("/x", "/a") else (200, "ok")) + res = refiner.refine_apply_service_policy( + str(art), "lab", "http://x", finding_id="f1", max_refine=1, out_dir=str(tmp_path), + records=_records("/x", "/a", "/b", "/c"), sim_threshold=0.9, log=lambda m: None) + assert res["passed"] is True # 50% is under a 90% threshold + + +# ---- no-regression: G3 adds a gate only when a sample exists ---- +def test_without_records_the_refiner_behaves_exactly_as_before(monkeypatch, tmp_path): + from vpcopilot import simulate + refiner, art = _g3_env(monkeypatch, tmp_path, CLEAN) + + def boom(rec, url): + raise AssertionError("no traffic sample was given — nothing should have been replayed") + monkeypatch.setattr(simulate, "_fire", boom) + res = refiner.refine_apply_service_policy( + str(art), "lab", "http://x", finding_id="f1", max_refine=3, out_dir=str(tmp_path), + log=lambda m: None) + assert res["passed"] is True and res["attempts"] == 1 + assert "blast_radius" not in res