diff --git a/.hermes.md b/.hermes.md index 69d20a5..af4bb30 100644 --- a/.hermes.md +++ b/.hermes.md @@ -38,9 +38,11 @@ Violin **must not** touch a target (no curl, nmap, browser, web_search for the t **Session-start/skill-load precondition** (code-enforced via `check-command`): +Launch with `hermes chat --skills pentest` when possible. If the profile was already started, load `pentest` before creating the marker below. Check the current state at any time with `python $HOME/.hermes/profiles/violin/scripts/violin_guard.py status --eng-dir "$ENG_DIR" --section skill`. + ```text printf 'skill-loaded: %s\n' "$(date -Iseconds)" > "$ENG_DIR/state/.skill-loaded-" -python $HOME/.hermes/profiles/violin/scripts/violin_guard.py check-skill-loaded --eng-dir "$ENG_DIR" --session-id "" +python $HOME/.hermes/profiles/violin/scripts/violin_guard.py status --eng-dir "$ENG_DIR" --section skill ``` Then pass the same session ID into every target-touching command check — `--eng-dir` is mandatory because it activates the PTT, history, hypothesis, and synchronization guards: @@ -66,6 +68,9 @@ The Violin agent **may**: - **Write scripts** — Create Python, Bash, PowerShell, or other scripts under `$ENG_DIR/exploits/` for exploitation, automation, or evidence processing. - **Collect evidence** — Save tool output, screenshots, and findings under `$ENG_DIR/evidence/` and in organised reports. - **Generate reports** — Produce structured penetration test reports with findings, risk ratings, evidence, and remediation steps. +- **Explain guard state cheaply** — Call `violin_status` before guessing at a phase or lock failure. It reports the current task/phase, phase requirements, pending commands, blockers, and next actions without mutation. +- **Review each batch once** — After a bounded batch, call `violin_review_batch` with the active PTT id, lifecycle status, and truthful evidence note. Include the optional finding object only when that batch's receipts support the finding. Retries reuse the same batch marker and finding. +- **Handle protected system files explicitly** — If Hermes' built-in file tool refuses a sensitive system path, stage the file under `$ENG_DIR/exploits/` and, only when the approved engagement requires it, use the host terminal with the necessary privilege to install it. Violin has no `grant-writelist` command; do not invent one. - **Announce intended actions** — Before each tool batch, phase transition, or major operation, tell the user what you plan to do, why, with which tool, and what evidence you expect. Wait for acknowledgment before executing. Use `clarify` or a plain message — do not skip straight to running commands. - **Summarise results** — After each logical tool batch, report: (a) what ran, (b) key results found or nothing notable, (c) evidence saved where. 3-5 lines max. Do not dump raw command output. - **Check in next-step** — After each sub-phase or completed batch, ask the user what to do next with concrete options. E.g. "DNS enumeration done. Found 3 subdomains. Next: tech detection with whatweb, or move to active scanning with nmap?" Do not advance the workflow silently. @@ -131,7 +136,7 @@ which # native Violin runs on **Hermes built-in toolsets** (see [`skills/pentest/SKILL.md §1`](./skills/pentest/SKILL.md#1-operating-model) for the capability inventory and [`README.md §Toolsets`](./README.md#enabled-toolsets) for the full matrix). The canonical command-gate logic lives in the top-level modules under `plugins/violin_guard/` and is callable two ways: - **CLI:** `python scripts/violin_guard.py check-command ...` (the path documented in SKILL.md §2). -- **Hermes plugin:** `plugins/violin_guard/` registers typed guard tools (`violin_check_command`, `violin_record_ptt`, `violin_record_hypothesis`, `violin_exec`, `violin_sync_done`, and adapters) that call the shared service directly. The plugin is required for target execution. +- **Hermes plugin:** `plugins/violin_guard/` registers typed guard tools (`violin_status`, `violin_check_command`, `violin_record_ptt`, `violin_record_hypothesis`, `violin_exec`, `violin_review_batch`, and adapters) that call the shared service directly. The plugin is required for target execution. Both entry points enforce the identical skill-load, active-PTT, history-freshness, hypothesis, and doc-sync gates. The executor itself writes exact history, but never updates PTT progress; that remains an explicit reviewed checkpoint after each bounded batch. Do **not** develop a third implementation. diff --git a/CHANGELOG.md b/CHANGELOG.md index 51dfef4..52e5e5c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,9 @@ ## 2.0.0 +- Added model-visible `violin_status` diagnostics, phase-aware 10/20-command sync windows, a 350-iteration profile budget, and atomic `violin_review_batch` reconciliation with optional receipt-backed finding output. +- Fixed explicit PTT task creation so the requested phase controls the row's actual table placement, and unified CLI/plugin PTT review state. +- Removed message-count heartbeat locks; executed-command heartbeat checks remain phase-aware and are suppressed during exploit-heavy phases. - Made the existing `violin_exec` contract explicit for every installed non-interactive Kali/Parrot CLI tool, and removed the partial target-tool name list from raw-terminal classification in favor of generic target-literal detection. - Reorganised the guard into focused top-level modules under `plugins/violin_guard/`, with separate history, result, execution, state, target, and service responsibilities. - Split web-injection and access-control playbooks into the on-demand `web-attacks` and `access-control` skills while keeping `pentest` as the engagement orchestrator. diff --git a/README.md b/README.md index 786dc24..1123efb 100644 --- a/README.md +++ b/README.md @@ -201,6 +201,9 @@ flowchart LR - **Evidence-first** — every finding backed by reproducible tool output, screenshots, request/response pairs - **Exploit-first validation** — no hypothesis advances to Validated without a verification command - **Stateful recovery** — phase summaries and checkpoints restore the current engagement after context compression without starting a new conversation +- **Self-explaining guard** — `violin_status` (or `python scripts/violin_guard.py status --eng-dir "$ENG_DIR"`) shows the active task and phase, pending commands and their required phases, phase requirements, skill state, blockers, and exact next actions without running a command +- **Phase-aware work windows** — RECON/VULN_RESEARCH allow 10 guarded commands per reviewed batch; EXPLOITATION/POST_EXPLOITATION/PRIVESC/FLAGS allow 20, and the Hermes profile budget is 350 tool iterations +- **One-call reconciliation** — `violin_review_batch` validates the completed batch, optionally writes its receipt-backed finding, updates the active PTT row once, and clears the batch lock last Full safety policy: `skills/pentest/references/standards.md`. Forbidden actions: `.hermes.md` §Forbidden Behaviour. @@ -242,6 +245,8 @@ python scripts/violin_guard.py check-release Validates the plugin manifest and registered tools, isolated Hermes-style plugin import, stale skill references, Ruff, and the full pytest suite. +Hermes skills are loaded on demand. Start the profile with `hermes chat --skills pentest` when the launcher supports arguments; otherwise load `pentest` immediately, then confirm the engagement marker with `python scripts/violin_guard.py status --eng-dir "$ENG_DIR" --section skill`. Hermes does not currently expose a distribution-level setting that can truthfully force-load a profile skill. + --- ## Optional: Kali Docker Container diff --git a/SOUL.md b/SOUL.md index 2314c9e..f825623 100644 --- a/SOUL.md +++ b/SOUL.md @@ -26,7 +26,7 @@ You are a senior security tester and reporting assistant. Be methodical, evidenc - Ask concise scoping questions when the target, authorisation, testing mode, or risk tolerance is unclear. - Maintain a clear trail from scope → method → evidence → finding → remediation. - Treat `skills/pentest/references/standards.md` as the authoritative safety policy for approval tiers, blocked actions, evidence handling, rate limits, and scope allowlists. -- **Use the `violin-guard` tools for all target-touching command execution.** Keep exactly one PTT task `[~]`; the guard blocks otherwise. `violin_exec` and `violin_exec_burst` write exact command history automatically, but never update PTT progress. At the end of the bounded batch, review results, explicitly update the active PTT row, and call `violin_sync_done`. Never ask the model to recreate normal command history. +- **Use the `violin-guard` tools for all target-touching command execution.** Keep exactly one PTT task `[~]`; the guard blocks otherwise. `violin_exec` and `violin_exec_burst` write exact command history automatically, but never update PTT progress. At the end of the bounded batch, review results and call `violin_review_batch` once, including a finding only when the batch receipts support one. Never ask the model to recreate normal command history. ## Workflow Drift Guard @@ -34,7 +34,7 @@ Detailed procedure lives in `skills/pentest/SKILL.md §2`; keep SOUL to hard inv - Bootstrap and scope come first: no target interaction until `$ENG_DIR`, `scope/scope.yaml`, `state/ptt.md`, `hypotheses.md`, and `state/history.md` exist and pass guard checks. - Target-touching commands use `violin_exec` or `violin_exec_burst`; `violin_exec` has no binary allowlist and is the single guarded boundary for any installed non-interactive Kali/Parrot CLI tool. Raw `terminal` is for host-local work and has best-effort target detection only. `execute_code` requires the Violin JSON audit header and is recorded against its engagement, but does not replace typed execution for target work. -- `sync_required` means reconcile the pending command's artifacts, then call `violin_sync_done`; do not retry target commands. +- `sync_required` means reconcile the pending command's artifacts, then call `violin_review_batch`; do not retry target commands. - `heartbeat_required` means re-read `skills/pentest/SKILL.md`, review scope/PTT/hypotheses/history, then call `violin_heartbeat_done`. - Never skip REPORTING or RETROSPECTIVE; record any gap explicitly. diff --git a/config.yaml b/config.yaml index 4bb8331..fdbe09f 100644 --- a/config.yaml +++ b/config.yaml @@ -2,6 +2,7 @@ # Hermes default and does not require a profile-specific provider or API key. agent: + max_turns: 350 service_tier: normal verbose: false reasoning_effort: medium diff --git a/plugins/violin_guard/__init__.py b/plugins/violin_guard/__init__.py index ad4f2ca..9a3d084 100644 --- a/plugins/violin_guard/__init__.py +++ b/plugins/violin_guard/__init__.py @@ -30,7 +30,7 @@ REGISTERED_TOOLS = [ "violin_exec", "violin_exec_status", "violin_exec_cancel", - "violin_sync_done", + "violin_review_batch", "violin_rebind_pending_batch", "violin_heartbeat_done", "violin_exec_burst", @@ -58,7 +58,12 @@ def register(ctx) -> None: ("violin_exec", schemas.EXEC_SCHEMA, service.handle_exec, "⚡"), ("violin_exec_status", schemas.EXEC_STATUS_SCHEMA, service.handle_exec_status, "i"), ("violin_exec_cancel", schemas.EXEC_CANCEL_SCHEMA, service.handle_exec_cancel, "x"), - ("violin_sync_done", schemas.SYNC_DONE_SCHEMA, service.handle_sync_done, "✅"), + ( + "violin_review_batch", + schemas.REVIEW_BATCH_SCHEMA, + service.handle_review_batch, + "✅", + ), ( "violin_rebind_pending_batch", schemas.REBIND_PENDING_BATCH_SCHEMA, @@ -160,7 +165,7 @@ def _on_session_reset_hook(session_id=None, eng_dir=None, **kwargs) -> None: def _on_session_finalize_hook(session_id=None, eng_dir=None, **kwargs) -> None: """Hook: session finalize. - Closeout gates are explicit (violin_sync_done / close command). On finalize + Closeout gates are explicit (violin_review_batch / close command). On finalize we leave a continuity marker so a fresh session can re-read pending state. """ if eng_dir: @@ -169,7 +174,7 @@ def _on_session_finalize_hook(session_id=None, eng_dir=None, **kwargs) -> None: if pending: state.set_heartbeat_pending( str(eng_dir), - "session finalized with a pending sync lock; run violin_sync_done", + "session finalized with a pending sync lock; run violin_review_batch", ) except Exception: pass diff --git a/plugins/violin_guard/command.py b/plugins/violin_guard/command.py index d5192c0..64fb634 100644 --- a/plugins/violin_guard/command.py +++ b/plugins/violin_guard/command.py @@ -427,7 +427,9 @@ def check_command(args: CheckCommandArgs) -> CheckResult: 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}" + f"requested phase is {phase.value}. Next: call violin_status, then close or " + "pause the current task and start one under the requested Phase heading with " + "violin_record_ptt" ) # 5. History staleness (duplicate detection) @@ -449,14 +451,15 @@ def check_command(args: CheckCommandArgs) -> CheckResult: # 7. Sync/heartbeat state sync_pending = state.get_pending_sync(str(eng_dir)) if sync_pending: - credit = state.sync_credit_remaining(str(eng_dir)) + credit = state.sync_credit_remaining(str(eng_dir), phase.value) last_command = (sync_pending.get("commands") or [{}])[-1].get( "command", sync_pending.get("command", "prior command") ) if credit == 0: result.add_error( f"prior command's artifacts not synced: {last_command} " - f"(phase: {sync_pending.get('phase')})" + f"(phase: {sync_pending.get('phase')}). Next: review the batch evidence and call " + "violin_review_batch with the active PTT task and a truthful note" ) else: result.add_info( @@ -465,10 +468,16 @@ def check_command(args: CheckCommandArgs) -> CheckResult: ) # 8. Sync-credit window exhausted - credit = state.sync_credit_remaining(str(eng_dir)) - result.infos.append(f"sync credit remaining: {credit}/{state.DEFAULT_SYNC_CREDIT}") + credit = state.sync_credit_remaining(str(eng_dir), phase.value) + credit_limit = int( + (sync_pending or {}).get("credit_limit") or state.sync_credit_limit(phase.value) + ) + result.infos.append(f"sync credit remaining: {credit}/{credit_limit}") if credit == 0: - result.add_error("sync-credit window exhausted — call violin_sync_done to reset") + result.add_error( + "sync-credit window exhausted; review the saved batch evidence, then call " + "violin_review_batch" + ) # 9. Heartbeat gate (set after every COMMAND_INTERVAL executed commands). # Execution owns the command count and creates the heartbeat lock after the @@ -479,8 +488,7 @@ def check_command(args: CheckCommandArgs) -> CheckResult: reason = state.get_heartbeat_reason(str(eng_dir)) detail = f": {reason}" if reason else "" result.add_error( - f"heartbeat pending{detail} — review engagement state, then run " - "violin_heartbeat_done" + f"heartbeat pending{detail} — review engagement state, then run violin_heartbeat_done" ) return result diff --git a/plugins/violin_guard/execution.py b/plugins/violin_guard/execution.py index baec230..14c5c32 100644 --- a/plugins/violin_guard/execution.py +++ b/plugins/violin_guard/execution.py @@ -245,7 +245,7 @@ def _monitor_background( def _commit_started_command(engagement: Path, command: str, phase: str, ptt_task_id: str) -> int: if state.is_local_bookkeeping_command(command): - return state.sync_credit_remaining(str(engagement)) + return state.sync_credit_remaining(str(engagement), phase) return _commit_guard_state(engagement, command, phase, ptt_task_id) @@ -454,7 +454,7 @@ def execute( def _commit_guard_state(eng_dir: Path, command: str, phase: str, ptt_task_id: str = "") -> int: state.record_ok_check(str(eng_dir), command, phase) - remaining = state.spend_sync_credit(str(eng_dir)) + remaining = state.spend_sync_credit(str(eng_dir), phase) state.mark_pending_sync(str(eng_dir), command, phase, ptt_task_id) count = state.tick_command(str(eng_dir)) from .phases import suppresses_heartbeat diff --git a/plugins/violin_guard/findings.py b/plugins/violin_guard/findings.py new file mode 100644 index 0000000..3c7662f --- /dev/null +++ b/plugins/violin_guard/findings.py @@ -0,0 +1,183 @@ +"""Structured finding creation from guarded execution receipts.""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import Any + +from . import state + +_FINDING_ID_RE = re.compile(r"FIND-(\d{3,})$") +_SEVERITIES = {"critical", "high", "medium", "low", "info"} + + +def _next_finding_id(directory: Path) -> str: + numbers = [] + for path in directory.glob("FIND-*.md"): + match = _FINDING_ID_RE.fullmatch(path.stem) + if match: + numbers.append(int(match.group(1))) + return f"FIND-{max(numbers, default=0) + 1:03d}" + + +def _batch_evidence(eng_dir: Path, pending: dict[str, Any]) -> list[str]: + unmatched = {str(item.get("command") or "") for item in pending.get("commands") or []} + evidence: list[str] = [] + manifests = sorted( + (eng_dir / "evidence" / "executions").glob("*.json"), + key=lambda path: path.stat().st_mtime_ns, + reverse=True, + ) + for manifest in manifests: + receipt = state.read_json(manifest) + command = str(receipt.get("command") or "") + if command not in unmatched: + continue + unmatched.remove(command) + for value in (receipt.get("evidence_paths") or {}).values(): + relative = str(value or "").strip() + if relative and (eng_dir / relative).is_file() and relative not in evidence: + evidence.append(relative) + return evidence + + +def _existing_batch_finding(directory: Path, batch_id: str) -> Path | None: + marker = f"- **Batch:** {batch_id}" + for path in sorted(directory.glob("FIND-*.md")): + try: + if marker in path.read_text(encoding="utf-8").splitlines(): + return path + except OSError: + continue + return None + + +def _validate_from_pending_batch( + eng_dir: str | Path, + pending: dict[str, Any], + *, + title: str, + severity: str, + description: str, + impact: str, + remediation: str, + finding_id: str = "", +) -> dict[str, Any]: + engagement = state.resolve_eng_dir(eng_dir) + values = { + "title": title.strip(), + "description": description.strip(), + "impact": impact.strip(), + "remediation": remediation.strip(), + } + severity_key = severity.strip().lower() + if not all(values.values()): + raise ValueError("title, description, impact, and remediation must be non-empty") + if severity_key not in _SEVERITIES: + raise ValueError("severity must be one of Critical, High, Medium, Low, or Info") + identifier = finding_id.strip().upper() + if identifier and not _FINDING_ID_RE.fullmatch(identifier): + raise ValueError("finding_id must use FIND-NNN format") + evidence = _batch_evidence(engagement, pending) + if not evidence: + raise ValueError("the current batch has no completed execution receipts to cite") + return { + **values, + "severity": severity_key, + "finding_id": identifier, + "evidence_paths": evidence, + } + + +def _create_from_pending_batch( + eng_dir: str | Path, + *, + title: str, + severity: str, + description: str, + impact: str, + remediation: str, + finding_id: str = "", + pending: dict[str, Any] | None = None, +) -> dict[str, Any]: + engagement = state.resolve_eng_dir(eng_dir) + pending = pending or state.get_pending_sync(engagement) + if not pending: + raise ValueError("no current execution batch; run guarded validation commands first") + draft = _validate_from_pending_batch( + engagement, + pending, + title=title, + severity=severity, + description=description, + impact=impact, + remediation=remediation, + finding_id=finding_id, + ) + + directory = engagement / "evidence" / "findings" + directory.mkdir(parents=True, exist_ok=True) + batch_id = str(pending.get("batch_id") or "") + existing = _existing_batch_finding(directory, batch_id) + if existing: + if draft["finding_id"] and draft["finding_id"] != existing.stem: + raise ValueError( + f"batch {batch_id} already has finding {existing.stem}; " + f"refusing requested {draft['finding_id']}" + ) + return { + "finding_id": existing.stem, + "path": existing.relative_to(engagement).as_posix(), + "evidence_paths": draft["evidence_paths"], + "batch_id": batch_id, + "reused": True, + } + + identifier = draft["finding_id"] or _next_finding_id(directory) + output = directory / f"{identifier}.md" + if output.exists(): + raise ValueError(f"finding already exists: {output}") + + commands = [str(item.get("command") or "") for item in pending.get("commands") or []] + lines = [ + f"# {identifier}: {draft['title']}", + "", + f"- **Severity:** {draft['severity'].title()}", + f"- **Batch:** {batch_id or 'unknown'}", + f"- **PTT task:** {pending.get('ptt_task_id') or 'unknown'}", + f"- **Phase:** {pending.get('phase') or 'unknown'}", + "", + "## Description", + "", + draft["description"], + "", + "## Impact", + "", + draft["impact"], + "", + "## Evidence", + "", + *[f"- `{path}`" for path in draft["evidence_paths"]], + "", + "## Reproduction commands", + "", + "```text", + *commands, + "```", + "", + "## Remediation", + "", + draft["remediation"], + "", + ] + temporary = output.with_suffix(output.suffix + ".tmp") + temporary.write_text("\n".join(lines), encoding="utf-8") + temporary.replace(output) + return { + "finding_id": identifier, + "path": output.relative_to(engagement).as_posix(), + "evidence_paths": draft["evidence_paths"], + "batch_id": batch_id, + "reused": False, + } diff --git a/plugins/violin_guard/plugin.yaml b/plugins/violin_guard/plugin.yaml index e660dcc..313cbc1 100644 --- a/plugins/violin_guard/plugin.yaml +++ b/plugins/violin_guard/plugin.yaml @@ -9,7 +9,7 @@ provides_tools: - violin_exec - violin_exec_status - violin_exec_cancel - - violin_sync_done + - violin_review_batch - violin_rebind_pending_batch - violin_heartbeat_done - violin_exec_burst diff --git a/plugins/violin_guard/ptt.py b/plugins/violin_guard/ptt.py index 45a5476..bf97821 100644 --- a/plugins/violin_guard/ptt.py +++ b/plugins/violin_guard/ptt.py @@ -200,7 +200,7 @@ def update_task(path: Path, task_id: str, status: str, note: str) -> PttTask: def create_task(path: Path, task_id: str, title: str, phase: str, note: str = "") -> PttTask: - """Append an explicitly requested untouched task to the canonical phase.""" + """Insert an explicitly requested untouched task into its canonical phase table.""" if not re.fullmatch(r"PT-[\w-]+", task_id): raise ValueError("task id must use the PT- prefix") canonical_phase = normalize_phase(phase).value @@ -208,12 +208,67 @@ def create_task(path: Path, task_id: str, title: str, phase: str, note: str = "" if any(task.id == task_id for task in tasks): raise ValueError(f"PTT task {task_id!r} already exists") content = path.read_text(encoding="utf-8") if path.exists() else "# Pentesting Task Tree\n" - heading = f"## Phase: {canonical_phase}" - if heading not in content: - content = ( - content.rstrip() - + f"\n\n{heading}\n\n| ID | Status | Task | Notes |\n|---|---|---|---|\n" + lines = content.splitlines() + phase_heading_index = None + for index, line in enumerate(lines): + match = re.match(r"^##\s+Phase:\s*(?P.+?)\s*$", line.strip(), re.IGNORECASE) + if not match: + continue + raw_phase = re.split(r"\s*\(", match.group("phase"), maxsplit=1)[0].strip() + try: + heading_phase = normalize_phase(raw_phase).value + except ValueError: + continue + if heading_phase == canonical_phase: + phase_heading_index = index + break + + if phase_heading_index is None: + lines.extend( + [ + "", + f"## Phase: {canonical_phase}", + "", + "| ID | Status | Task | Notes |", + "|---|---|---|---|", + ] ) - content = content.rstrip() + f"\n| {task_id} | [ ] | {title.strip()} | {note.strip()} |\n" - path.write_text(content, encoding="utf-8") + phase_heading_index = len(lines) - 4 + + next_heading_index = next( + ( + index + for index in range(phase_heading_index + 1, len(lines)) + if re.match(r"^##\s+Phase:", lines[index].strip(), re.IGNORECASE) + ), + len(lines), + ) + table_start = next( + ( + index + for index in range(phase_heading_index + 1, next_heading_index) + if lines[index].lstrip().startswith("|") + ), + None, + ) + if table_start is None: + raise ValueError(f"phase {canonical_phase} has no task table") + + table_end = table_start + while table_end + 1 < next_heading_index and lines[table_end + 1].lstrip().startswith("|"): + table_end += 1 + column_count = len([cell for cell in lines[table_start].strip().strip("|").split("|")]) + if column_count < 4: + raise ValueError(f"phase {canonical_phase} task table must have at least four columns") + + def clean_cell(value: str) -> str: + return value.strip().replace("|", "\\|").replace("\n", " ") + + cells = [task_id, "[ ]", clean_cell(title)] + cells.extend([""] * (column_count - 4)) + cells.append(clean_cell(note)) + lines.insert(table_end + 1, "| " + " | ".join(cells) + " |") + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_text("\n".join(lines) + "\n", encoding="utf-8") + temporary.replace(path) return next(task for task in parse_ptt(path) if task.id == task_id) diff --git a/plugins/violin_guard/schemas.py b/plugins/violin_guard/schemas.py index c9c3432..4beb55b 100644 --- a/plugins/violin_guard/schemas.py +++ b/plugins/violin_guard/schemas.py @@ -118,14 +118,37 @@ EXEC_SCHEMA = { }, } -SYNC_DONE_SCHEMA = { - "description": "Verify explicit batch reconciliation. Command history is written automatically, but the active PTT row must be reviewed and updated after the batch; the executor cannot satisfy this checkpoint. Clears the lock only when both artifacts are fresh.", +REVIEW_BATCH_SCHEMA = { + "description": "Review the current completed batch, optionally create one receipt-backed finding, update the active PTT task, and release the sync lock. All inputs are validated before mutation; the lock clears last.", "parameters": { "type": "object", "properties": { - "eng_dir": {"type": "string", "description": "Engagement directory"}, + "eng_dir": {"type": "string"}, + "id": {"type": "string", "description": "Active PTT task id"}, + "status": { + "type": "string", + "enum": ["[~]", "[x]", "[!]", "[-]"], + }, + "note": {"type": "string", "description": "Truthful result/evidence review"}, + "finding": { + "type": "object", + "description": "Optional structured finding derived only from this batch", + "properties": { + "finding_id": {"type": "string", "description": "Optional FIND-NNN id"}, + "title": {"type": "string"}, + "severity": { + "type": "string", + "enum": ["Critical", "High", "Medium", "Low", "Info"], + }, + "description": {"type": "string"}, + "impact": {"type": "string"}, + "remediation": {"type": "string"}, + }, + "required": ["title", "severity", "description", "impact", "remediation"], + "additionalProperties": False, + }, }, - "required": ["eng_dir"], + "required": ["eng_dir", "id", "status", "note"], "additionalProperties": False, }, } @@ -155,7 +178,7 @@ REBIND_PENDING_BATCH_SCHEMA = { } HEARTBEAT_DONE_SCHEMA = { - "description": f"Call AFTER heartbeat review: re-read skills/pentest/SKILL.md and review scope.yaml / state/ptt.md / hypotheses.md / state/history.md. Cadence is {state.COMMAND_INTERVAL} target commands or {state.MESSAGE_INTERVAL} message ticks; exploitation/post-exploitation suppresses heartbeat. Clears heartbeat lock so violin_exec may release the next command.", + "description": f"Call AFTER heartbeat review: re-read skills/pentest/SKILL.md and review scope.yaml / state/ptt.md / hypotheses.md / state/history.md. Cadence is {state.COMMAND_INTERVAL} executed target commands; exploitation/post-exploitation/PRIVESC/FLAGS suppress heartbeat. Clears heartbeat lock so violin_exec may release the next command.", "parameters": { "type": "object", "properties": { @@ -168,7 +191,7 @@ HEARTBEAT_DONE_SCHEMA = { EXEC_BURST_SCHEMA = { "name": "violin_exec_burst", - "description": "Single-approval bounded command batch. Requires one unambiguous [~] PTT task. Every completed command is appended to history automatically, but the executor never updates PTT progress. Review the batch, update the active PTT row explicitly, then call violin_sync_done. Use for recon and exploit/race batches; never raw terminal for targets.", + "description": "Single-approval bounded command batch. Requires one unambiguous [~] PTT task. Every completed command is appended to history automatically, but the executor never updates PTT progress. Review the batch once with violin_review_batch. Use for recon and exploit/race batches; never raw terminal for targets.", "parameters": { "type": "object", "properties": { @@ -363,7 +386,7 @@ TARGET_SCHEMA = { STATUS_SCHEMA = { "name": "violin_status", - "description": "One-shot engagement health read: bootstrap completeness, skill-load freshness, pending doc-sync, heartbeat-pending, sync credit remaining, and command/message counts. Mutates no state.", + "description": "Cheap one-shot explanation of the current task and phase, per-phase command requirements, pending batch commands, blockers, exact next actions, skill-load state, heartbeat state, and phase-aware sync credit. Mutates no state.", "parameters": { "type": "object", "properties": { @@ -376,7 +399,7 @@ STATUS_SCHEMA = { "description": "explicit skill-load marker path (else $ENG_DIR/.skill-loaded)", }, }, - "required": [], + "required": ["eng_dir"], "additionalProperties": False, }, } diff --git a/plugins/violin_guard/service.py b/plugins/violin_guard/service.py index 6441a45..ce7a980 100644 --- a/plugins/violin_guard/service.py +++ b/plugins/violin_guard/service.py @@ -4,12 +4,13 @@ from __future__ import annotations import json import os +import re import shlex from functools import wraps from pathlib import Path +from . import bootstrap, execution, findings, hypotheses, ptt, state from . import command as cmd_module -from . import execution, hypotheses, ptt, state from .adapters import ( build_ffuf, build_httpx, @@ -19,6 +20,7 @@ from .adapters import ( ) from .command import CheckCommandArgs from .history import history_contains +from .phases import Phase, requires_hypothesis, suppresses_heartbeat from .targets import resolve_target # --------------------------------------------------------------------------- @@ -92,75 +94,30 @@ def handle_record_ptt(a, **kwargs): note = (a.get("note") or "").strip() status = a.get("status", "[~]") - # --- Self-certify guard (audit P0-sync) --------------------------------- if not task or not note: - raise ValueError("task id and non-empty review note required") - if not pending: - if not any(item.id == task for item in doc): - created = ptt.create_task( - _eng_path(eng_dir) / "state" / "ptt.md", - task, - a.get("title") or task, - a.get("phase") or "RECON", - note, - ) - doc = ptt.parse_ptt(_eng_path(eng_dir) / "state" / "ptt.md") - if status == "[ ]": - return _json("ok", task_id=created.id, task_created=True) - existing = next((item for item in doc if item.id == task), None) - if existing and status in {"[x]", "[-]"}: - if existing.status != "[~]": - raise ValueError("only the active [~] task may be closed outside a batch") - ptt.update_task(_eng_path(eng_dir) / "state" / "ptt.md", task, status, note) - return _json("ok", task_id=task, task_closed=True) - return _start_ptt_task(_eng_path(eng_dir) / "state" / "ptt.md", doc, task, status, note) - validation = ptt.validate_ptt(doc) - if validation.errors: - raise ValueError("PTT must have exactly one valid active task before review") - active = ptt.find_active_task(doc) - captured_task = pending.get("ptt_task_id") - if not captured_task: + raise ValueError("task id and non-empty lifecycle note required") + if pending: raise ValueError( - "pending batch has no captured PTT task; refusing legacy self-certification" + "a target batch is pending; use violin_review_batch instead of violin_record_ptt" ) - # A human may update ptt.md directly after a batch. If there is one - # phase-compatible active task, reconcile that deliberate state edit - # here instead of forcing an opaque multi-step rebind ceremony. - if active and active.id != captured_task: - phases = { - str(item.get("phase") or pending.get("phase") or "") - for item in pending.get("commands") or [] - } - {""} - if all(ptt.task_matches_phase(active, phase) for phase in phases): - state.rebind_pending_sync( - eng_dir, - expected_batch_id=str(pending.get("batch_id") or ""), - current_task_id=str(captured_task), - replacement_task_id=active.id, - note="automatic rebind after direct PTT edit", - ) - pending = state.get_pending_sync(eng_dir) or pending - captured_task = active.id - if task != captured_task: - raise ValueError(f"reviewed task {task!r} does not match batch task {captured_task!r}") - if not active or active.id != captured_task: - raise ValueError( - f"reviewed task {task!r} is not the active task; resolve the active task first" + if not any(item.id == task for item in doc): + created = ptt.create_task( + _eng_path(eng_dir) / "state" / "ptt.md", + task, + a.get("title") or task, + a.get("phase") or "RECON", + note, ) - batch_id = pending.get("batch_id") - if batch_id and batch_id not in note: - note = f"{note} [reviewed-batch:{batch_id}]" - for item in pending.get("commands") or []: - cmd = item.get("command") - if cmd and not history_contains(eng_dir, cmd): - raise ValueError( - f"pending command not yet in history.md: {cmd!r}; " - "the batch must finish before review" - ) - - ptt.update_task(_eng_path(eng_dir) / "state" / "ptt.md", task, status, note) - state.mark_ptt_reviewed(eng_dir, task, note) - return _json("ok", task_id=task, batch_id=pending.get("batch_id")) + doc = ptt.parse_ptt(_eng_path(eng_dir) / "state" / "ptt.md") + if status == "[ ]": + return _json("ok", task_id=created.id, task_created=True) + existing = next((item for item in doc if item.id == task), None) + if existing and status in {"[x]", "[-]"}: + if existing.status != "[~]": + raise ValueError("only the active [~] task may be closed outside a batch") + ptt.update_task(_eng_path(eng_dir) / "state" / "ptt.md", task, status, note) + return _json("ok", task_id=task, task_closed=True) + return _start_ptt_task(_eng_path(eng_dir) / "state" / "ptt.md", doc, task, status, note) def _start_ptt_task(ptt_path: Path, tasks, task_id: str, status: str, note: str) -> str: @@ -210,24 +167,150 @@ def _scope_hosts(eng_dir: str) -> set[str] | None: return scope_hosts(data) or None +def _task_row_contains(path: Path, task_id: str, marker: str) -> bool: + for line in path.read_text(encoding="utf-8").splitlines(): + match = re.match(r"^\|\s*(PT-[\w-]+)\s*\|", line.strip()) + if match and match.group(1) == task_id: + return marker in line + return False + + +def _validate_review_batch(a: dict, pending: dict) -> dict: + eng_dir = str(a.get("eng_dir") or "") + task_id = str(a.get("id") or "").strip() + note = str(a.get("note") or "").strip() + status = str(a.get("status") or "").strip() + if not task_id or not note: + raise ValueError("active task id and non-empty review note are required") + if status not in {"[~]", "[x]", "[!]", "[-]"}: + raise ValueError("status must be one of [~], [x], [!], or [-]") + + batch_id = str(pending.get("batch_id") or "").strip() + captured_task = str(pending.get("ptt_task_id") or "").strip() + if not batch_id or not captured_task: + raise ValueError("pending batch is missing its batch or PTT task identity") + if task_id != captured_task: + raise ValueError(f"reviewed task {task_id!r} does not match batch task {captured_task!r}") + + ptt_path = _eng_path(eng_dir) / "state" / "ptt.md" + tasks = ptt.parse_ptt(ptt_path) + selected = next((item for item in tasks if item.id == task_id), None) + if selected is None: + raise ValueError(f"batch task {task_id!r} is missing from the PTT") + marker = f"[reviewed-batch:{batch_id}]" + already_recorded = selected.status == status and _task_row_contains(ptt_path, task_id, marker) + if not already_recorded: + validation = ptt.validate_ptt(tasks) + if validation.errors: + raise ValueError("PTT must have exactly one valid active task before batch review") + active = ptt.find_active_task(tasks) + if not active or active.id != task_id: + raise ValueError(f"batch task {task_id!r} must be the sole active [~] task") + phases = { + str(item.get("phase") or pending.get("phase") or "") + for item in pending.get("commands") or [] + } - {""} + incompatible = sorted( + phase for phase in phases if not ptt.task_matches_phase(active, phase) + ) + if incompatible: + raise ValueError( + f"batch task {task_id!r} is not phase-compatible with " + ", ".join(incompatible) + ) + + for item in pending.get("commands") or []: + command = str(item.get("command") or "") + if command and not history_contains(eng_dir, command): + raise ValueError( + f"pending command not yet in exact history: {command!r}; " + "wait for execution completion before review" + ) + + finding = a.get("finding") + if finding is not None: + if not isinstance(finding, dict): + raise ValueError("finding must be an object when supplied") + findings._validate_from_pending_batch( + eng_dir, + pending, + title=str(finding.get("title") or ""), + severity=str(finding.get("severity") or ""), + description=str(finding.get("description") or ""), + impact=str(finding.get("impact") or ""), + remediation=str(finding.get("remediation") or ""), + finding_id=str(finding.get("finding_id") or ""), + ) + return { + "batch_id": batch_id, + "task_id": task_id, + "status": status, + "note": note, + "marker": marker, + "already_recorded": already_recorded, + "ptt_path": ptt_path, + "finding": finding, + } + + @_serialise_errors -def handle_sync_done(a, **kwargs): +def handle_review_batch(a, **kwargs): + """Review one completed batch, optionally record a finding, and release its lock.""" + + eng_dir = str(a.get("eng_dir") or "").strip() + if not eng_dir: + raise ValueError("eng_dir is required") + engagement = _eng_path(eng_dir) + review_lock = engagement / "state" / "review-batch.json" try: - p = state.get_pending_sync(a["eng_dir"]) - if not p: - return _json("ok", message="nothing pending") - if not p.get("ptt_reviewed"): - return _json("sync_required", error="explicit PTT review required") - for item in p.get("commands") or []: - if not history_contains(a["eng_dir"], item.get("command", "")): + with state.lock_file(review_lock): + pending = state.get_pending_sync(engagement) + if not pending: return _json( - "sync_required", - error="all pending commands must exist in exact history before sync", + "ok", + batch_id=None, + task_id=None, + task_status=None, + released=True, + finding=None, + finding_path=None, + message="nothing pending", ) - state.clear_pending_sync(a["eng_dir"]) - return _json("ok", batch_id=p.get("batch_id")) - except Exception as e: - return _json("error", error=str(e)) + context = _validate_review_batch(a, pending) + finding_result = None + finding = context["finding"] + if finding is not None: + finding_result = findings._create_from_pending_batch( + engagement, + pending=pending, + title=str(finding.get("title") or ""), + severity=str(finding.get("severity") or ""), + description=str(finding.get("description") or ""), + impact=str(finding.get("impact") or ""), + remediation=str(finding.get("remediation") or ""), + finding_id=str(finding.get("finding_id") or ""), + ) + if not context["already_recorded"]: + review_note = f"{context['note']} {context['marker']}" + ptt.update_task( + context["ptt_path"], context["task_id"], context["status"], review_note + ) + state.clear_pending_sync(engagement) + return _json( + "ok", + batch_id=context["batch_id"], + task_id=context["task_id"], + task_status=context["status"], + released=True, + finding=finding_result, + finding_path=finding_result.get("path") if finding_result else None, + ) + except (OSError, ValueError) as exc: + return _json( + "blocked", + released=False, + error=str(exc), + next_action="Resolve the reported batch, PTT, history, or finding issue and retry violin_review_batch", + ) def _rebind_fields(a) -> tuple[str, str, str, str, str]: @@ -516,11 +599,121 @@ def handle_target(a, **kwargs): @_serialise_errors def handle_status(a, **kwargs): + if not str(a.get("eng_dir") or "").strip(): + raise ValueError("eng_dir is required") + eng_dir = state.resolve_eng_dir(a.get("eng_dir", "")) + bootstrap_result = bootstrap.check_bootstrap(eng_dir, auto_repair=False) + tasks = ptt.parse_ptt(eng_dir / "state" / "ptt.md") + ptt_result = ptt.validate_ptt(tasks) + active = ptt.find_active_task(tasks) if not ptt_result.errors else None + current_phase = active.phase if active else None + pending = state.get_pending_sync(eng_dir) + credit_limit = int( + (pending or {}).get("credit_limit") or state.sync_credit_limit(current_phase) + ) + credit = state.sync_credit_remaining(eng_dir, current_phase) + counts = state.read_counts(eng_dir) + session_id = state.resolve_session_id(eng_dir) + marker = eng_dir / "state" / f".skill-loaded-{session_id}" if session_id else None + skill_loaded = bool(marker and marker.is_file()) + + blockers = [ + { + "code": "bootstrap", + "reason": error, + "next_action": "Run check-bootstrap and repair the named engagement artifact", + } + for error in bootstrap_result.errors + ] + if not session_id: + blockers.append( + { + "code": "skill_session_unknown", + "reason": "No session id is recorded for the skill-load gate", + "next_action": "Load pentest, then create its marker for the current session", + } + ) + elif not skill_loaded: + blockers.append( + { + "code": "skill_not_loaded", + "reason": f"Pentest skill marker is missing for session {session_id}", + "next_action": f"Load pentest, then create {marker}", + } + ) + blockers.extend( + { + "code": "ptt", + "reason": error, + "next_action": "Use violin_record_ptt to leave exactly one phase-compatible [~] task", + } + for error in ptt_result.errors + ) + if pending and credit == 0: + blockers.append( + { + "code": "sync_required", + "reason": "The bounded command batch is complete and still locked", + "next_action": ( + "Review its evidence, then call violin_review_batch with the active task" + ), + } + ) + heartbeat_pending = state.has_heartbeat_pending(eng_dir) + if heartbeat_pending and not (current_phase and suppresses_heartbeat(Phase(current_phase))): + blockers.append( + { + "code": "heartbeat_required", + "reason": state.get_heartbeat_reason(eng_dir) or "Periodic review is pending", + "next_action": "Review engagement state, then call violin_heartbeat_done", + } + ) + + phase_requirements = { + phase.value: { + "ptt_phase": "EXPLOITATION" if phase is Phase.POST_EXPLOITATION else phase.value, + "hypothesis_required": requires_hypothesis(phase), + "sync_window": state.sync_credit_limit(phase.value), + "heartbeat_enabled": not suppresses_heartbeat(phase), + } + for phase in Phase + } + pending_commands = [ + {"command": item.get("command", ""), "required_phase": item.get("phase", "")} + for item in (pending or {}).get("commands") or [] + ] return _json( - "ok", - sync_pending=state.has_pending_sync(a["eng_dir"]), - sync_credit_remaining=state.sync_credit_remaining(a["eng_dir"]), - command_count=state.read_counts(a["eng_dir"])["commands"], + "blocked" if blockers else "ok", + engagement=str(eng_dir), + current_task=active.id if active else None, + current_task_title=active.title if active else None, + current_phase=current_phase, + command_phase_rule=( + "Every target command must declare the active task phase; POST_EXPLOITATION uses an " + "EXPLOITATION PTT task" + ), + phase_requirements=phase_requirements, + blockers=blockers, + pending_batch={ + "batch_id": (pending or {}).get("batch_id"), + "task_id": (pending or {}).get("ptt_task_id"), + "ptt_reviewed": bool((pending or {}).get("ptt_reviewed")), + "commands": pending_commands, + } + if pending + else None, + sync_credit_remaining=credit, + sync_credit_limit=credit_limit, + heartbeat_pending=heartbeat_pending, + heartbeat_reason=state.get_heartbeat_reason(eng_dir), + command_count=counts["commands"], + message_count=counts["messages"], + skill={ + "name": "pentest", + "session_id": session_id or None, + "loaded": skill_loaded, + "marker": str(marker) if marker else None, + }, ) diff --git a/plugins/violin_guard/state.py b/plugins/violin_guard/state.py index 0dee419..8283d5c 100644 --- a/plugins/violin_guard/state.py +++ b/plugins/violin_guard/state.py @@ -17,8 +17,15 @@ from filelock import FileLock DEFAULT_SYNC_CREDIT = 5 COMMAND_INTERVAL = 50 -MESSAGE_INTERVAL = 60 MAX_BURST_COMMANDS = 20 +PHASE_SYNC_CREDIT = { + "RECON": 10, + "VULN_RESEARCH": 10, + "EXPLOITATION": 20, + "POST_EXPLOITATION": 20, + "PRIVESC": 20, + "FLAGS": 20, +} # Local tools LOCAL_TOOLS = {"echo", "true", "false", "printf", "pwd", "ls", "cat", "date"} @@ -141,16 +148,22 @@ def _sync_path(eng_dir: str | Path) -> Path: return _state_dir(eng_dir) / _SYNC_FILE -def sync_credit_remaining(eng_dir: str | Path) -> int: +def sync_credit_limit(phase: str | None = None) -> int: + key = str(phase or "").strip().upper().replace("-", "_") + return PHASE_SYNC_CREDIT.get(key, DEFAULT_SYNC_CREDIT) + + +def sync_credit_remaining(eng_dir: str | Path, phase: str | None = None) -> int: data = read_json(_sync_path(eng_dir)) - return max(0, data.get("credit", DEFAULT_SYNC_CREDIT)) + return max(0, data.get("credit", sync_credit_limit(phase))) -def spend_sync_credit(eng_dir: str | Path) -> int: +def spend_sync_credit(eng_dir: str | Path, phase: str) -> int: path = _sync_path(eng_dir) def spend(data: dict[str, Any]) -> int: - credit = max(0, data.get("credit", DEFAULT_SYNC_CREDIT) - 1) + starting_credit = data.get("credit", sync_credit_limit(phase)) + credit = max(0, starting_credit - 1) data["credit"] = credit return credit @@ -182,6 +195,7 @@ def mark_pending_sync( or datetime.now(UTC).isoformat().replace("+00:00", "Z"), "ptt_task_id": task_id, "ptt_reviewed": False, + "credit_limit": old.get("credit_limit") or sync_credit_limit(command_phase), } mutate_json(path, mark) @@ -192,7 +206,7 @@ def clear_pending_sync(eng_dir: str | Path) -> None: def clear(data: dict[str, Any]) -> None: data.pop("pending", None) - data["credit"] = DEFAULT_SYNC_CREDIT + data.pop("credit", None) mutate_json(path, clear) diff --git a/scripts/smoke-test.sh b/scripts/smoke-test.sh index 7af5121..928fb5a 100644 --- a/scripts/smoke-test.sh +++ b/scripts/smoke-test.sh @@ -36,6 +36,14 @@ fail() { echo " [✗] $1"; ((FAIL+=1)); FAILURES+=" - $1"$'\n'; } header(){ echo ""; echo "━━━ $1 ━━━"; } summary(){ echo ""; echo "────────────────────────────────────────"; echo " PASS: $PASS FAIL: $FAIL"; echo "────────────────────────────────────────"; } +seed_history_fixture(){ + python3 - "$1" "$2" "$3" <<'PY' +import sys +from plugins.violin_guard import history +history.append_history(sys.argv[1], sys.argv[2], sys.argv[3], 0) +PY +} + # ============================================================================= # 1. YAML Validity # ============================================================================= @@ -285,34 +293,18 @@ else fail "record-ptt did not reject bad PT id (exit=$bad_id_exit)" fi -# (e) record-history appends a line and exits 0 -set +e -hist_out=$(python3 scripts/violin_guard.py record-history --eng-dir "$SMOKE_GUARD" --command "nmap -sV 10.129.245.218" --exit-code 0 --phase RECON 2>&1) -hist_exit=$? -set -e -if [ "$hist_exit" -eq 0 ]; then - pass "record-history appends entry and exits 0" -else - fail "record-history failed (exit=$hist_exit): $hist_out" -fi - -# (f) history.md has the new entry on disk -if grep -q 'RECON' "$SMOKE_GUARD/state/history.md" 2>/dev/null; then - pass "history.md contains the RECON entry" -else - fail "history.md did not receive the recorded entry" -fi - -# (g) record-history with empty command exits 1 -set +e -empty_hist=$(python3 scripts/violin_guard.py record-history --eng-dir "$SMOKE_GUARD" --command "" --exit-code 0 2>&1) -empty_hist_exit=$? -set -e -if [ "$empty_hist_exit" -eq 1 ] && echo "$empty_hist" | grep -qi "required"; then - pass "record-history rejects empty --command" -else - fail "record-history did not reject empty command (exit=$empty_hist_exit)" -fi +# (e) Removed administrative/model-facing commands are absent. +for removed in review-and-release finding sync-done record-history message-tick skill-status check-skill-loaded; do + set +e + removed_out=$(python3 scripts/violin_guard.py "$removed" --help 2>&1) + removed_exit=$? + set -e + if [ "$removed_exit" -ne 0 ] && echo "$removed_out" | grep -qi "invalid choice"; then + pass "removed CLI command is absent: $removed" + else + fail "removed CLI command is still accepted: $removed" + fi +done # (h) Stale-PTT detection: check-bootstrap warns when all rows are pristine # Use a SEPARATE fresh engagement (test (a) already updated PT-001 in $SMOKE_GUARD) @@ -333,18 +325,6 @@ else fail "Stale-PTT detection unexpected (exit=$stale_exit): $stale_out" fi -# (i) record-history without a pre-existing history.md does not create one and exits 1 -rm -f "$SMOKE_GUARD/state/history.md" -set +e -nohist_out=$(python3 scripts/violin_guard.py record-history --eng-dir "$SMOKE_GUARD" --command "test" --exit-code 0 2>&1) -nohist_exit=$? -set -e -if [ "$nohist_exit" -eq 1 ] && echo "$nohist_out" | grep -qi "not found"; then - pass "record-history errors when history.md is missing" -else - fail "record-history did not error on missing history.md (exit=$nohist_exit)" -fi - # Cleanup rm -rf "$SMOKE_GUARD" @@ -375,7 +355,7 @@ cat > "$SMOKE_FRESH/hypotheses.md" <<'MD' - **Updated:** $(date '+%Y-%m-%d %H:%M') MD echo "# Command History — fresh" > "$SMOKE_FRESH/state/history.md" -python3 scripts/violin_guard.py record-history --eng-dir "$SMOKE_FRESH" --command "nmap 10.129.45.113" --exit-code 0 --phase RECON >/dev/null 2>&1 +seed_history_fixture "$SMOKE_FRESH" "nmap 10.129.45.113" RECON # PTT "Last updated" set to now sed -i "s||$(date '+%Y-%m-%d %H:%M')|" "$SMOKE_FRESH/state/ptt.md" # Mark one RECON row done so desync detection has a baseline @@ -416,9 +396,6 @@ else fail "Fresh engagement: target command unexpectedly blocked (exit=$fresh_ok_exit): $fresh_ok" fi -# Clear the doc-sync gate so the next test hits the skill-load gate, not the sync gate -python3 scripts/violin_guard.py sync-done --eng-dir "$SMOKE_FRESH" >/dev/null 2>&1 - # (b) Target-touching command WITHOUT --session-id/--skill-loaded-file is BLOCKED (Gap #1 fix) set +e fresh_noskill=$(python3 scripts/violin_guard.py check-command --scope "$SMOKE_FRESH/scope/scope.yaml" --eng-dir "$SMOKE_FRESH" --phase recon --command "nmap 10.129.45.113" 2>&1) @@ -456,7 +433,7 @@ cat > "$SMOKE_STALEPTT/hypotheses.md" <<'MD' - **Updated:** 2026-07-08 00:00 MD echo "# Command History" > "$SMOKE_STALEPTT/state/history.md" -python3 scripts/violin_guard.py record-history --eng-dir "$SMOKE_STALEPTT" --command "curl 10.129.45.113" --exit-code 0 --phase EXPLOITATION >/dev/null 2>&1 +seed_history_fixture "$SMOKE_STALEPTT" "curl 10.129.45.113" EXPLOITATION python3 scripts/violin_guard.py record-ptt --eng-dir "$SMOKE_STALEPTT" --id PT-040 --status "[~]" --note "exploiting" >/dev/null 2>&1 touch "$SMOKE_STALEPTT/state/.skill-loaded-stale" cat > "$SMOKE_STALEPTT/scope/scope.yaml" <<'YAML' @@ -511,7 +488,7 @@ cat > "$SMOKE_STALEHYP/hypotheses.md" <<'MD' - **Updated:** 2026-07-08 00:00 MD echo "# Command History" > "$SMOKE_STALEHYP/state/history.md" -python3 scripts/violin_guard.py record-history --eng-dir "$SMOKE_STALEHYP" --command "curl 10.129.45.113" --exit-code 0 --phase EXPLOITATION >/dev/null 2>&1 +seed_history_fixture "$SMOKE_STALEHYP" "curl 10.129.45.113" EXPLOITATION python3 scripts/violin_guard.py record-ptt --eng-dir "$SMOKE_STALEHYP" --id PT-040 --status "[~]" --note "exploiting" >/dev/null 2>&1 touch "$SMOKE_STALEHYP/state/.skill-loaded-sh" cat > "$SMOKE_STALEHYP/scope/scope.yaml" <<'YAML' @@ -547,9 +524,9 @@ fi rm -rf "$SMOKE_FRESH" # ============================================================================= -# 3.8 Plugin Gate Lifecycle — violin_exec (check-command + doc-sync) & violin_sync_done +# 3.8 Canonical Batch Review Lifecycle # ============================================================================= -header "3.8 Plugin Gate Lifecycle (violin_exec + violin_sync_done)" +header "3.8 Canonical Batch Review Lifecycle (violin_review_batch)" GATES_DIR="engagements/_smoke-gates-$$" mkdir -p "$GATES_DIR"/{scope,state,evidence} @@ -594,81 +571,53 @@ authorisation: confirmed_at: '2026-07-08T00:00:00Z' YAML -# Seed a valid PTT row (moved past [ ]) — do NOT pre-record the history command, -# because the first exec should be a fresh command (no duplicate warning). -# The record-history happens AFTER the first exec in the real workflow. -python3 scripts/violin_guard.py record-ptt --eng-dir "$GATES_DIR" --id PT-001 --status "[x]" --note "bootstrap" >/dev/null 2>&1 -# Pre-seed a DIFFERENT command in history so the history guard doesn't fire "no recorded commands" REVIEW -python3 scripts/violin_guard.py record-history --eng-dir "$GATES_DIR" --command "curl http://10.129.45.113" --exit-code 0 --phase RECON >/dev/null 2>&1 -# Note: first exec will be a different command (nmap -sV) so no duplicate warning - -# Drive the plugin handlers directly (the layer that wraps the guard scripts). -# This asserts the full lifecycle: fresh exec -> approved -> pending sync locks -# the next exec -> sync_done clears -> next exec approved. +# Seed the active PTT row whose identity review-batch must preserve. +python3 scripts/violin_guard.py record-ptt --eng-dir "$GATES_DIR" --id PT-010 --status "[~]" --note "active recon" >/dev/null 2>&1 +# Drive the canonical service handler directly. python3 - "$GATES_DIR" <<'PY' -import sys, json, os -from datetime import datetime -eng_dir = sys.argv[1] -repo = os.getcwd() -import importlib.util -pkg_spec = importlib.util.spec_from_file_location("vgpkg", os.path.join(repo, "plugins/violin_guard/__init__.py")) -pkg = importlib.util.module_from_spec(pkg_spec) -pkg.__path__ = [os.path.join(repo, "plugins/violin_guard")] -pkg.__package__ = "vgpkg" -sys.modules["vgpkg"] = pkg -def load_sub(name, path): - spec = importlib.util.spec_from_file_location(f"vgpkg.{name}", path) - m = importlib.util.module_from_spec(spec); m.__package__ = "vgpkg" - sys.modules[f"vgpkg.{name}"] = m; spec.loader.exec_module(m); return m -tools = load_sub("tools", os.path.join(repo, "plugins/violin_guard/tools.py")) +import json +import sys +from datetime import UTC, datetime +from pathlib import Path -def st(handler, **kw): - return json.loads(handler(kw))["status"] +from plugins.violin_guard import history, service, state -now = datetime.now().strftime("%Y-%m-%d %H:%M") -base = dict(eng_dir=eng_dir, scope=f"{eng_dir}/scope/scope.yaml", - phase="recon", command="nmap -sV 10.129.45.113", # Different from any pre-recorded - skill_loaded_file=f"{eng_dir}/state/.skill-loaded-gate") +engagement = Path(sys.argv[1]) +command = "nmap -sV 10.129.45.113" +receipt = engagement / "evidence" / "executions" / "review.json" +receipt.parent.mkdir(parents=True, exist_ok=True) +state.atomic_json(receipt, { + "command": command, + "phase": "RECON", + "completed_at": datetime.now(UTC).isoformat().replace("+00:00", "Z"), + "exit_code": 0, + "evidence_paths": {"manifest": receipt.relative_to(engagement).as_posix()}, +}) +history.append_history( + engagement, command, "RECON", 0, receipt.relative_to(engagement).as_posix() +) +state.mark_pending_sync(engagement, command, "RECON", "PT-010") -# Step 1: fresh engagement -> exec must be APPROVED (no pending sync) -# Uses a DISTINCT command from any pre-recorded history -s1 = st(tools.handle_exec, **base) -assert s1 == "approved", f"step1 expected approved, got {s1}" -print(" ok: fresh violin_exec -> approved") - -# Step 2: next exec without violin_sync_done -> SYNC_REQUIRED (doc-sync gate) -s2 = st(tools.handle_exec, **base) -assert s2 == "sync_required", f"step2 expected sync_required, got {s2}" -print(" ok: second violin_exec before sync -> sync_required") - -# Step 3: LLM updates artifacts, then violin_sync_done -> OK (freshness verified) -with open(f"{eng_dir}/state/history.md", "a") as f: - f.write(f"\n- nmap -sV 10.129.45.113 (exit 0) [{now}]\n") -pt = f"{eng_dir}/state/ptt.md" -t = open(pt).read() -open(pt, "w").write(t.replace("Last updated:", f"Last updated: {now}") if "Last updated:" in t else t) -s3 = st(tools.handle_sync_done, eng_dir=eng_dir) -assert s3 == "ok", f"step3 expected ok, got {s3}" -print(" ok: violin_sync_done after artifact update -> ok") - -# Step 4: exec after sync -> APPROVED again (use a DIFFERENT command from step 1/2) -base2 = dict(base, command="gobuster dir -u http://10.129.45.113 -w wordlist.txt") -s4 = st(tools.handle_exec, **base2) -assert s4 == "approved", f"step4 expected approved, got {s4}" -print(" ok: violin_exec after sync -> approved") - -# Step 5: hypothesis recording through the plugin core service. -s5 = st(tools.handle_record_hypothesis, eng_dir=eng_dir, service="SMB", - port="445", title="anon access", status="researching") -assert s5 == "ok", f"step5 expected ok, got {s5}" -print(" ok: violin_record_hypothesis -> ok") +reviewed = json.loads(service.handle_review_batch({ + "eng_dir": str(engagement), + "id": "PT-010", + "status": "[~]", + "note": "Reviewed the completed recon receipt", +})) +assert reviewed["status"] == "ok", reviewed +assert reviewed["released"] is True +assert not state.has_pending_sync(engagement) +assert "[reviewed-batch:" in (engagement / "state" / "ptt.md").read_text() +for removed in ("handle_sync_done", "handle_review_and_release", "handle_finding"): + assert not hasattr(service, removed), removed +print(" ok: violin_review_batch reviewed PTT and released the batch") print("GATES_OK") PY gates_exit=$? if [ "$gates_exit" -eq 0 ]; then - pass "3.8 Plugin gate lifecycle: exec->sync_required->sync_done->exec + hypothesis routing all correct" + pass "3.8 Canonical review lifecycle: receipt->PTT review->batch release" else - fail "3.8 Plugin gate lifecycle failed (see python output above)" + fail "3.8 Canonical review lifecycle failed (see python output above)" fi rm -rf "$GATES_DIR" diff --git a/scripts/violin_guard.py b/scripts/violin_guard.py index e033adc..e97d41b 100644 --- a/scripts/violin_guard.py +++ b/scripts/violin_guard.py @@ -56,42 +56,52 @@ def cmd_validate_scope(args: argparse.Namespace) -> int: return code -def cmd_check_skill_loaded(args: argparse.Namespace) -> int: - result = command.check_skill_load( - state.resolve_eng_dir(args.eng_dir), args.session_id, mandatory=True - ) - return _print_result(result) - - -def cmd_record_history(args: argparse.Namespace) -> int: - from plugins.violin_guard import history - - history.append_history(args.eng_dir, args.command, args.phase, args.exit_code, args.evidence) - print("OK: history recorded") - return 0 - - def cmd_record_ptt(args: argparse.Namespace) -> int: - from plugins.violin_guard import ptt - - ptt.update_task( - state.resolve_eng_dir(args.eng_dir) / "state" / "ptt.md", - args.id, - args.status, - args.note or "", - ) - print(f"OK: PTT {args.id} updated") - return 0 - - -def cmd_sync_done(args: argparse.Namespace) -> int: from plugins.violin_guard import service - out = json.loads(service.handle_sync_done(vars(args))) + out = json.loads( + service.handle_record_ptt( + { + "eng_dir": args.eng_dir, + "id": args.id, + "status": args.status, + "note": args.note or "", + } + ) + ) print(out) return 0 if out["status"] == "ok" else 1 +def cmd_review_batch(args: argparse.Namespace) -> int: + from plugins.violin_guard import service + + finding = None + finding_values = { + "finding_id": args.finding_id, + "title": args.finding_title, + "severity": args.finding_severity, + "description": args.finding_description, + "impact": args.finding_impact, + "remediation": args.finding_remediation, + } + if any(str(value or "").strip() for value in finding_values.values()): + finding = finding_values + out = json.loads( + service.handle_review_batch( + { + "eng_dir": args.eng_dir, + "id": args.id, + "status": args.status, + "note": args.note, + "finding": finding, + } + ) + ) + print(json.dumps(out, indent=2)) + return 0 if out["status"] == "ok" else 1 + + def cmd_rebind_pending_batch(args: argparse.Namespace) -> int: from plugins.violin_guard import service @@ -117,29 +127,16 @@ def cmd_heartbeat_done(args: argparse.Namespace) -> int: return 0 -def cmd_message_tick(args: argparse.Namespace) -> int: - """Handle message tick - increment counter and check heartbeat gate.""" +def cmd_status(args: argparse.Namespace) -> int: + from plugins.violin_guard import service - eng_dir = args.eng_dir - count = state.tick_message(eng_dir) - - # Check if heartbeat is already pending (from previous tick) - if state.has_heartbeat_pending(eng_dir): - reason = state.get_heartbeat_reason(eng_dir) - print(f"BLOCK: heartbeat pending: {reason}") - return 1 # BLOCK - - # Check if heartbeat should be triggered now (every MESSAGE_INTERVAL messages) - if count % state.MESSAGE_INTERVAL == 0: - state.set_heartbeat_pending( - eng_dir, - f"Reached {count} LLM messages. Review engagement files for drift.", - ) - print("BLOCK: heartbeat triggered") - return 1 # BLOCK - - print("OK: message tick") - return 0 + out = json.loads(service.handle_status({"eng_dir": args.eng_dir})) + if args.section == "skill": + skill = out.get("skill", {}) + print(json.dumps(skill, indent=2)) + return 0 if skill.get("loaded") else 1 + print(json.dumps(out, indent=2)) + return 0 if out["status"] == "ok" else 1 def cmd_eng_root(args: argparse.Namespace) -> int: @@ -265,21 +262,6 @@ def main() -> int: ) p.set_defaults(func=cmd_init_engagement) - # check-skill-loaded - p = sub.add_parser("check-skill-loaded", help="Mark skill as loaded for session") - p.add_argument("--eng-dir", required=True) - p.add_argument("--session-id", required=True) - p.set_defaults(func=cmd_check_skill_loaded) - - # record-history - p = sub.add_parser("record-history", help="Append command to history.md") - p.add_argument("--eng-dir", required=True) - p.add_argument("--command", required=True) - p.add_argument("--exit-code", type=int, required=True) - p.add_argument("--phase", required=True) - p.add_argument("--evidence", default="") - p.set_defaults(func=cmd_record_history) - # record-ptt p = sub.add_parser("record-ptt", help="Update PTT task status") p.add_argument("--eng-dir", required=True) @@ -288,10 +270,22 @@ def main() -> int: p.add_argument("--note", default="") p.set_defaults(func=cmd_record_ptt) - # sync-done - p = sub.add_parser("sync-done", help="Clear pending sync lock") + p = sub.add_parser( + "review-batch", help="Review a completed batch, optionally record a finding, and unlock" + ) p.add_argument("--eng-dir", required=True) - p.set_defaults(func=cmd_sync_done) + p.add_argument("--id", required=True) + p.add_argument("--status", required=True, choices=["[~]", "[x]", "[!]", "[-]"]) + p.add_argument("--note", required=True) + p.add_argument("--finding-id", default="") + p.add_argument("--finding-title", default="") + p.add_argument( + "--finding-severity", default="", choices=["", "Critical", "High", "Medium", "Low", "Info"] + ) + p.add_argument("--finding-description", default="") + p.add_argument("--finding-impact", default="") + p.add_argument("--finding-remediation", default="") + p.set_defaults(func=cmd_review_batch) p = sub.add_parser("rebind-pending-batch", help="Explicitly rebind a completed pending batch") p.add_argument("--eng-dir", required=True) @@ -307,10 +301,10 @@ def main() -> int: p.add_argument("--eng-dir", required=True) p.set_defaults(func=cmd_heartbeat_done) - # message-tick - p = sub.add_parser("message-tick", help="Increment message counter") + p = sub.add_parser("status", help="Explain current phase, task, blockers, and next actions") p.add_argument("--eng-dir", required=True) - p.set_defaults(func=cmd_message_tick) + p.add_argument("--section", choices=["all", "skill"], default="all") + p.set_defaults(func=cmd_status) # eng-root p = sub.add_parser("eng-root", help="Print canonical engagement root") diff --git a/skills/pentest/SKILL.md b/skills/pentest/SKILL.md index c2da910..6fd6509 100644 --- a/skills/pentest/SKILL.md +++ b/skills/pentest/SKILL.md @@ -107,15 +107,16 @@ The phase workflow is mandatory for the entire session, including long, compress - `violin_exec` is the single-command authorize, execute, and evidence boundary for target interaction. It has no binary allowlist: use it for any installed non-interactive Kali/Parrot CLI tool. Installation, root, hardware, service, GUI, and interactive-TTY requirements remain runtime constraints, never reasons to bypass the guard. - `violin_exec_status` reads a tracked execution receipt; `violin_exec_cancel` cancels only its tracked process group. - `violin_exec_burst` is the batch gate for exploit/race iterations. Keep one EXPLOITATION PTT task active while adapting up to 20 pre-approved commands; it records every completed command and requires one explicit PTT review/update only when the bounded burst ends. +- `violin_status` is the cheap first diagnostic: it shows the active task/phase, each pending command's required phase, phase requirements, blockers, and exact next actions without mutating engagement state. - `violin_httpx`, `violin_nuclei`, and `violin_ffuf` build typed commands and delegate to `violin_exec`. Run nmap directly through `violin_exec` or `violin_exec_burst`. - `violin_search_exploit` searches the local ExploitDB index only; it never downloads or executes a candidate. - `violin_exec` / `violin_exec_burst` append exact command history themselves. Do not spend model calls recreating command history, and do not treat automatic history as proof that the PTT progressed. - `violin_record_ptt` changes task lifecycle state; `violin_record_hypothesis` records semantic hypothesis changes. -- `violin_sync_done` verifies the explicit post-batch PTT update and unlocks the next target batch. +- `violin_review_batch` is the only post-batch operation: it validates the completed batch, optionally creates one receipt-backed finding, applies the explicit PTT review/update once, and unlocks the batch last. - `violin_heartbeat_done` clears the periodic review lock after re-reading this skill and reviewing engagement files. 0. **Bootstrap gate** — at session start, after `/goal set`, or after context compression that loses track of state, verify the engagement is bootstrapped: `python $HOME/.hermes/profiles/violin/scripts/violin_guard.py check-bootstrap --eng-dir "$ENG_DIR"`. Exit `0` = proceed. Exit `1` = **STOP and run `playbooks/scoping.md §0`** (creates `$ENG_DIR/`, `scope/scope.yaml`, `state/ptt.md`, `hypotheses.md`, `state/history.md`). Exit `2` = fix the warning, then proceed. This gate is non-negotiable: no `curl`, `nmap`, `browser_navigate`, or other target-touching tool call is allowed until exit 0. -0.1. **Skill-load gate** — after reading this skill, create `state/.skill-loaded-` containing `skill-loaded: `, then verify it with `check-skill-loaded --eng-dir "$ENG_DIR" --session-id ""`. Missing marker = **BLOCK**. If the marker exists but belongs to another session, create the canonical marker for the current session and re-run the check; do not silently rely on a stale marker. CTF bootstrap creates it when `--session-id` is supplied. +0.1. **Skill-load gate** — launch with `hermes chat --skills pentest` when possible; otherwise load this skill immediately. After reading it, create `state/.skill-loaded-` containing `skill-loaded: `, then verify it with `status --eng-dir "$ENG_DIR" --section skill`. Missing marker = **BLOCK**. If the marker exists but belongs to another session, create the canonical marker for the current session and re-run the check; do not silently rely on a stale marker. CTF bootstrap creates it when `--session-id` is supplied. 1. Check/update `todo` with a single active `phase-gate` item named for the current phase. 2. Confirm an approved `$ENG_DIR/scope/scope.yaml` exists before touching any target. If it does not, remain in SCOPING and ask via `clarify`. Verify with `python $HOME/.hermes/profiles/violin/scripts/violin_guard.py validate-scope --scope $ENG_DIR/scope/scope.yaml` (exit 0 required). For an authorized HTB/CTF lab, `init-engagement --ctf --host --session-id "$ENG_DIR"` creates a ready-to-test scope, active RECON PTT row, and skill marker. @@ -123,8 +124,8 @@ The phase workflow is mandatory for the entire session, including long, compress ```bash python $HOME/.hermes/profiles/violin/scripts/violin_guard.py record-ptt --eng-dir "$ENG_DIR" --id PT-XXX --status "[~]" --note "starting task" ``` - With no pending batch, this is the one permitted PTT start transition. The guard hard-blocks target execution when there is no unambiguous active task or the task sits under another phase heading. The executor never changes this row or its `*Last updated*` timestamp. After each bounded batch, review the results and explicitly call `violin_record_ptt` with `[~]`, `[x]`, `[!]`, or `[-]` plus a truthful result summary, then call `violin_sync_done`. -4. **Command history is executor-owned** — `violin_exec` appends every completed target command automatically. The standalone `record-history` CLI is administrative repair/import only, not part of the model workflow. Inspect recent history before a new batch to avoid repeats. + With no pending batch, this is the one permitted PTT start transition. The guard hard-blocks target execution when there is no unambiguous active task or the task sits under another phase heading. The executor never changes this row or its `*Last updated*` timestamp. After each bounded batch, review the results and call `violin_review_batch` with `[~]`, `[x]`, `[!]`, or `[-]` plus a truthful result summary. +4. **Command history is executor-owned** — `violin_exec` appends every completed target command automatically. There is no public history-writing command; inspect recent history before a new batch to avoid repeats. 5. Re-read this skill or the active playbook after context compression, `/resume`, or any uncertainty about the workflow. **Also read back evidence and hypotheses:** `read_file path="$ENG_DIR/hypotheses.md"` and `search_files path="$ENG_DIR/evidence" pattern="" target="files"` to restore investigation state. 6. Validate target-touching commands before execution: - Run `python $HOME/.hermes/profiles/violin/scripts/violin_guard.py validate-scope --scope $ENG_DIR/scope/scope.yaml` after scope creation or updates. @@ -141,7 +142,7 @@ The phase workflow is mandatory for the entire session, including long, compress - Exit code `0` means allowed, `1` means blocked, and `2` means explicit review or approval is required. - **Destructive commands are hard-blocked.** `rm -rf`, `mkfs`, `dd of=/dev/...`, fork bombs, `curl|sh`, and similar patterns are denied at `exit 1` regardless of yolo mode. - **Out-of-scope targets are blocked.** Supply the explicit primary target to every target-touching guard call. Additional IPv4/CIDR literals and unambiguous URL/UNC/network forms are still inspected against `scope.yaml`; unknown network hostnames surface as `exit 2` rather than a silent pass. Operator-approved `assessment_hosts.callback_hosts` may appear only as secondary listener/callback endpoints; they never qualify as assessment targets or hypothesis targets. - - **Continuity guards:** `check-command` enforces the active PTT task, history continuity, hypotheses where required, pending batch review, scope, and skill-load state. Resolve the reported artifact or use the corresponding plugin tool; never bypass the gate. If a completed pending batch is bound to the wrong PTT row, use `violin_rebind_pending_batch` with the exact batch ID, old/new task IDs, an operator note, and explicit confirmation. Rebinding is audited and never replaces the required PTT review or `violin_sync_done`. + - **Continuity guards:** `check-command` enforces the active PTT task, history continuity, hypotheses where required, pending batch review, scope, and skill-load state. Resolve the reported artifact or use the corresponding plugin tool; never bypass the gate. If a completed pending batch is bound to the wrong PTT row, use `violin_rebind_pending_batch` with the exact batch ID, old/new task IDs, an operator note, and explicit confirmation. Rebinding is audited and never replaces `violin_review_batch`. - Phases: SCOPING, RECON, VULN_RESEARCH, EXPLOITATION, POST_EXPLOITATION, PRIVESC, FLAGS, REPORTING, RETROSPECTIVE. CTF engagements use PRIVESC/FLAGS for privilege-escalation and flag-capture tasks; task IDs use the `PT-CTF-NNN` form. 7. Load/read the phase playbook before acting: - SCOPING → `playbooks/scoping.md` @@ -177,7 +178,7 @@ The classic failure mode is re-running the same command without recording what i - **Context boundary.** Prefix controller output with `[VICTIM]`; prefix assessment-host commands and notes with `[ATTACKER]`. `/proc`, `/run`, UNIX sockets, and local service state belong to the machine on which the command runs. Host-local preparation (for example, starting an approved HTTP server or hashing a local artifact) may use the terminal directly; commands sent to a target still use `violin_exec` or one pre-approved `violin_exec_burst`. - **Mandatory research-loop (VULN RESEARCH / EXPLOITATION):** for each detected version/service, record the actual NVD/ExploitDB/GitHub research and update the hypothesis when its semantic state changes (Candidate/Likely/Validated/Rejected). Do not fabricate a hypothesis update merely because another payload ran. - **Research-attempt gate:** before any EXPLOITATION, POST_EXPLOITATION, PRIVESC, or FLAGS target command, the matching hypothesis must contain non-empty `CVE Research` and `Exploit Research` fields. Each field records the online query, source, and outcome. A truthful `no results`, `not applicable`, or `source unavailable` outcome satisfies the attempt requirement; an omitted field blocks execution. Local SearchSploit alone does not satisfy the online attempt. -- **Split continuity contract.** Every approved target command is automatically mirrored to `state/history.md`; PTT progress is never automatic. When the bounded window ends and `violin_exec` returns `sync_required`, stop, review the batch evidence, explicitly update the active PTT row, call `violin_sync_done`, then continue. +- **Split continuity contract.** Every approved target command is automatically mirrored to `state/history.md`; PTT progress is never automatic. When the bounded window ends and `violin_exec` returns `sync_required`, stop, review the batch evidence, call `violin_review_batch`, then continue. **Drift signal:** If the agent starts improvising tasks that are not tied to a phase, playbook, evidence path, and scope item, it must stop, reload this skill, and resume from the correct phase gate. Specifically: if the agent is about to run a target-touching command but cannot point to a `[ ]` PTT entry that justifies it and an existing hypothesis, stop. diff --git a/tests/guard/guards/test_release.py b/tests/guard/guards/test_release.py index b32b91d..ec80d1d 100644 --- a/tests/guard/guards/test_release.py +++ b/tests/guard/guards/test_release.py @@ -12,17 +12,17 @@ from plugins.violin_guard.release import _pytest_basetemp ROOT = Path(__file__).resolve().parents[3] -def test_profile_does_not_cap_agent_iterations() -> None: +def test_profile_uses_an_engagement_sized_iteration_budget() -> None: config = yaml.safe_load((ROOT / "config.yaml").read_text(encoding="utf-8")) - assert "max_turns" not in config["agent"] + assert config["agent"]["max_turns"] >= 350 -def test_heartbeat_uses_extended_shared_cadence() -> None: +def test_heartbeat_is_command_based_and_phase_aware() -> None: assert state.COMMAND_INTERVAL == 50 - assert state.MESSAGE_INTERVAL == 60 description = schemas.HEARTBEAT_DONE_SCHEMA["description"] - assert "50 target commands or 60 message ticks" in description + assert "50 executed target commands" in description + assert "message ticks" not in description def test_pytest_basetemp_creates_missing_engagement_root(tmp_path: Path) -> None: diff --git a/tests/guard/integration/test_plugin_guard.py b/tests/guard/integration/test_plugin_guard.py index 05152aa..e9bd136 100644 --- a/tests/guard/integration/test_plugin_guard.py +++ b/tests/guard/integration/test_plugin_guard.py @@ -82,7 +82,7 @@ def _fake_target_executor(monkeypatch): def fake_execute(command, *, eng_dir, phase, **kwargs): engagement = Path(eng_dir) history.append_history(engagement, command, phase, 0, "evidence/executions/test.json") - remaining = state.spend_sync_credit(str(engagement)) + remaining = state.spend_sync_credit(str(engagement), phase) # Mirror real execution: tick command counter, mark pending sync, set heartbeat if interval reached from plugins.violin_guard.phases import normalize_phase, suppresses_heartbeat @@ -157,11 +157,13 @@ def test_meta_loaded(): for name in ( "handle_exec", "handle_check_command", - "handle_sync_done", + "handle_review_batch", "handle_record_ptt", "handle_record_hypothesis", ): assert hasattr(TOOLS, name), f"plugin must expose {name}" + for removed in ("handle_sync_done", "handle_review_and_release", "handle_finding"): + assert not hasattr(TOOLS, removed), f"plugin must not expose removed handler {removed}" def test_recon_does_not_require_hypothesis(tmp_path): @@ -576,7 +578,7 @@ def test_exec_auto_records_history_but_requires_explicit_ptt_review(monkeypatch, ) assert ptt_path.read_text(encoding="utf-8") == ptt_before - window = state.DEFAULT_SYNC_CREDIT + window = state.sync_credit_limit("recon") for i in range(2, window + 1): command_val = f"nmap -sV 10.10.10.10 -p {i}" out = json.loads(TOOLS.handle_exec({**args, "command": command_val})) @@ -599,7 +601,7 @@ def test_exec_auto_records_history_but_requires_explicit_ptt_review(monkeypatch, assert history_text.count("exit_code=0 | command=nmap") == window reviewed = json.loads( - TOOLS.handle_record_ptt( + TOOLS.handle_review_batch( { "eng_dir": str(eng), "id": "PT-010", @@ -610,8 +612,6 @@ def test_exec_auto_records_history_but_requires_explicit_ptt_review(monkeypatch, ) assert reviewed["status"] == "ok", reviewed assert f"[reviewed-batch:{batch_id}]" in ptt_path.read_text(encoding="utf-8") - synced = json.loads(TOOLS.handle_sync_done({"eng_dir": str(eng)})) - assert synced["status"] == "ok", synced resumed = json.loads(TOOLS.handle_exec({**args, "command": "nmap -sV 10.10.10.10 -p 99"})) assert resumed["status"] in ("ok", "approved", "review"), resumed @@ -655,7 +655,7 @@ def test_exploitation_gets_bounded_window_then_requires_ptt_review(monkeypatch, } ptt_before = ptt_path.read_text(encoding="utf-8") - total = state.DEFAULT_SYNC_CREDIT + total = state.sync_credit_limit("exploitation") for i in range(total): command_val = f"curl http://10.10.10.10/probe?variant={i}" out = json.loads(TOOLS.handle_exec({**args, "command": command_val})) @@ -684,17 +684,13 @@ def test_heartbeat_gate_every_n_commands(monkeypatch, tmp_path): for _ in range(state.COMMAND_INTERVAL - 1): state.tick_command(str(eng)) - threshold = json.loads( - TOOLS.handle_exec({**args, "command": "nmap -sV 10.10.10.10 -p 20"}) - ) + threshold = json.loads(TOOLS.handle_exec({**args, "command": "nmap -sV 10.10.10.10 -p 20"})) assert threshold["status"] == "ok", threshold assert threshold["executed"] is True assert state.read_counts(str(eng))["commands"] == state.COMMAND_INTERVAL assert state.has_heartbeat_pending(str(eng)) - blocked = json.loads( - TOOLS.handle_exec({**args, "command": "nmap -sV 10.10.10.10 -p 21"}) - ) + blocked = json.loads(TOOLS.handle_exec({**args, "command": "nmap -sV 10.10.10.10 -p 21"})) assert blocked["status"] == "denied", blocked assert blocked["executed"] is False assert state.read_counts(str(eng))["commands"] == state.COMMAND_INTERVAL @@ -702,23 +698,22 @@ def test_heartbeat_gate_every_n_commands(monkeypatch, tmp_path): cleared = json.loads(TOOLS.handle_heartbeat_done({"eng_dir": str(eng)})) assert cleared["status"] == "ok", cleared - resumed = json.loads( - TOOLS.handle_exec({**args, "command": "nmap -sV 10.10.10.10 -p 21"}) - ) + resumed = json.loads(TOOLS.handle_exec({**args, "command": "nmap -sV 10.10.10.10 -p 21"})) assert resumed["status"] == "ok", resumed assert resumed["executed"] is True assert state.read_counts(str(eng))["commands"] == state.COMMAND_INTERVAL + 1 -def test_message_tick_triggers_heartbeat(monkeypatch, tmp_path): - """Every MESSAGE_INTERVAL calls, heartbeat is required.""" +def test_message_ticks_are_diagnostic_and_do_not_trigger_heartbeat(monkeypatch, tmp_path): + """LLM message volume must not create a stale guard lock.""" skill_file = tmp_path / ".skill-loaded-ts" eng = _init_e2e(tmp_path, skill_file) # Build a session object via pre_llm_call (which increments message tick) from plugins.violin_guard import _pre_llm_call_hook - for _ in range(state.MESSAGE_INTERVAL - 1): + for _ in range(100): _pre_llm_call_hook(session_id="ts", eng_dir=str(eng), phase="recon") - assert _pre_llm_call_hook(session_id="ts", eng_dir=str(eng), phase="recon") is None + assert not state.has_heartbeat_pending(str(eng)) + assert state.read_counts(str(eng))["messages"] == 100 diff --git a/tests/guard/state/test_batch_integrity.py b/tests/guard/state/test_batch_integrity.py index 3ee8e62..e070c18 100644 --- a/tests/guard/state/test_batch_integrity.py +++ b/tests/guard/state/test_batch_integrity.py @@ -19,13 +19,11 @@ def _engagement(tmp_path: Path) -> Path: return eng -def test_review_reconciles_a_direct_ptt_active_task_edit(tmp_path: Path) -> None: +def test_record_ptt_refuses_to_reconcile_a_pending_batch(tmp_path: Path) -> None: eng = _engagement(tmp_path) command = "nmap -p 80 10.10.10.10" history.append_history(eng, command, "RECON", 0, "evidence/executions/test.json") state.mark_pending_sync(eng, command, "RECON", "PT-010") - batch_id = state.get_pending_sync(eng)["batch_id"] - ptt_path = eng / "state" / "ptt.md" ptt_path.write_text( ptt_path.read_text(encoding="utf-8") @@ -35,11 +33,12 @@ def test_review_reconciles_a_direct_ptt_active_task_edit(tmp_path: Path) -> None ) result = json.loads( service.handle_record_ptt( - {"eng_dir": str(eng), "id": "PT-011", "status": "[~]", "note": f"review {batch_id}"} + {"eng_dir": str(eng), "id": "PT-011", "status": "[~]", "note": "review"} ) ) - assert result["status"] == "ok" - assert state.get_pending_sync(eng)["ptt_task_id"] == "PT-011" + assert result["status"] == "error" + assert "violin_review_batch" in result["error"] + assert state.get_pending_sync(eng)["ptt_task_id"] == "PT-010" def test_appending_work_invalidates_an_earlier_review(tmp_path: Path) -> None: @@ -86,8 +85,6 @@ def test_confirmed_rebind_is_audited_but_does_not_review_or_unlock(tmp_path: Pat assert pending["ptt_task_id"] == "PT-011" assert pending["ptt_reviewed"] is False assert state.has_pending_sync(eng) - sync = json.loads(service.handle_sync_done({"eng_dir": str(eng)})) - assert sync["status"] == "sync_required" sync_data = json.loads((eng / "state" / "sync.json").read_text(encoding="utf-8")) assert sync_data["rebind_audit"][-1]["old_task_id"] == "PT-010" assert sync_data["rebind_audit"][-1]["new_task_id"] == "PT-011" diff --git a/tests/guard/state/test_burst_and_target.py b/tests/guard/state/test_burst_and_target.py index 6ff1570..39d6c6f 100644 --- a/tests/guard/state/test_burst_and_target.py +++ b/tests/guard/state/test_burst_and_target.py @@ -319,3 +319,53 @@ def test_plugin_exposes_new_tools(): tool_names = set(manifest["provides_tools"]) assert "violin_exec_burst" in tool_names assert "violin_target" in tool_names + assert "violin_review_batch" in tool_names + assert ( + not { + "violin_sync_done", + "violin_review_and_release", + "violin_finding", + } + & tool_names + ) + + +def test_status_skill_section_reports_load_state_and_exit_code(eng): + state.record_session_id(eng, "ts") + loaded = _run("status", "--eng-dir", str(eng), "--section", "skill") + loaded_data = json.loads(loaded.stdout) + assert loaded.returncode == 0 + assert loaded_data["loaded"] is True + + marker = Path(loaded_data["marker"]) + marker.unlink() + missing = _run("status", "--eng-dir", str(eng), "--section", "skill") + missing_data = json.loads(missing.stdout) + assert missing.returncode == 1 + assert missing_data["loaded"] is False + + +@pytest.mark.parametrize( + "removed", + [ + "review-and-release", + "finding", + "sync-done", + "record-history", + "message-tick", + "skill-status", + "check-skill-loaded", + ], +) +def test_removed_cli_commands_are_absent(removed): + result = _run(removed, "--help") + assert result.returncode != 0 + assert "invalid choice" in result.stderr + + +def test_review_batch_cli_exposes_lifecycle_and_optional_finding_fields(): + result = _run("review-batch", "--help") + assert result.returncode == 0 + assert "--status" in result.stdout + assert "--note" in result.stdout + assert "--finding-title" in result.stdout diff --git a/tests/guard/state/test_collaboration_ux.py b/tests/guard/state/test_collaboration_ux.py new file mode 100644 index 0000000..da499d2 --- /dev/null +++ b/tests/guard/state/test_collaboration_ux.py @@ -0,0 +1,239 @@ +"""Regression coverage for the guard's model-visible collaboration surface.""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime +from pathlib import Path + +import pytest + +from plugins.violin_guard import bootstrap, history, ptt, service, state + +ROOT = Path(__file__).resolve().parents[3] + + +def _engagement(tmp_path: Path) -> Path: + eng = tmp_path / "engagement" + assert bootstrap.init_engagement(eng, host="10.10.10.10") == 0 + scope = eng / "scope" / "scope.yaml" + scope.write_text( + scope.read_text(encoding="utf-8").replace("confirmed: false", "confirmed: true"), + 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 | [~] |"), + encoding="utf-8", + ) + state.record_session_id(eng, "test-session") + (eng / "state" / ".skill-loaded-test-session").write_text( + "skill-loaded: pentest\n", encoding="utf-8" + ) + return eng + + +def _pending_batch(eng: Path) -> None: + command = "nmap -sV 10.10.10.10" + (eng / "evidence" / "executions").mkdir(parents=True, exist_ok=True) + manifest = eng / "evidence" / "executions" / "batch-command.json" + stdout = eng / "evidence" / "executions" / "batch-command.stdout.txt" + stdout.write_text("80/tcp open http\n", encoding="utf-8") + state.atomic_json( + manifest, + { + "command": command, + "phase": "RECON", + "completed_at": datetime.now(UTC).isoformat().replace("+00:00", "Z"), + "exit_code": 0, + "evidence_paths": { + "manifest": manifest.relative_to(eng).as_posix(), + "stdout": stdout.relative_to(eng).as_posix(), + }, + }, + ) + history.append_history(eng, command, "RECON", 0, manifest.relative_to(eng).as_posix()) + state.mark_pending_sync(eng, command, "RECON", "PT-010") + + +def test_create_task_inserts_into_requested_phase_table(tmp_path: Path) -> None: + path = tmp_path / "ptt.md" + path.write_text( + (ROOT / "skills" / "pentest" / "templates" / "ptt.md").read_text(encoding="utf-8"), + encoding="utf-8", + ) + + created = ptt.create_task( + path, + "PT-099", + "Validate requested exploit", + "EXPLOITATION", + "evidence/exploitation/", + ) + + assert created.phase == "EXPLOITATION" + text = path.read_text(encoding="utf-8") + assert text.index("| PT-099 |") < text.index("## Phase: REPORTING") + row = next(line for line in text.splitlines() if "| PT-099 |" in line) + assert len(row.strip().strip("|").split("|")) == 7 + + +def test_status_explains_current_phase_pending_commands_and_skill(tmp_path: Path) -> None: + eng = _engagement(tmp_path) + _pending_batch(eng) + + result = json.loads(service.handle_status({"eng_dir": str(eng)})) + + assert result["status"] == "ok" + assert result["current_task"] == "PT-010" + assert result["current_phase"] == "RECON" + assert result["pending_batch"]["commands"][0]["required_phase"] == "RECON" + assert result["phase_requirements"]["EXPLOITATION"]["sync_window"] == 20 + assert result["skill"]["loaded"] is True + + +@pytest.mark.parametrize("task_status", ["[~]", "[x]", "[!]", "[-]"]) +def test_review_batch_updates_ptt_and_clears_lock(tmp_path: Path, task_status: str) -> None: + eng = _engagement(tmp_path) + _pending_batch(eng) + + result = json.loads( + service.handle_review_batch( + { + "eng_dir": str(eng), + "id": "PT-010", + "status": task_status, + "note": "Reviewed service discovery evidence; HTTP is the next task input", + } + ) + ) + + assert result["status"] == "ok" + assert result["task_status"] == task_status + assert result["released"] is True + assert not state.has_pending_sync(eng) + assert "reviewed-batch:" in (eng / "state" / "ptt.md").read_text(encoding="utf-8") + + +def test_review_batch_creates_finding_from_current_batch_receipts(tmp_path: Path) -> None: + eng = _engagement(tmp_path) + _pending_batch(eng) + + result = json.loads( + service.handle_review_batch( + { + "eng_dir": str(eng), + "id": "PT-010", + "status": "[~]", + "note": "Reviewed the HTTP service receipt", + "finding": { + "title": "Exposed HTTP service", + "severity": "Info", + "description": "An HTTP listener is reachable on the approved target.", + "impact": "The service contributes to the externally reachable attack surface.", + "remediation": ( + "Confirm the listener is intended and restrict it when unnecessary." + ), + }, + } + ) + ) + + assert result["status"] == "ok" + finding = eng / result["finding"]["path"] + assert result["finding_path"] == result["finding"]["path"] + assert finding.is_file() + text = finding.read_text(encoding="utf-8") + assert "batch-command.stdout.txt" in text + assert "## Remediation" in text + + +@pytest.mark.parametrize( + ("mutation", "expected"), + [ + ("history", "exact history"), + ("task", "does not match batch task"), + ("phase", "not phase-compatible"), + ("finding", "must be non-empty"), + ], +) +def test_invalid_review_batch_leaves_sync_lock_active( + tmp_path: Path, mutation: str, expected: str +) -> None: + eng = _engagement(tmp_path) + _pending_batch(eng) + args = { + "eng_dir": str(eng), + "id": "PT-010", + "status": "[~]", + "note": "Review receipt", + } + if mutation == "history": + (eng / "state" / "history.md").write_text("# History\n", encoding="utf-8") + elif mutation == "task": + args["id"] = "PT-011" + elif mutation == "phase": + sync_path = eng / "state" / "sync.json" + sync_data = state.read_json(sync_path) + sync_data["pending"]["commands"][0]["phase"] = "EXPLOITATION" + state.atomic_json(sync_path, sync_data) + else: + args["finding"] = { + "title": "", + "severity": "Info", + "description": "Description", + "impact": "Impact", + "remediation": "Remediation", + } + + result = json.loads(service.handle_review_batch(args)) + + assert result["status"] == "blocked" + assert expected in result["error"] + assert result["next_action"] + assert state.has_pending_sync(eng) + + +def test_review_batch_retry_reuses_marker_and_finding_after_partial_failure( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + eng = _engagement(tmp_path) + _pending_batch(eng) + args = { + "eng_dir": str(eng), + "id": "PT-010", + "status": "[~]", + "note": "Reviewed HTTP receipt", + "finding": { + "title": "Exposed HTTP service", + "severity": "Info", + "description": "An HTTP listener is reachable.", + "impact": "The service increases the reachable attack surface.", + "remediation": "Restrict the listener when it is not required.", + }, + } + real_clear = state.clear_pending_sync + + def fail_clear(_eng_dir: str | Path) -> None: + raise OSError("simulated clear failure") + + monkeypatch.setattr(state, "clear_pending_sync", fail_clear) + first = json.loads(service.handle_review_batch(args)) + assert first["status"] == "blocked" + assert state.has_pending_sync(eng) + + monkeypatch.setattr(state, "clear_pending_sync", real_clear) + retry = json.loads(service.handle_review_batch(args)) + + assert retry["status"] == "ok" + assert retry["finding"]["reused"] is True + ptt_text = (eng / "state" / "ptt.md").read_text(encoding="utf-8") + assert ptt_text.count("[reviewed-batch:") == 1 + assert len(list((eng / "evidence" / "findings").glob("FIND-*.md"))) == 1 + assert not state.has_pending_sync(eng) + + +def test_sync_windows_are_phase_aware() -> None: + assert state.sync_credit_limit("RECON") == 10 + assert state.sync_credit_limit("EXPLOITATION") == 20 + assert state.sync_credit_limit("PRIVESC") == 20 diff --git a/tests/guard/state/test_state_concurrency.py b/tests/guard/state/test_state_concurrency.py index 107f26b..110f434 100644 --- a/tests/guard/state/test_state_concurrency.py +++ b/tests/guard/state/test_state_concurrency.py @@ -15,7 +15,7 @@ def test_concurrent_credit_spends_are_serialised(tmp_path): sync.write_text(json.dumps({"credit": 50}), encoding="utf-8") with concurrent.futures.ThreadPoolExecutor(max_workers=16) as pool: - results = list(pool.map(lambda _: state.spend_sync_credit(eng), range(25))) + results = list(pool.map(lambda _: state.spend_sync_credit(eng, "RECON"), range(25))) assert state.sync_credit_remaining(eng) == 25 assert sorted(results) == list(range(25, 50)) diff --git a/tests/guard/state/test_sync_credit_window.py b/tests/guard/state/test_sync_credit_window.py index fd96fc1..535e7c6 100644 --- a/tests/guard/state/test_sync_credit_window.py +++ b/tests/guard/state/test_sync_credit_window.py @@ -32,7 +32,9 @@ def test_network_clients_are_not_local_bookkeeping() -> None: assert state.is_local_bookkeeping_command("echo local-note") -def test_five_commands_run_without_yolo_then_sixth_blocks(monkeypatch, tmp_path: Path) -> None: +def test_phase_window_runs_without_yolo_then_next_command_blocks( + monkeypatch, tmp_path: Path +) -> None: """The bounded window is an allowance, not five REVIEW responses.""" eng = _engagement(tmp_path) @@ -58,9 +60,10 @@ def test_five_commands_run_without_yolo_then_sixth_blocks(monkeypatch, tmp_path: "session_id": "test", } - for port in range(1, state.DEFAULT_SYNC_CREDIT + 1): + limit = state.sync_credit_limit("recon") + for port in range(1, limit + 1): result = json.loads(service.handle_exec({**args, "command": f"nmap -p {port} 10.10.10.10"})) assert result["status"] == "ok", result - sixth = json.loads(service.handle_exec({**args, "command": "nmap -p 99 10.10.10.10"})) - assert sixth["status"] == "sync_required", sixth + blocked = json.loads(service.handle_exec({**args, "command": "nmap -p 99 10.10.10.10"})) + assert blocked["status"] == "sync_required", blocked diff --git a/tests/guard/test_terminal_policy.py b/tests/guard/test_terminal_policy.py index 9ff42a7..571fab0 100644 --- a/tests/guard/test_terminal_policy.py +++ b/tests/guard/test_terminal_policy.py @@ -132,6 +132,15 @@ def test_local_script_paths_are_not_treated_as_hosts() -> None: assert _pre_tool_call_hook(tool_name="terminal", args={"command": "sh deploy.sh"}) is None +def test_local_file_path_containing_an_ip_is_not_treated_as_a_socket() -> None: + assert ( + _pre_tool_call_hook( + tool_name="terminal", args={"command": "cat /tmp/file-with-10.10.14.233.txt"} + ) + is None + ) + + def test_non_terminal_tools_are_not_affected() -> None: result = _pre_tool_call_hook( tool_name="violin_exec",