External/violin
1
0
Fork 0
mirror of https://github.com/Strategic-Automation/violin.git synced 2026-08-14 12:33:37 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge P0 PTT hypothesis phase enforcement

Browse Source
This commit is contained in:
Violin
2026-07-13 08:41:03 +01:00
parent bcc57fe534 ac5848a85a
commit 683f393ac7
6 changed files with 80 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/violin_guard/core/command.py
+34 -1
View File
@@ -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)
plugins/violin_guard/core/ptt.py
+22 -12
View File
@@ -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<phase>.+?)\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
tests/guard/test_burst_and_target.py
+2 -2
View File
@@ -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))
tests/guard/test_correctness_roadmap_1_1_1.py
+8 -1
View File
@@ -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(
tests/guard/test_plugin_guard.py
+13 -6
View File
@@ -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-<session-id>``; 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",
)
tests/guard/test_sync_credit_window.py
+1 -1
View File
@@ -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