feat(hooks): enforce the guard from a PreToolUse hook

The catalog and zerto_check_tool are advice. An MCP server cannot see or
block another server's tool calls, so a model that skips the guard is not
stopped by anything, and the change lands with no checkpoint behind it.

A PreToolUse hook runs in the host, where the tool call actually pauses.
That turns capture-before-execute from a convention into something the
host enforces.

  read-only                      no decision, runs
  mutating, checkpoint confirmed no decision, plus additionalContext
                                 telling the model which checkpoint to
                                 recover from
  mutating, checkpoint failed    DENY, the call never happens
  mutating, VM unknown to Zerto  prompt, nothing to rewind to
  mutating, no VM in the args    prompt, the catalog's vm_arg missed
  unlisted tool                  prompt

Decisions go out as JSON rather than exit 2. Exit 2 blocks
unconditionally but discards the JSON, and with it the reason, so the
model would be refused without being told why.

A lookup miss is not a refusal. Asking Zerto for a plain hostname by
vmIdentifier returns HTTP 400, and an early version reported that as
"could not tag a checkpoint", which denied changes to machines Zerto had
simply never heard of. Those are now separated: unknown VM prompts, a VM
Zerto knows but will not tag denies.

Timeouts are budgeted for the slow path. A successful tag took about 4s,
but a refusal took 63s, because wait_for_tag spends 45s waiting for a
checkpoint that will never arrive on an AWS or Azure protected VPG. If
the host's timeout fires first it cancels the hook and discards its
output, and the call proceeds unguarded, so the hook's own budget (150s)
stays under the configured one (180s): better to deny than be cancelled.

Verified against a live ZVM 10.9.10. jp-ubuntu tagged checkpoint 7180 and
was allowed; win2019-1, whose VPG is protected at an AWS site where
tagged checkpoints are unsupported, was denied. Both held under
permission_mode bypassPermissions, which is when an agent is most likely
running unattended.

Broad excepts in hooks/ are deliberate and scoped in pyproject: a hook
that raises breaks the tool call it exists to protect.

pytest 60 passed (12 new).

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_016yVfC5nvZowoLFnEGWhLGn

hooks/README.md

@@ -0,0 +1,72 @@
+# PreToolUse guard hook
+
+Makes the rewind guard enforceable instead of advisory.
+
+The MCP server cannot see another server's tool calls, so `zerto_check_tool` and
+`zerto_guard_before_mutate` only work if the model chooses to call them. A model
+that skips the step is not stopped by anything. A `PreToolUse` hook runs in the
+host, where the tool call genuinely pauses, so a failed checkpoint stops the
+change.
+
+## Decisions
+
+| situation | decision | effect |
+|---|---|---|
+| tool in `read_only_tools` | none | runs, no checkpoint |
+| mutating, checkpoint confirmed | none, plus `additionalContext` | runs, and the model is told which checkpoint to recover from |
+| mutating, checkpoint failed | `deny` | **the call never happens** |
+| mutating, VM unknown to Zerto | `prompt` | human decides; nothing to rewind to |
+| mutating, no VM in the arguments | `prompt` | catalog's `vm_arg` did not match |
+| unlisted tool | `prompt` | nobody said it was read-only |
+
+`ZERTO_HOOK_UNKNOWN` switches the unlisted case to `allow` or `deny`.
+
+## Install
+
+```bash
+cp hooks/settings.example.json /tmp/x        # then merge the hooks block into
+                                             # .claude/settings.json
+export ZERTO_REWIND_CONFIG=/abs/path/config.json
+```
+
+Both paths in the command must be absolute, and the interpreter must be the
+venv that has this package installed.
+
+## Timeouts, which matter here
+
+The hook is synchronous: the host waits. That is the point, because the
+checkpoint has to exist before the change does.
+
+Budget for the slow path, not the fast one. A successful tag took about 4s
+against a healthy VPG, but a **refusal took 63s**: `wait_for_tag` waits 45s for
+a checkpoint that is never going to appear, which is exactly what happens on a
+VPG whose protected site is AWS or Azure, where tagged checkpoints are not
+supported.
+
+So:
+
+- `timeout` in settings.json: 180 (seconds)
+- `ZERTO_HOOK_GUARD_TIMEOUT`: 150 (seconds), kept under it
+
+If the host's timeout fires first it cancels the hook and **discards its
+output**, and the tool call carries on through the normal permission flow. A
+timeout is therefore a silent failure of the guard, which is why the hook's own
+budget is the smaller of the two: it would rather deny than be cancelled.
+
+## Verified behaviour
+
+Against a live ZVM 10.9.10, all six rows of the table above. Two worth naming:
+
+- `jp-ubuntu` (healthy, local VPG) tagged checkpoint 7180 and allowed the call,
+  passing the tag back through `additionalContext`.
+- `win2019-1` (VPG `CMH-AWS-1`, protected site AWS) was **denied**: Zerto cannot
+  insert a tagged checkpoint there, so the change would not have been
+  recoverable.
+
+Both decisions held with `permission_mode: bypassPermissions`. A hook still
+blocks when the user has turned permissions off, which is when an agent is most
+likely to be running unattended.
+
+## Log
+
+`~/.zerto-guard-hook.log`, or `ZERTO_HOOK_LOG`. One line per decision.

