From 6042ea9179519bb4f8867b180a375baaf02351a2 Mon Sep 17 00:00:00 2001 From: PipeSlinger Date: Mon, 12 Oct 2026 06:13:41 -0500 Subject: [PATCH 1/3] fix(hooks): update PreToolUse matcher to include mcp__.* tool invocations --- hooks/hooks.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hooks/hooks.json b/hooks/hooks.json index eb8a961..275a9e8 100644 --- a/hooks/hooks.json +++ b/hooks/hooks.json @@ -14,7 +14,7 @@ ], "PreToolUse": [ { - "matcher": "Bash|Edit|Write|Agent", + "matcher": "Bash|Edit|Write|Agent|mcp__.*", "hooks": [ { "type": "command", From a7a6d97d35d88b57cc97d8849638d4c42dcbd2ba Mon Sep 17 00:00:00 2001 From: PipeSlinger Date: Mon, 12 Oct 2026 06:15:06 -0500 Subject: [PATCH 2/3] feat(guards): inspect MCP tool invocations and record refusal telemetry --- core/guard_engine.py | 83 +++++++++++++++++++++++++++++++++++++++----- 1 file changed, 75 insertions(+), 8 deletions(-) diff --git a/core/guard_engine.py b/core/guard_engine.py index 652c7c8..6dacefe 100644 --- a/core/guard_engine.py +++ b/core/guard_engine.py @@ -1,4 +1,4 @@ -"""Guard engine — evaluate whether a tool call is allowed in the current step.""" +"""Guard engine - evaluate whether a tool call is allowed in the current step.""" import json import os @@ -8,6 +8,9 @@ from . import event_store, state_machine +# Standard keys that carry executable commands or scripts in MCP and CLI tool inputs +COMMAND_KEYS = ("command", "cmd", "script", "query", "eval", "code") + def _approved_commands(project_dir: str | None) -> set[str]: """Per-run pre-accept allowlist: exact commands an operator vetted for THIS run. @@ -22,6 +25,72 @@ def _approved_commands(project_dir: str | None) -> set[str]: return set() +def _extract_command_and_target(tool_name: str, tool_input: dict) -> tuple[str, str | None]: + """Extract command text or target file from hook input. + + Inspects Bash, Edit/Write, and MCP tool invocations (mcp__*, ctx_execute, etc.). + """ + if not isinstance(tool_input, dict): + return "", None + + # Standard tool types + if tool_name == "Bash": + return tool_input.get("command", ""), None + if tool_name in ("Edit", "Write"): + return tool_input.get("file_path", ""), tool_input.get("file_path") + + # MCP tool invocations (e.g. mcp__* or custom execute tools) + if tool_name.startswith("mcp__") or "execute" in tool_name.lower(): + # Look for direct command strings + for k in COMMAND_KEYS: + val = tool_input.get(k) + if isinstance(val, str) and val.strip(): + return val, None + + # Look for file path edits through MCP tools + if "file_path" in tool_input and isinstance(tool_input["file_path"], str): + return tool_input["file_path"], tool_input["file_path"] + + return "", None + + +def _log_refusal(tool_name: str, command: str, reason: str, rule: str, project_dir: str | None = None) -> None: + """Record refusal telemetry to agentry events stream if configured.""" + # 1. Weft event store (local project events.jsonl) + # Recorded in evaluate() via event_store.append("wf.guard_blocked") + + # 2. Agentry events stream (~/.local/state/agentry/events.jsonl) + # Reads build_id / task from AGENT_P_TAGS environment variable if present + agentry_dir = Path.home() / ".local" / "state" / "agentry" + agent_p_tags = os.environ.get("AGENT_P_TAGS", "") + + # Extract build_id and task from AGENT_P_TAGS (e.g., 'build_id=xyz,task=abc') + tags = {} + if agent_p_tags: + for part in agent_p_tags.split(","): + if "=" in part: + k, v = part.split("=", 1) + tags[k.strip()] = v.strip() + + payload = { + "event": "guard", + "tool": tool_name, + "refused_input": command, + "rule": rule, + "reason": reason, + "build_id": tags.get("build_id"), + "task": tags.get("task"), + } + + try: + agentry_dir.mkdir(parents=True, exist_ok=True) + log_file = agentry_dir / "events.jsonl" + with open(log_file, "a", encoding="utf-8") as f: + f.write(json.dumps(payload) + "\n") + except OSError: + pass + + def evaluate(hook_input: dict, project_dir: str | None = None) -> dict | None: """Check if the tool call is blocked by a guard from a non-current step. @@ -32,20 +101,17 @@ def evaluate(hook_input: dict, project_dir: str | None = None) -> dict | None: return None tool_name = hook_input.get("tool_name", "") - command = "" - if tool_name == "Bash": - command = hook_input.get("tool_input", {}).get("command", "") - elif tool_name in ("Edit", "Write"): - command = hook_input.get("tool_input", {}).get("file_path", "") + tool_input = hook_input.get("tool_input", {}) + command, _ = _extract_command_and_target(tool_name, tool_input) if not command: return None # Pre-accept escape hatch: an operator-vetted exact command bypasses every guard - # for this one run. Exact-match only — if the agent deviates from the approved + # for this one run. Exact-match only - if the agent deviates from the approved # string at all, it stays blocked. This is the sole path a destructive/publish # command reaches the system, and it requires a human's explicit approval. - if tool_name == "Bash" and command.strip() in _approved_commands(project_dir): + if (tool_name == "Bash" or tool_name.startswith("mcp__")) and command.strip() in _approved_commands(project_dir): return None current_id = state["current_step"] @@ -74,6 +140,7 @@ def evaluate(hook_input: dict, project_dir: str | None = None) -> dict | None: workflow_id=state["workflow_id"], project_dir=project_dir, ) + _log_refusal(tool_name, command, msg, pattern, project_dir) return {"blocked": True, "reason": msg} except re.error as e: print(f"[weft] Warning: invalid guard pattern '{pattern}': {e}", file=sys.stderr) From a4185ed0e1a862d9a185313fe9cfcae70e7205c8 Mon Sep 17 00:00:00 2001 From: PipeSlinger Date: Mon, 12 Oct 2026 06:26:05 -0500 Subject: [PATCH 3/3] test(guards): add tests for MCP tool guards and refusal telemetry in tests/test_mcp_guards.py --- tests/test_mcp_guards.py | 64 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 64 insertions(+) create mode 100644 tests/test_mcp_guards.py diff --git a/tests/test_mcp_guards.py b/tests/test_mcp_guards.py new file mode 100644 index 0000000..3cd1aa2 --- /dev/null +++ b/tests/test_mcp_guards.py @@ -0,0 +1,64 @@ +"""Tests for MCP tool guard detection and refusal telemetry.""" + +import json +from pathlib import Path +import pytest + +from core import guard_engine, state_machine + + +class TestMcpGuards: + def test_mcp_tool_command_is_guarded(self, project_dir, guarded_template): + """MCP tool call carrying a shell command is caught by the guard.""" + pdir = str(project_dir) + state_machine.start_workflow(guarded_template, "s1", pdir) + + result = guard_engine.evaluate( + { + "tool_name": "mcp__context_mode__ctx_execute", + "tool_input": {"command": "git push origin main"} + }, + pdir, + ) + assert result is not None + assert result["blocked"] is True + + def test_mcp_tool_cmd_key_is_guarded(self, project_dir, guarded_template): + """MCP tool call using 'cmd' or 'script' is caught by the guard.""" + pdir = str(project_dir) + state_machine.start_workflow(guarded_template, "s1", pdir) + + result = guard_engine.evaluate( + { + "tool_name": "mcp__runner__run", + "tool_input": {"cmd": "kubectl apply -f pod.yaml"} + }, + pdir, + ) + assert result is not None + assert result["blocked"] is True + + def test_logs_refusal_event_with_agent_p_tags(self, project_dir, guarded_template, monkeypatch, tmp_path): + """Refusal telemetry records to ~/.local/state/agentry/events.jsonl with AGENT_P_TAGS.""" + pdir = str(project_dir) + state_machine.start_workflow(guarded_template, "s1", pdir) + + monkeypatch.setenv("AGENT_P_TAGS", "build_id=bld_4910,task=refactor_core") + monkeypatch.setattr(Path, "home", lambda: tmp_path) + + result = guard_engine.evaluate( + {"tool_name": "Bash", "tool_input": {"command": "git push origin main"}}, + pdir, + ) + assert result is not None and result["blocked"] is True + + agentry_log = tmp_path / ".local" / "state" / "agentry" / "events.jsonl" + assert agentry_log.exists() + lines = [json.loads(line) for line in agentry_log.read_text(encoding="utf-8").splitlines()] + assert len(lines) == 1 + refusal = lines[0] + assert refusal["event"] == "guard" + assert refusal["tool"] == "Bash" + assert refusal["refused_input"] == "git push origin main" + assert refusal["build_id"] == "bld_4910" + assert refusal["task"] == "refactor_core"