diff --git a/plugins/violin_guard/__init__.py b/plugins/violin_guard/__init__.py index 34148cf..04025b4 100644 --- a/plugins/violin_guard/__init__.py +++ b/plugins/violin_guard/__init__.py @@ -6,13 +6,22 @@ Hermes plugin registration entry point. from __future__ import annotations import contextlib +import json +from pathlib import Path from . import ( code_execution_audit, + ptt, schemas, # noqa: E402 service, # noqa: E402 state, # noqa: E402 ) +from .skill_receipts import ( + advance_context_generation, + binding_readiness, + record_binding_turn, + record_delivery_turn, +) from .terminal_policy import block_terminal_command __all__ = ["register", "TOOLS", "REGISTERED_TOOLS", "tools"] @@ -20,6 +29,23 @@ __all__ = ["register", "TOOLS", "REGISTERED_TOOLS", "tools"] TOOLS = service tools = service +_SESSION_ENGAGEMENTS: dict[str, str] = {} +_TARGET_TOOLS = { + "violin_exec", + "violin_exec_burst", + "violin_httpx", + "violin_nuclei", + "violin_ffuf", + "violin_listener", +} +_BROWSER_TARGET_TOOLS = { + "browser_navigate", + "browser_click", + "browser_type", + "browser_select", + "browser_scroll", +} + # Tool names registered with the Hermes plugin loader. Kept in sync with the # registration tuple below; the release gate compares these against # plugin.yaml's provides_tools. @@ -119,6 +145,14 @@ def _pre_tool_call_hook(tool_name=None, args=None, **kwargs): commands must use ``violin_exec`` or ``violin_exec_burst``. """ args = args if isinstance(args, dict) else {} + session_id = str(kwargs.get("session_id") or args.get("session_id") or "") + eng_dir = str(args.get("eng_dir") or "") + if session_id and eng_dir: + _SESSION_ENGAGEMENTS[session_id] = eng_dir + if tool_name in _TARGET_TOOLS or tool_name in _BROWSER_TARGET_TOOLS: + blocked = _check_turn_binding(tool_name, args, kwargs) + if blocked: + return {"action": "block", "message": blocked} if tool_name == "execute_code": _metadata, message = code_execution_audit.validate_source(args.get("code")) return None if message is None else {"action": "block", "message": message} @@ -133,6 +167,26 @@ def _post_tool_call_hook(tool_name=None, args=None, result=None, duration_ms=0, if tool_name == "execute_code" and isinstance(args, dict): with contextlib.suppress(Exception): code_execution_audit.record_completion(args.get("code"), result, duration_ms) + if tool_name not in {"violin_record_ptt", "violin_review_batch"}: + return + args = args if isinstance(args, dict) else {} + eng_dir = str(args.get("eng_dir") or "") + turn_id = str(kwargs.get("turn_id") or "") + if not eng_dir or not turn_id: + return + try: + payload = json.loads(result) if isinstance(result, str) else result + if not isinstance(payload, dict): + return + skill = payload.get("skill") or {} + delivery_id = str(skill.get("delivery_id") or "") + if payload.get("status") == "skill_prepared" and delivery_id: + record_delivery_turn(eng_dir, delivery_id=delivery_id, turn_id=turn_id) + task_id = str(payload.get("task_id") or payload.get("binding_task_id") or "") + if payload.get("status") == "ok" and task_id: + record_binding_turn(eng_dir, task_id=task_id, turn_id=turn_id) + except Exception: + return # --------------------------------------------------------------------------- @@ -152,14 +206,46 @@ def _pre_llm_call_hook(session_id=None, eng_dir=None, **kwargs): with contextlib.suppress(Exception): state.tick_message(str(eng_dir)) state.record_session_id(str(eng_dir), session_id) + if session_id: + _SESSION_ENGAGEMENTS[str(session_id)] = str(eng_dir) return None def _on_session_reset_hook(session_id=None, eng_dir=None, **kwargs) -> None: """Hook: session reset (context compression, /goal set, etc.).""" + eng_dir = eng_dir or _SESSION_ENGAGEMENTS.get(str(session_id or "")) if eng_dir: with contextlib.suppress(Exception): state.tick_message(str(eng_dir)) + advance_context_generation(str(eng_dir), str(session_id or "reset")) + + +def _check_turn_binding(tool_name: str, args: dict, hook: dict) -> str | None: + """Stop target/browser activity until an earlier-turn receipt is bound.""" + + session_id = str(hook.get("session_id") or args.get("session_id") or "") + eng_dir = str(args.get("eng_dir") or _SESSION_ENGAGEMENTS.get(session_id) or "") + if not eng_dir: + return ( + f"{tool_name} needs an engagement associated through violin_status or a Violin tool " + "before browser or target activity is allowed" + ) + tasks = ptt.parse_ptt(Path(eng_dir) / "state" / "ptt.md") + active = ptt.find_active_task(tasks) + if not active: + return "target activity requires exactly one active [~] PTT task" + binding, reason = binding_readiness(eng_dir, task_id=active.id, session_id=session_id) + if reason: + return f"target activity blocked: {reason}; select and prepare a routed skill first" + turn_id = str(hook.get("turn_id") or "") + if turn_id and turn_id in { + str(binding.get("bound_turn_id") or ""), + str((binding or {}).get("delivered_turn_id") or ""), + }: + return ( + "target activity is blocked in the skill delivery/binding turn; retry on the next turn" + ) + return None def _on_session_finalize_hook(session_id=None, eng_dir=None, **kwargs) -> None: diff --git a/plugins/violin_guard/service.py b/plugins/violin_guard/service.py index 066f9a6..c138620 100644 --- a/plugins/violin_guard/service.py +++ b/plugins/violin_guard/service.py @@ -22,7 +22,14 @@ from .adapters import ( from .command import CheckCommandArgs from .history import history_contains from .phases import Phase, requires_hypothesis, suppresses_heartbeat -from .skill_receipts import HermesSkillViewAdapter, bind_task, complete_delivery, prepare_delivery +from .skill_policy import resolve_skill_route +from .skill_receipts import ( + HermesSkillViewAdapter, + bind_task, + binding_readiness, + complete_delivery, + prepare_delivery, +) from .targets import resolve_target # --------------------------------------------------------------------------- @@ -146,6 +153,7 @@ def handle_record_ptt(a, **kwargs): "digest": digest, "content": viewed.content, "error": viewed.error, + "delivery_id": reservation.id, }, ) if reservation.status == "preparing": @@ -378,6 +386,7 @@ def handle_review_batch(a, **kwargs): "digest": digest, "content": viewed.content, "error": viewed.error, + "delivery_id": reservation.id, }, ) if reservation.status == "preparing": @@ -418,6 +427,7 @@ def handle_review_batch(a, **kwargs): released=True, finding=finding_result, finding_path=finding_result.get("path") if finding_result else None, + binding_task_id=(context["task_id"] if skill else None), ) except (OSError, ValueError) as exc: return _json( @@ -728,7 +738,13 @@ def handle_status(a, **kwargs): 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()) + legacy_marker = str(marker) if marker and marker.is_file() else None + binding, binding_reason = ( + binding_readiness(eng_dir, task_id=active.id, session_id=session_id) + if active and session_id + else (None, "no active task or session") + ) + route = resolve_skill_route(current_phase or "RECON") blockers = [ { @@ -742,16 +758,16 @@ def handle_status(a, **kwargs): 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", + "reason": "No session id is recorded for receipt-backed skill delivery", + "next_action": "Use violin_record_ptt to select and prepare a routed skill", } ) - elif not skill_loaded: + elif active and binding_reason: 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}", + "code": "skill_binding_required", + "reason": binding_reason, + "next_action": "Prepare the routed skill, then repeat the PTT update next turn", } ) blockers.extend( @@ -822,10 +838,14 @@ def handle_status(a, **kwargs): 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, + "binding": binding, + "binding_ready": binding_reason is None, + "binding_reason": binding_reason, + "route_candidates": list(route.allowed), + "legacy_marker": legacy_marker, + "legacy_marker_status": "obsolete" if legacy_marker else "absent", + "recovery": "Select a route candidate with violin_record_ptt and retry next turn", }, runtime=runtime_backend.runtime_readiness(eng_dir), ) diff --git a/plugins/violin_guard/skill_receipts.py b/plugins/violin_guard/skill_receipts.py index fd44d96..8732b3e 100644 --- a/plugins/violin_guard/skill_receipts.py +++ b/plugins/violin_guard/skill_receipts.py @@ -26,7 +26,10 @@ __all__ = [ "bind_task", "complete_delivery", "get_binding", + "binding_readiness", "get_delivery", + "record_binding_turn", + "record_delivery_turn", "prepare_delivery", "prepare_review_readiness", ] @@ -302,6 +305,60 @@ def get_binding(eng_dir: str | Path, task_id: str) -> dict[str, Any] | None: return data["bindings"].get(task_id) +def binding_readiness( + eng_dir: str | Path, *, task_id: str, session_id: str +) -> tuple[dict[str, Any] | None, str | None]: + """Return the active receipt binding, or a precise fail-closed reason.""" + + data, recovered = _load(_path(eng_dir)) + if recovered: + return None, "skill receipt state is unavailable; prepare the selected skill again" + context = data.get("context") or {} + binding = (data.get("bindings") or {}).get(task_id) + if not binding: + return None, "the active PTT task has no delivered skill binding" + if str(context.get("session_id") or "") != session_id: + return None, "the skill binding belongs to a different session" + if int(binding.get("context_generation", -1)) != int(context.get("generation") or 0): + return None, "the skill binding is stale after a context reset" + delivery = (data.get("deliveries") or {}).get(binding.get("delivery_id")) + if not delivery or delivery.get("status") != "delivered": + return None, "the bound skill delivery is not ready" + if delivery.get("bundle_digest") != binding.get("bundle_digest"): + return None, "the bound skill digest no longer matches its delivery" + return {**binding, "delivered_turn_id": delivery.get("delivered_turn_id")}, None + + +def record_delivery_turn(eng_dir: str | Path, *, delivery_id: str, turn_id: str) -> None: + """Associate a successful ``skill_view`` result with its Hermes turn.""" + + if not turn_id: + return + + def record(data: dict[str, Any]) -> None: + entry = data["deliveries"].get(delivery_id) + if entry and entry.get("status") == "delivered": + entry["delivered_turn_id"] = turn_id + entry["updated_at"] = _now() + + _mutate(eng_dir, record) + + +def record_binding_turn(eng_dir: str | Path, *, task_id: str, turn_id: str) -> None: + """Associate a binding commit with its Hermes turn.""" + + if not turn_id: + return + + def record(data: dict[str, Any]) -> None: + binding = data["bindings"].get(task_id) + if binding: + binding["bound_turn_id"] = turn_id + binding["bound_at"] = _now() + + _mutate(eng_dir, record) + + def prepare_review_readiness( eng_dir: str | Path, *, diff --git a/scripts/violin_guard.py b/scripts/violin_guard.py index e97d41b..2c5f091 100644 --- a/scripts/violin_guard.py +++ b/scripts/violin_guard.py @@ -134,7 +134,7 @@ def cmd_status(args: argparse.Namespace) -> int: if args.section == "skill": skill = out.get("skill", {}) print(json.dumps(skill, indent=2)) - return 0 if skill.get("loaded") else 1 + return 0 if skill.get("binding_ready") else 1 print(json.dumps(out, indent=2)) return 0 if out["status"] == "ok" else 1 diff --git a/tests/guard/state/test_burst_and_target.py b/tests/guard/state/test_burst_and_target.py index 6d984fe..ccba1c6 100644 --- a/tests/guard/state/test_burst_and_target.py +++ b/tests/guard/state/test_burst_and_target.py @@ -361,17 +361,20 @@ def test_plugin_exposes_new_tools(): def test_status_skill_section_reports_load_state_and_exit_code(eng): state.record_session_id(eng, "ts") + bind_active_task(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 + assert loaded_data["binding_ready"] is True + assert loaded_data["legacy_marker_status"] == "obsolete" - marker = Path(loaded_data["marker"]) + marker = Path(loaded_data["legacy_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 + assert missing.returncode == 0 + assert missing_data["binding_ready"] is True + assert missing_data["legacy_marker_status"] == "absent" @pytest.mark.parametrize( diff --git a/tests/guard/state/test_collaboration_ux.py b/tests/guard/state/test_collaboration_ux.py index da499d2..5320f8f 100644 --- a/tests/guard/state/test_collaboration_ux.py +++ b/tests/guard/state/test_collaboration_ux.py @@ -9,6 +9,7 @@ from pathlib import Path import pytest from plugins.violin_guard import bootstrap, history, ptt, service, state +from tests.guard.receipt_fixture import bind_active_task ROOT = Path(__file__).resolve().parents[3] @@ -30,6 +31,7 @@ def _engagement(tmp_path: Path) -> Path: (eng / "state" / ".skill-loaded-test-session").write_text( "skill-loaded: pentest\n", encoding="utf-8" ) + bind_active_task(eng, "test-session") return eng @@ -89,7 +91,8 @@ def test_status_explains_current_phase_pending_commands_and_skill(tmp_path: Path 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 + assert result["skill"]["binding_ready"] is True + assert result["skill"]["legacy_marker_status"] in {"absent", "obsolete"} @pytest.mark.parametrize("task_status", ["[~]", "[x]", "[!]", "[-]"]) diff --git a/tests/guard/test_terminal_policy.py b/tests/guard/test_terminal_policy.py index 30162f5..d03c155 100644 --- a/tests/guard/test_terminal_policy.py +++ b/tests/guard/test_terminal_policy.py @@ -5,7 +5,9 @@ from __future__ import annotations import pytest from plugins.violin_guard import ( + _on_session_reset_hook, _post_tool_call_hook, + _pre_llm_call_hook, _pre_tool_call_hook, bootstrap, register, @@ -172,14 +174,68 @@ def test_local_file_path_containing_an_ip_is_not_treated_as_a_socket() -> None: ) -def test_non_terminal_tools_are_not_affected() -> None: +def test_target_tools_require_an_engagement_binding() -> None: result = _pre_tool_call_hook( tool_name="violin_exec", args={"command": "nmap -sV 10.10.10.10"}, session_id="test-session", ) - assert result is None + assert result["action"] == "block" + assert "engagement associated" in result["message"] + + +def test_skill_delivery_turn_blocks_target_and_browser_until_next_turn(tmp_path) -> None: + eng = _engagement(tmp_path) + _pre_llm_call_hook(session_id="test", eng_dir=str(eng)) + _post_tool_call_hook( + tool_name="violin_record_ptt", + args={"eng_dir": str(eng), "id": "PT-010"}, + result='{"status":"ok","task_id":"PT-010"}', + turn_id="turn-bind", + ) + + blocked = _pre_tool_call_hook( + tool_name="violin_exec", + args={"eng_dir": str(eng), "session_id": "test"}, + session_id="test", + turn_id="turn-bind", + ) + assert blocked["action"] == "block" + assert "next turn" in blocked["message"] + + browser_blocked = _pre_tool_call_hook( + tool_name="browser_navigate", + args={"url": "https://10.10.10.10"}, + session_id="test", + turn_id="turn-bind", + ) + assert browser_blocked["action"] == "block" + + assert ( + _pre_tool_call_hook( + tool_name="browser_navigate", + args={"url": "https://10.10.10.10"}, + session_id="test", + turn_id="turn-next", + ) + is None + ) + + +def test_session_reset_invalidates_active_skill_binding(tmp_path) -> None: + eng = _engagement(tmp_path) + _pre_llm_call_hook(session_id="test", eng_dir=str(eng)) + _on_session_reset_hook(session_id="test") + + blocked = _pre_tool_call_hook( + tool_name="violin_exec", + args={"eng_dir": str(eng), "session_id": "test"}, + session_id="test", + turn_id="after-reset", + ) + assert blocked["action"] == "block" + assert "stale after a context reset" in blocked["message"] def _engagement(tmp_path):