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"