From 46bad5c1a82c184c55170cbb5eeee8ccdf89bf03 Mon Sep 17 00:00:00 2001 From: Violin Date: Wed, 12 Aug 2026 07:09:51 +0100 Subject: [PATCH] refactor(guard): de-benchmark production code; route-level coverage obligations - Remove challenge_ids seeding (vulnerability-name leak) from scope; seed engagement.coverage_obligations (client-style in-scope endpoints) + generic engagement.audit_mode flag instead - plugins/violin_guard now contains zero benchmark/run-id references; gates are framework-owned and audit-mode-gated - Cross-engagement guard blocks ANY foreign engagement dir, not just benchmark-run-* - Anti-cheat regression test asserts no vuln names in generated scope --- benchmark/run.py | 33 ++++++--- plugins/violin_guard/command.py | 20 +++--- plugins/violin_guard/handlers/base.py | 10 +-- plugins/violin_guard/handlers/ptt_handlers.py | 70 ++++++++++--------- plugins/violin_guard/receipt_integrity.py | 2 +- tests/guard/state/test_workflow_contract.py | 40 +++++------ tests/guard/test_benchmark_runner.py | 24 +++++++ 7 files changed, 122 insertions(+), 77 deletions(-) diff --git a/benchmark/run.py b/benchmark/run.py index 86f0266..1c0ae93 100644 --- a/benchmark/run.py +++ b/benchmark/run.py @@ -188,22 +188,33 @@ def _scope_for_target(target: str) -> dict: brief = brief_path.read_text(encoding="utf-8").strip() if brief: scope.setdefault("engagement", {})["brief"] = brief - # Benchmark challenge IDs: seed the scored challenge ids into scope as - # framework-owned coverage targets. The VULN_RESEARCH coverage gate then - # requires an evidence-backed cell per challenge so the agent cannot - # declare "coverage finalized" while specific scored endpoints (e.g. - # by-color filter, users/, uploads/fetch-url) were never probed. - challenges_path = fixture_path / "challenges.json" - if isinstance(benchmark, dict) and challenges_path.is_file(): + # Engagement audit mode: a generic "structured engagement" flag the + # framework gates on. Benchmark harness sets it; a real client engagement + # with strict record-keeping can set the same flag. No benchmark concept + # leaks into framework logic. + scope.setdefault("engagement", {})["audit_mode"] = True + # Coverage obligations: route-level API scope a client would provide at + # kickoff (in-scope endpoints), NOT vulnerability names — the agent must + # still discover what is vulnerable. Derived from the target's endpoint + # inventory; the framework's coverage gate then requires an evidence-backed + # matrix cell per obligation. + obligations_path = fixture_path.parent / "challenges.json" + if obligations_path.is_file(): try: challenges = json.loads( - challenges_path.read_text(encoding="utf-8") + obligations_path.read_text(encoding="utf-8") ).get("challenges", []) except (OSError, ValueError): challenges = [] - benchmark.pop("challenge_ids", None) - if challenges: - benchmark["challenge_ids"] = [ch["id"] for ch in challenges if ch.get("id")] + obligations: list[str] = [] + for ch in challenges: + endpoint = ch.get("endpoint") or "" + candidates = endpoint if isinstance(endpoint, list) else [endpoint] + for item in candidates: + item = str(item or "").strip() + if item and item not in obligations: + obligations.append(item) + scope.setdefault("engagement", {})["coverage_obligations"] = obligations return scope diff --git a/plugins/violin_guard/command.py b/plugins/violin_guard/command.py index 292872c..8670817 100644 --- a/plugins/violin_guard/command.py +++ b/plugins/violin_guard/command.py @@ -365,11 +365,10 @@ def check_http_proof_flags(command: str) -> CheckResult: SKILL.md §4 mandates `-i` or `-sv` when testing HTTP endpoints so evidence files carry empirical status lines. Receipts from plain `curl -s` (no `-i`) - make `has_decisive_proof` fail and burn real confirmations at scoring time - (benchmark-run-20260811_175611: only 6/194 receipts carried HTTP/1.1). - Exempted: status probes (`-w %{http_code}`), HEAD (`-I`), header dumps - (`-D`), and offline captures (`-o`/`-O`/`> file`) that are not interactive - HTTP evidence. + make `has_decisive_proof` fail at validation time and burn real + confirmations. Exempted: status probes (`-w %{http_code}`), HEAD (`-I`), + header dumps (`-D`), and offline captures (`-o`/`-O`/`> file`) that are + not interactive HTTP evidence. """ result = CheckResult() if not _HTTP_CLIENT_RE.search(command) or not _HTTP_URL_RE.search(command): @@ -387,13 +386,18 @@ def check_http_proof_flags(command: str) -> CheckResult: def check_cross_engagement_paths(command: str, active_eng_dir: Path) -> CheckResult: - """Block commands that reference a foreign engagement directory under engagements/.""" + """Block commands that reference a foreign engagement directory under engagements/. + + The active engagement may be referenced (the agent legitimately reads its + own evidence); any OTHER engagement directory is off-limits — whether from + a previous run or a different client — to keep engagements isolated. + """ result = CheckResult() - pattern = r"(?i)(?:[/\\]|^)engagements[/\\](benchmark-run-[a-zA-Z0-9_-]+|benchmark-run)\b" + pattern = r"(?i)(?:[/\\]|^)engagements[/\\]([a-zA-Z0-9_-]+)\b" active_name = active_eng_dir.name for match in re.finditer(pattern, command): ref_name = match.group(1) - if ref_name != active_name and ref_name.startswith("benchmark-run-"): + if ref_name != active_name: result.add_error( f"cross-engagement path access blocked: command references foreign engagement directory '{ref_name}' " f"while active engagement is '{active_name}'" diff --git a/plugins/violin_guard/handlers/base.py b/plugins/violin_guard/handlers/base.py index b8fcc63..1e2b22f 100644 --- a/plugins/violin_guard/handlers/base.py +++ b/plugins/violin_guard/handlers/base.py @@ -51,11 +51,11 @@ def _result(r) -> dict[str, list[str]]: def _log_guard_friction(eng_dir: Path, result, command: str) -> None: """Append a framework_feedback.md row when the guard blocks or reviews. - Only writes when state/framework_feedback.md already exists — the - benchmark runner creates it at engagement init. Real engagements without - the file are untouched. Recording here means friction is captured at the - moment it happens, with zero agent bookkeeping, so the agent never has to - reconstruct what was blocked from memory at the end of the run. + Only writes when state/framework_feedback.md already exists — engagement + initialization creates it. Engagements without the file are untouched. + Recording here means friction is captured at the moment it happens, with + zero agent bookkeeping, so the agent never has to reconstruct what was + blocked from memory at the end of the run. """ feedback = eng_dir / "state" / "framework_feedback.md" if not feedback.exists(): diff --git a/plugins/violin_guard/handlers/ptt_handlers.py b/plugins/violin_guard/handlers/ptt_handlers.py index d6b880b..86b6345 100644 --- a/plugins/violin_guard/handlers/ptt_handlers.py +++ b/plugins/violin_guard/handlers/ptt_handlers.py @@ -63,7 +63,9 @@ def _validate_phase_exit(engagement: Path, task_id: str, status: str) -> None: scope_data = ( yaml.safe_load(scope_path.read_text(encoding="utf-8")) if scope_path.is_file() else {} ) - if isinstance(scope_data, dict) and (scope_data.get("benchmark") or {}).get("mode") is True: + if isinstance(scope_data, dict) and ( + (scope_data.get("engagement") or {}).get("audit_mode") is True + ): matrix_path = engagement / "state" / "coverage-matrix.yaml" if not matrix_path.is_file(): raise ValueError( @@ -73,19 +75,23 @@ def _validate_phase_exit(engagement: Path, task_id: str, status: str) -> None: entries = matrix.get("coverage") if isinstance(matrix, dict) else None if not isinstance(entries, dict) or not entries: raise ValueError("coverage matrix must contain a non-empty coverage mapping") - challenge_ids = (scope_data.get("benchmark") or {}).get("challenge_ids") or [] - covered_ids = ( - {str(name).lower() for name in entries} - | {str(entry.get("challenge_id") or "").lower() for entry in entries.values() if isinstance(entry, dict)} - ) + obligations = (scope_data.get("engagement") or {}).get("coverage_obligations") or [] + cell_texts = [ + f"{str(name).lower()} {str(entry.get('evidence_or_reason') or '').lower()}" + for name, entry in entries.items() + if isinstance(entry, dict) + ] unresolved_coverage = [] - # Scored challenges must map to a matrix cell — the 2026-08-11 - # 184338 run self-declared ssrf/redirects/rate_limits "tested" or - # N/A while ssrf-*, no-rate-limiting, open-redirect, sqli-color- - # filter and user-enumeration were never probed at all. - for cid in challenge_ids: - if str(cid).strip().lower() not in covered_ids: - unresolved_coverage.append(f"{cid} (no coverage-matrix cell)") + # Client-provided in-scope endpoints must map to a matrix cell. + # Route-level obligations keep the agent honest about coverage + # without leaking what is vulnerable — the same requirement a + # strict client engagement would hold it to. + for obligation in obligations: + obligation = str(obligation).strip().lower() + if not obligation: + continue + if not any(obligation in text for text in cell_texts): + unresolved_coverage.append(f"{obligation} (no coverage-matrix cell)") for name, entry in entries.items(): if not isinstance(entry, dict): unresolved_coverage.append(name) @@ -97,10 +103,10 @@ def _validate_phase_exit(engagement: Path, task_id: str, status: str) -> None: continue # not_applicable must be backed by an evidence file: a bare # reason ("no rate-limit behavior observed") is a memory - # reconstruction, the exact false-positive factory. The - # 2026-08-11 run self-declared rate_limits N/A against a real - # challenge because it never probed. blocked must name the - # guard that prevented testing. + # reconstruction, the exact false-positive factory. Self- + # declaring N/A against a live endpoint without probing it is + # how real findings are missed. blocked must name the guard + # that prevented testing. if status == "not_applicable" and "evidence/" not in reason: unresolved_coverage.append(f"{name} (not_applicable without evidence file)") elif status == "blocked" and "guard" not in reason.lower(): @@ -125,10 +131,10 @@ def _validate_phase_exit(engagement: Path, task_id: str, status: str) -> None: # A Rejected hypothesis disposed as not_implemented must still carry a # real executed test and evidence. "N/A - placeholder" with no evidence # means the cheapest discriminating test never ran — the agent disposed - # it from the conversation (the 2026-08-11 161722 run rejected H-001 - # (admin/admin login check) this way and weak-admin-creds was never - # tested). Surface-mapping hypotheses (e.g. "API surface enumeration") - # pass when they cite real bundle/probe evidence. + # it from the conversation (e.g. rejecting a default-credential login + # check this way while the endpoint was never probed). Surface-mapping + # hypotheses (e.g. "API surface enumeration") pass when they cite real + # bundle/probe evidence. if not unresolved: untested_disposals = [] for item in board: @@ -157,11 +163,10 @@ def _validate_phase_exit(engagement: Path, task_id: str, status: str) -> None: "VULN_RESEARCH cannot close with unresolved hypotheses: " + ", ".join(unresolved) ) # Canonization gate: a Validated hypothesis without a linked FIND file - # is invisible to the scorer ("technical proof exists but no Validated - # hypothesis cites it"). The 2026-08-11 165714 run wrote 8 FIND files - # and 9 matching evidence bundles yet scored 0/20 formalized because no - # hypothesis carried Linked findings. Require 1:N links now, at phase - # close, not at REPORTING (closeout skips REPORTING tasks). + # is invisible to downstream reporting/validation — evidence exists but + # no written claim ties it to a canonical finding, so the finding is + # lost at closeout. Require 1:N links now, at phase close, not at + # REPORTING (closeout skips REPORTING tasks). uncanonized = [] for item in board: if item.canonical_status() != "Validated": @@ -189,13 +194,14 @@ def _validate_phase_exit(engagement: Path, task_id: str, status: str) -> None: scope_data = ( yaml.safe_load(scope_path.read_text(encoding="utf-8")) if scope_path.is_file() else {} ) - # Benchmark mode: REPORTING may not close unless the run actually - # reached EXPLOITATION or later. The 2026-08-11 regression logged - # every command as phase=recon, declared RECON "coverage finalized", - # skipped PT-102/PT-103, and jumped straight to REPORTING — scoring - # 6/20 with 14 challenges never touched. The agent cannot fake this: + # Audit mode: REPORTING may not close unless the engagement actually + # reached EXPLOITATION or later. A run that logs everything as recon, + # declares coverage finalized, and jumps straight to REPORTING has + # skipped the entire validation phase. The agent cannot fake this: # evidence only accumulates by running commands in a later phase. - if isinstance(scope_data, dict) and (scope_data.get("benchmark") or {}).get("mode") is True: + if isinstance(scope_data, dict) and ( + (scope_data.get("engagement") or {}).get("audit_mode") is True + ): history_path = engagement / "state" / "history.md" reached_later_phase = False if history_path.is_file(): diff --git a/plugins/violin_guard/receipt_integrity.py b/plugins/violin_guard/receipt_integrity.py index 254d4bb..81d0cd1 100644 --- a/plugins/violin_guard/receipt_integrity.py +++ b/plugins/violin_guard/receipt_integrity.py @@ -1,4 +1,4 @@ -"""Authenticate benchmark execution receipts and their evidence artifacts.""" +"""Authenticate execution receipts and their evidence artifacts.""" from __future__ import annotations diff --git a/tests/guard/state/test_workflow_contract.py b/tests/guard/state/test_workflow_contract.py index 0433cfa..1d8c6ab 100644 --- a/tests/guard/state/test_workflow_contract.py +++ b/tests/guard/state/test_workflow_contract.py @@ -166,14 +166,14 @@ def test_ptt_notes_redact_credentials_before_persisting() -> None: assert "[REDACTED_API_KEY]" in redacted -def test_benchmark_vulnerability_research_exit_requires_dispositioned_matrix( +def test_audit_mode_vulnerability_research_exit_requires_dispositioned_matrix( tmp_path: Path, ) -> None: engagement = tmp_path / "engagement" assert bootstrap.init_engagement(engagement, host="10.10.10.10") == 0 scope = engagement / "scope" / "scope.yaml" scope.write_text( - scope.read_text(encoding="utf-8") + "\nbenchmark:\n mode: true\n", + scope.read_text(encoding="utf-8") + "\nengagement:\n audit_mode: true\n", encoding="utf-8", ) matrix = engagement / "state" / "coverage-matrix.yaml" @@ -202,12 +202,12 @@ def test_reporting_exit_requires_canonical_finding(tmp_path: Path) -> None: _validate_phase_exit(engagement, "PT-050", "[x]") -def test_reporting_exit_blocks_recon_only_run_in_benchmark_mode(tmp_path: Path) -> None: +def test_reporting_exit_blocks_recon_only_run_in_audit_mode(tmp_path: Path) -> None: engagement = tmp_path / "engagement" assert bootstrap.init_engagement(engagement, host="10.10.10.10") == 0 scope = engagement / "scope" / "scope.yaml" scope.write_text( - scope.read_text(encoding="utf-8") + "\nbenchmark:\n mode: true\n", + scope.read_text(encoding="utf-8") + "\nengagement:\n audit_mode: true\n", encoding="utf-8", ) history = engagement / "state" / "history.md" @@ -220,12 +220,12 @@ def test_reporting_exit_blocks_recon_only_run_in_benchmark_mode(tmp_path: Path) _validate_phase_exit(engagement, "PT-050", "[x]") -def test_reporting_exit_allows_exploitation_history_in_benchmark_mode(tmp_path: Path) -> None: +def test_reporting_exit_allows_exploitation_history_in_audit_mode(tmp_path: Path) -> None: engagement = tmp_path / "engagement" assert bootstrap.init_engagement(engagement, host="10.10.10.10") == 0 scope = engagement / "scope" / "scope.yaml" scope.write_text( - scope.read_text(encoding="utf-8") + "\nbenchmark:\n mode: true\n", + scope.read_text(encoding="utf-8") + "\nengagement:\n audit_mode: true\n", encoding="utf-8", ) history = engagement / "state" / "history.md" @@ -257,7 +257,7 @@ def test_vuln_research_exit_requires_evidence_for_not_applicable_coverage( assert bootstrap.init_engagement(engagement, host="10.10.10.10") == 0 scope = engagement / "scope" / "scope.yaml" scope.write_text( - scope.read_text(encoding="utf-8") + "\nbenchmark:\n mode: true\n", + scope.read_text(encoding="utf-8") + "\nengagement:\n audit_mode: true\n", encoding="utf-8", ) matrix = engagement / "state" / "coverage-matrix.yaml" @@ -277,7 +277,7 @@ def test_vuln_research_exit_accepts_evidence_backed_not_applicable(tmp_path: Pat assert bootstrap.init_engagement(engagement, host="10.10.10.10") == 0 scope = engagement / "scope" / "scope.yaml" scope.write_text( - scope.read_text(encoding="utf-8") + "\nbenchmark:\n mode: true\n", + scope.read_text(encoding="utf-8") + "\nengagement:\n audit_mode: true\n", encoding="utf-8", ) matrix = engagement / "state" / "coverage-matrix.yaml" @@ -298,7 +298,7 @@ def test_vuln_research_exit_blocks_not_implemented_rejection_without_evidence( assert bootstrap.init_engagement(engagement, host="10.10.10.10") == 0 scope = engagement / "scope" / "scope.yaml" scope.write_text( - scope.read_text(encoding="utf-8") + "\nbenchmark:\n mode: true\n", + scope.read_text(encoding="utf-8") + "\nengagement:\n audit_mode: true\n", encoding="utf-8", ) matrix = engagement / "state" / "coverage-matrix.yaml" @@ -332,7 +332,7 @@ def test_vuln_research_exit_accepts_surface_mapping_rejection_with_evidence( assert bootstrap.init_engagement(engagement, host="10.10.10.10") == 0 scope = engagement / "scope" / "scope.yaml" scope.write_text( - scope.read_text(encoding="utf-8") + "\nbenchmark:\n mode: true\n", + scope.read_text(encoding="utf-8") + "\nengagement:\n audit_mode: true\n", encoding="utf-8", ) matrix = engagement / "state" / "coverage-matrix.yaml" @@ -363,7 +363,7 @@ def test_vuln_research_exit_blocks_validated_hypothesis_without_linked_finding( assert bootstrap.init_engagement(engagement, host="10.10.10.10") == 0 scope = engagement / "scope" / "scope.yaml" scope.write_text( - scope.read_text(encoding="utf-8") + "\nbenchmark:\n mode: true\n", + scope.read_text(encoding="utf-8") + "\nengagement:\n audit_mode: true\n", encoding="utf-8", ) matrix = engagement / "state" / "coverage-matrix.yaml" @@ -393,7 +393,7 @@ def test_vuln_research_exit_accepts_validated_hypothesis_with_linked_finding( assert bootstrap.init_engagement(engagement, host="10.10.10.10") == 0 scope = engagement / "scope" / "scope.yaml" scope.write_text( - scope.read_text(encoding="utf-8") + "\nbenchmark:\n mode: true\n", + scope.read_text(encoding="utf-8") + "\nengagement:\n audit_mode: true\n", encoding="utf-8", ) matrix = engagement / "state" / "coverage-matrix.yaml" @@ -451,18 +451,18 @@ def test_validated_hypothesis_rejects_escaping_or_empty_evidence(tmp_path: Path) def test_vuln_research_exit_blocks_uncharted_scored_challenges(tmp_path: Path) -> None: - """Coverage completeness: every seeded challenge_id needs a matrix cell. + """Coverage completeness: every in-scope endpoint needs a matrix cell. - Mirrors benchmark-run-20260811_184338 where the agent closed VULN_RESEARCH - with ssrf-*/no-rate-limiting/user-enumeration never probed while - self-declaring ssrf/redirects/rate_limits 'tested' or N/A. + Client-provided in-scope endpoints (fetch-url, login) must map to a + coverage-matrix cell — self-declared 'tested'/N/A coverage of related + categories is not enough. """ engagement = tmp_path / "engagement" assert bootstrap.init_engagement(engagement, host="10.10.10.10") == 0 scope = engagement / "scope" / "scope.yaml" scope.write_text( scope.read_text(encoding="utf-8") - + "\nbenchmark:\n mode: true\n challenge_ids:\n - ssrf-fetch-url\n - no-rate-limiting\n", + + "\nengagement:\n audit_mode: true\n coverage_obligations:\n - GET /api/v1/uploads/fetch-url\n - POST /api/v1/auth/login\n", encoding="utf-8", ) matrix = engagement / "state" / "coverage-matrix.yaml" @@ -480,7 +480,7 @@ def test_vuln_research_exit_blocks_aspirational_tested_narrative(tmp_path: Path) assert bootstrap.init_engagement(engagement, host="10.10.10.10") == 0 scope = engagement / "scope" / "scope.yaml" scope.write_text( - scope.read_text(encoding="utf-8") + "\nbenchmark:\n mode: true\n", + scope.read_text(encoding="utf-8") + "\nengagement:\n audit_mode: true\n", encoding="utf-8", ) matrix = engagement / "state" / "coverage-matrix.yaml" @@ -498,12 +498,12 @@ def test_vuln_research_exit_accepts_challenge_cells_with_artifact(tmp_path: Path scope = engagement / "scope" / "scope.yaml" scope.write_text( scope.read_text(encoding="utf-8") - + "\nbenchmark:\n mode: true\n challenge_ids:\n - no-rate-limiting\n", + + "\nengagement:\n audit_mode: true\n coverage_obligations:\n - POST /api/v1/auth/login\n", encoding="utf-8", ) matrix = engagement / "state" / "coverage-matrix.yaml" matrix.write_text( - "coverage:\n no-rate-limiting:\n status: not_applicable\n evidence_or_reason: 'evidence/vuln-research/rate_na.txt - 429 never observed'\n", + "coverage:\n no-rate-limiting:\n status: not_applicable\n evidence_or_reason: 'evidence/vuln-research/rate_na.txt - 429 never observed; POST /api/v1/auth/login probed 20x'\n", encoding="utf-8", ) _validate_phase_exit(engagement, "PT-030", "[x]") # no exception diff --git a/tests/guard/test_benchmark_runner.py b/tests/guard/test_benchmark_runner.py index c86a7a1..b72e6ec 100644 --- a/tests/guard/test_benchmark_runner.py +++ b/tests/guard/test_benchmark_runner.py @@ -96,6 +96,30 @@ def test_init_benchmark_engagement_seeds_engage_brief_into_scope(tmp_path: Path) assert "register" in brief.lower() +def test_init_benchmark_engagement_seeds_route_obligations_not_vuln_names( + tmp_path: Path, +) -> None: + """Coverage obligations are endpoint scope, never vulnerability names. + + The framework must not leak what is vulnerable (no challenge ids like + 'ssrf-fetch-url') — only client-style in-scope endpoints a real engagement + brief would contain. + """ + eng_dir = tmp_path / "eng" + init_benchmark_engagement(eng_dir, "http://test-target.local:8080") + + scope = yaml.safe_load((eng_dir / "scope" / "scope.yaml").read_text(encoding="utf-8")) + engagement = scope.get("engagement") or {} + assert engagement.get("audit_mode") is True + obligations = [str(o) for o in engagement.get("coverage_obligations") or []] + assert "POST /api/v1/auth/login" in obligations + assert "GET /api/v1/uploads/fetch-url" in obligations + # Anti-cheat: no challenge/vulnerability identifiers anywhere in scope. + text = (eng_dir / "scope" / "scope.yaml").read_text(encoding="utf-8").lower() + assert "ssrf" not in text + assert "challenge_ids" not in text + + def test_closeout_detection_requires_all_artifacts(tmp_path: Path) -> None: eng_dir = tmp_path / "eng" init_benchmark_engagement(eng_dir, "https://target.local")