From ac5848a85abb7919731c10aaae05a971433e68ef Mon Sep 17 00:00:00 2001 From: Violin Date: Mon, 13 Jul 2026 08:41:03 +0100 Subject: [PATCH] Enforce PTT phase and hypothesis relevance --- plugins/violin_guard/core/command.py | 35 ++++++++++++++++++- plugins/violin_guard/core/ptt.py | 34 +++++++++++------- tests/guard/test_burst_and_target.py | 4 +-- tests/guard/test_correctness_roadmap_1_1_1.py | 9 ++++- tests/guard/test_plugin_guard.py | 19 ++++++---- tests/guard/test_sync_credit_window.py | 2 +- 6 files changed, 80 insertions(+), 23 deletions(-) diff --git a/plugins/violin_guard/core/command.py b/plugins/violin_guard/core/command.py index 29a0513..b0d1752 100644 --- a/plugins/violin_guard/core/command.py +++ b/plugins/violin_guard/core/command.py @@ -334,7 +334,9 @@ def _matches_network(candidate: str, networks: list[ipaddress._BaseNetwork]) -> network = ipaddress.ip_network(candidate, strict=False) except ValueError: return False - return any(network.version == allowed.version and network.subnet_of(allowed) for allowed in networks) + return any( + network.version == allowed.version and network.subnet_of(allowed) for allowed in networks + ) def check_scope_targets(scope_path: Path, command: str) -> CheckResult: @@ -447,6 +449,31 @@ def check_hypothesis_freshness(eng_dir: Path, phase: Phase, command: str) -> Hyp result.add_error(f"phase {phase.value} requires at least one hypothesis in hypotheses.md") return result + acceptable_phases = { + Phase.VULN_RESEARCH: {Phase.VULN_RESEARCH}, + Phase.EXPLOITATION: {Phase.VULN_RESEARCH, Phase.EXPLOITATION}, + Phase.POST_EXPLOITATION: {Phase.EXPLOITATION, Phase.POST_EXPLOITATION}, + Phase.PRIVESC: {Phase.EXPLOITATION, Phase.POST_EXPLOITATION, Phase.PRIVESC}, + Phase.FLAGS: {Phase.PRIVESC, Phase.FLAGS}, + }.get(phase, {phase}) + targets = set(_extract_target_candidates(command)) + relevant = [] + for hypothesis in hyps: + if hypothesis.canonical_status() == "Rejected" or not hypothesis.target: + continue + try: + hypothesis_phase = normalize_phase(hypothesis.phase) + except ValueError: + continue + target = _normalise_scope_host(hypothesis.target) + if hypothesis_phase in acceptable_phases and (not targets or target in targets): + relevant.append(hypothesis) + if not relevant: + result.add_error( + f"phase {phase.value} requires a non-rejected hypothesis matching the command target" + ) + return result + # Check for stale hypotheses (no update in 48h) stale = 0 now = datetime.now(UTC) @@ -529,6 +556,12 @@ def check_command(args: CheckCommandArgs) -> CheckResult: result.warnings.extend(ptt_validation.warnings) if ptt_validation.active_task: result.infos.append(f"active PTT task: {ptt_validation.active_task}") + active_task = ptt.find_active_task(ptt_validation.tasks) + if active_task and not ptt.task_matches_phase(active_task, phase): + result.add_error( + f"active PTT task {active_task.id} belongs to {active_task.phase or 'no phase'}; " + f"requested phase is {phase.value}" + ) # 5. History staleness (duplicate detection) hist_result = check_history_staleness(eng_dir, args.command) diff --git a/plugins/violin_guard/core/ptt.py b/plugins/violin_guard/core/ptt.py index 1aabcbb..0819c60 100644 --- a/plugins/violin_guard/core/ptt.py +++ b/plugins/violin_guard/core/ptt.py @@ -10,12 +10,15 @@ from dataclasses import dataclass, field from datetime import UTC, datetime from pathlib import Path +from .phases import Phase, normalize_phase + __all__ = [ "PttTask", "PttValidationResult", "parse_ptt", "validate_ptt", "find_active_task", + "task_matches_phase", "is_stale", "update_task", ] @@ -43,6 +46,7 @@ class PttTask: title: str note: str = "" updated: str = "" + phase: str = "" def to_markdown(self) -> str: datetime.now(UTC).strftime("%Y-%m-%d %H:%M") @@ -76,7 +80,14 @@ def parse_ptt(path: Path) -> list[PttTask]: return [] content = path.read_text(encoding="utf-8") tasks = [] + current_phase = "" for line in content.splitlines(): + heading = re.match(r"^##\s+Phase:\s*(?P.+?)\s*$", line.strip(), re.IGNORECASE) + if heading: + current_phase = ( + heading.group("phase").strip().upper().replace("-", "_").replace(" ", "_") + ) + continue m = _PTT_RE.match(line.strip()) if m: tasks.append( @@ -85,6 +96,7 @@ def parse_ptt(path: Path) -> list[PttTask]: status=m.group("status").strip(), title=m.group("title").strip(), note=m.group("note").strip(), + phase=current_phase, ) ) return tasks @@ -126,6 +138,13 @@ def find_active_task(tasks: list[PttTask]) -> PttTask | None: return None +def task_matches_phase(task: PttTask, phase: Phase | str) -> bool: + """Whether a task belongs to the requested execution phase.""" + requested = normalize_phase(phase) if isinstance(phase, str) else phase + expected = Phase.EXPLOITATION if requested is Phase.POST_EXPLOITATION else requested + return task.phase == expected.value + + def is_stale(path: Path) -> bool: """True if every task is still [ ] (pristine).""" tasks = parse_ptt(path) @@ -164,18 +183,9 @@ def update_task(path: Path, task_id: str, status: str, note: str) -> PttTask: cells[1] = status if len(cells) >= 4: cells[-1] = note - new_line = ( - "| " - + " | ".join( - ( - cells[0], - cells[1], - cells[2] if len(cells) > 2 else "", - cells[-1] if len(cells) > 3 else "", - ) - ) - + " |" - ) + # Keep every original column. EXPLOITATION rows contain hypothesis, + # validation-command, and patch columns that must not be flattened. + new_line = "| " + " | ".join(cells) + " |" lines = content.splitlines() lines[target_idx] = new_line diff --git a/tests/guard/test_burst_and_target.py b/tests/guard/test_burst_and_target.py index 4037d03..81e61e6 100644 --- a/tests/guard/test_burst_and_target.py +++ b/tests/guard/test_burst_and_target.py @@ -55,7 +55,7 @@ def eng(tmp_path): ) ptt = d / "state" / "ptt.md" ptt.write_text( - ptt.read_text(encoding="utf-8").replace("| PT-001 | [ ] |", "| PT-001 | [~] |"), + ptt.read_text(encoding="utf-8").replace("| PT-010 | [ ] |", "| PT-010 | [~] |"), encoding="utf-8", ) return d @@ -235,7 +235,7 @@ def test_plugin_exec_burst_accepts_inline_commands(monkeypatch, tmp_path): ) ptt = d / "state" / "ptt.md" ptt.write_text( - ptt.read_text(encoding="utf-8").replace("| PT-001 | [ ] |", "| PT-001 | [~] |"), + ptt.read_text(encoding="utf-8").replace("| PT-010 | [ ] |", "| PT-010 | [~] |"), encoding="utf-8", ) _patch_burst(monkeypatch, str(d)) diff --git a/tests/guard/test_correctness_roadmap_1_1_1.py b/tests/guard/test_correctness_roadmap_1_1_1.py index 2c58f13..946fa8b 100644 --- a/tests/guard/test_correctness_roadmap_1_1_1.py +++ b/tests/guard/test_correctness_roadmap_1_1_1.py @@ -75,7 +75,7 @@ def _init_e2e(tmp_path, skill_file, allowed=("recon", "vuln-research", "exploita ) ptt = eng / "state" / "ptt.md" ptt.write_text( - ptt.read_text(encoding="utf-8").replace("| PT-001 | [ ] |", "| PT-001 | [~] |"), + ptt.read_text(encoding="utf-8").replace("| PT-010 | [ ] |", "| PT-010 | [~] |"), encoding="utf-8", ) return eng @@ -198,6 +198,13 @@ def test_post_exploitation_requires_scope_and_skill_load(tmp_path): ), encoding="utf-8", ) + ptt_path = eng / "state" / "ptt.md" + ptt_path.write_text( + ptt_path.read_text(encoding="utf-8") + .replace("| PT-010 | [~] |", "| PT-010 | [x] |") + .replace("| PT-042 | [ ] |", "| PT-042 | [~] |"), + encoding="utf-8", + ) res = command.check_command( command.CheckCommandArgs( diff --git a/tests/guard/test_plugin_guard.py b/tests/guard/test_plugin_guard.py index a1699df..5a36857 100644 --- a/tests/guard/test_plugin_guard.py +++ b/tests/guard/test_plugin_guard.py @@ -130,7 +130,7 @@ def _init_e2e(tmp_path, skill_file): The skill-load gate requires a session-scoped marker at ``$ENG_DIR/state/.skill-loaded-``; passing ``--session-id`` makes the CLI compute that canonical path itself, so we write there. We - also pre-mark PT-001 as in-progress so the PTT staleness guard (which BLOCKs + also pre-mark PT-010 as in-progress so the PTT phase gate (which BLOCKs until at least one PT row has moved past ``[ ]``) does not reject the very first recon command — this mirrors a normal SCOPING->RECON handoff. """ @@ -148,7 +148,7 @@ def _init_e2e(tmp_path, skill_file): # At least one PTT row must have advanced so the staleness guard passes. ptt_path = eng / "state" / "ptt.md" ptt_path.write_text( - ptt_path.read_text(encoding="utf-8").replace("| PT-001 | [ ] |", "| PT-001 | [~] |"), + ptt_path.read_text(encoding="utf-8").replace("| PT-010 | [ ] |", "| PT-010 | [~] |"), encoding="utf-8", ) return eng @@ -209,6 +209,13 @@ def test_recon_does_not_require_hypothesis(tmp_path): ), encoding="utf-8", ) + ptt_path = eng / "state" / "ptt.md" + ptt_path.write_text( + ptt_path.read_text(encoding="utf-8") + .replace("| PT-010 | [~] |", "| PT-010 | [x] |") + .replace("| PT-030 | [ ] |", "| PT-030 | [~] |"), + encoding="utf-8", + ) research2 = command.check_command( command.CheckCommandArgs( command="nmap -sV 10.10.10.10", @@ -251,7 +258,7 @@ def test_first_command_requires_an_active_ptt_task(tmp_path): eng = _init_e2e(tmp_path, skill_file) ptt_path = eng / "state" / "ptt.md" ptt_path.write_text( - ptt_path.read_text(encoding="utf-8").replace("| PT-001 | [~] |", "| PT-001 | [ ] |"), + ptt_path.read_text(encoding="utf-8").replace("| PT-010 | [~] |", "| PT-010 | [ ] |"), encoding="utf-8", ) @@ -275,7 +282,7 @@ def test_multiple_active_ptt_tasks_block_target_execution(tmp_path): eng = _init_e2e(tmp_path, skill_file) ptt_path = eng / "state" / "ptt.md" ptt_path.write_text( - ptt_path.read_text(encoding="utf-8").replace("| PT-010 | [ ] |", "| PT-010 | [~] |"), + ptt_path.read_text(encoding="utf-8").replace("| PT-011 | [ ] |", "| PT-011 | [~] |"), encoding="utf-8", ) result = command.check_command( @@ -406,7 +413,7 @@ def test_exec_auto_records_history_but_requires_explicit_ptt_review(monkeypatch, TOOLS.handle_record_ptt( { "eng_dir": str(eng), - "id": "PT-001", + "id": "PT-010", "status": "[~]", "note": f"batch reviewed (batch_id {batch_id})", } @@ -427,7 +434,7 @@ def test_exploitation_gets_bounded_window_then_requires_ptt_review(monkeypatch, ptt_path = eng / "state" / "ptt.md" ptt_path.write_text( ptt_path.read_text(encoding="utf-8") - .replace("| PT-001 | [~] |", "| PT-001 | [x] |") + .replace("| PT-010 | [~] |", "| PT-010 | [x] |") .replace("| PT-042 | [ ] |", "| PT-042 | [~] |"), encoding="utf-8", ) diff --git a/tests/guard/test_sync_credit_window.py b/tests/guard/test_sync_credit_window.py index 2845014..4ece925 100644 --- a/tests/guard/test_sync_credit_window.py +++ b/tests/guard/test_sync_credit_window.py @@ -19,7 +19,7 @@ def _engagement(tmp_path: Path) -> Path: (eng / "state" / ".skill-loaded-test").write_text("skill-loaded: test\n", encoding="utf-8") ptt_path = eng / "state" / "ptt.md" ptt_path.write_text( - ptt_path.read_text(encoding="utf-8").replace("| PT-001 | [ ] |", "| PT-001 | [~] |"), + ptt_path.read_text(encoding="utf-8").replace("| PT-010 | [ ] |", "| PT-010 | [~] |"), encoding="utf-8", ) return eng