From 630167de3f800395d15d4e3fa1112e9c06560020 Mon Sep 17 00:00:00 2001 From: henleda Date: Mon, 27 Jul 2026 13:35:31 -0500 Subject: [PATCH] feat(drift): I2 pre-apply drift and conflict detection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Compare live LB vs last snapshot vs what is about to be pushed, read-only throughout, and gate the apply on the result. `drift.preflight()` is called by BOTH apply paths — `apply.apply_from_scan` and `refiner.refine_apply_service_policy`. The refiner is the default for `--from-scan` and the console's Mitigate button, so gating only the former would have gated nothing anyone uses. The roadmap's conflict criterion was wrong, and running it live is what proved it. It assumed the new policy is appended after the attached ones; both attach paths replace the oneof with exactly one policy, so this tool never leaves two service policies attached and there is no earlier policy to be shadowed by. The first live run produced a confident false positive on banknimbus-dev. Replaced by the two real behaviours underneath it: * shadowing, in its only true scope — an ALLOW inside the policy being applied that matches the exploit before its DENY. Refuses; `--force` overrides; reuses `lint_service_policy` so drift and the linter cannot disagree about FIRST_MATCH. Degrades to a warning on the refine path, because reordering rules is what refine exists to do. * displacement — that same wholesale replacement silently DETACHES whatever was attached, a live loss of protection nothing was reporting. Warned and audited, never refused: replacing the previous band-aid is the normal flow. `no_change` returns before the XC create, not merely before `ApplyContext.load()`, so a skipped apply leaves no LB PUT, no snapshot, no run artifact and no stray policy object. It is determined for service_policy only — there the proposed end state is exact. For the six LB-wide toggles presence is detectable but presence is not parameter equality, so those report `already_attached` and are never auto-skipped. Six new audit actions, all category `gate`, with the decision as their outcome (`refused` / `overridden` / `no_change` / `displaced` / `drift_detected`) — never passed/failed, because there was no apply to score. `simulate_override` from G2 was missing from the export model and is added with them. Surfaces: `vpcopilot drift --lb ` (exit 1 on conflict), `GET /api/drift`, a drift strip on the console's Mitigate step, and an inline "apply anyway" button on a refusal. The strip renders nothing when the check cannot run, so a credential-free screenshot capture stays clean. No regressions: an LB already carrying a foreign policy still mitigates; dry runs and `create_only` are never gated; behaviour with no snapshot and no conflict is unchanged. Each pinned by a test. Verified live against the tenant: field-level diff off a real 2026-07-24 snapshot, displacement warning, shadowed-DENY refusal (no LB change, no stray policy), `no_change`, and `--force`. 341 tests, ruff clean. Co-Authored-By: Claude Opus 5 (1M context) --- ROADMAP.md | 67 ++-- docs/AUDIT.md | 21 +- docs/DEMO.md | 6 +- docs/USAGE.md | 31 ++ src/vpcopilot/apply.py | 61 ++-- src/vpcopilot/cli.py | 59 +++- src/vpcopilot/console/app.py | 31 +- src/vpcopilot/console/static/index.html | 44 ++- src/vpcopilot/drift.py | 278 +++++++++++++++++ src/vpcopilot/export.py | 16 +- src/vpcopilot/refiner.py | 15 +- tests/conftest.py | 5 + tests/test_console_drift.py | 112 +++++++ tests/test_drift.py | 386 ++++++++++++++++++++++++ tests/test_export.py | 29 ++ 15 files changed, 1106 insertions(+), 55 deletions(-) create mode 100644 src/vpcopilot/drift.py create mode 100644 tests/test_console_drift.py create mode 100644 tests/test_drift.py diff --git a/ROADMAP.md b/ROADMAP.md index 2e92a9f..fb59174 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -382,29 +382,50 @@ holds only while a human keeps checking. still accumulate state and pollute the malicious-user telemetry the demo points at, so a minimum interval between replays of the same finding belongs in the design. -- [ ] **I2** Drift and conflict detection. (M, P1) - `snapshots/` already captures the pre-change LB state. Turn that into a comparison run - before apply: what is on the LB now versus what the last run left versus what is about to - be pushed. - - Acceptance: re-applying an unchanged control reports `no_change` and writes nothing; a - hand edit made in the XC console since the last apply reports as a field-level diff; an - earlier ALLOW rule shadowing the new DENY blocks apply as a conflict; drift runs - read-only. - - Surfaces: `src/vpcopilot/drift.py`, `vpcopilot drift --lb `, `GET /api/drift`, - a drift check before `engine.ApplyContext.load()` takes the snapshot. - - **Reconciled:** `engine.SafeApply` is not a symbol — "SafeApply spine" is prose for - `engine.py`, which exports `RollbackError`, `protected_lbs`, `guard_lb`, `ApplyContext`, - `poll_until`, `safe_rollback`. The snapshot is taken in `ApplyContext.load()`. - - **`no_change` must return before `ApplyContext.load()`**, not merely before the mutation: - `load()` writes `out/lb_snapshot.json` *and* a fresh `out/snapshots/-.json` on every - call (`engine.py:63-72`), and `self_test()` always issues an idempotent PUT to XC - (`engine.py:75-83`). A drift check therefore does its own `xc.get_lb()`. - - **Partly built:** `ApplyContext.load()` already writes per-LB timestamped snapshots to - `out/snapshots/` and `export.py` already bundles them, so the raw material exists. Note also - that `lint_service_policy` already implements an *exploit-relative* FIRST_MATCH shadow check - (`apply.py:116-152`) — it returns early without an exploit request (`apply.py:130-131`), so - the reusable piece for a drift comparison is the `_matches` + FIRST_MATCH walk, not the - function as-is. +- [x] **I2** Drift and conflict detection. (M, P1) — **DONE:** `drift.py` compares live LB vs last + snapshot vs proposed, read-only throughout. `drift.preflight()` is the pre-apply gate, called by + **both** apply paths — `apply.apply_from_scan` and `refiner.refine_apply_service_policy` (the + latter is the default for `--from-scan` and the console's Mitigate button, so gating only the + former would have gated nothing anyone uses). Surfaces: `vpcopilot drift --lb ` (exit 1 on + conflict), `GET /api/drift`, `--force` / console **apply anyway**. Six new audit actions, all + category `gate`. Verified live against `banknimbus-dev`. + - **Acceptance, as met:** + - `no_change` — met. Reports and writes nothing: no LB PUT, no `snapshots/`, no + `lb_snapshot.json`, and no stray policy object, because the gate runs before the XC *create*, + not just before `ApplyContext.load()`. + - field-level diff of a hand edit — met. Verified against a real 2026-07-24 snapshot of + `banknimbus-dev`: 12 dotted-path changes, correctly attributed. + - drift runs read-only — met, and pinned by tests on both the CLI and the endpoint (the + endpoint is polled from the browser; a version that wrote a snapshot would corrupt the run + dir just by someone opening a page). + - **"an earlier ALLOW rule shadowing the new DENY blocks apply as a conflict" — the criterion + as written was wrong, and running it live is what proved it.** It assumes the new policy is + appended after the attached ones. Both attach paths do the opposite — they replace the oneof + with exactly one policy (`apply.py`, `refiner.py`: `active_service_policies = {"policies": + [{ns, name}]}`) — so this tool never leaves two service policies attached and there is no + earlier policy to be shadowed by. The first live run produced a confident false positive on + `banknimbus-dev`. Replaced by the two real behaviours underneath it: + - **shadowing, in its only true scope** — an ALLOW *inside the policy being applied* that + matches the exploit before its DENY. This **refuses** (`--force` overrides), and reuses + `lint_service_policy` rather than re-deriving FIRST_MATCH so the two can never disagree. + On the refine path it degrades to a warning: reordering rules is what the refine loop + exists to do, and refusing there would break the default flow to protect it from a problem + it already fixes. + - **displacement** — that same wholesale replacement silently *detaches* whatever was + attached, which is a live loss of protection nothing was reporting. Warned and audited + (`policy_displaced`, carrying whether the displaced policy is what currently blocks this + exploit), never refused: replacing the previous band-aid is the normal flow and refusing + would break every second apply. Follows the G2 precedent — warn with an audited override, + not a machine veto. + - **Deliberate partial:** `no_change` is determined for `service_policy` only. There the proposed + end state is exact (that policy name in `active_service_policies`), so "unchanged" is a fact. + For the six LB-wide toggles, presence is detectable by inverting `detach_control` but presence + is **not** parameter equality — a `rate_limit` already on at 100/MINUTE would read as unchanged + while you push 5/MINUTE. Silently skipping a real change is worse than re-applying an identical + one, so those report `already_attached` with a note and are never auto-skipped. + - **No regressions:** an LB already carrying a foreign policy still mitigates (warn + proceed); + dry runs are never gated; `create_only` is never gated; behaviour with no snapshot and no + conflict is byte-identical to before. Each pinned by a test. --- diff --git a/docs/AUDIT.md b/docs/AUDIT.md index 56b5c09..4f0bec8 100644 --- a/docs/AUDIT.md +++ b/docs/AUDIT.md @@ -18,6 +18,11 @@ Source of truth: `src/vpcopilot/audit.py` (the sink), `runmeta.py` (identity + p attaching/enabling a band-aid on an LB, a refine-and-retry loop, opening a code-fix PR, retiring a band-aid, and a rollback that failed. One JSON object per line in `/audit.log`, never rewritten. +Also recorded: every **gate decision** — the pre-apply drift check refusing, being overridden, +skipping an unchanged apply, or reporting that an apply detached another policy. Nothing changed on +the LB in those cases, which is exactly why they belong here. "We looked and refused" and "we looked +and overrode it anyway" are questions an auditor asks, and only the log can answer them. + That scope is the whole band-aid path, but it is not literally every XC write the tool can make: the lab/teardown utilities `vpcopilot lab-create` (`lab.py`) and `vpcopilot xc-rm` (`cli.py`) mutate XC outside the finding lifecycle and write **no** audit record. They are setup/cleanup helpers, not @@ -86,7 +91,7 @@ detail = {k: v for k, v in detail.items() if k not in _STAMPED} Everything else is per-action detail. -### The 15 actions +### The 21 actions | action | category | what it means | key detail fields | |---|---|---|---| @@ -105,6 +110,12 @@ Everything else is per-action detail. | `open_pr` | cure | The code-fix PR (the cure) was opened on GitHub | `finding_id` `finding` `repo` `url` `number` | | `rollback_failed` | rollback | The LB could **not** be confirmed restored to its pre-apply snapshot after N retries. The one entry that must never be anonymous — the LB may be left in a changed state | `finding_id` `lb` `namespace` `reason` | | `apply_timing` | timing | Wall-clock + outcome for one console Mitigate click. Feeds MTTM on the hero panel and policy quality in the model benchmark | `control` `finding_id` `passed` `elapsed_s` `attempts` `before_after` `unfixable` `reason` `kept` | +| `drift_detected` | gate | The live LB differs from the snapshot the last apply left — someone edited it outside this tool. Carries the full field-level diff | `lb` `policy` `snapshot` `changes` | +| `policy_displaced` | gate | Attaching replaces `active_service_policies` wholesale, so this apply **detached** another service policy. `protects_exploit` is true when the displaced policy is what currently blocks this finding's exploit | `lb` `policy` `displaced` `rules` `denies` `protects_exploit` | +| `apply_skipped_no_change` | gate | The policy asked for was already the attached one. Nothing was pushed — no LB PUT, no snapshot, no run artifact | `lb` `policy` | +| `drift_block` | gate | The apply was **refused**: an ALLOW inside the policy matched the exploit before its DENY, so under FIRST_MATCH the band-aid would have attached cleanly and blocked nothing | `lb` `policy` `conflicts` | +| `drift_override` | gate | The same conflict, applied anyway via `--force` / the console's **apply anyway**. The override is the point of the entry | `lb` `policy` `conflicts` | +| `simulate_override` | gate | A policy that shadow simulation flagged as over-broad was promoted anyway | `finding_id` `policy` `lb` `block_rate` `threshold` `reason` | Notes read from the source: @@ -122,6 +133,14 @@ Notes read from the source: - `apply_timing` is written by the console only, and only when `dry_run` is false (`console/app.py::_run_action`). A CLI-driven run has none — MTTM and `elapsed_s` are simply absent, not zero. +- The six **gate** actions are the only ones that record a moment when the LB did **not** change. + Their `outcome` in the normalized export is the decision itself — `refused`, `overridden`, + `no_change`, `displaced`, `drift_detected` — never `passed`/`failed`, because there was no apply + to score. Dry runs remain unlogged: a gate decision is a real decision about a real apply, a dry + run is a preview. +- `policy_displaced` is a **warning, not a veto**. Every apply replaces whatever was attached, so + refusing would break every second apply; it is said out loud and written down instead. If you are + auditing what protection an LB lost over time, this is the entry to read. ### Reading it raw diff --git a/docs/DEMO.md b/docs/DEMO.md index c269876..c7450c2 100644 --- a/docs/DEMO.md +++ b/docs/DEMO.md @@ -128,6 +128,8 @@ To regenerate them: rebuild the dataset with `python3 demo/build_demo_out.py`, r `#simulate`, `#mitigate` and `#retire` steps plus `demo/out/report.html` at 1200px wide / 2× device pixel ratio. Point the console at a **credential-free** `.env` when you do (`VPCOPILOT_ENV=…`): with XC creds -loaded, the hero band renders a deep link carrying your tenant hostname and namespace, and that -would ship in the image. `build_demo_out.py` curates `actor`/`host`/`out_dir` in the fixture for the +loaded, the hero band renders a deep link carrying your tenant hostname and namespace, and the +Mitigate step's drift strip renders the target LB's live control set — both would ship in the image. +Without creds the drift strip renders nothing at all, which is why it does not appear in +`4-mitigate.png`. `build_demo_out.py` curates `actor`/`host`/`out_dir` in the fixture for the same reason — no real machine identity in a shared dataset. diff --git a/docs/USAGE.md b/docs/USAGE.md index dfc90fb..5f97344 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -62,6 +62,34 @@ agent to correct the spec (using the exact exploit + legit requests), and retrie required". The working refined spec is written back to the artifact. `--no-refine` for single-shot. Configurable in the console's action bar (**refine** + **attempts**). +### Pre-apply drift check (I2) + +Before anything is created or attached, a **read-only** comparison runs against the live LB. It +answers three questions and short-circuits or warns accordingly: + +```sh +vpcopilot drift --lb # what changed since the last apply +vpcopilot drift --lb --control service_policy --policy --finding +``` + +| finding | what happens | override | +|---|---|---| +| the policy is **already the attached one** | reports `no_change` and **writes nothing** — no LB PUT, no snapshot, no run artifact | `--force` | +| a field on the LB **changed since the last snapshot** (someone edited it in the XC console) | field-level diff printed and audited as `drift_detected`; the apply continues | — | +| applying will **detach** another service policy | warned and audited as `policy_displaced`, including whether the displaced policy is what currently blocks this exploit; the apply continues | — | +| an **ALLOW inside the policy being applied** matches the exploit before its DENY | refused — under FIRST_MATCH the band-aid would attach cleanly and block nothing | `--force`, or `--refine` (default), which reorders it instead | + +`drift` exits `1` when it finds a conflict, so it composes in a script. The check never PUTs, never +writes a snapshot, and never touches the run directory — it is safe to run or poll at any time. + +**Why displacement is a warning and not a refusal.** Attaching replaces `active_service_policies` +wholesale, so every apply detaches whatever was there. Replacing the previous band-aid is the +normal flow; refusing would break every second apply. It is said out loud and written down instead. + +In the console the same check runs inside the Mitigate job, so its warnings appear in the live log. +A refusal renders an **apply anyway** button, and `no_change` renders as its own outcome rather +than a pass or a fail. + ## 5. Open the code-fix PR (the cure) ```sh vpcopilot pr --repo owner/name --base --path-prefix [--finding ] [--dry-run] @@ -206,6 +234,9 @@ VPCOPILOT_DEFAULT_PREFIX= # usually empty - **Guardrails:** `PROTECTED_POLICIES` (the `nimbus-*` demo policies) can't be created/deleted; protected LBs (`VPCOPILOT_PROTECTED_LBS`, default `nimbus-www`) can't be mutated without `--allow-protected-lb`. +- **Pre-apply drift check:** before any create or attach, the live LB is compared against the last + snapshot and against what is about to be pushed — see [Pre-apply drift check](#pre-apply-drift-check-i2). + Re-applying an unchanged policy writes nothing; a policy that could never fire is refused. - **Reversible:** every apply snapshots the LB and rolls back on validation failure (or by default). Every change is written to the append-only audit log — the finding that justified it, the control and the XC object, the load balancer and its namespace, whether it was kept or rolled diff --git a/src/vpcopilot/apply.py b/src/vpcopilot/apply.py index ec22f5d..109cbc1 100644 --- a/src/vpcopilot/apply.py +++ b/src/vpcopilot/apply.py @@ -125,13 +125,35 @@ def artifact_spec(spec: dict) -> dict: return inner if isinstance(inner, dict) and inner else spec +def rule_matches(rule_spec: dict, exploit: dict) -> bool: + """Does one rule's path+method matcher cover this request? + + Extracted from `lint_service_policy` unchanged so `drift.py` can run the same FIRST_MATCH walk + ACROSS the policies already attached to an LB, not just within one spec. Behaviour is + deliberately identical, including that `exact_values` is treated as a prefix — loosening or + tightening it here would silently change what the linter reports.""" + import re + path = exploit.get("path", "") or "" + method = (exploit.get("method") or "GET").upper() + p = rule_spec.get("path") or {} + ok = any(path == v or path.startswith(v) + for v in (p.get("prefix_values") or []) + (p.get("exact_values") or [])) + for rx in (p.get("regex_values") or []): + try: + if re.search(rx, path): + ok = True + except re.error: + pass + methods = [m.upper() for m in ((rule_spec.get("http_method") or {}).get("methods") or [])] + return ok and (not methods or method in methods) + + def lint_service_policy(spec: dict, exploit: dict | None) -> list[str]: """Deterministic pre-apply lint — catch a service_policy that won't actually block the exploit, BEFORE any live LB round-trip. Under FIRST_MATCH the FIRST rule whose path+method match the exploit decides its fate; if that's an ALLOW (a bad rule order, or a DENY whose path is a wrong guess so an allow-all catches the exploit first), the exploit sails through. Returns issue strings ([] = looks correct).""" - import re spec = artifact_spec(spec) # accept a {metadata, spec} artifact, exactly as apply_from_scan does rules = (spec.get("rule_list") or {}).get("rules") or [] if not any((r.get("spec") or {}).get("action") == "DENY" for r in rules): @@ -141,22 +163,9 @@ def lint_service_policy(spec: dict, exploit: dict | None) -> list[str]: path = exploit.get("path", "") or "" method = (exploit.get("method") or "GET").upper() - def _matches(rs: dict) -> bool: - p = rs.get("path") or {} - ok = any(path == v or path.startswith(v) - for v in (p.get("prefix_values") or []) + (p.get("exact_values") or [])) - for rx in (p.get("regex_values") or []): - try: - if re.search(rx, path): - ok = True - except re.error: - pass - methods = [m.upper() for m in ((rs.get("http_method") or {}).get("methods") or [])] - return ok and (not methods or method in methods) - for r in rules: # FIRST_MATCH: the first path+method match decides rs = r.get("spec") or {} - if _matches(rs): + if rule_matches(rs, exploit): if rs.get("action") == "ALLOW": return [f"an ALLOW rule matches the exploit {method} {path} before any DENY — " "FIRST_MATCH lets it through (fix the rule order or the DENY path)"] @@ -356,11 +365,16 @@ def rollback(): # B3: verified, retried rollback (LB restored to snapshot or Ro def apply_from_scan(artifact_path: str, lb: str, target_url: str, *, name: str | None = None, create_only: bool = False, dry_run: bool = False, keep: bool = False, allow_protected: bool = False, probe: bool = False, retries: int = 8, - wait_seconds: int = 8, finding_id: str | None = None, + wait_seconds: int = 8, finding_id: str | None = None, force: bool = False, out_dir: str = "out", log: Callable = print) -> dict: """End-to-end from a generated artifact: create the policy in XC (if missing), then attach -> validate -> rollback via apply_service_policy. Guarded against clobbering a - protected policy.""" + protected policy. + + I2 — a read-only drift check runs FIRST, before `ApplyContext.load()`: `load()` writes both + `lb_snapshot.json` and a fresh `snapshots/-.json` on every call and `self_test()` always + PUTs, so "reports no_change and writes nothing" has to short-circuit earlier than the mutation. + `force=True` re-applies anyway (to re-validate a policy that is already attached).""" xc = XC() art = json.loads(Path(artifact_path).read_text()) # Normalize to an XC create body: {metadata:{name,namespace,...}, spec:{...}}. @@ -382,6 +396,19 @@ def apply_from_scan(artifact_path: str, lb: str, target_url: str, *, name: str | if src_meta.get(k) is not None: body["metadata"][k] = src_meta[k] + # I2 — the pre-apply gate runs before the CREATE, not just before the attach: a run that is + # about to be refused should not leave a stray policy object behind in the tenant. It also has + # to precede `ApplyContext.load()`, which writes a snapshot on every call, so "no_change writes + # nothing" is true of the run directory as well as the LB. `create_only` makes no attachment, + # so there is nothing for it to gate. + if not dry_run and not create_only: + from .drift import preflight + d = preflight(lb, policy_name, out_dir=out_dir, force=force, xc=xc, log=log, spec=spec, + exploit=(_load_probe(out_dir, finding_id) or {}).get("exploit")) + if d is not None: + return {"mode": "no_change", "control": "service_policy", "policy": policy_name, + "passed": None, "drift": d} + if xc.service_policy_exists(policy_name): log(f"policy '{policy_name}' already exists — not overwriting") elif dry_run: diff --git a/src/vpcopilot/cli.py b/src/vpcopilot/cli.py index 4a77556..b280d17 100644 --- a/src/vpcopilot/cli.py +++ b/src/vpcopilot/cli.py @@ -157,6 +157,7 @@ def apply( probe_login_path: str = typer.Option(None, "--probe-login-path", help="login endpoint path, default /api/login (or VPCOPILOT_PROBE_LOGIN_PATH)"), probe_token: str = typer.Option(None, "--probe-token", help="bearer token for validation instead of user/pass (or VPCOPILOT_PROBE_TOKEN)"), finding: str = typer.Option(None, "--finding", help="finding id whose probe (out/probes.json) validates this policy; overrides the ledger lookup"), + force: bool = typer.Option(False, "--force", help="apply despite the pre-apply drift check: re-attach a policy that is already attached, or push past a conflicting ALLOW on another attached policy"), out: str = typer.Option("out", help="output directory"), ): """Gated apply: (create from scan) -> snapshot -> self-test -> attach -> validate -> refine/rollback.""" @@ -174,10 +175,10 @@ def apply( from .refiner import refine_apply_service_policy res = refine_apply_service_policy(from_scan, lb, url, name=name, keep=keep, allow_protected=allow_protected_lb, max_refine=refine_attempts, - finding_id=finding, out_dir=out, log=logf) + finding_id=finding, force=force, out_dir=out, log=logf) elif from_scan: from .apply import apply_from_scan - res = apply_from_scan(from_scan, lb, url, name=name, create_only=create_only, **kw) + res = apply_from_scan(from_scan, lb, url, name=name, create_only=create_only, force=force, **kw) else: from .apply import apply_service_policy res = apply_service_policy(lb, policy, url, **kw) @@ -611,6 +612,60 @@ def ledger(out: str = typer.Option("out", help="output directory")): rprint(t) +@app.command() +def drift( + lb: str = typer.Option(..., "--lb", help="load balancer to inspect"), + out: str = typer.Option("out", help="run directory holding the snapshots to compare against"), + control: str = typer.Option(None, "--control", help="the control you are about to apply"), + policy: str = typer.Option(None, "--policy", help="service-policy name you are about to attach"), + finding: str = typer.Option(None, "--finding", help="use this finding's exploit for the shadowing check"), + artifact: str = typer.Option(None, "--artifact", help="the generated policy artifact you are about to apply; without it the shadowing check has no rules to walk"), +): + """What is on the LB now, versus what the last run left, versus what you are about to push. + + Read-only: no PUT, no snapshot written, nothing in the run dir touched.""" + import json as _json + from pathlib import Path as _Path + + from .apply import _load_probe + from .drift import check + + load_dotenv() + exploit = (_load_probe(out, finding) or {}).get("exploit") if finding else None + art = artifact or (str(_Path(out, "policies", f"service_policy.{policy}.json")) if policy else None) + spec = _json.loads(_Path(art).read_text()) if art and _Path(art).is_file() else None + r = check(lb, out_dir=out, control=control, policy_name=policy, exploit=exploit, spec=spec, + log=lambda m: rprint(f"[dim]{m}[/dim]")) + lvs = r["live_vs_snapshot"] + t = Table(title=f"drift · {lb}") + for c in ["field", "last snapshot", "live now"]: + t.add_column(c) + for c in lvs["changes"]: + t.add_row(c["path"], _json.dumps(c["was"])[:60], _json.dumps(c["now"])[:60]) + if lvs["changes"]: + rprint(t) + elif lvs["snapshot"]: + rprint(f"[green]no drift[/green] since {lvs['snapshot']}") + else: + rprint("[dim]no snapshot for this LB yet — nothing to compare against[/dim]") + rprint(f"attached controls: {', '.join(r['attached']) or '(none)'}") + if r["proposed"]["no_change"]: + rprint(f"[yellow]no_change[/yellow] — {r['proposed']['policy_name']} is already attached") + elif r["proposed"]["note"]: + rprint(f"[dim]{r['proposed']['note']}[/dim]") + for d in r["displaces"]: + detail = "contents unreadable" if d.get("unreadable") else f"{d['rules']} rule(s), {d['denies']} DENY" + rprint(f"[yellow]⚠ applying will DETACH[/yellow] '{d['policy']}' ({detail})" + + (" [red]— and it is what currently blocks this exploit[/red]" + if d["protects_exploit"] else "")) + for c in r["conflicts"]: + rprint(f"[red]CONFLICT[/red] {c['reason']}") + for w in r["warnings"]: + rprint(f"[yellow]⚠ {w}[/yellow]") + if not r["ok_to_apply"]: + raise typer.Exit(code=1) + + @app.command(name="xc-status") def xc_status(lb: str = typer.Option("vpcopilot-lab", help="HTTP LB name")): """Read-only: confirm XC auth and show the LB's service-policy config + existing policies.""" diff --git a/src/vpcopilot/console/app.py b/src/vpcopilot/console/app.py index 94c7718..6ff7374 100644 --- a/src/vpcopilot/console/app.py +++ b/src/vpcopilot/console/app.py @@ -242,6 +242,24 @@ def runs(): return {"current": str(OUT), "runs": find_runs(".")} +@app.get("/api/drift") +def drift_ep(lb: str, control: str | None = None, policy: str | None = None, + finding: str | None = None): + """I2: what is on the LB now vs what the last run left vs what is about to be pushed. + Read-only — it never PUTs and never writes a snapshot, so it is safe to poll from the UI.""" + load_dotenv(ENV_PATH, override=True) + from ..apply import _load_probe + from ..drift import check + try: + exploit = (_load_probe(str(OUT), finding) or {}).get("exploit") if finding else None + art = OUT / "policies" / f"service_policy.{policy}.json" if policy else None + spec = json.loads(art.read_text()) if art and art.is_file() else None + return check(lb, out_dir=str(OUT), control=control, policy_name=policy, + exploit=exploit, spec=spec, log=lambda m: None) + except Exception as e: # noqa: BLE001 + raise HTTPException(400, str(e)) + + @app.get("/api/audit-verify") def audit_verify(): """J2: check the bundle this run last wrote (`/audit-bundle.zip`) against its own manifest. @@ -618,6 +636,7 @@ class ActionReq(BaseModel): refine_attempts: int | None = None allow_protected_lb: bool = False allow_overbroad: bool = False # G2: apply anyway when simulation flagged the policy as too broad + force: bool = False # I2: apply anyway when the pre-apply drift check objects _jobs: dict[str, dict] = {} # job_id -> {state, log, result, error, control, finding_id} @@ -650,9 +669,11 @@ def _dispatch_action(body: ActionReq, log): from ..refiner import refine_apply_service_policy return refine_apply_service_policy(art, body.lb, body.url, finding_id=body.finding_id, name=body.policy_name, keep=body.keep, allow_protected=body.allow_protected_lb, - max_refine=body.refine_attempts, config_path=_active_config, out_dir=str(OUT), log=log) + max_refine=body.refine_attempts, config_path=_active_config, force=body.force, + out_dir=str(OUT), log=log) return A.apply_from_scan(art, body.lb, body.url, name=body.policy_name, dry_run=body.dry_run, - keep=body.keep, allow_protected=body.allow_protected_lb, out_dir=str(OUT), log=log) + keep=body.keep, allow_protected=body.allow_protected_lb, force=body.force, + out_dir=str(OUT), log=log) if c == "malicious_user": return A.apply_malicious_user(body.lb, **kw) if c == "rate_limit": @@ -732,6 +753,7 @@ class ApplyReq(BaseModel): refine: bool = True refine_attempts: int | None = None allow_protected_lb: bool = False + force: bool = False @app.post("/api/apply") @@ -743,10 +765,11 @@ def do_apply(body: ApplyReq): from ..refiner import refine_apply_service_policy return refine_apply_service_policy(art, body.lb, body.url, name=body.name, keep=body.keep, allow_protected=body.allow_protected_lb, - max_refine=body.refine_attempts, out_dir=str(OUT), log=lambda m: None) + max_refine=body.refine_attempts, force=body.force, + out_dir=str(OUT), log=lambda m: None) from ..apply import apply_from_scan return apply_from_scan(art, body.lb, body.url, name=body.name, create_only=body.create_only, - dry_run=body.dry_run, keep=body.keep, + dry_run=body.dry_run, keep=body.keep, force=body.force, allow_protected=body.allow_protected_lb, out_dir=str(OUT), log=lambda m: None) except Exception as e: # noqa: BLE001 diff --git a/src/vpcopilot/console/static/index.html b/src/vpcopilot/console/static/index.html index 7428399..de8b480 100644 --- a/src/vpcopilot/console/static/index.html +++ b/src/vpcopilot/console/static/index.html @@ -200,6 +200,7 @@

④ Mitigate live with XC

Apply each recommended band-aid to the load balancer and validate it against the finding's real exploit — attach → validate → refine → keep or roll back. Turn off dry-run in Run settings to make it live. “✓ covered by …” means a sibling finding's band-aid already protects this one (same file + control). Only per-request controls (service_policy, api_schema) block a single fired exploit; rate-limit / bot / malicious-user / data-guard are behavioral and validate at config level.

+
Loading…
@@ -427,9 +428,32 @@ applies every band-aid one at a time under Run settings — dry-run rehearses, off goes live; continues past failures. ` : ""; mitigateBox.innerHTML = rows ? bar + `${rows}
findingsevmitigate
` : '
No verified findings — run a scan.
'; + loadDrift(); +} +// I2: the state of the target LB *before* you click anything. Read-only — /api/drift never PUTs and +// never writes a snapshot, so it is safe to call every time this section renders. +async function loadDrift(){ + const box=document.getElementById("driftBox"), st=S(); + if(!box) return; + if(!st.lb){ box.innerHTML=""; return; } + // Silent when the check cannot run — no XC creds loaded, LB not in this namespace. This strip is + // a convenience readout; the gate that actually matters runs inside the apply, where a failure + // is a failure. A red banner on a page you are merely looking at would be noise. + let d; try{ d=await jget("/api/drift?lb="+encodeURIComponent(st.lb)); } + catch(e){ box.innerHTML=""; return; } + const lvs=d.live_vs_snapshot||{}, ch=lvs.changes||[]; + const bits=[`${esc(d.lb)} currently has: ${(d.attached||[]).map(esc).join(", ")||"no controls attached"}`]; + if(ch.length){ + bits.push(`⚠ ${ch.length} field(s) changed on this LB since the last apply — ` + + `someone edited it outside this tool`); + bits.push(`
`
+      + ch.map(c=>`${esc(c.path)}: ${esc(JSON.stringify(c.was))} → ${esc(JSON.stringify(c.now))}`).join("\n")
+      + `
`); + } else if(lvs.snapshot){ bits.push('· no drift since the last snapshot'); } + box.innerHTML=`
${bits.join(" ")}
`; } // one apply → resolves ONLY when the job finishes, so a batch never fires two live applies at once -function applyOne(fid, control, policyName){ +function applyOne(fid, control, policyName, force){ return new Promise(resolve=>{ const st=S(); const box=document.getElementById("job-"+fid); if(!box){ resolve({ok:false}); return; } @@ -446,7 +470,7 @@ } jpost("/api/action",{control,finding_id:fid,policy_name:policyName,lb:st.lb,url:st.url, dry_run:st.dry,keep:st.keep,refine:st.refine,refine_attempts:st.refineAttempts,allow_protected_lb:st.allow, - allow_overbroad: !!(sim && sim.blocked_promotion)}) + allow_overbroad: !!(sim && sim.blocked_promotion), force: !!force}) .then(job=>{ let n=0; // lines already appended for this job — same full-transcript + ?since= tail as the scan log const poll=setInterval(async()=>{ @@ -454,7 +478,13 @@ pushLog(logEl, s.log); n = s.log_total ?? (n + (s.log||[]).length); if(s.state==="running"){ statEl.textContent=`applying ${esc(control)}…`; return; } clearInterval(poll); - if(s.state==="error"){ box.className="jobbox err"; statEl.textContent="error: "+s.error; loadHero(); resolve({ok:false}); return; } + if(s.state==="error"){ + box.className="jobbox err"; statEl.textContent="error: "+s.error; + // I2: the drift guard refuses rather than vetoes — offer the documented override inline. + if(/force=True/.test(s.error||"") && !force) + statEl.innerHTML+=` `; + loadHero(); resolve({ok:false}); return; } renderApplyResult(box,statEl,control,s.result||{}); loadHero(); const r=s.result||{}; resolve({ok: r.passed!==false && r.config_enabled!==false, result:r}); // mirror renderApplyResult's success test @@ -496,8 +526,14 @@ function renderApplyResult(box,statEl,control,res){ const ok = res.passed!==false && res.config_enabled!==false; box.className="jobbox "+(ok?"ok":"err"); + if(res.mode==="no_change"){ // I2: nothing was pushed, so pass/fail has nothing to report on + box.className="jobbox"; + statEl.innerHTML=`no change — ${esc(res.policy||control)} is already the ` + + `attached service policy; nothing was pushed. Use “apply anyway” to re-validate it.`; + return; + } const bits=[`${esc(res.mode||control)}`]; - if(res.passed!==undefined) bits.push(`passed=${res.passed}`); + if(res.passed!==undefined && res.passed!==null) bits.push(`passed=${res.passed}`); if(res.config_enabled!==undefined) bits.push(`enabled=${res.config_enabled}`); if(res.attempts>1) bits.push(`self-healed in ${res.attempts} attempts`); else if(res.attempts) bits.push(`${res.attempts} attempt`); diff --git a/src/vpcopilot/drift.py b/src/vpcopilot/drift.py new file mode 100644 index 0000000..3d2a6db --- /dev/null +++ b/src/vpcopilot/drift.py @@ -0,0 +1,278 @@ +"""I2 — drift and conflict detection: what is on the LB *now*, versus what the last run left, +versus what is about to be pushed. + +`ApplyContext.load()` already writes a per-LB timestamped snapshot before every change. That makes +the "what did we leave behind" half free; this turns it into a comparison you can run on its own, +and a guard the apply path can consult **before** it touches anything. + +Three questions, answered separately because they fail differently: + +1. **live vs last snapshot** — has someone hand-edited the LB in the XC console since we last + applied? Reported as a field-level diff, because "the LB changed" is useless and + `rate_limit.requests: 100 -> 5` is not. +2. **live vs proposed** — is what we are about to push already there? +3. **displacement and shadowing** — what does applying cost, and will the new policy actually fire? + +The roadmap wrote (3) as "an earlier ALLOW on an already-attached policy shadows the new DENY". +Running it against a live LB showed that cannot happen: both attach paths REPLACE the oneof with +exactly one policy — + + new_spec["active_service_policies"] = {"policies": [{"namespace": ..., "name": policy_name}]} + +— so this tool never leaves two service policies attached, and there is no earlier policy to be +shadowed by. Chasing the criterion as written produced a confident false positive on a real LB. + +What is real, and what replaced it: + +* **displacement** — that same replacement silently DETACHES whatever was attached. On a live LB + that is a loss of protection nothing was reporting. Warned and audited, never refused: replacing + the previous band-aid is the normal flow, and refusing would break every second apply. +* **shadowing, in its only true scope** — an ALLOW *inside the policy being applied* that matches + the exploit before its DENY does. `lint_service_policy` already walks exactly that, so this reuses + it rather than re-deriving it. + +Read-only, always: no PUT, no snapshot written, no run artifact touched. A drift check that +mutated the thing it inspects would be worthless as a pre-flight.""" +from __future__ import annotations + +import copy +import json +from pathlib import Path +from typing import Callable + +from .apply import rule_matches +from .controls import CONTROLS, detach_control + + +def save_snapshot(out_dir: str, lb: str, lb_obj: dict) -> str: + """Write a snapshot in the same place and shape `ApplyContext.load()` does — so a drift check + has something to compare against even on a run that has not applied anything yet.""" + snaps = Path(out_dir, "snapshots") + snaps.mkdir(parents=True, exist_ok=True) + from datetime import datetime, timezone + ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S") + p = snaps / f"{lb}-{ts}.json" + p.write_text(json.dumps(lb_obj, indent=2)) + return str(p) + + +def latest_snapshot(out_dir: str, lb: str) -> Path | None: + """The newest snapshot for THIS lb. Names sort lexically because the timestamp is + `%Y%m%dT%H%M%S`; `lb_snapshot.json` is deliberately ignored — it is overwritten by whichever + LB was applied last and cannot be attributed.""" + snaps = Path(out_dir, "snapshots") + if not snaps.is_dir(): + return None + files = sorted(snaps.glob(f"{lb}-*.json")) + return files[-1] if files else None + + +def diff_specs(was: dict, now: dict, *, _prefix: str = "") -> list[dict]: + """Field-level differences as dotted paths. `[{path, was, now}]`, sorted, with a whole subtree + reported at the point it appears or disappears rather than exploded leaf by leaf.""" + out: list[dict] = [] + for key in sorted(set(was) | set(now)): + path = f"{_prefix}{key}" + a, b = was.get(key), now.get(key) + if a == b: + continue + if isinstance(a, dict) and isinstance(b, dict): + out.extend(diff_specs(a, b, _prefix=f"{path}.")) + else: + out.append({"path": path, "was": a, "now": b}) + return out + + +def _attached_service_policies(spec: dict) -> list[dict]: + return ((spec.get("active_service_policies") or {}).get("policies") or []) + + +def _control_present(spec: dict, control: str) -> bool: + """Is this control attached at all? Determined by inverting `detach_control` — the registry is + the single source of truth for what "attached" means per control, so this cannot drift from + what rollback and retire do.""" + if control not in CONTROLS: + return False + probe = copy.deepcopy(spec) + detach_control(probe, control) + return probe != spec + + +def check(lb: str, *, out_dir: str = "out", control: str | None = None, + policy_name: str | None = None, exploit: dict | None = None, spec: dict | None = None, + xc=None, log: Callable = print) -> dict: + """Compare live / last-snapshot / proposed. Never mutates anything.""" + if xc is None: + from .xc import XC + xc = XC() + lb_obj = xc.get_lb(lb) + live = lb_obj.get("spec", {}) or {} + + snap_path = latest_snapshot(out_dir, lb) + changes: list[dict] = [] + if snap_path: + try: + was = (json.loads(snap_path.read_text()).get("spec") or {}) + changes = diff_specs(was, live) + except json.JSONDecodeError: + changes = [] + report = { + "lb": lb, + "namespace": getattr(xc, "ns", ""), + "live_vs_snapshot": { + "snapshot": str(snap_path) if snap_path else None, + "drifted": bool(changes), + "changes": changes, + }, + "attached": sorted(c for c in CONTROLS if _control_present(live, c)), + "proposed": {"control": control, "policy_name": policy_name, + "no_change": False, "already_attached": False, "note": ""}, + "displaces": [], + "conflicts": [], + "conflicts_checked": False, + "warnings": [], + "ok_to_apply": True, + } + + if control: + present = _control_present(live, control) + report["proposed"]["already_attached"] = present + if control == "service_policy" and policy_name: + # The proposed end state is exact — this policy name attached — so "unchanged" is a + # fact, not an inference, and skipping is safe. + names = [p.get("name") for p in _attached_service_policies(live)] + report["proposed"]["no_change"] = names == [policy_name] + if present and not report["proposed"]["no_change"]: + report["proposed"]["note"] = f"a different service policy is attached: {names}" + elif present: + # Presence is NOT parameter equality: a rate_limit already on at 100/MINUTE would read + # as unchanged while you push 5/MINUTE. Silently skipping a real change is worse than + # re-applying an identical one, so this reports and never claims no_change. + report["proposed"]["note"] = ( + f"{control} is already attached, but its parameters are not compared — " + "re-applying will push the new values") + + if control == "service_policy" and not report["proposed"]["no_change"]: + # What this apply COSTS. Attaching replaces the oneof wholesale, so every currently + # attached policy other than the one being applied is about to be detached. + for ref in _attached_service_policies(live): + name = ref.get("name") + if not name or name == policy_name: + continue + entry = {"policy": name, "denies": 0, "rules": 0, "protects_exploit": False} + try: + obj = xc.get_service_policy(name) + except Exception as e: # noqa: BLE001 — a policy we cannot read is a warning, not a crash + report["warnings"].append(f"could not read attached policy '{name}': {e}") + entry["unreadable"] = True + report["displaces"].append(entry) + continue + rules = ((obj.get("spec") or obj).get("rule_list") or {}).get("rules") or [] + entry["rules"] = len(rules) + entry["denies"] = sum(1 for r in rules if (r.get("spec") or {}).get("action") == "DENY") + if exploit: + for rule in rules: # FIRST_MATCH: the first matching rule decides this policy + rs = rule.get("spec") or {} + if rule_matches(rs, exploit): + entry["protects_exploit"] = rs.get("action") == "DENY" + break + report["displaces"].append(entry) + + if control == "service_policy" and spec is not None and exploit: + # Shadowing in its only true scope: inside the policy being applied. Reuses the A9 linter + # so drift and the linter can never disagree about what FIRST_MATCH means. + from .apply import lint_service_policy + report["conflicts_checked"] = True + for issue in lint_service_policy(spec, exploit): + if "ALLOW rule matches the exploit" in issue: + report["conflicts"].append({"policy": policy_name, "kind": "shadowed_deny", + "reason": issue}) + + report["ok_to_apply"] = not report["conflicts"] + return report + + +def summarize(report: dict) -> str: + """One line per finding, for a log sink or a CLI that already has a table.""" + lines = [] + lvs = report["live_vs_snapshot"] + if lvs["drifted"]: + lines.append(f"drift: {len(lvs['changes'])} field(s) changed on {report['lb']} since " + f"{Path(lvs['snapshot']).name if lvs['snapshot'] else 'the last snapshot'}") + for c in lvs["changes"]: + lines.append(f" {c['path']}: {json.dumps(c['was'])} -> {json.dumps(c['now'])}") + if report["proposed"]["no_change"]: + lines.append(f"no_change: {report['proposed']['policy_name']} is already the attached policy") + elif report["proposed"]["note"]: + lines.append(f"note: {report['proposed']['note']}") + for c in report["conflicts"]: + lines.append(f"conflict: {c['reason']}") + for w in report["warnings"]: + lines.append(f"⚠ {w}") + return "\n".join(lines) or f"no drift on {report['lb']}" + + +def preflight(lb: str, policy_name: str, *, out_dir: str, exploit: dict | None = None, + spec: dict | None = None, refine: bool = False, force: bool = False, + xc=None, log: Callable = print) -> dict | None: + """The pre-apply gate, in ONE place because both apply paths need it — `apply_from_scan` and + `refine_apply_service_policy`, the latter being the default for the CLI's `--from-scan` and the + console's Mitigate button. Two copies of a safety check is one copy that eventually stops + matching. + + Returns the drift report when the caller should short-circuit with `no_change`, or None to + proceed. Raises only when applying would provably accomplish nothing. + + `refine=True` downgrades the shadowed-DENY refusal to a warning: correcting rule order is + precisely what the refine loop exists to do, and refusing there would break the default path + to protect it from a problem it already fixes. + + Every outcome that changes what happens is audited — the refusal, the override, the skip, and + the detachment. A guard whose decisions are invisible cannot be reviewed after the fact.""" + from .audit import record as _audit + rep = check(lb, out_dir=out_dir, control="service_policy", policy_name=policy_name, + exploit=exploit, spec=spec, xc=xc, log=log) + lvs = rep["live_vs_snapshot"] + if lvs["drifted"]: + log(f" ⚠ drift: {len(lvs['changes'])} field(s) on {lb} changed since the last apply — " + "someone edited it outside this tool") + for c in lvs["changes"][:6]: + log(f" {c['path']}: {json.dumps(c['was'])[:60]} -> {json.dumps(c['now'])[:60]}") + _audit(out_dir, "drift_detected", lb=lb, policy=policy_name, + snapshot=lvs["snapshot"], changes=lvs["changes"]) + for w in rep["warnings"]: + log(f" ⚠ {w}") + + for d in rep["displaces"]: + # Not a veto — replacing the previous band-aid IS the normal flow. But it is a live loss of + # protection, so it is said out loud and written down rather than happening quietly. + detail = (f"{d['rules']} rule(s), {d['denies']} DENY" if not d.get("unreadable") + else "contents unreadable") + log(f" ⚠ applying will DETACH '{d['policy']}' ({detail})") + if d["protects_exploit"]: + log(f" — and '{d['policy']}' is currently what blocks this exploit") + _audit(out_dir, "policy_displaced", lb=lb, policy=policy_name, displaced=d["policy"], + denies=d["denies"], rules=d["rules"], protects_exploit=d["protects_exploit"]) + + if rep["conflicts"]: + first = rep["conflicts"][0] + log(f" ⚠ conflict: {first['reason']}") + if refine: + log(" — refine will reorder it at apply; continuing") + elif not force: + _audit(out_dir, "drift_block", lb=lb, policy=policy_name, conflicts=rep["conflicts"]) + raise RuntimeError(f"refusing to apply '{policy_name}': {first['reason']}. " + "Pass force=True (CLI --force, console “apply anyway”) to apply it " + "regardless.") + else: + _audit(out_dir, "drift_override", lb=lb, policy=policy_name, conflicts=rep["conflicts"]) + log(" ⚠ forced: applying past the conflict — the DENY may never fire") + + if rep["proposed"]["no_change"] and not force: + log(f"no_change — '{policy_name}' is already the attached service policy on {lb}; " + "nothing to push") + _audit(out_dir, "apply_skipped_no_change", lb=lb, policy=policy_name) + return rep + if rep["proposed"]["note"]: + log(f" note: {rep['proposed']['note']}") + return None diff --git a/src/vpcopilot/export.py b/src/vpcopilot/export.py index 1814801..9158910 100644 --- a/src/vpcopilot/export.py +++ b/src/vpcopilot/export.py @@ -38,6 +38,11 @@ def _quiet(_msg: str) -> None: "create_api_definition": "create", "refine_apply": "refine", "apply_timing": "timing", "open_pr": "cure", "retire": "retire", "rollback_failed": "rollback", + # Gate decisions. Nothing changed on the LB, which is exactly why they belong in the export: + # "we looked and refused" and "we looked and overrode" are both answers an auditor needs, and + # only the log has them. + "drift_detected": "gate", "policy_displaced": "gate", "drift_block": "gate", + "drift_override": "gate", "apply_skipped_no_change": "gate", "simulate_override": "gate", } # The XC control each action acted on, where the action name alone implies it. CONTROL = { @@ -46,6 +51,9 @@ def _quiet(_msg: str) -> None: "apply_rate_limit": "rate_limit", "apply_bot_defense": "bot_defense", "apply_waf": "waf", "create_app_firewall": "waf", "apply_data_guard": "waf_data_guard", "apply_api_schema": "api_schema", "create_api_definition": "api_schema", + "drift_detected": "service_policy", "policy_displaced": "service_policy", + "drift_block": "service_policy", "drift_override": "service_policy", + "apply_skipped_no_change": "service_policy", "simulate_override": "service_policy", } # Flat CSV columns, in reading order: when · who · what · why · where · outcome. COLUMNS = ["ts", "run_id", "actor", "host", "category", "action", "finding_id", "title", @@ -76,7 +84,13 @@ def _outcome(e: dict) -> str: uses yet another mix), so coalesce rather than trust one name.""" fixed = {"rollback_failed": "rollback_failed", "retire": "retired", "open_pr": "pr_opened", "create_service_policy": "created", "create_app_firewall": "created", - "create_api_definition": "created"}.get(e.get("action", "")) + "create_api_definition": "created", + # Gate decisions: the outcome IS the decision. Falling through to the pass/fail + # coalescing below would label a refusal "recorded", which reads like a no-op. + "drift_detected": "drift_detected", "policy_displaced": "displaced", + "drift_block": "refused", "drift_override": "overridden", + "apply_skipped_no_change": "no_change", + "simulate_override": "overridden"}.get(e.get("action", "")) if fixed: return fixed if e.get("unfixable"): diff --git a/src/vpcopilot/refiner.py b/src/vpcopilot/refiner.py index 4bb9408..2648435 100644 --- a/src/vpcopilot/refiner.py +++ b/src/vpcopilot/refiner.py @@ -71,6 +71,7 @@ def refine_apply_service_policy(artifact_path: str, lb: str, target_url: str, *, 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, + force: bool = False, 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. @@ -82,7 +83,12 @@ def refine_apply_service_policy(artifact_path: str, lb: str, target_url: str, *, 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.""" + skipped entirely and this behaves exactly as it did before G3. + + I2 — the same read-only pre-apply drift check `apply_from_scan` runs happens here too, because + this (not `apply_from_scan`) is the default path for both the CLI's `--from-scan` and the + console's Mitigate button. A guard the primary UX skips is not a guard. `force=True` bypasses + it.""" 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.") @@ -99,6 +105,13 @@ def refine_apply_service_policy(artifact_path: str, lb: str, target_url: str, *, finding = _load_finding(out_dir, finding_id) probe = _load_probe(out_dir, finding_id) + from .drift import preflight + d = preflight(lb, policy_name, out_dir=out_dir, force=force, xc=xc, log=log, spec=spec, + exploit=(probe or {}).get("exploit"), refine=True) + if d is not None: + return {"passed": None, "mode": "no_change", "policy": policy_name, "attempts": 0, + "drift": d} + lb_obj = xc.get_lb(lb) orig_spec = lb_obj.get("spec", {}) base_meta = {k: lb_obj["metadata"][k] for k in META_KEYS if k in lb_obj.get("metadata", {})} diff --git a/tests/conftest.py b/tests/conftest.py index 35b48d6..7f2046c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -45,6 +45,11 @@ def list_service_policies(self): def service_policy_exists(self, name): return name in self.service_policies + def get_service_policy(self, name): + if name not in self.service_policies: # the real client raises XCError on a 404 + raise RuntimeError(f"GET service_policy {name} -> 404") + return copy.deepcopy(self.service_policies[name]) + def create_service_policy(self, obj): self.service_policies[obj["metadata"]["name"]] = copy.deepcopy(obj) return obj diff --git a/tests/test_console_drift.py b/tests/test_console_drift.py new file mode 100644 index 0000000..6fa2cc8 --- /dev/null +++ b/tests/test_console_drift.py @@ -0,0 +1,112 @@ +"""I2 console surface: `GET /api/drift` answers the pre-apply question read-only, and the apply +endpoints thread `force` so the operator can override from the UI. + +The endpoint must stay read-only under every input — it is polled from the browser, so a version +that wrote a snapshot would corrupt the run dir just by someone opening a page.""" +import json + +from fastapi.testclient import TestClient + +from vpcopilot.console import app as A + + +def _client(): + return TestClient(A.app) + + +def _artifact(out, name="deny-x", shadowed=False): + (out / "policies").mkdir(parents=True, exist_ok=True) + deny = {"action": "DENY", "path": {"exact_values": ["/users/v1/register"]}, + "http_method": {"methods": ["POST"]}} + allow = {"action": "ALLOW", "path": {"prefix_values": ["/"]}} + rules = [allow, deny] if shadowed else [deny, allow] + (out / "policies" / f"service_policy.{name}.json").write_text(json.dumps( + {"metadata": {"name": name}, "spec": {"rule_list": {"rules": [{"spec": r} for r in rules]}}})) + + +def _probe(out, fid="f1"): + (out / "probes.json").write_text(json.dumps( + [{"finding_id": fid, "exploit": {"method": "POST", "path": "/users/v1/register"}}])) + + +def test_drift_reports_the_field_level_diff(tmp_path, monkeypatch, fake_xc): + monkeypatch.setattr(A, "OUT", tmp_path) + monkeypatch.setattr("vpcopilot.drift.XC", lambda *a, **k: fake_xc, raising=False) + monkeypatch.setattr("vpcopilot.xc.XC", lambda *a, **k: fake_xc) + from vpcopilot import drift + drift.save_snapshot(str(tmp_path), "lab", {"spec": {"rate_limit": {"requests": 100}}}) + fake_xc.lb["spec"] = {"rate_limit": {"requests": 5}} + body = _client().get("/api/drift?lb=lab").json() + assert body["live_vs_snapshot"]["drifted"] is True + assert body["live_vs_snapshot"]["changes"] == [ + {"path": "rate_limit.requests", "was": 100, "now": 5}] + + +def test_drift_picks_up_this_runs_artifact_for_the_shadowing_check(tmp_path, monkeypatch, fake_xc): + """The browser sends a policy NAME, not a spec. Without the endpoint resolving that to the + artifact on disk the shadowing check has no rules to walk and silently reports nothing.""" + monkeypatch.setattr(A, "OUT", tmp_path) + monkeypatch.setattr("vpcopilot.xc.XC", lambda *a, **k: fake_xc) + _artifact(tmp_path, shadowed=True) + _probe(tmp_path) + body = _client().get("/api/drift?lb=lab&control=service_policy&policy=deny-x&finding=f1").json() + assert body["conflicts_checked"] is True + assert body["conflicts"] and body["conflicts"][0]["kind"] == "shadowed_deny" + assert body["ok_to_apply"] is False + + +def test_a_correctly_ordered_policy_is_clean(tmp_path, monkeypatch, fake_xc): + monkeypatch.setattr(A, "OUT", tmp_path) + monkeypatch.setattr("vpcopilot.xc.XC", lambda *a, **k: fake_xc) + _artifact(tmp_path) + _probe(tmp_path) + body = _client().get("/api/drift?lb=lab&control=service_policy&policy=deny-x&finding=f1").json() + assert body["ok_to_apply"] is True and not body["conflicts"] + + +def test_drift_reports_what_applying_would_detach(tmp_path, monkeypatch, fake_xc): + monkeypatch.setattr(A, "OUT", tmp_path) + monkeypatch.setattr("vpcopilot.xc.XC", lambda *a, **k: fake_xc) + fake_xc.service_policies["someone-elses"] = {"spec": {"rule_list": {"rules": [ + {"spec": {"action": "DENY", "path": {"prefix_values": ["/admin"]}}}]}}} + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "someone-elses"}]}} + _artifact(tmp_path) + body = _client().get("/api/drift?lb=lab&control=service_policy&policy=deny-x").json() + assert [d["policy"] for d in body["displaces"]] == ["someone-elses"] + + +def test_the_endpoint_writes_nothing(tmp_path, monkeypatch, fake_xc): + """Polled from the browser — a version that wrote a snapshot would corrupt the run dir just by + someone opening the page.""" + monkeypatch.setattr(A, "OUT", tmp_path) + monkeypatch.setattr("vpcopilot.xc.XC", lambda *a, **k: fake_xc) + _artifact(tmp_path) + _probe(tmp_path) + before = sorted(p.name for p in tmp_path.iterdir()) + _client().get("/api/drift?lb=lab&control=service_policy&policy=deny-x&finding=f1") + assert sorted(p.name for p in tmp_path.iterdir()) == before + assert fake_xc.put_lb_calls == [] + + +def test_an_unknown_lb_is_a_400_not_a_500(tmp_path, monkeypatch, fake_xc): + monkeypatch.setattr(A, "OUT", tmp_path) + + def boom(*a, **k): + raise RuntimeError("404 load balancer not found") + monkeypatch.setattr(fake_xc, "get_lb", boom) + monkeypatch.setattr("vpcopilot.xc.XC", lambda *a, **k: fake_xc) + assert _client().get("/api/drift?lb=nope").status_code == 400 + + +def test_the_apply_endpoints_accept_force(tmp_path, monkeypatch): + """The console's override has to reach the guard — a UI button wired to a field the request + model drops would look like it worked and change nothing.""" + monkeypatch.setattr(A, "OUT", tmp_path) + seen = {} + monkeypatch.setattr("vpcopilot.refiner.refine_apply_service_policy", + lambda *a, **k: seen.update(k) or {"passed": True}) + _artifact(tmp_path) + r = _client().post("/api/apply", json={"artifact": "service_policy.deny-x.json", + "lb": "lab", "url": "http://x", "force": True}) + assert r.status_code == 200 and seen["force"] is True diff --git a/tests/test_drift.py b/tests/test_drift.py new file mode 100644 index 0000000..840db11 --- /dev/null +++ b/tests/test_drift.py @@ -0,0 +1,386 @@ +"""I2 — drift and conflict detection: what is on the LB now, versus what the last run left, versus +what is about to be pushed. + +Read-only throughout: `drift` never PUTs, never writes a snapshot, and never touches the run's +artifacts. Offline against FakeXC.""" +import json + +import pytest + +from vpcopilot import drift + +EXPLOIT = {"method": "POST", "path": "/users/v1/register"} + + +def _policy(*rules): + return {"metadata": {"name": "p"}, "spec": {"rule_list": {"rules": [{"spec": r} for r in rules]}}} + + +DENY_REGISTER = {"action": "DENY", "path": {"exact_values": ["/users/v1/register"]}, + "http_method": {"methods": ["POST"]}} +ALLOW_ALL = {"action": "ALLOW", "path": {"prefix_values": ["/"]}} + + +# ---- live vs the last snapshot: did someone hand-edit the LB? ---- +def test_no_snapshot_yet_is_not_drift(fake_xc, tmp_path): + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc) + assert r["live_vs_snapshot"]["drifted"] is False + assert r["live_vs_snapshot"]["snapshot"] is None + + +def test_an_unchanged_lb_shows_no_drift(fake_xc, tmp_path): + drift.save_snapshot(str(tmp_path), "lab", fake_xc.get_lb("lab")) + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc) + assert r["live_vs_snapshot"]["drifted"] is False and not r["live_vs_snapshot"]["changes"] + + +def test_a_hand_edit_since_the_last_apply_is_a_field_level_diff(fake_xc, tmp_path): + drift.save_snapshot(str(tmp_path), "lab", fake_xc.get_lb("lab")) + fake_xc.lb["spec"]["app_firewall"] = {"name": "someone-added-this"} # edited in the XC console + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc) + assert r["live_vs_snapshot"]["drifted"] is True + ch = r["live_vs_snapshot"]["changes"] + assert any(c["path"] == "app_firewall" and c["was"] is None for c in ch) + + +def test_a_removed_field_is_reported_too(fake_xc, tmp_path): + fake_xc.lb["spec"]["rate_limit"] = {"requests": 5} + drift.save_snapshot(str(tmp_path), "lab", fake_xc.get_lb("lab")) + del fake_xc.lb["spec"]["rate_limit"] + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc) + assert any(c["path"] == "rate_limit" and c["now"] is None + for c in r["live_vs_snapshot"]["changes"]) + + +def test_a_nested_change_reports_the_dotted_path(fake_xc, tmp_path): + fake_xc.lb["spec"]["rate_limit"] = {"requests": 100, "unit": "MINUTE"} + drift.save_snapshot(str(tmp_path), "lab", fake_xc.get_lb("lab")) + fake_xc.lb["spec"]["rate_limit"]["requests"] = 5 + ch = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc)["live_vs_snapshot"]["changes"] + assert any(c["path"] == "rate_limit.requests" and c["was"] == 100 and c["now"] == 5 for c in ch) + + +# ---- live vs proposed: is this already what is on the LB? ---- +def test_reapplying_the_same_service_policy_is_no_change(fake_xc, tmp_path): + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "deny-x"}]}} + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc, + control="service_policy", policy_name="deny-x") + assert r["proposed"]["no_change"] is True + + +def test_applying_a_different_policy_is_a_change(fake_xc, tmp_path): + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "deny-other"}]}} + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc, + control="service_policy", policy_name="deny-x") + assert r["proposed"]["no_change"] is False + + +def test_an_lb_wide_control_reports_presence_but_never_claims_no_change(fake_xc, tmp_path): + """Presence is not parameter equality: a rate_limit already attached at 100/MINUTE would read + as 'unchanged' while you push 5/MINUTE. Silently skipping a real change is worse than the bug + this item fixes, so these are reported and NOT auto-skipped.""" + fake_xc.lb["spec"]["rate_limit"] = {"requests": 100, "unit": "MINUTE"} + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc, control="rate_limit") + assert r["proposed"]["already_attached"] is True + assert r["proposed"]["no_change"] is False + assert "parameter" in r["proposed"]["note"].lower() + + +# ---- displacement: what applying COSTS ---- +# The roadmap wrote this as "an earlier ALLOW on an attached policy shadows the new DENY". A live +# run disproved it: both attach paths replace `active_service_policies` with exactly one policy, so +# two are never attached at once and there is no earlier policy to be shadowed by. What the +# replacement DOES do is silently detach whatever was there — which nothing reported until now. +def test_applying_reports_the_policy_it_will_detach(fake_xc, tmp_path): + fake_xc.service_policies["existing"] = _policy(DENY_REGISTER, ALLOW_ALL) + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "existing"}]}} + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc, control="service_policy", + policy_name="deny-x", exploit=EXPLOIT) + assert [d["policy"] for d in r["displaces"]] == ["existing"] + assert r["displaces"][0] == {"policy": "existing", "rules": 2, "denies": 1, + "protects_exploit": True} + + +def test_displacement_never_blocks_the_apply(fake_xc, tmp_path): + """Replacing the previous band-aid IS the normal flow — refusing would break every second + apply. It is warned and audited, not vetoed.""" + fake_xc.service_policies["existing"] = _policy(DENY_REGISTER, ALLOW_ALL) + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "existing"}]}} + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc, control="service_policy", + policy_name="deny-x", exploit=EXPLOIT) + assert r["ok_to_apply"] is True and not r["conflicts"] + + +def test_a_displaced_policy_that_does_not_block_the_exploit_says_so(fake_xc, tmp_path): + fake_xc.service_policies["existing"] = _policy(ALLOW_ALL) + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "existing"}]}} + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc, control="service_policy", + policy_name="deny-x", exploit=EXPLOIT) + assert r["displaces"][0]["protects_exploit"] is False + + +def test_replacing_a_policy_with_itself_displaces_nothing(fake_xc, tmp_path): + fake_xc.service_policies["deny-x"] = _policy(ALLOW_ALL) + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "deny-x"}]}} + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc, control="service_policy", + policy_name="deny-x", exploit=EXPLOIT) + assert r["displaces"] == [] and not r["conflicts"] + + +# ---- shadowing, in its only true scope: inside the policy being applied ---- +def test_an_allow_before_the_deny_in_the_proposed_policy_is_a_conflict(fake_xc, tmp_path): + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc, control="service_policy", + policy_name="deny-x", exploit=EXPLOIT, + spec={"rule_list": {"rules": [{"spec": ALLOW_ALL}, {"spec": DENY_REGISTER}]}}) + assert r["conflicts"] and r["conflicts"][0]["kind"] == "shadowed_deny" + assert r["ok_to_apply"] is False + + +def test_the_right_rule_order_is_not_a_conflict(fake_xc, tmp_path): + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc, control="service_policy", + policy_name="deny-x", exploit=EXPLOIT, + spec={"rule_list": {"rules": [{"spec": DENY_REGISTER}, {"spec": ALLOW_ALL}]}}) + assert not r["conflicts"] and r["ok_to_apply"] is True + + +def test_shadowing_agrees_with_the_linter(fake_xc, tmp_path): + """drift reuses `lint_service_policy` rather than re-deriving FIRST_MATCH, so the two can never + disagree about what shadows what.""" + from vpcopilot.apply import lint_service_policy + spec = {"rule_list": {"rules": [{"spec": ALLOW_ALL}, {"spec": DENY_REGISTER}]}} + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc, control="service_policy", + policy_name="deny-x", exploit=EXPLOIT, spec=spec) + assert r["conflicts"][0]["reason"] in lint_service_policy(spec, EXPLOIT) + + +def test_no_exploit_means_no_shadowing_claim(fake_xc, tmp_path): + """Shadowing is relative to a concrete request. Without one there is nothing to decide, and + guessing would be worse than saying so.""" + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc, control="service_policy", + policy_name="deny-x", + spec={"rule_list": {"rules": [{"spec": ALLOW_ALL}, {"spec": DENY_REGISTER}]}}) + assert not r["conflicts"] and r["conflicts_checked"] is False + + +def test_an_unreadable_attached_policy_is_reported_not_raised(fake_xc, tmp_path): + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "vanished"}]}} + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc, control="service_policy", + policy_name="deny-x", exploit=EXPLOIT) + assert any("vanished" in w for w in r["warnings"]) + + +# ---- read-only ---- +def test_drift_never_writes_and_never_puts(fake_xc, tmp_path): + drift.save_snapshot(str(tmp_path), "lab", fake_xc.get_lb("lab")) + before = sorted(p.name for p in tmp_path.rglob("*")) + drift.check("lab", out_dir=str(tmp_path), xc=fake_xc, control="service_policy", + policy_name="deny-x", exploit=EXPLOIT) + assert sorted(p.name for p in tmp_path.rglob("*")) == before + assert fake_xc.put_lb_calls == [] + + +def test_it_reads_the_newest_snapshot(fake_xc, tmp_path): + snaps = tmp_path / "snapshots" + snaps.mkdir() + (snaps / "lab-20260101T000000.json").write_text(json.dumps({"spec": {"old": True}})) + (snaps / "lab-20260727T120000.json").write_text(json.dumps({"spec": {"no_service_policies": {}}})) + (snaps / "other-20260727T130000.json").write_text(json.dumps({"spec": {"wrong-lb": True}})) + r = drift.check("lab", out_dir=str(tmp_path), xc=fake_xc) + assert r["live_vs_snapshot"]["snapshot"].endswith("lab-20260727T120000.json") + assert r["live_vs_snapshot"]["drifted"] is False + + +# ---- the matcher extracted from the linter must behave identically ---- +def test_rule_matches_is_the_same_logic_the_linter_uses(): + from vpcopilot.apply import lint_service_policy, rule_matches + assert rule_matches(DENY_REGISTER, EXPLOIT) is True + assert rule_matches({"action": "DENY", "path": {"exact_values": ["/other"]}}, EXPLOIT) is False + # the linter still catches an ALLOW shadowing the exploit inside one spec + bad = {"rule_list": {"rules": [{"spec": ALLOW_ALL}, {"spec": DENY_REGISTER}]}} + assert lint_service_policy(bad, EXPLOIT) + + +# ---- the pre-apply guard ---- +def _artifact(tmp_path, name="deny-x"): + art = tmp_path / f"service_policy.{name}.json" + art.write_text(json.dumps({"metadata": {"name": name}, "spec": {"rule_list": {"rules": [ + {"spec": DENY_REGISTER}, {"spec": ALLOW_ALL}]}}})) + return str(art) + + +def _guard_env(monkeypatch, fake_xc, noop_sleep): + from vpcopilot import apply as A + import vpcopilot.engine as engine + real = engine.ApplyContext.__post_init__ + + def patched(self): + real(self) + self.sleep = noop_sleep + monkeypatch.setattr(engine.ApplyContext, "__post_init__", patched) + monkeypatch.setattr(A, "XC", lambda *a, **k: fake_xc) + monkeypatch.setenv("VPCOPILOT_PROTECTED_LBS", "nimbus-www") + return A + + +def test_reapplying_the_attached_policy_reports_no_change_and_writes_nothing( + monkeypatch, fake_xc, tmp_path, noop_sleep): + A = _guard_env(monkeypatch, fake_xc, noop_sleep) + fake_xc.service_policies["deny-x"] = _policy(DENY_REGISTER, ALLOW_ALL) + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "deny-x"}]}} + res = A.apply_from_scan(_artifact(tmp_path), "lab", "http://x", out_dir=str(tmp_path), + log=lambda m: None) + assert res["mode"] == "no_change" + assert fake_xc.put_lb_calls == [] # nothing pushed + assert not (tmp_path / "snapshots").exists() # and no snapshot written + assert not (tmp_path / "lb_snapshot.json").exists() + + +def test_force_reapplies_a_policy_that_is_already_attached(monkeypatch, fake_xc, tmp_path, noop_sleep): + A = _guard_env(monkeypatch, fake_xc, noop_sleep) + monkeypatch.setattr(A, "_run_validation", + lambda *a, **k: {"exploit_status": 403, "exploit_blocked": True, "legit_ok": True}) + fake_xc.service_policies["deny-x"] = _policy(DENY_REGISTER, ALLOW_ALL) + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "deny-x"}]}} + res = A.apply_from_scan(_artifact(tmp_path), "lab", "http://x", out_dir=str(tmp_path), + force=True, log=lambda m: None) + assert res["mode"] != "no_change" and fake_xc.put_lb_calls + + +def test_a_shadowed_deny_blocks_the_apply(monkeypatch, fake_xc, tmp_path, noop_sleep): + """Rule order that makes the band-aid a no-op is caught before the policy is even created — + not after a live attach, a validate, and a rollback.""" + A = _guard_env(monkeypatch, fake_xc, noop_sleep) + (tmp_path / "probes.json").write_text(json.dumps([{"finding_id": "f1", "exploit": EXPLOIT}])) + art = tmp_path / "service_policy.shadowed.json" + art.write_text(json.dumps({"metadata": {"name": "shadowed"}, "spec": {"rule_list": {"rules": [ + {"spec": ALLOW_ALL}, {"spec": DENY_REGISTER}]}}})) + with pytest.raises(RuntimeError, match="FIRST_MATCH"): + A.apply_from_scan(str(art), "lab", "http://x", finding_id="f1", + out_dir=str(tmp_path), log=lambda m: None) + assert fake_xc.put_lb_calls == [] + assert "shadowed" not in fake_xc.service_policies # nor a stray object left in the tenant + + +def test_displacement_warns_but_still_applies(monkeypatch, fake_xc, tmp_path, noop_sleep): + """No regressions: an LB that already carries someone else's policy must still be mitigable.""" + A = _guard_env(monkeypatch, fake_xc, noop_sleep) + monkeypatch.setattr(A, "_run_validation", + lambda *a, **k: {"exploit_status": 403, "exploit_blocked": True, "legit_ok": True}) + fake_xc.service_policies["someone-elses"] = _policy(DENY_REGISTER, ALLOW_ALL) + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "someone-elses"}]}} + res = A.apply_from_scan(_artifact(tmp_path), "lab", "http://x", out_dir=str(tmp_path), + log=lambda m: None) + assert res["mode"] != "no_change" and fake_xc.put_lb_calls + assert "policy_displaced" in _actions(tmp_path) + + +def test_a_dry_run_is_never_gated(monkeypatch, fake_xc, tmp_path, noop_sleep): + """Dry-run changes nothing, so there is nothing for a pre-apply guard to prevent.""" + A = _guard_env(monkeypatch, fake_xc, noop_sleep) + fake_xc.service_policies["deny-x"] = _policy(DENY_REGISTER, ALLOW_ALL) + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "deny-x"}]}} + res = A.apply_from_scan(_artifact(tmp_path), "lab", "http://x", dry_run=True, + out_dir=str(tmp_path), log=lambda m: None) + assert res["mode"] != "no_change" + + +def test_an_unrelated_attached_policy_does_not_trigger_no_change(monkeypatch, fake_xc, tmp_path, noop_sleep): + A = _guard_env(monkeypatch, fake_xc, noop_sleep) + monkeypatch.setattr(A, "_run_validation", + lambda *a, **k: {"exploit_status": 403, "exploit_blocked": True, "legit_ok": True}) + fake_xc.service_policies["something-else"] = _policy(DENY_REGISTER) + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "something-else"}]}} + res = A.apply_from_scan(_artifact(tmp_path), "lab", "http://x", out_dir=str(tmp_path), + log=lambda m: None) + assert res["mode"] != "no_change" + + +# ---- the shared preflight, and the audit trail it leaves ---- +def _actions(tmp_path): + from vpcopilot import audit + return [e["action"] for e in audit.load(str(tmp_path))] + + +def test_the_refiner_is_gated_too(monkeypatch, fake_xc, tmp_path, noop_sleep): + """The refiner — not apply_from_scan — is the default path for both `--from-scan` and the + console's Mitigate button. A guard the primary UX skips is not a guard.""" + import vpcopilot.refiner as R + monkeypatch.setattr(R, "XC", lambda *a, **k: fake_xc) + drift.save_snapshot(str(tmp_path), "lab", {"spec": {}}) + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "deny-x"}]}} + res = R.refine_apply_service_policy(_artifact(tmp_path), "lab", "http://x", + out_dir=str(tmp_path), log=lambda m: None) + assert res["mode"] == "no_change" and fake_xc.put_lb_calls == [] + assert "drift_detected" in _actions(tmp_path) + + +def test_the_refiner_warns_about_a_shadowed_deny_instead_of_refusing(fake_xc, tmp_path): + """Reordering rules is exactly what the refine loop exists to do. Refusing on the refine path + would break the default flow to protect it from a problem it already fixes.""" + lines = [] + assert drift.preflight("lab", "deny-x", out_dir=str(tmp_path), exploit=EXPLOIT, refine=True, + spec={"rule_list": {"rules": [{"spec": ALLOW_ALL}, + {"spec": DENY_REGISTER}]}}, + xc=fake_xc, log=lines.append) is None + assert any("refine will reorder" in ln for ln in lines) + assert "drift_block" not in _actions(tmp_path) + + +def test_a_refusal_is_audited(fake_xc, tmp_path): + with pytest.raises(RuntimeError): + drift.preflight("lab", "deny-x", out_dir=str(tmp_path), exploit=EXPLOIT, + spec={"rule_list": {"rules": [{"spec": ALLOW_ALL}, + {"spec": DENY_REGISTER}]}}, + xc=fake_xc, log=lambda m: None) + assert "drift_block" in _actions(tmp_path) + + +def test_forcing_past_a_conflict_is_audited(fake_xc, tmp_path): + """Overriding is allowed; overriding silently is not — the whole point of the audit trail.""" + assert drift.preflight("lab", "deny-x", out_dir=str(tmp_path), exploit=EXPLOIT, force=True, + spec={"rule_list": {"rules": [{"spec": ALLOW_ALL}, + {"spec": DENY_REGISTER}]}}, + xc=fake_xc, log=lambda m: None) is None + acts = _actions(tmp_path) + assert "drift_override" in acts and "drift_block" not in acts + + +def test_displacing_a_live_policy_is_audited(fake_xc, tmp_path): + fake_xc.service_policies["someone-elses"] = _policy(DENY_REGISTER, ALLOW_ALL) + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "someone-elses"}]}} + assert drift.preflight("lab", "deny-x", out_dir=str(tmp_path), exploit=EXPLOIT, + xc=fake_xc, log=lambda m: None) is None + from vpcopilot import audit + ev = [e for e in audit.load(str(tmp_path)) if e["action"] == "policy_displaced"] + assert ev and ev[0]["displaced"] == "someone-elses" and ev[0]["protects_exploit"] is True + + +def test_a_skip_and_hand_edited_drift_are_both_audited(fake_xc, tmp_path): + drift.save_snapshot(str(tmp_path), "lab", {"spec": {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "deny-x"}]}}}) + fake_xc.lb["spec"] = {"active_service_policies": {"policies": [ + {"namespace": fake_xc.ns, "name": "deny-x"}]}, "rate_limit": {"requests": 5}} + rep = drift.preflight("lab", "deny-x", out_dir=str(tmp_path), xc=fake_xc, log=lambda m: None) + assert rep is not None and rep["proposed"]["no_change"] + acts = _actions(tmp_path) + assert "apply_skipped_no_change" in acts and "drift_detected" in acts + + +def test_preflight_still_writes_no_lb_change(fake_xc, tmp_path): + drift.preflight("lab", "deny-x", out_dir=str(tmp_path), exploit=EXPLOIT, + xc=fake_xc, log=lambda m: None) + assert fake_xc.put_lb_calls == [] and not (tmp_path / "snapshots").exists() diff --git a/tests/test_export.py b/tests/test_export.py index 496d20c..da63b02 100644 --- a/tests/test_export.py +++ b/tests/test_export.py @@ -187,3 +187,32 @@ def test_bundle_all_folders_each_run_under_an_index(tmp_path): idx = json.loads(z.read("index.json")) assert {r["folder"] for r in idx["runs"]} == {"out-a", "out-b"} assert all(r["events"] == 1 for r in idx["runs"]) + + +# ---- I2: gate decisions are evidence too ---- +def test_gate_decisions_survive_the_export(tmp_path): + """Nothing changed on the LB, which is exactly why these belong in the export: "we looked and + refused" and "we looked and overrode" are answers only the log has.""" + from vpcopilot import audit + from vpcopilot.export import build_audit_events + for act in ("drift_detected", "policy_displaced", "drift_block", "drift_override", + "apply_skipped_no_change", "simulate_override"): + audit.record(str(tmp_path), act, lb="lab", policy="deny-x", finding_id="f1") + ev = {e["action"]: e for e in build_audit_events(str(tmp_path))} + assert len(ev) == 6 + assert all(e["category"] == "gate" for e in ev.values()) + assert ev["drift_block"]["outcome"] == "refused" + assert ev["drift_override"]["outcome"] == "overridden" + assert ev["apply_skipped_no_change"]["outcome"] == "no_change" + assert ev["policy_displaced"]["outcome"] == "displaced" + # and they carry the control, so a service-policy filter does not lose them + assert all(e["control"] == "service_policy" for e in ev.values()) + + +def test_a_refusal_is_not_labelled_recorded(tmp_path): + """Falling through to the pass/fail coalescing would label a refusal "recorded", which reads + like a no-op — the opposite of what happened.""" + from vpcopilot import audit + from vpcopilot.export import build_audit_events + audit.record(str(tmp_path), "drift_block", lb="lab", policy="deny-x") + assert build_audit_events(str(tmp_path))[0]["outcome"] == "refused"