feat: implement violin_guard plugin with secure execution adapters, terminal policies, and pentest skill frameworks

plugins/violin_guard/adapters.py

@@ -313,7 +313,7 @@ def search_exploit(args: dict) -> dict[str, Any]:
         return {
             "available": False,
             "tool": "searchsploit",
-            "message": "searchsploit is not installed or not on PATH",
+            "message": "searchsploit is not installed or not on PATH; install exploitdb via 'apt install exploitdb'",
             "candidates": [],
             "online_corroboration_required": True,
             "executed_candidates": False,

plugins/violin_guard/bootstrap.py

@@ -141,6 +141,9 @@ def _ctf_scope(host: str) -> dict:
                 "banner grabbing",
                 "version detection",
                 "vulnerability scanning",
+                "vulnerability research",
+                "cve-research",
+                "exploitdb",
                 "exploit validation (in-scope, non-destructive)",
                 "privilege escalation",
                 "flag capture (user.txt, root.txt)",

plugins/violin_guard/code_execution_audit.py

@@ -38,7 +38,9 @@ def parse_metadata(source: object) -> tuple[dict[str, str] | None, str | None]:
     if not isinstance(raw, dict) or set(raw) != _REQUIRED_FIELDS:
         return (
             None,
-            "execute_code metadata must contain exactly eng_dir, phase, target, and session_id",
+            "execute_code metadata must contain exactly eng_dir, phase, target, and session_id. "
+            'Header format (line 1 of code): # violin: {"eng_dir":"<path>","phase":"<phase>","target":"<target>","session_id":"<session_id>"} '
+            "(obtain session_id via violin_status)",
         )
     if not all(isinstance(raw[name], str) and raw[name].strip() for name in _REQUIRED_FIELDS):
         return None, "execute_code metadata values must be non-empty strings"

plugins/violin_guard/command.py

@@ -209,23 +209,44 @@ def _normalise_action(value: object) -> str:
     return " ".join(str(value).strip().lower().replace("_", " ").replace("/", " ").split())
 
 
+def _is_action_permitted(allowed_items: Any, phase_actions: frozenset[str]) -> bool:
+    for item in allowed_items:
+        raw_str = str(item)
+        norm_full = _normalise_action(raw_str)
+        if norm_full in phase_actions:
+            return True
+        without_parens = re.sub(r"\(.*?\)", "", raw_str)
+        norm_clean = _normalise_action(without_parens)
+        if norm_clean in phase_actions:
+            return True
+        for act in phase_actions:
+            if act in norm_full or act in norm_clean:
+                return True
+    return False
+
+
 def check_scope_authorization(scope: dict[str, Any] | None, phase: Phase) -> CheckResult:
     """Ensure the approved rules of engagement allow the requested phase."""
     result = CheckResult()
     if not isinstance(scope, dict):
         return result
     roe = scope.get("rules_of_engagement") or {}
-    allowed = {_normalise_action(item) for item in roe.get("allowed_actions", []) or []}
+    raw_allowed = roe.get("allowed_actions", []) or []
     forbidden = {_normalise_action(item) for item in roe.get("forbidden_actions", []) or []}
     actions = _PHASE_ACTIONS[phase]
     if forbidden & actions:
         result.add_error(
             f"phase {phase.value} conflicts with scope.rules_of_engagement.forbidden_actions"
         )
-    if not allowed & actions:
+    if not _is_action_permitted(raw_allowed, actions):
+        allowed_options = sorted(actions)
+        formatted_options = ", ".join(f"'{act}'" for act in allowed_options)
+        current_str = ", ".join(f"'{item}'" for item in raw_allowed) or "none"
         result.add_error(
-            f"phase {phase.value} is not permitted by scope.rules_of_engagement.allowed_actions; "
-            f"add an allowed_actions entry containing one of: {', '.join(sorted(actions))}"
+            f"phase {phase.value} is not permitted by scope.rules_of_engagement.allowed_actions "
+            f"(current allowed_actions: [{current_str}]). "
+            f"Select and add one of the following valid action strings for {phase.value} to "
+            f"rules_of_engagement.allowed_actions in scope/scope.yaml (one of: [{formatted_options}])"
         )
     return result
 
@@ -276,6 +297,22 @@ def check_local_artifact_paths(command: str) -> CheckResult:
     return result
 
 
+def check_cross_engagement_paths(command: str, active_eng_dir: Path) -> CheckResult:
+    """Block commands that reference a foreign engagement directory under engagements/."""
+    result = CheckResult()
+    pattern = r"(?i)(?:[/\\]|^)engagements[/\\](benchmark-run-[a-zA-Z0-9_-]+|benchmark-run)\b"
+    active_name = active_eng_dir.name
+    for match in re.finditer(pattern, command):
+        ref_name = match.group(1)
+        if ref_name != active_name and ref_name.startswith("benchmark-run-"):
+            result.add_error(
+                f"cross-engagement path access blocked: command references foreign engagement directory '{ref_name}' "
+                f"while active engagement is '{active_name}'"
+            )
+            break
+    return result
+
+
 def check_skill_binding(eng_dir: Path, task_id: str, session_id: str, phase: Phase) -> CheckResult:
     """Require a delivered, current-context receipt binding for target work."""
     result = CheckResult()
@@ -319,11 +356,29 @@ def check_hypothesis_freshness(
         return result
 
     acceptable_phases = {
-        Phase.VULN_RESEARCH: {Phase.VULN_RESEARCH},
-        Phase.EXPLOITATION: {Phase.VULN_RESEARCH, Phase.EXPLOITATION},
-        Phase.POST_EXPLOITATION: {Phase.EXPLOITATION, Phase.POST_EXPLOITATION},
-        Phase.PRIVESC: {Phase.EXPLOITATION, Phase.POST_EXPLOITATION, Phase.PRIVESC},
-        Phase.FLAGS: {Phase.PRIVESC, Phase.FLAGS},
+        Phase.VULN_RESEARCH: {Phase.RECON, Phase.VULN_RESEARCH},
+        Phase.EXPLOITATION: {Phase.RECON, Phase.VULN_RESEARCH, Phase.EXPLOITATION},
+        Phase.POST_EXPLOITATION: {
+            Phase.RECON,
+            Phase.VULN_RESEARCH,
+            Phase.EXPLOITATION,
+            Phase.POST_EXPLOITATION,
+        },
+        Phase.PRIVESC: {
+            Phase.RECON,
+            Phase.VULN_RESEARCH,
+            Phase.EXPLOITATION,
+            Phase.POST_EXPLOITATION,
+            Phase.PRIVESC,
+        },
+        Phase.FLAGS: {
+            Phase.RECON,
+            Phase.VULN_RESEARCH,
+            Phase.EXPLOITATION,
+            Phase.POST_EXPLOITATION,
+            Phase.PRIVESC,
+            Phase.FLAGS,
+        },
     }.get(phase, {phase})
     scope_path = eng_dir / "scope" / "scope.yaml"
     scope_data = validate_scope(scope_path).scope_data if scope_path.exists() else None
@@ -354,14 +409,21 @@ def check_hypothesis_freshness(
             relevant.append(hypothesis)
     if not relevant:
         eligible = [
-            f"H-{h.id}@{normalise_target(h.target)}"
+            f"H-{h.id}@{normalise_target(h.target)}[phase:{h.phase}]"
             for h in hyps
             if h.canonical_status() != "Rejected" and h.target
         ]
-        msg = f"phase {phase.value} requires a non-rejected hypothesis matching the command target"
+        msg = (
+            f"phase {phase.value} requires a non-rejected hypothesis matching the command target and acceptable phase "
+            f"(acceptable phases for {phase.value}: {', '.join(p.value for p in sorted(acceptable_phases, key=lambda x: x.value))})"
+        )
         if norm_hyp_id:
             msg += f" (linked H-{norm_hyp_id.zfill(3)})"
-        msg += f"; parsed targets: {', '.join(sorted(targets)) or 'none'}; available hypotheses: {', '.join(eligible) or 'none'}"
+        msg += (
+            f"; parsed target(s): {', '.join(sorted(targets)) or 'none'}; "
+            f"available hypotheses: {', '.join(eligible) or 'none'}. "
+            f"To update a hypothesis's phase or status, use violin_record_hypothesis or edit hypotheses.md."
+        )
         result.add_error(msg)
         return result
 
@@ -480,11 +542,11 @@ def check_command(args: CheckCommandArgs) -> CheckResult:
         result.infos.append(f"active PTT task: {ptt_validation.active_task}")
         active_task = ptt.find_active_task(ptt_validation.tasks)
         if active_task and not ptt.task_matches_phase(active_task, phase):
+            task_phase_display = active_task.phase or "RECON (unspecified '## Phase:' header)"
             result.add_error(
-                f"active PTT task {active_task.id} belongs to {active_task.phase or 'no phase'}; "
-                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"
+                f"active PTT task {active_task.id} phase is '{task_phase_display}' (heading-derived from '## Phase:' section in state/ptt.md); "
+                f"requested phase is '{phase.value}'. Next action: call violin_status, then update state/ptt.md so task {active_task.id} sits under a '## Phase: {phase.value}' header "
+                f"or pass phase='{phase.value}' when updating task status via violin_record_ptt."
             )
         if active_task and active_task.note:
             hyp_match = re.search(r"\bH-\d+\b", active_task.note, re.IGNORECASE)

plugins/violin_guard/handlers/ptt_handlers.py

@@ -44,7 +44,10 @@ def _start_ptt_task(
     """Arm one untouched, phase-bound task before the first target command."""
 
     if status != "[~]":
-        raise ValueError("without a pending batch, only [~] may start a PTT task")
+        raise ValueError(
+            f"invalid status {status!r}; starting a task via violin_record_ptt requires status='[~]' "
+            "(use bracket tokens '[~]', '[x]', '[-]', '[!]', not English status words like 'in_progress')"
+        )
     active = ptt.find_active_task(tasks)
     if active and active.id != task_id:
         resolved_dir = ptt_path.parent.parent if eng_dir is None else Path(eng_dir)

plugins/violin_guard/hypotheses.py

@@ -317,7 +317,10 @@ def validate_hypothesis_record(
         _normalize_status(str(fields.get("status") or "Candidate")) == "Validated"
         and not str(fields.get("runtime_evidence") or "").strip()
     ):
-        errors.append("Validated requires runtime_evidence; source evidence alone is not proof")
+        errors.append(
+            "status='Validated' requires the 'runtime_evidence' field (e.g. 'evidence/executions/001-command.json' "
+            "or 'evidence/exploitation/poc.txt'); source evidence alone is not proof"
+        )
     return errors
 
 

plugins/violin_guard/ptt.py

@@ -19,6 +19,7 @@ __all__ = [
     "find_active_task",
     "task_matches_phase",
     "update_task",
+    "sync_ptt",
 ]
 
 
@@ -189,6 +190,14 @@ def update_task(path: Path, task_id: str, status: str, note: str) -> PttTask:
 
     lines = content.splitlines()
     lines[target_idx] = new_line
+
+    # Also synchronize any top summary bullet list lines (e.g. - [ ] PT-101 ...)
+    bullet_re = re.compile(r"^(\s*-\s*)\[[ x~!-]\](\s+" + re.escape(task_id) + r"\b.*)")
+    for i, line in enumerate(lines):
+        m = bullet_re.match(line)
+        if m:
+            lines[i] = f"{m.group(1)}{status}{m.group(2)}"
+
     path.write_text(
         "\n".join(lines) + ("\n" if content and not content.endswith("\n") else ""),
         encoding="utf-8",
@@ -274,4 +283,39 @@ def create_task(path: Path, task_id: str, title: str, phase: str, note: str = ""
     temporary = path.with_suffix(path.suffix + ".tmp")
     temporary.write_text("\n".join(lines) + "\n", encoding="utf-8")
     temporary.replace(path)
+    sync_ptt(path)
     return next(task for task in parse_ptt(path) if task.id == task_id)
+
+
+def sync_ptt(path: Path) -> list[PttTask]:
+    """Synchronize top-level summary checklist items (- [ ] PT-XXX) with table row statuses."""
+    if not path.exists():
+        return []
+    tasks = parse_ptt(path)
+    if not tasks:
+        return []
+    task_statuses = {t.id: t.status for t in tasks}
+
+    content = path.read_text(encoding="utf-8")
+    lines = content.splitlines()
+    modified = False
+
+    bullet_re = re.compile(r"^(\s*-\s*)\[[ x~!-]\](\s+(?P<id>PT-[\w-]+)\b.*)")
+    for i, line in enumerate(lines):
+        m = bullet_re.match(line)
+        if m:
+            t_id = m.group("id")
+            if t_id in task_statuses:
+                new_status = task_statuses[t_id]
+                new_line = f"{m.group(1)}{new_status}{m.group(2)}"
+                if new_line != line:
+                    lines[i] = new_line
+                    modified = True
+
+    if modified:
+        path.write_text(
+            "\n".join(lines) + ("\n" if content and not content.endswith("\n") else ""),
+            encoding="utf-8",
+        )
+        tasks = parse_ptt(path)
+    return tasks

plugins/violin_guard/schemas.py

@@ -33,7 +33,13 @@ class RecordPttArgsModel(BaseModel):
 
     eng_dir: str
     id: str
-    status: str = ""
+    status: str = Field(
+        "",
+        description=(
+            "Task lifecycle status token: '[~]' (active/start), '[x]' (completed), "
+            "'[-]' (cancelled), '[!]' (blocked). Must be a literal bracket token (not 'in_progress')."
+        ),
+    )
     note: str = ""
     skill: str = Field(..., description="Selected Violin skill required before task activation")
     technique: str = Field(..., description="Concrete technique required before task activation")
@@ -98,11 +104,17 @@ class RecordHypothesisArgsModel(BaseModel):
     entry_point: str = ""
     data_flow: str = ""
     source_evidence: str = ""
-    runtime_evidence: str = ""
+    runtime_evidence: str = Field(
+        "",
+        description=(
+            "Required when status is Validated. Path to runtime execution receipt or evidence file "
+            "(e.g. evidence/executions/001-command.json, evidence/exploitation/poc.txt)."
+        ),
+    )
 
 
 class ExecArgsModel(BaseModel):
-    """Authorize and execute one target command using any installed non-interactive Kali/Parrot CLI tool; there is no binary allowlist. Requires one unambiguous [~] PTT task. Scope, phase, hypothesis, history, evidence, timeout, and sync gates still apply, and runtime requirements such as installation, root, hardware, services, GUI, or a TTY are not bypassed. The tool appends exact command history but never updates PTT progress. Hard BLOCK and sync_required never create a process."""
+    """Authorize and execute one target command using any installed non-interactive Kali/Parrot CLI tool; there is no binary allowlist. Commands execute under POSIX shell (/bin/sh, dash on Debian/Ubuntu containers). Builtins like 'source' do not exist in POSIX shell ('source: not found'); use '. file.env' or 'export $(cat file.env)' / 'export $(grep -v "^#" file | xargs)' to load environment variables. Multi-command syntax (&&, ;) is supported, but bash-isms (source, [[ ]], <()) will fail. Requires one unambiguous [~] PTT task. Scope, phase, hypothesis, history, evidence, timeout, and sync gates still apply, and runtime requirements such as installation, root, hardware, services, GUI, or a TTY are not bypassed. The tool appends exact command history but never updates PTT progress. Hard BLOCK and sync_required never create a process."""
 
     model_config = ConfigDict(extra="forbid")
 
@@ -178,7 +190,7 @@ class RebindPendingBatchArgsModel(BaseModel):
 
 
 class HeartbeatDoneArgsModel(BaseModel):
-    """Call AFTER heartbeat review."""
+    """Call AFTER heartbeat review. Clear sequence: 1) violin_status -> 2) violin_review_batch (if pending batch exists) -> 3) violin_heartbeat_done(eng_dir=...)."""
 
     model_config = ConfigDict(extra="forbid")
 
@@ -186,7 +198,7 @@ class HeartbeatDoneArgsModel(BaseModel):
 
 
 class ExecBurstArgsModel(BaseModel):
-    """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."""
+    """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. Sync credit limits per phase apply (Recon: 5, Vuln Research: 10, Exploitation: 10, Post-Exploitation: 20 per sync window) and are shared across execution tools. If a burst is denied with 'insufficient sync credit for burst: need N, have M', split the command set into smaller bursts (size <= M) and review the batch via violin_review_batch to refresh sync credit. Use for recon and exploit/race batches; never raw terminal for targets."""
 
     model_config = ConfigDict(extra="allow")
 
@@ -369,8 +381,9 @@ REBIND_PENDING_BATCH_SCHEMA = to_tool_schema(RebindPendingBatchArgsModel)
 HEARTBEAT_DONE_SCHEMA = to_tool_schema(
     HeartbeatDoneArgsModel,
     description=(
-        f"Call AFTER heartbeat review: re-read skills/pentest/SKILL.md and review scope.yaml /"
-        f" state/ptt.md / hypotheses.md / state/history.md. Cadence is {state.COMMAND_INTERVAL}"
+        f"Call AFTER heartbeat review. Required clear sequence: 1) violin_status -> 2) violin_review_batch "
+        f"(if pending batch exists) -> 3) violin_heartbeat_done(eng_dir=...). Re-read skills/pentest/SKILL.md "
+        f"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."
     ),

plugins/violin_guard/skill_policy.py

@@ -172,6 +172,8 @@ _VULNERABILITY_ROUTES = {
     "auth-bypass": "access-control",
     "authentication": "access-control",
     "authorization": "access-control",
+    "broken-access-control": "access-control",
+    "broken-object-level-authorization": "access-control",
     "idor": "access-control",
     "jwt": "access-control",
     "command-injection": "web-attacks",
@@ -179,7 +181,9 @@ _VULNERABILITY_ROUTES = {
     "sqli": "web-attacks",
     "sql-injection": "web-attacks",
     "ssrf": "web-attacks",
+    "server-side-request-forgery": "web-attacks",
     "xss": "web-attacks",
+    "cross-site-scripting": "web-attacks",
     "source-analysis": "audit-context-building",
     "static-analysis": "semgrep",
     "sarif": "sarif-parsing",
@@ -284,7 +288,10 @@ def resolve_skill_route(
         selected = _PHASE_DEFAULTS[canonical_phase]
     mismatch: list[str] = list(catalog_errors)
     if raw_vulnerability and raw_vulnerability not in _VULNERABILITY_ROUTES:
-        mismatch.append(f"unknown vulnerability class: {vulnerability_class}")
+        valid_classes = ", ".join(sorted(_VULNERABILITY_ROUTES.keys()))
+        mismatch.append(
+            f"unknown vulnerability class: {vulnerability_class}; valid classes are: {valid_classes}"
+        )
     if raw_source and raw_source not in _SOURCE_ROUTES:
         mismatch.append(f"unknown candidate source: {candidate_source}")
     allowed = () if mismatch else (selected,)

plugins/violin_guard/state.py

@@ -47,13 +47,25 @@ def _eng_root() -> Path:
     override = os.environ.get("VIOLIN_ENG_ROOT", "").strip()
     if override:
         return Path(override).expanduser().resolve()
-    # <profile>/plugins/violin_guard/state.py -> <profile>
-    return Path(__file__).resolve().parents[2]
+    container_root = Path("/violin")
+    if container_root.exists() and (container_root / "engagements").exists():
+        return container_root.resolve()
+    cwd = Path.cwd().resolve()
+    for parent in (cwd, cwd.parent, cwd.parent.parent):
+        if (parent / "engagements").exists():
+            return parent
+    candidate = Path(__file__).resolve().parents[2]
+    if (candidate / "engagements").exists():
+        return candidate
+    return cwd
 
 
 def resolve_eng_dir(eng_dir: str | Path) -> Path:
     """Resolve an engagement directory path (absolute or relative to profile root)."""
     if not str(eng_dir).strip() or str(eng_dir).strip() == ".":
+        env_eng = os.environ.get("ENG_DIR", "").strip()
+        if env_eng:
+            return Path(env_eng).expanduser().resolve()
         cwd = Path.cwd().resolve()
         if (cwd / "scope" / "scope.yaml").exists() or (cwd / "hypotheses.md").exists():
             return cwd
@@ -119,13 +131,8 @@ def lock_file(path: Path):
     """Acquire an exclusive advisory lock on ``path`` for the duration of a ``with`` block."""
     lock_path = path.with_suffix(path.suffix + ".lock")
     ensure_dir(lock_path.parent)
-    try:
-        with FileLock(str(lock_path), timeout=20):
-            yield
-    finally:
-        with contextlib.suppress(OSError):
-            if lock_path.exists():
-                lock_path.unlink()
+    with FileLock(str(lock_path), timeout=20):
+        yield
 
 
 def read_json(path: Path) -> dict[str, Any]:

plugins/violin_guard/terminal_policy.py

@@ -253,6 +253,16 @@ def _is_local_compilation_or_test(seg: CommandSegment) -> bool:
     return "py_compile" in seg.raw_text or "pytest" in lower_words or "unittest" in lower_words
 
 
+def _is_local_package_import_check(seg: CommandSegment) -> bool:
+    """Return True if command is a local package availability probe (e.g. python3 -c 'import requests')."""
+    if seg.executable not in _SCRIPT_INTERPRETERS:
+        return False
+    if _url_hosts(seg.raw_text) or _IPV4_RE.search(seg.raw_text):
+        return False
+    text = seg.raw_text.strip()
+    return bool(re.search(r"""(?:python|python3)\s+-c\s+["']\s*import\s+[\w\s,.]+\s*["']""", text))
+
+
 def _block_terminal_segment(seg: CommandSegment) -> str | None:
     segment_text = seg.raw_text
     executable = seg.executable
@@ -260,7 +270,11 @@ def _block_terminal_segment(seg: CommandSegment) -> str | None:
     if _NETWORK_PATH_RE.search(segment_text):
         return _message("network socket path detected in the raw terminal command")
 
-    if executable in _SCRIPT_INTERPRETERS and _NETWORK_MODULE_RE.search(segment_text):
+    if (
+        executable in _SCRIPT_INTERPRETERS
+        and _NETWORK_MODULE_RE.search(segment_text)
+        and not _is_local_package_import_check(seg)
+    ):
         return _message("network-capable script primitive detected in the raw terminal command")
 
     # Package/source retrieval is allowed for local setup (for example git