diff --git a/plugins/violin_guard/command.py b/plugins/violin_guard/command.py index 1c25079..381a46b 100644 --- a/plugins/violin_guard/command.py +++ b/plugins/violin_guard/command.py @@ -42,6 +42,12 @@ __all__ = [ # Argument / Result dataclasses # --------------------------------------------------------------------------- +# Grace window for the record-as-you-go recency gate: evidence newer than the +# hypothesis board's last update by more than this many seconds blocks further +# target commands. 15 minutes is generous enough for burst timing/clock skew +# while still catching run-long bookkeeping deferral. +_RECORD_AS_YOU_GO_GRACE = 15 * 60 + @dataclass class CheckCommandArgs: @@ -508,6 +514,49 @@ def check_hypothesis_freshness( if stale: result.add_warning(f"hypothesis guard: {stale} hypothesis(es) not updated in 48h") + # Recency gate — record-as-you-go enforcement. If the newest execution + # evidence is NEWER than the hypothesis board's last update, the agent is + # deferring bookkeeping and will reconstruct results from conversation + # memory later (the false-positive factory). Block further commands until + # the result is recorded on the board via violin_record_hypothesis. + # Bursts preflight every command before any executes, so this never + # false-fires mid-burst; a grace window absorbs same-burst timing skew. + exec_dir = eng_dir / "evidence" / "executions" + newest_evidence = 0.0 + if exec_dir.is_dir(): + for path in exec_dir.iterdir(): + if path.suffix == ".json" and not path.name.endswith((".lock", ".tmp")): + try: + newest_evidence = max(newest_evidence, path.stat().st_mtime) + except OSError: + continue + if newest_evidence: + for h in relevant: + if not h.updated: + continue + raw = h.updated.strip() + candidate = raw.removesuffix(" UTC").removesuffix("Z").strip() + updated_ts = None + for fmt in ("%Y-%m-%d %H:%M", "%Y-%m-%dT%H:%M", "%Y-%m-%d %H:%M:%S"): + try: + updated_ts = datetime.strptime(candidate, fmt) + break + except ValueError: + continue + if updated_ts is None: + continue + updated_ts = updated_ts.replace(tzinfo=UTC) + # evidence mtime is naive epoch — treat as UTC for comparison + if newest_evidence > updated_ts.timestamp() + _RECORD_AS_YOU_GO_GRACE: + result.add_error( + f"hypothesis H-{h.id} has not been updated since the latest execution " + "evidence — record the batch result on the hypothesis board NOW via " + "violin_record_hypothesis (status, Test Response, Runtime Evidence, " + "Updated) before running further commands. Deferred bookkeeping forces " + "memory-based reconstruction and is how false positives are born." + ) + break # one clear blocker per check is enough + return result diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/guard/__init__.py b/tests/guard/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/guard/test_recency_gate.py b/tests/guard/test_recency_gate.py new file mode 100644 index 0000000..5cb2618 --- /dev/null +++ b/tests/guard/test_recency_gate.py @@ -0,0 +1,98 @@ +"""Tests for the record-as-you-go recency gate (deferred bookkeeping blocker).""" + +from __future__ import annotations + +import os +import time +from pathlib import Path + +from plugins.violin_guard import command +from plugins.violin_guard.command import Phase + +_STALE_HYP = """### H-001: Queue service validation +- **Target:** 10.129.47.140:1515 +- **Status:** Validated +- **Phase:** EXPLOITATION +- **CVE Research:** web_search queue 1515 CVE; NVD; no results +- **Exploit Research:** web_search queue 1515 exploit; GitHub; no results +- **Updated:** 2026-08-01 10:00 +""" + +_FRESH_HYP = """### H-001: Queue service validation +- **Target:** 10.129.47.140:1515 +- **Status:** Validated +- **Phase:** EXPLOITATION +- **CVE Research:** web_search queue 1515 CVE; NVD; no results +- **Exploit Research:** web_search queue 1515 exploit; GitHub; no results +- **Updated:** {updated} +""" + + +def _make_engagement(tmp_path: Path, hyp_text: str, evidence_age: float) -> Path: + eng = tmp_path / "eng" + eng.mkdir(parents=True, exist_ok=True) + (eng / "hypotheses.md").write_text(hyp_text, encoding="utf-8") + exec_dir = eng / "evidence" / "executions" + exec_dir.mkdir(parents=True) + receipt = exec_dir / "2026-08-10T120000-deadbeef-exec.json" + receipt.write_text('{"command": "test"}', encoding="utf-8") + # age the evidence file: now - evidence_age seconds + stamp = time.time() - evidence_age + os.utime(receipt, (stamp, stamp)) + return eng + + +def test_recency_gate_blocks_when_board_stale(tmp_path: Path) -> None: + """Evidence 2h old, board updated a month ago -> block further commands.""" + eng = _make_engagement(tmp_path, _STALE_HYP, evidence_age=2 * 3600) + result = command.check_hypothesis_freshness( + eng, Phase.EXPLOITATION, "python3 exploit.py 10.129.47.140 1515" + ) + assert any("not been updated since" in err for err in result.errors) + assert "violin_record_hypothesis" in " ".join(result.errors) + + +def test_recency_gate_passes_when_board_fresh(tmp_path: Path) -> None: + """Board updated after the latest evidence -> gate silent.""" + now = time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime()) + eng = _make_engagement( + tmp_path, _FRESH_HYP.format(updated=now), evidence_age=60 + ) + result = command.check_hypothesis_freshness( + eng, Phase.EXPLOITATION, "python3 exploit.py 10.129.47.140 1515" + ) + assert not any("not been updated since" in err for err in result.errors) + + +def test_recency_gate_grace_window(tmp_path: Path) -> None: + """Evidence slightly newer than board (within grace) must not block.""" + # board updated 5 min ago, evidence 10 min ago -> evidence is OLDER + now = time.time() + updated_str = time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(now - 300)) + eng = _make_engagement( + tmp_path, _FRESH_HYP.format(updated=updated_str), evidence_age=600 + ) + result = command.check_hypothesis_freshness( + eng, Phase.EXPLOITATION, "python3 exploit.py 10.129.47.140 1515" + ) + assert not any("not been updated since" in err for err in result.errors) + + +def test_recency_gate_noop_without_evidence(tmp_path: Path) -> None: + """No execution evidence -> gate never fires (recon/early phases).""" + eng = tmp_path / "eng" + eng.mkdir(parents=True, exist_ok=True) + (eng / "hypotheses.md").write_text(_STALE_HYP, encoding="utf-8") + result = command.check_hypothesis_freshness( + eng, Phase.EXPLOITATION, "python3 exploit.py 10.129.47.140 1515" + ) + assert not any("not been updated since" in err for err in result.errors) + + +def test_recency_gate_recon_phases_untouched(tmp_path: Path) -> None: + """Recon does not require hypotheses at all — gate must stay silent.""" + eng = _make_engagement(tmp_path, _STALE_HYP, evidence_age=2 * 3600) + result = command.check_hypothesis_freshness( + eng, Phase.RECON, "nmap -p- 10.129.47.140" + ) + assert not result.errors