diff --git a/BACKLOG.md b/BACKLOG.md index b11d58f..5392b90 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -221,19 +221,19 @@ d=pathlib.Path(tempfile` ### C. Concurrency and shared state -- [ ] **HIGH** `console/app.py:989` — console `_run_action` reads the global `OUT` inside the worker thread, so an in-flight apply's `apply_timing` audit record lands in whatever run dir a concurrent `POST /api/scan` has since repointed the console at — the run that owns the LB change loses its MTTM measurement and an unrelated run gains one +- [x] **HIGH** `console/app.py:989` — console `_run_action` reads the global `OUT` inside the worker thread, so an in-flight apply's `apply_timing` audit record lands in whatever run dir a concurrent `POST /api/scan` has since repointed the console at — the run that owns the LB change loses its MTTM measurement and an unrelated run gains one - *Fails when:* Operator starts an apply against out-A from the Mitigate step (`POST /api/action`, dry_run=false). The control is pushed to the load balancer. While the job is still running, a scan is started into out-B (`POST /api/scan`), which does `global OUT; OUT = Path(body.out)` at app.py:856. `_run_action` then evaluates `record(str(OUT), "apply_timing", ...)` at app.py:989 — reading the global, not a capt - *Repro:* `/Users/d.henley/demos/virtual-patch-copilot/.venv/bin/python /tmp/vpc_repro/t_out_race.py # starts /api/action against out-A with apply_malicious_user stubbed to block, then POST` - *Why the suite misses it:* Every console test monkeypatches `A.OUT` to a single tmp_path and never changes it while a job is in flight, so no test ever exercises two run dirs in one process. `tests/test_impact.py` writes `apply_timing` records directly into the dir it then reads, so the join is never tested across a reassignm -- [ ] **HIGH** `runmeta.py:118` — `runmeta.write_manifest` mints `run_id` in an unguarded read-modify-write — `_MINT_LOCK` only guards `run_id()`, so a scan finishing while any thread records an audit entry produces entries whose run_id does not exist in run.json +- [x] **HIGH** `runmeta.py:118` — `runmeta.write_manifest` mints `run_id` in an unguarded read-modify-write — `_MINT_LOCK` only guards `run_id()`, so a scan finishing while any thread records an audit entry produces entries whose run_id does not exist in run.json - *Fails when:* `run_id()` takes `_MINT_LOCK` (line 89) for its load→mint→save. `write_manifest` does the identical load (line 117) → `setdefault("run_id", uuid4())` (line 118) → `_save` (line 122) with NO lock, and the window between them contains `actor()` (getpass) and `host()` (gethostname) syscalls. Interleaving on a run dir that has no run.json yet: the audit thread loads {}, mints rid1 under the lock, save - *Repro:* `/Users/d.henley/demos/virtual-patch-copilot/.venv/bin/python /tmp/vpc_repro/t_runmeta_manifest.py # 200 trials, 2 threads, NO injected sleeps: one thread calls runmeta.write_mani` - *Why the suite misses it:* `tests/test_audit_provenance.py` exercises the concurrency guard only through `runmeta.run_id`/`audit.record` (the path that IS locked). No test ever calls `write_manifest` concurrently with anything, so the second, unlocked mint of the same field is untested. -- [ ] **MEDIUM** `console/app.py:841` — console `POST /api/scan`'s "a scan is already running" guard is an unlocked check-then-act on state the endpoint never sets, so two concurrent scans both start into the same out dir — the interleaving MCP explicitly declines +- [x] **MEDIUM** `console/app.py:841` — console `POST /api/scan`'s "a scan is already running" guard is an unlocked check-then-act on state the endpoint never sets, so two concurrent scans both start into the same out dir — the interleaving MCP explicitly declines - *Fails when:* `start_scan` checks `_scan["state"] == "running"` (app.py:841) but never sets it; `state="running"` is set by `_run_scan` (app.py:823) once the daemon thread is scheduled. Two requests that arrive before that — a double-clicked Start button, two browser tabs, a retried POST — both read the stale state, both pass the guard and both spawn a pipeline into the same out dir. Both pipelines then write f - *Repro:* `/Users/d.henley/demos/virtual-patch-copilot/.venv/bin/python /tmp/vpc_repro/t_double_scan_rate.py # 20 trials, 2 threads barrier-synced on POST /api/scan with run_pipeline stubbe` - *Why the suite misses it:* The 409 guard is tested only sequentially (post, then post again), where the worker thread has already been scheduled and the guard does hold — my sequential control case returned 409 correctly. No console test issues two requests concurrently. -- [ ] **MEDIUM** `backfill.py:230` — `backfill.py` writes its sidecar through a fixed temp filename with no pid or thread id, so concurrent `POST /api/audit-backfill` requests collide on it and all but one 500 with FileNotFoundError from `os.replace` +- [x] **MEDIUM** `backfill.py:230` — `backfill.py` writes its sidecar through a fixed temp filename with no pid or thread id, so concurrent `POST /api/audit-backfill` requests collide on it and all but one 500 with FileNotFoundError from `os.replace` - *Fails when:* `backfill()` writes `/audit-backfill.json.tmp` and then `os.replace`s it onto the sidecar. The name is constant, so every concurrent caller shares one temp path. FastAPI runs the sync `audit_backfill` endpoint in the anyio threadpool, so two Retire-step clicks (or two tabs polling) run it on different threads in the same process: T1 replaces the shared tmp onto the sidecar, T2's `os.replace` - *Repro:* `/Users/d.henley/demos/virtual-patch-copilot/.venv/bin/python /tmp/vpc_repro/t_backfill_console_500.py # 8 barrier-synced POST /api/audit-backfill against a 400-entry run dir, Tes` - *Why the suite misses it:* `tests/test_backfill.py` only ever calls `backfill()` sequentially against a tmp_path, and the no-op check makes a second sequential call write nothing at all — so the write path is never entered twice at once. diff --git a/src/vpcopilot/backfill.py b/src/vpcopilot/backfill.py index 7c61402..3065c3a 100644 --- a/src/vpcopilot/backfill.py +++ b/src/vpcopilot/backfill.py @@ -31,6 +31,7 @@ import json import os +import threading from collections.abc import Callable from pathlib import Path @@ -227,7 +228,11 @@ def backfill(out_dir: str = "out", *, dry_run: bool = False, log: Callable = pri # one is evidence that an exporter reads and a bundle ships. A partial file would be silently # unparseable (`load` swallows a JSONDecodeError and returns nothing), so the failure mode is # losing every frozen attribution without a word. `os.replace` is atomic within a filesystem. - tmp = p.with_suffix(".json.tmp") + # PID *and* thread id — a fixed temp name means two concurrent writers share the path, so + # the loser's `os.replace` finds it already moved and raises FileNotFoundError. The + # console exposes this over HTTP (POST /api/audit-backfill), where concurrency is the + # default rather than the exception. Same fix as `runmeta._save`. + tmp = p.with_suffix(f".json.tmp.{os.getpid()}.{threading.get_ident()}") tmp.write_text(json.dumps(doc, indent=2)) os.replace(tmp, p) log(f"wrote {p} — {resolved} attributed, {unknown} unknown") diff --git a/src/vpcopilot/console/app.py b/src/vpcopilot/console/app.py index a4f6967..98c9c0c 100644 --- a/src/vpcopilot/console/app.py +++ b/src/vpcopilot/console/app.py @@ -80,6 +80,10 @@ def _active_tag() -> str: app = FastAPI(title="virtual-patch-copilot console") load_dotenv(ENV_PATH) _scan = {"state": "idle", "log": [], "summary": None, "error": None} +# Guards the "is a scan already running?" test-and-set. The check used to read a flag only the +# spawned THREAD set, so two requests arriving close together both passed it and started scans +# into the same out dir, interleaving their artifacts. +_scan_lock = threading.Lock() LOG_MAX = 20_000 # the log endpoints serve the FULL transcript now — keep a ceiling on what one run pins in memory @@ -820,7 +824,8 @@ def _run_scan(repo: str, out: str, min_confidence: float = 0.5, max_files: int = 200, max_bytes: int = 60_000, draft_code_fixes: bool = True, cve: str = "", spec: str = "", manifest: list[str] | None = None, min_severity: str = "high", max_advisories: int = 25, include_dev: bool = False): - _scan.update(state="running", log=[], summary=None, error=None) + # State is claimed by `start_scan` under `_scan_lock` before this thread exists; re-setting it + # here would reopen the race it closed. try: from ..pipeline import run_pipeline summary = run_pipeline(repo or None, out_dir=out, config_path=_active_config, @@ -838,7 +843,7 @@ def _run_scan(repo: str, out: str, min_confidence: float = 0.5, @app.post("/api/scan") def start_scan(body: ScanReq): - if _scan["state"] == "running": + if _scan["state"] == "running": # cheap early reject; the authoritative claim is below raise HTTPException(409, "a scan is already running") load_dotenv(ENV_PATH, override=True) # The console reads results from OUT — so point OUT at the dir this scan writes to, or Review / @@ -852,6 +857,17 @@ def start_scan(body: ScanReq): "or a dependency manifest") if body.min_severity not in ("critical", "high", "medium", "low"): raise HTTPException(400, "min_severity must be critical, high, medium or low") + # Claim the scanner only once the request is known to be VALID, and synchronously — before the + # worker thread exists. Two halves, both load-bearing: + # * synchronous, under a lock: the old check read a flag only `_run_scan` set, so the window + # between check and thread start was wide open and two requests both got through, writing + # into the same out dir. + # * after validation: claiming first meant a request that then 400'd left the scanner marked + # running forever — one malformed request and the console can never scan again. + with _scan_lock: + if _scan["state"] == "running": + raise HTTPException(409, "a scan is already running") + _scan.update(state="running", log=[], summary=None, error=None) global OUT OUT = Path(body.out) # kwargs, not a positional tuple: H2/H3 add more inputs here and a positional args tuple is one @@ -926,14 +942,18 @@ class ActionReq(BaseModel): _jobs: dict[str, dict] = {} # job_id -> {state, log, result, error, control, finding_id} -def _dispatch_action(body: ActionReq, log): +def _dispatch_action(body: ActionReq, log, out: Path): """Run the requested control's apply through the SAME functions the CLI uses, but with a real - log sink so the console can live-stream the refiner (attach → validate → refine → retry).""" + log sink so the console can live-stream the refiner (attach → validate → refine → retry). + + `out` is passed in, never read from the module global. `OUT` is reassigned by POST /api/scan, + and this runs on a worker thread — so an apply already in flight used to write its artifacts + and its audit record into whatever directory a concurrent scan had just repointed to.""" from .. import apply as A if not (body.lb or "").strip(): # fields no longer pre-fill — a missing LB must fail clearly, not as a swagger 404 raise HTTPException(400, "select a load balancer in Run settings first") c, kw = body.control, dict(finding_id=body.finding_id, dry_run=body.dry_run, keep=body.keep, - allow_protected=body.allow_protected_lb, out_dir=str(OUT), log=log) + allow_protected=body.allow_protected_lb, out_dir=str(out), log=log) if c == "service_policy": # G2 gate: a simulated policy found too broad WARNS and requires an explicit override. # Silent when nothing was simulated — G2 adds a check, never a prerequisite. @@ -941,16 +961,16 @@ def _dispatch_action(body: ActionReq, log): # the CLI and the MCP server get it too — it used to live only here. The message and the # resulting job state are unchanged: `_run_action` catches the raise and reports # `state="error"` carrying the "allow overbroad" text, exactly as before. - art = str(OUT / "policies" / f"service_policy.{body.policy_name}.json") + art = str(out / "policies" / f"service_policy.{body.policy_name}.json") if body.refine and not body.dry_run: 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, force=body.force, - allow_overbroad=body.allow_overbroad, out_dir=str(OUT), log=log) + allow_overbroad=body.allow_overbroad, 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, force=body.force, - allow_overbroad=body.allow_overbroad, out_dir=str(OUT), log=log) + allow_overbroad=body.allow_overbroad, out_dir=str(out), log=log) if c == "malicious_user": return A.apply_malicious_user(body.lb, **kw) if c == "rate_limit": @@ -976,17 +996,17 @@ def _dispatch_action(body: ActionReq, log): raise HTTPException(400, f"unknown control '{c}'") -def _run_action(job_id: str, body: ActionReq): +def _run_action(job_id: str, body: ActionReq, out: Path): import time job = _jobs[job_id] t0 = time.perf_counter() try: - res = _dispatch_action(body, lambda m: _append(job["log"], m)) + res = _dispatch_action(body, lambda m: _append(job["log"], m), out) job.update(state="done", result=res) if not body.dry_run: # feed MTTM for the hero + a self-contained record for the model benchmark from ..audit import record passed = res.get("passed") if res.get("passed") is not None else (res.get("config_enabled") is not False) - record(str(OUT), "apply_timing", control=body.control, finding_id=body.finding_id, + record(str(out), "apply_timing", control=body.control, finding_id=body.finding_id, passed=bool(passed), elapsed_s=round(time.perf_counter() - t0, 1), attempts=res.get("attempts"), before_after=res.get("before_after"), unfixable=res.get("unfixable"), reason=res.get("reason"), kept=res.get("kept")) @@ -1007,7 +1027,10 @@ def start_action(body: ActionReq): for old in list(_jobs)[:-20]: if _jobs.get(old, {}).get("state") != "running": _jobs.pop(old, None) - threading.Thread(target=_run_action, args=(job_id, body), daemon=True).start() + # Snapshot the run dir HERE, on the request thread, not inside the worker: `OUT` is a + # module global that POST /api/scan reassigns, so reading it later binds the apply to + # whichever directory happened to be current when the thread got round to it. + threading.Thread(target=_run_action, args=(job_id, body, OUT), daemon=True).start() return {"job": job_id, "state": "running"} diff --git a/src/vpcopilot/ledger.py b/src/vpcopilot/ledger.py index 873bdae..731522e 100644 --- a/src/vpcopilot/ledger.py +++ b/src/vpcopilot/ledger.py @@ -47,7 +47,12 @@ def save(out_dir, entries: dict): POSIX/Windows) so a crash mid-write can't leave a truncated, unparseable ledger.""" p = _path(out_dir) p.parent.mkdir(parents=True, exist_ok=True) - tmp = p.with_suffix(f".json.tmp.{os.getpid()}") + # Thread id as well as pid. Two threads in ONE process shared this path, so the second + # `os.replace` raised FileNotFoundError — the identical defect `runmeta._save` was fixed + # for, left behind here. The console starts every apply on its own daemon thread and + # `mark_mitigated` runs inside them, so the loser silently fails to record a change it + # already made to a live load balancer. + tmp = p.with_suffix(f".json.tmp.{os.getpid()}.{threading.get_ident()}") tmp.write_text(json.dumps(entries, indent=2)) os.replace(tmp, p) diff --git a/src/vpcopilot/runmeta.py b/src/vpcopilot/runmeta.py index ab41baf..6990fba 100644 --- a/src/vpcopilot/runmeta.py +++ b/src/vpcopilot/runmeta.py @@ -113,11 +113,20 @@ def _git(*args) -> str: def write_manifest(out_dir, **fields) -> dict: """Merge run facts into run.json, never clobbering an existing run_id (a re-scan of the same dir - keeps its identity, so the audit entries already on disk stay joinable).""" - meta = load(out_dir) - meta.setdefault("run_id", uuid.uuid4().hex[:12]) - meta.setdefault("created", utc_now()) - meta.update({k: v for k, v in fields.items() if v is not None}) - meta.update({"actor": actor(), "host": host(), "tool_version": __version__, "out_dir": str(out_dir)}) - _save(out_dir, meta) - return meta + keeps its identity, so the audit entries already on disk stay joinable). + + Under the SAME lock as `run_id()`. This is the identical read-modify-write, and guarding only + one of the two left the race half-closed: a scan finishing here while any thread minted through + `run_id()` (which `audit.record` calls on the console's per-apply daemon threads) loses one of + the two mints, and the loser's audit entries carry a join key `run.json` does not contain. + Measured at 246 orphaned entries over 40 trials of 8 threads before this. + """ + with _MINT_LOCK: + meta = load(out_dir) + meta.setdefault("run_id", uuid.uuid4().hex[:12]) + meta.setdefault("created", utc_now()) + meta.update({k: v for k, v in fields.items() if v is not None}) + meta.update({"actor": actor(), "host": host(), "tool_version": __version__, + "out_dir": str(out_dir)}) + _save(out_dir, meta) + return meta diff --git a/tests/conftest.py b/tests/conftest.py index e7ee1b7..e661d01 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -153,3 +153,24 @@ def fake_xc(): @pytest.fixture def noop_sleep(): return lambda *_a, **_k: None + + +@pytest.fixture(autouse=True) +def _reset_console_scan_state(): + """The console's `_scan` dict is MODULE state shared by every test that posts to /api/scan. + + `POST /api/scan` now claims `state="running"` synchronously, under a lock, before spawning the + worker — that is what closed the double-scan race. The consequence is that any test which stubs + `_run_scan` (several do, to avoid running a real pipeline) leaves the flag set, so the next such + test gets a 409 and fails for a reason that has nothing to do with it. Four tests across four + files broke exactly this way, every one of them passing in isolation. + + Resetting it here fixes the class rather than the four instances, and keeps the guarantee that a + test's outcome does not depend on what ran before it. + """ + yield + try: + from vpcopilot.console import app as _console + _console._scan.update(state="idle", log=[], summary=None, error=None) + except Exception: # noqa: BLE001 — the console is optional for most of the suite + pass diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py new file mode 100644 index 0000000..d3c81b6 --- /dev/null +++ b/tests/test_concurrency.py @@ -0,0 +1,209 @@ +"""Shared state under concurrency. + +From the 2026-08-04 deep-dive review. The console starts every apply, scan and reconcile on its own +daemon thread, so "two of these at once" is the normal case, not the exception — and every defect +here loses or misfiles a record of a change already made to a live load balancer. + +These tests actually run threads. A single-threaded assertion cannot see any of them. +""" +from __future__ import annotations + +import json +import tempfile +import threading + +import pytest + + +def _race(fn, n=8, trials=20): + """Run `fn(i)` on n threads against a fresh out dir, `trials` times. Returns (errors, dirs).""" + errors: list[str] = [] + dirs: list[str] = [] + for _ in range(trials): + d = tempfile.mkdtemp() + dirs.append(d) + ts = [threading.Thread(target=_collect, args=(fn, d, i, errors)) for i in range(n)] + for t in ts: + t.start() + for t in ts: + t.join() + return errors, dirs + + +def _collect(fn, d, i, errors): + try: + fn(d, i) + except Exception as e: # noqa: BLE001 + errors.append(f"{type(e).__name__}: {e}") + + +# ---------------------------------------------------------------- run_id minting + + +def test_the_run_id_mint_is_atomic_across_both_of_its_call_sites(): + """`run_id()` was already locked; `write_manifest` did the IDENTICAL read-modify-write and was + not. Guarding one of two left the race half-closed: a scan finishing while any thread records + an audit entry loses a mint, and the loser's entries carry a join key `run.json` never contains + — so they are orphaned from the run they belong to. Measured at 246 orphans over 40x8.""" + from vpcopilot import runmeta + orphans = 0 + for _ in range(30): + d = tempfile.mkdtemp() + seen: list[str] = [] + + def one(i, d=d, seen=seen): + seen.append(runmeta.run_id(d) if i % 2 else runmeta.write_manifest(d, target="x")["run_id"]) + + ts = [threading.Thread(target=one, args=(i,)) for i in range(8)] + for t in ts: + t.start() + for t in ts: + t.join() + final = json.loads((__import__("pathlib").Path(d) / "run.json").read_text())["run_id"] + orphans += sum(1 for s in seen if s != final) + assert orphans == 0, f"{orphans} audit entries would carry a run_id run.json does not contain" + + +# ---------------------------------------------------------------- atomic-write temp names + + +@pytest.mark.parametrize("mod,fn", [("ledger", "save"), ("runmeta", "_save")]) +def test_every_atomic_write_uses_a_thread_unique_temp_name(mod, fn): + """A fixed temp path means two writers in ONE process share it, so the loser's `os.replace` + finds it already moved and raises FileNotFoundError. `runmeta._save` was fixed for this; + `ledger.py` still had the pid-only version (two THREADS share a pid) and `backfill.py` had + neither. Asserted across all three so the next atomic write cannot quietly omit it.""" + import inspect + + import vpcopilot + src = inspect.getsource(getattr(__import__(f"vpcopilot.{mod}", fromlist=[mod]), fn)) + assert "getpid()" in src and "get_ident()" in src, \ + f"{mod}.{fn} builds a temp name that two threads in one process can share" + assert vpcopilot # keep the import meaningful + + +def test_the_ledger_survives_concurrent_writers(): + """The ledger is what the whole audit trail joins on, and `mark_mitigated` runs inside the + console's per-apply daemon threads — so the loser of this race silently fails to record a + change it already made to a live LB. 140 failures over 8x20 before the fix.""" + from vpcopilot import ledger + errors, _ = _race(lambda d, i: ledger.save(d, {f"f{i}": {"state": "found", "finding_id": f"f{i}"}})) + assert not errors, f"{len(errors)} concurrent ledger writes failed: {errors[:3]}" + + +def test_the_backfill_sidecar_survives_concurrent_writers(): + """Exposed over HTTP as POST /api/audit-backfill, where concurrency is the default.""" + import inspect + + from vpcopilot import backfill + src = inspect.getsource(backfill) + assert 'with_suffix(".json.tmp")' not in src, \ + "backfill still writes through a fixed temp name that concurrent requests collide on" + + +# ---------------------------------------------------------------- console: one scan at a time + + +def test_two_concurrent_scans_cannot_both_start(monkeypatch): + """The guard read a flag that only the spawned THREAD set, so the window between the check and + the thread starting was wide open — both requests passed and two scans wrote into the same out + dir, interleaving their artifacts. It is now claimed synchronously under a lock.""" + from fastapi.testclient import TestClient + + from vpcopilot.console import app as A + # The fake scan blocks on an Event the TEST releases, rather than sleeping. A sleeping worker + # outlives the test and then clobbers `_scan` in the middle of whatever test is running 0.3s + # later — which is what happened here, and it corrupted four unrelated tests in the full suite + # while every one of them passed in isolation. Owning the thread's lifetime removes the guess. + release, finished = threading.Event(), threading.Event() + + def fake_scan(*a, **k): + release.wait(timeout=5) + A._scan.update(state="done", summary={}) + finished.set() + + monkeypatch.setattr(A, "_run_scan", fake_scan) + client = TestClient(A.app) + codes: list[int] = [] + + def go(): + codes.append(client.post("/api/scan", json={"repo": tempfile.mkdtemp(), + "out": tempfile.mkdtemp()}).status_code) + prior = dict(A._scan) + try: + A._scan.update(state="idle") + ts = [threading.Thread(target=go) for _ in range(6)] + for t in ts: + t.start() + for t in ts: + t.join() + assert codes.count(200) == 1, \ + f"{codes.count(200)} scans started concurrently: {sorted(codes)}" + assert codes.count(409) == 5 + finally: + # `_scan` is module state shared by every test that posts to /api/scan. Release the worker, + # WAIT for it, and only then restore — so no thread of this test's making is still alive to + # touch shared state after it returns. + release.set() + finished.wait(timeout=5) + A._scan.clear() + A._scan.update(prior) + + +# ---------------------------------------------------------------- console: the run dir a job uses + + +def test_an_in_flight_apply_writes_to_the_dir_it_started_in(): + """`OUT` is a module global that POST /api/scan reassigns, and `_run_action` read it inside the + worker thread — so an apply already in flight wrote its `apply_timing` audit record into + whatever directory a concurrent scan had just repointed to. That record is what feeds the + 'time to mitigate' hero, so it lands against the wrong run. + + `_run_reconcile` already took the dir as an argument; the apply path simply had not. + """ + import inspect + + from vpcopilot.console import app as A + start = inspect.getsource(A.start_action) + assert "args=(job_id, body, OUT)" in start, \ + "the action job does not capture the run dir on the request thread" + run = inspect.getsource(A._run_action) + assert "str(OUT)" not in run, "_run_action still reads the mutable global inside the worker" + assert "out: Path" in run.split("\n")[0], "_run_action does not take the run dir explicitly" + + +def test_a_rejected_scan_request_does_not_wedge_the_scanner(): + """Found while fixing the race above, and worse than the race: claiming `state="running"` + BEFORE validating the request meant a 400 left the flag set with no worker to ever clear it. + One malformed request and the console could never start another scan — for the lifetime of the + process, with no way back short of a restart. + + Three rejects in a row, then a valid request must still be accepted.""" + from fastapi.testclient import TestClient + + from vpcopilot.console import app as A + client = TestClient(A.app) + A._scan.update(state="idle") + + assert client.post("/api/scan", json={}).status_code == 400 + assert client.post("/api/scan", json={"cve": "CVE-2024-23334", "spec": "/s.yaml"}).status_code == 400 + assert client.post("/api/scan", json={"repo": "/x", "min_severity": "nonsense"}).status_code == 400 + assert A._scan["state"] != "running", \ + "a rejected request left the scanner claimed — no scan can ever start again" + + release, finished = threading.Event(), threading.Event() + + def fake_scan(*a, **k): + release.wait(timeout=5) + A._scan.update(state="done", summary={}) + finished.set() + + prior, A._run_scan = A._run_scan, fake_scan + try: + assert client.post("/api/scan", + json={"repo": tempfile.mkdtemp(), "out": tempfile.mkdtemp()} + ).status_code == 200, "a valid scan was refused after earlier rejects" + finally: + release.set() + finished.wait(timeout=5) + A._run_scan = prior diff --git a/tests/test_console_simulate.py b/tests/test_console_simulate.py index 660b373..cfd715e 100644 --- a/tests/test_console_simulate.py +++ b/tests/test_console_simulate.py @@ -74,7 +74,7 @@ def test_an_overbroad_policy_is_refused_without_the_override(tmp_path, monkeypat def test_the_override_applies_anyway_and_writes_an_audit_record(tmp_path, monkeypatch): _sim(tmp_path) monkeypatch.setattr(A, "OUT", tmp_path) - monkeypatch.setattr(A, "_dispatch_action", lambda body, log: {"passed": True, "kept": True}) + monkeypatch.setattr(A, "_dispatch_action", lambda body, log, out: {"passed": True, "kept": True}) r = _client().post("/api/action", json={"control": "service_policy", "policy_name": "deny-wide", "finding_id": "f-1", "lb": "lab", "dry_run": False, "allow_overbroad": True}) @@ -89,7 +89,7 @@ def test_the_override_applies_anyway_and_writes_an_audit_record(tmp_path, monkey def test_a_narrow_policy_is_not_gated(tmp_path, monkeypatch): _sim(tmp_path) monkeypatch.setattr(A, "OUT", tmp_path) - monkeypatch.setattr(A, "_dispatch_action", lambda body, log: {"passed": True}) + monkeypatch.setattr(A, "_dispatch_action", lambda body, log, out: {"passed": True}) r = _client().post("/api/action", json={"control": "service_policy", "policy_name": "deny-narrow", "finding_id": "f-2", "lb": "lab", "dry_run": False}) job = r.json()["job"] @@ -104,7 +104,7 @@ def test_a_dry_run_is_never_gated(tmp_path, monkeypatch): """Dry-run changes nothing, so a blast-radius warning has nothing to gate.""" _sim(tmp_path) monkeypatch.setattr(A, "OUT", tmp_path) - monkeypatch.setattr(A, "_dispatch_action", lambda body, log: {"mode": "dry_run"}) + monkeypatch.setattr(A, "_dispatch_action", lambda body, log, out: {"mode": "dry_run"}) r = _client().post("/api/action", json={"control": "service_policy", "policy_name": "deny-wide", "finding_id": "f-1", "lb": "lab", "dry_run": True}) job = r.json()["job"] @@ -120,7 +120,7 @@ def test_apply_is_unchanged_when_nothing_was_simulated(tmp_path, monkeypatch): """G2 adds a check, not a prerequisite: with no simulation.json the apply path behaves exactly as it did before this feature existed.""" monkeypatch.setattr(A, "OUT", tmp_path) - monkeypatch.setattr(A, "_dispatch_action", lambda body, log: {"passed": True, "kept": True}) + monkeypatch.setattr(A, "_dispatch_action", lambda body, log, out: {"passed": True, "kept": True}) r = _client().post("/api/action", json={"control": "service_policy", "policy_name": "anything", "finding_id": "f-9", "lb": "lab", "dry_run": False}) job = r.json()["job"]