feat(hooks): enforce the guard from a PreToolUse hook (#9)
This commit was merged in pull request #9.
This commit is contained in:
Executable
+247
@@ -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()
|
||||
Reference in New Issue
Block a user