hooks/settings.example.json

@@ -0,0 +1,17 @@
+{
+  "_comment": "Copy the hooks block into .claude/settings.json (project) or ~/.claude/settings.json (all projects). Use absolute paths.",
+  "hooks": {
+    "PreToolUse": [
+      {
+        "matcher": "mcp__.*",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "/home/you/zerto-ai-rewind/.venv/bin/python /home/you/zerto-ai-rewind/hooks/zerto_guard_hook.py",
+            "timeout": 180
+          }
+        ]
+      }
+    ]
+  }
+}

hooks/zerto_guard_hook.py

@@ -0,0 +1,247 @@
+#!/usr/bin/env python3
+"""Claude Code PreToolUse hook: capture before execute, enforced by the host.
+
+The MCP server cannot see another server's tool calls, so the guard it exposes
+is advice the model may skip. This hook sits in the host, where the call really
+does pause, so a failed checkpoint stops the change instead of merely
+suggesting it should.
+
+Decisions:
+
+  read-only tool            allow, no checkpoint
+  mutating, guard ok        allow, and tell the model which checkpoint to use
+  mutating, guard failed    DENY. No checkpoint means no rewind, so no change.
+  mutating, VM unprotected  prompt. Zerto cannot rewind it; a human decides.
+  unknown tool              prompt. Nobody said it was read-only.
+
+Exit 0 always, with the decision in stdout JSON. Exit 2 would block
+unconditionally and ignore the JSON, which loses the reason text.
+
+Install (project .claude/settings.json). Give it room: the guard inserts a
+tagged checkpoint and waits for the Zerto task to reach Completed.
+
+  {"hooks": {"PreToolUse": [{
+     "matcher": "mcp__.*",
+     "hooks": [{"type": "command",
+                "command": "/path/to/.venv/bin/python /path/to/hooks/zerto_guard_hook.py",
+                "timeout": 180}]}]}}
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import re
+import sys
+import time
+
+LOG = os.environ.get("ZERTO_HOOK_LOG", os.path.expanduser("~/.zerto-guard-hook.log"))
+# Seconds the hook will wait for the checkpoint. Must stay under the hook
+# timeout configured in settings.json, or the host cancels us and the tool
+# call proceeds unguarded through the normal permission flow.
+GUARD_TIMEOUT_S = float(os.environ.get("ZERTO_HOOK_GUARD_TIMEOUT", "150"))
+UNKNOWN_DECISION = os.environ.get("ZERTO_HOOK_UNKNOWN", "prompt")  # prompt | allow | deny
+
+
+def log(msg: str) -> None:
+    try:
+        with open(LOG, "a") as fh:
+            fh.write(f"{time.strftime('%Y-%m-%dT%H:%M:%S')} {msg}\n")
+    except Exception:
+        pass
+
+
+def emit(decision: str | None, reason: str = "", context: str = "") -> None:
+    """Write the PreToolUse decision and leave. Never raises."""
+    if decision is None and not context:
+        sys.exit(0)  # no opinion; normal permission flow applies
+    out: dict = {"hookEventName": "PreToolUse"}
+    if decision:
+        out["permissionDecision"] = decision
+        out["permissionDecisionReason"] = reason
+    if context:
+        out["additionalContext"] = context
+    print(json.dumps({"hookSpecificOutput": out}))
+    sys.exit(0)
+
+
+def split_mcp_name(tool_name: str) -> tuple[str, str] | None:
+    """mcp__<server>__<tool> -> (server, tool). Server names may contain _."""
+    m = re.match(r"^mcp__(.+?)__(.+)$", tool_name or "")
+    return (m.group(1), m.group(2)) if m else None
+
+
+def resolve_vm(tool_input: dict, vm_arg: str) -> str:
+    """Pull the VM identifier out of the pending call's arguments."""
+    if vm_arg and tool_input.get(vm_arg):
+        return str(tool_input[vm_arg])
+    # vm_arg is the catalog's answer, but fall back to the usual suspects so a
+    # slightly-wrong catalog entry degrades to a prompt rather than a crash.
+    for key in (
+        "host",
+        "hostname",
+        "computer_name",
+        "computerName",
+        "vm",
+        "vm_name",
+        "target",
+        "limit",
+        "address",
+        "server",
+    ):
+        if tool_input.get(key):
+            return str(tool_input[key])
+    return ""
+
+
+async def run_guard(query: str, change_id: str, action: str) -> dict:
+    from zerto_rewind_mcp.checkpoints import make_tag, tag_vpgs
+    from zerto_rewind_mcp.client import ZertoClient, ZertoError
+    from zerto_rewind_mcp.config import load_config
+    from zerto_rewind_mcp.protection import find_from_rows
+
+    settings = load_config()
+    client = ZertoClient(
+        base_url=str(settings["zerto_url"]),
+        username=str(settings["username"]),
+        password=str(settings.get("password") or ""),
+        client_id=str(settings.get("client_id") or "zerto-client"),
+        verify_tls=bool(settings.get("verify_tls")),
+    )
+    try:
+        # Two different failures hide here. "Zerto has never heard of this VM"
+        # is not the same as "Zerto refused to tag it", and they get different
+        # answers, so a lookup miss must not surface as a refusal. Looking up a
+        # plain hostname by vmIdentifier returns HTTP 400, which is a miss.
+        rows = []
+        for kwargs in ({"vm_name": query}, {"vm_identifier": query}):
+            try:
+                rows = await client.get_vms(**kwargs)
+            except ZertoError:
+                rows = []
+            if rows:
+                break
+        if not rows:
+            return {
+                "ok": False,
+                "protected": False,
+                "message": f"Zerto has no VM matching {query!r}.",
+            }
+        result = find_from_rows(query, rows)
+        if result.outcome != "ok" or not result.taggable_vpgs:
+            return {"ok": False, "protected": False, "message": result.message}
+        tag = make_tag(
+            "claude-code",
+            change_id,
+            action=action,
+            vm_name=result.vm.vm_name if result.vm else None,
+        )
+        out = await tag_vpgs(client, result, tag)
+        out["protected"] = True
+        return out
+    except ZertoError as exc:
+        # Reached Zerto and it refused, or never reached it at all. Either way
+        # there is no checkpoint, so this is not a pass.
+        return {"ok": False, "protected": True, "message": str(exc)}
+    finally:
+        try:
+            await client.aclose()
+        except Exception:
+            pass
+
+
+def main() -> None:
+    try:
+        payload = json.loads(sys.stdin.read() or "{}")
+    except Exception as exc:
+        log(f"unparseable stdin: {exc}")
+        emit(None)
+
+    tool_name = payload.get("tool_name") or ""
+    tool_input = payload.get("tool_input") or {}
+    parts = split_mcp_name(tool_name)
+    if not parts:
+        emit(None)  # not an MCP tool; this hook has nothing to say
+    server, tool = parts
+
+    try:
+        from zerto_rewind_mcp.catalog import MUTATING, READ_ONLY
+        from zerto_rewind_mcp.config import load_catalog, load_config
+
+        catalog = load_catalog(load_config())
+        verdict = catalog.classify(server, tool)
+        entry = catalog.get(server, tool)
+    except Exception as exc:
+        log(f"catalog unavailable ({exc}); staying out of the way")
+        emit(None)
+
+    if verdict == READ_ONLY:
+        log(f"{tool_name}: read-only, allowed")
+        emit(None)
+
+    if verdict != MUTATING:
+        log(f"{tool_name}: unknown -> {UNKNOWN_DECISION}")
+        emit(
+            UNKNOWN_DECISION,
+            f"{server}/{tool} is not a known read-only tool and is not in the "
+            "Zerto mutating catalog, so no checkpoint was taken. If it changes a "
+            "protected VM, that change cannot be rewound. Allow it?",
+        )
+
+    vm = resolve_vm(tool_input, entry.vm_arg if entry else "")
+    if not vm:
+        log(f"{tool_name}: mutating but no VM in args {sorted(tool_input)}")
+        emit(
+            "prompt",
+            f"{server}/{tool} is a guest-mutating tool, but no VM could be read "
+            f"from its arguments (expected {entry.vm_arg if entry else '?'}). "
+            "No checkpoint was taken. Allow it?",
+        )
+
+    change_id = f"{payload.get('session_id', 'session')[:8]}-{payload.get('tool_use_id', '')[-8:]}"
+    action = f"{server}/{tool} on {vm}"
+    try:
+        guard = asyncio.run(asyncio.wait_for(run_guard(vm, change_id, action), GUARD_TIMEOUT_S))
+    except TimeoutError:
+        log(f"{tool_name}: guard timed out after {GUARD_TIMEOUT_S}s -> deny")
+        emit(
+            "deny",
+            f"Zerto did not confirm a tagged checkpoint for {vm} within "
+            f"{GUARD_TIMEOUT_S:.0f}s, so the change cannot be rewound. Refusing.",
+        )
+    except Exception as exc:
+        log(f"{tool_name}: guard raised {type(exc).__name__}: {exc} -> deny")
+        emit("deny", f"The Zerto guard failed for {vm}: {exc}. No checkpoint, so refusing.")
+
+    if guard.get("ok"):
+        tagged = (guard.get("tagged") or [{}])[0]
+        log(f"{tool_name}: tagged cp {tagged.get('checkpoint_id')} on {vm}, allowed")
+        emit(
+            None,
+            context=(
+                f"Zerto tagged checkpoint {tagged.get('checkpoint_id')} on VPG "
+                f"{tagged.get('vpg_name')} for {vm} before this call, tag "
+                f"{guard.get('tag')!r}. If this change breaks the guest, recover from "
+                "that checkpoint with zerto_recover_file."
+            ),
+        )
+
+    if not guard.get("protected", True):
+        log(f"{tool_name}: {vm} not protected -> prompt")
+        emit(
+            "prompt",
+            f"{vm} is not protected by Zerto, so this change cannot be rewound. "
+            f"({guard.get('message', '')}) Allow it anyway?",
+        )
+
+    log(f"{tool_name}: guard failed on {vm} -> deny :: {guard.get('message')}")
+    emit(
+        "deny",
+        f"Zerto could not tag a checkpoint for {vm}, so this change would not be "
+        f"recoverable: {guard.get('message')}",
+    )
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -41,3 +41,8 @@ ignore = ["ASYNC240"]
 asyncio_mode = "auto"
 testpaths = ["tests"]
 pythonpath = ["src"]
+
+[tool.ruff.lint.per-file-ignores]
+# A PreToolUse hook runs inside the host's tool call. If it raises, it breaks
+# the call it was meant to protect, so it swallows everything and logs instead.
+"hooks/*.py" = ["BLE001", "S110"]

tests/test_guard_hook.py

@@ -0,0 +1,70 @@
+"""The PreToolUse hook's pure helpers. The decision paths are exercised live."""
+
+import importlib.util
+from pathlib import Path
+
+import pytest
+
+spec = importlib.util.spec_from_file_location(
+    "zerto_guard_hook", Path(__file__).resolve().parent.parent / "hooks" / "zerto_guard_hook.py"
+)
+hook = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(hook)
+
+
+@pytest.mark.parametrize(
+    ("tool_name", "expected"),
+    [
+        ("mcp__ssh__exec", ("ssh", "exec")),
+        ("mcp__winrm__run_ps", ("winrm", "run_ps")),
+        # server names may themselves contain underscores; split on the FIRST __
+        ("mcp__plugin_my_thing__write_file", ("plugin_my_thing", "write_file")),
+        ("mcp__ob1__OB1__search_thoughts", ("ob1", "OB1__search_thoughts")),
+        ("Bash", None),
+        ("", None),
+    ],
+)
+def test_split_mcp_name(tool_name, expected):
+    assert hook.split_mcp_name(tool_name) == expected
+
+
+def test_resolve_vm_prefers_the_catalog_arg():
+    assert hook.resolve_vm({"host": "a", "computer_name": "b"}, "computer_name") == "b"
+
+
+def test_resolve_vm_falls_back_when_catalog_arg_is_wrong():
+    # a slightly wrong catalog entry should degrade to a prompt, not a crash
+    assert hook.resolve_vm({"computer_name": "web01"}, "host") == "web01"
+
+
+def test_resolve_vm_returns_empty_when_no_vm_present():
+    assert hook.resolve_vm({"command": "ls"}, "host") == ""
+
+
+def test_emit_with_no_decision_and_no_context_exits_silently(capsys):
+    with pytest.raises(SystemExit) as exc:
+        hook.emit(None)
+    assert exc.value.code == 0
+    assert capsys.readouterr().out == ""
+
+
+def test_emit_deny_shape(capsys):
+    import json
+
+    with pytest.raises(SystemExit) as exc:
+        hook.emit("deny", "no checkpoint")
+    assert exc.value.code == 0  # exit 2 would discard the JSON and lose the reason
+    out = json.loads(capsys.readouterr().out)["hookSpecificOutput"]
+    assert out["hookEventName"] == "PreToolUse"
+    assert out["permissionDecision"] == "deny"
+    assert out["permissionDecisionReason"] == "no checkpoint"
+
+
+def test_emit_context_without_decision(capsys):
+    import json
+
+    with pytest.raises(SystemExit):
+        hook.emit(None, context="checkpoint 7180")
+    out = json.loads(capsys.readouterr().out)["hookSpecificOutput"]
+    assert "permissionDecision" not in out
+    assert out["additionalContext"] == "checkpoint 7180"