Compare commits

..
6 Commits
Author SHA1 Message Date
justinandClaude Opus 5 3e7b777e53 demo: read the recovered content instead of a path
zerto_recover_file no longer writes the file to this host and hands back a
path, because a path here means nothing to a caller elsewhere and letting
the caller choose it was an arbitrary write. Both drivers still read
rec["path"], so they broke.

They now use the returned content. A small recovered_bytes() helper in each
driver decodes the text or base64 form, so the Windows copy-back keeps
shipping exact bytes rather than letting PowerShell rewrite line endings,
which is the bug that put a stray CR in an earlier take.

The Linux driver writes the bytes to a local file before scp, since scp
needs something on disk to send.

Verified against a live recovery: the helper returns 164 bytes whose sha256
matches the one the server reported, and the keys the on-screen show()
filter uses are all still present.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_016yVfC5nvZowoLFnEGWhLGn
2026-09-22 19:41:52 -04:00
justinandClaude Opus 5 e802d86196 demo: stop claiming backup is the only other copy
The narration said the changed file's "only other copy is in last night's
backup". That is false, and it gives away the argument the demo exists to
make: Zerto has a copy from seconds before the change. That is the whole
point.

d5 now draws the contrast instead of conceding it. Backup has last night,
hours old. Zerto has seconds before the change.

d2 had the same problem in a quieter form, asserting there was "no other
copy of that file anywhere" while Zerto was already protecting the machine.
Cut, since d5 carries the comparison.

Also fixes the pronunciation map that made the voice say "vem". The
replace map takes a phrase and a pronunciation, so {"VM": "vee em"} is
spoken as one run-together word. Expanded to "virtual machine".

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_016yVfC5nvZowoLFnEGWhLGn
2026-09-22 19:39:54 -04:00
justinandClaude Opus 5 2a35397e99 demo: commit the recording harness
The demo tooling only existed in a session scratch directory, which is
temporary. This puts it in the repo so the video can be rebuilt.

Terminal only: tmux drives a two pane session, asciinema records it, agg
renders it, ffmpeg encodes it. The left pane runs the loop against a live
ZVM, the right pane polls the VPG journal so the tagged checkpoint appears
on camera as it lands.

Narration is synthesised per beat and aligned to marks the driver writes
while it runs, rather than to predicted timings, so a slow API call or an
FLR mount that takes longer than usual does not drift the audio. Two
things that has to respect are written down in the README: agg's
idle-time-limit must exceed the longest pause or it compresses idle time
and breaks the wall-clock mapping, and each beat holds for its narration
length so no line is cut off.

demo_win.json carries live guest credentials, so only an example with
placeholders is committed and the real file is gitignored, along with the
generated wav, cast, gif and mp4.

The xAI key path and voice id come from the environment now instead of
being hardcoded to one machine.

Also records the narration gotchas that cost time: mapping an acronym to
run-together phonetics ({"VM": "vee em"}) is spoken as one word, "vem";
prose written for the page sounds robotic read aloud; and volumedetect
reports no samples when aimed at a file whose first stream is video,
which makes a working audio track look silent.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_016yVfC5nvZowoLFnEGWhLGn
2026-09-22 19:39:54 -04:00
claude bf3b8e1a05 fix(recover): return the file content, not a path on this host (#10) 2026-09-22 19:39:43 -04:00
claude 8baeceac32 feat(hooks): enforce the guard from a PreToolUse hook (#9) 2026-09-22 19:39:24 -04:00
claude 3d487e0850 docs: tagged checkpoints do work on cloud-protected VPGs (#11) 2026-09-22 19:39:06 -04:00
14 changed files with 581 additions and 27 deletions
+16 -1
View File
@@ -64,7 +64,22 @@ Git never had the file. RPO is the journal, not last night's backup.
vSphere ZVM 10.x and ZCA on AWS/Azure, same REST paths. HVM is out (separate swagger). Failover Live is not a tool. vSphere ZVM 10.x and ZCA on AWS/Azure, same REST paths. HVM is out (separate swagger). Failover Live is not a tool.
Tagged checkpoints cannot be inserted when the **protected** site is Azure or AWS (Zerto API). Point this server at the vSphere protected ZVM. Tagged checkpoints **do** work when the protected site is Azure or AWS. The 9.0 API
reference says they cannot be inserted; that is wrong on 10.9.10, where both were
accepted and the task reached `Completed`.
What differs is latency and granularity, both set by the **protected** site:
| protected at | journal gap | tag visible after |
|---|---|---|
| vSphere | 5s | ~4s |
| Azure | 60s | ~34s |
| AWS | 630s | ~128s |
The tagged checkpoint is also stamped about 30s *after* the insert request, so on a
cloud-protected VPG a prompt mutation can land *inside* the checkpoint meant to
precede it. Recover from the newest checkpoint that already existed when the guard
ran, not from the tag.
## Not this product ## Not this product
+17 -4
View File
@@ -16,6 +16,17 @@ CYAN, GREEN, RED, YEL, MAG = "\033[36m", "\033[32m", "\033[31m", "\033[33m", "\0
SPEED = float(os.environ.get("DEMO_SPEED", "1.0")) SPEED = float(os.environ.get("DEMO_SPEED", "1.0"))
def recovered_bytes(rec: dict) -> bytes:
"""zerto_recover_file returns the content now, not a path on the MCP host.
The server may be on another machine, so a path there is of no use here.
"""
import base64
if rec.get("encoding") == "base64":
return base64.b64decode(rec["content"])
return rec["content"].encode()
def w(s=""): def w(s=""):
sys.stdout.write(s + "\n"); sys.stdout.flush() sys.stdout.write(s + "\n"); sys.stdout.flush()
@@ -212,8 +223,8 @@ async def main():
if not rec.get("ok"): if not rec.get("ok"):
w(f"{RED} FLR failed{RST}"); return 1 w(f"{RED} FLR failed{RST}"); return 1
beat(1.5) beat(1.5)
w(f"{YEL} $ cat {rec['path']}{RST}") w(f"{YEL} recovered {rec['bytes']} bytes, sha256 {rec['sha256'][:16]}...{RST}")
body = Path(rec["path"]).read_text() body = recovered_bytes(rec).decode()
for line in body.splitlines(): for line in body.splitlines():
w(f"{GREEN}{BOLD} | {line}{RST}") w(f"{GREEN}{BOLD} | {line}{RST}")
beat(2.5) beat(2.5)
@@ -222,11 +233,13 @@ async def main():
step("Copy it back to the guest") step("Copy it back to the guest")
note("This part is scp, not Zerto. Zerto got the bytes back.") note("This part is scp, not Zerto. Zerto got the bytes back.")
target = f"{CFG['user']}@{CFG['host']}" target = f"{CFG['user']}@{CFG['host']}"
local_copy = SP / rec["name"]
local_copy.write_bytes(recovered_bytes(rec))
scp = (["sshpass", "-p", CFG["password"]] if CFG.get("password") else []) + \ scp = (["sshpass", "-p", CFG["password"]] if CFG.get("password") else []) + \
["scp", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null", ["scp", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null",
"-o", "LogLevel=ERROR"] + (["-i", CFG["key"]] if CFG.get("key") else []) + \ "-o", "LogLevel=ERROR"] + (["-i", CFG["key"]] if CFG.get("key") else []) + \
[rec["path"], f"{target}:{CFG['guest_path']}"] [str(local_copy), f"{target}:{CFG['guest_path']}"]
w(f"{YEL} $ scp {Path(rec['path']).name} {target}:{CFG['guest_path']}{RST}") w(f"{YEL} $ scp {rec['name']} {target}:{CFG['guest_path']}{RST}")
subprocess.run(scp, capture_output=True, text=True, timeout=60) subprocess.run(scp, capture_output=True, text=True, timeout=60)
guest(f"cat {CFG['guest_path']}") guest(f"cat {CFG['guest_path']}")
beat(1.5) beat(1.5)
+13 -2
View File
@@ -22,6 +22,17 @@ T0 = time.time()
STEP = [0] STEP = [0]
def recovered_bytes(rec: dict) -> bytes:
"""zerto_recover_file returns the content now, not a path on the MCP host.
The server may be on another machine, so a path there is of no use here.
"""
import base64
if rec.get("encoding") == "base64":
return base64.b64decode(rec["content"])
return rec["content"].encode()
def w(s=""): def w(s=""):
sys.stdout.write(s + "\n"); sys.stdout.flush() sys.stdout.write(s + "\n"); sys.stdout.flush()
@@ -155,7 +166,7 @@ async def main():
show(rec, keep=["ok", "bytes", "flr_path", "unmount"]) show(rec, keep=["ok", "bytes", "flr_path", "unmount"])
if not rec.get("ok"): if not rec.get("ok"):
w(f"{RED} FLR failed{RST}"); return 1 w(f"{RED} FLR failed{RST}"); return 1
body = Path(rec["path"]).read_text() body = recovered_bytes(rec).decode()
w(f"{YEL} recovered file:{RST}") w(f"{YEL} recovered file:{RST}")
for line in body.splitlines(): for line in body.splitlines():
w(f"{GREEN}{BOLD} | {line}{RST}") w(f"{GREEN}{BOLD} | {line}{RST}")
@@ -166,7 +177,7 @@ async def main():
# endings and appends one, which makes the restored file differ # endings and appends one, which makes the restored file differ
# from what Zerto handed back. WriteAllBytes does not. # from what Zerto handed back. WriteAllBytes does not.
import base64 import base64
blob = Path(rec["path"]).read_bytes() blob = recovered_bytes(rec)
b64 = base64.b64encode(blob).decode() b64 = base64.b64encode(blob).decode()
guest_ps( guest_ps(
f"[IO.File]::WriteAllBytes('{CFG['guest_path']}'," f"[IO.File]::WriteAllBytes('{CFG['guest_path']}',"
+93
View File
@@ -0,0 +1,93 @@
# 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 vSphere-protected VPG, but a **refusal took 63s**, because
`wait_for_tag` spends 45s before giving up.
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**. The deny path
works, but see the known issue below: that particular denial was wrong.
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.
## Known issue: false denials on cloud-protected VPGs
The `win2019-1` denial above was a **false negative**, and it is worth
understanding before relying on this hook in an estate with cloud-protected
workloads.
`wait_for_tag` gives up after a hardcoded 45s. That is generous for a
vSphere-protected VPG, which checkpoints every 5s and surfaces a tag in about
4s. It is far too short elsewhere: a tag takes ~34s to appear on an
Azure-protected VPG and ~128s on an AWS-protected one, because journal cadence
is set by the protected site (5s vSphere, 60s Azure, 630s AWS).
So the hook denied the change, and the checkpoint landed anyway. It is in the
journal as `cp 56`, timestamped a minute after the hook reported failure. The
guard told the agent there was no rewind point while Zerto was in the middle of
creating one.
That failure mode is worse than the one the hook guards against, because it is
silent and looks correct: legitimate work is refused on every cloud-protected VM
while the log reads like the guard is doing its job.
Until `wait_for_tag` becomes cadence-aware, scope this hook's matcher to
vSphere-protected workloads.
+17
View File
@@ -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
}
]
}
]
}
}
+247
View File
@@ -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()
+5
View File
@@ -41,3 +41,8 @@ ignore = ["ASYNC240"]
asyncio_mode = "auto" asyncio_mode = "auto"
testpaths = ["tests"] testpaths = ["tests"]
pythonpath = ["src"] 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"]
+4 -1
View File
@@ -70,7 +70,10 @@ so a stuck session blocks the next recovery. Find it with
## Facts that bite ## Facts that bite
- A tagged checkpoint is crash-consistent, not app-quiesced. - A tagged checkpoint is crash-consistent, not app-quiesced.
- Tagged checkpoints are not supported when the **protected** site is Azure or AWS. Talk to the vSphere protected ZVM. - Tagged checkpoints work on Azure and AWS protected VPGs, but they appear late:
~34s (Azure) and ~128s (AWS) versus ~4s on vSphere, and the checkpoint is stamped
about 30s after you ask for it. On those VPGs the tag can end up *after* your
change, so treat the newest checkpoint that already existed as the rewind point.
- 10.9 FLR Operator RBAC fails; Administrator is the documented workaround. - 10.9 FLR Operator RBAC fails; Administrator is the documented workaround.
- FLR cannot run during clone, test, live failover, or EJC. - FLR cannot run during clone, test, live failover, or EJC.
- Linux FLR: files >1.5GB are a bad idea; some characters in names are refused. - Linux FLR: files >1.5GB are a bad idea; some characters in names are refused.
+4 -1
View File
@@ -93,7 +93,10 @@ async def wait_for_tag(
raise ZertoError( raise ZertoError(
f"Tagged checkpoint {tag!r} did not appear on VPG {vpg_identifier} " f"Tagged checkpoint {tag!r} did not appear on VPG {vpg_identifier} "
f"within {timeout_s:.0f}s. Do not mutate. " f"within {timeout_s:.0f}s. Do not mutate. "
"If the protected site is Azure or AWS, tagged checkpoints are not supported." "On a cloud-protected VPG the tag routinely takes longer than this to appear "
"(measured ~34s on Azure, ~128s on AWS), so this timeout may simply be too "
"short rather than the insert having failed. Check the Zerto task before "
"assuming it did not land."
) )
+1 -2
View File
@@ -137,8 +137,7 @@ def find_from_rows(query: str, rows: list[dict[str, Any]]) -> FindResult:
outcome="none", outcome="none",
query=query, query=query,
message=( message=(
f"VM {vm.vm_name} ({vm.vm_identifier}) has no VPG. " f"VM {vm.vm_name} ({vm.vm_identifier}) has no VPG. Unprotected: refuse the change."
"Unprotected: refuse the change."
), ),
vm=vm, vm=vm,
) )
+52 -14
View File
@@ -2,6 +2,8 @@
from __future__ import annotations from __future__ import annotations
import base64
import hashlib
import json import json
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -32,6 +34,20 @@ _catalog: MutatingCatalog | None = None
_settings: dict[str, Any] = {} _settings: dict[str, Any] = {}
# Files above this are refused rather than streamed through a tool result.
# FLR is for a config file or a dropped directory; a disk image belongs on the
# whole-VM ladder in docs/recover-ladder.md.
MAX_RECOVER_BYTES = 1_048_576
def _encode_recovered(blob: bytes) -> dict[str, Any]:
"""Text where it is text, base64 otherwise, so the caller can just use it."""
try:
return {"encoding": "text", "content": blob.decode("utf-8")}
except UnicodeDecodeError:
return {"encoding": "base64", "content": base64.b64encode(blob).decode("ascii")}
def _dump(payload: Any) -> str: def _dump(payload: Any) -> str:
return json.dumps(payload, indent=2, default=str) return json.dumps(payload, indent=2, default=str)
@@ -133,8 +149,9 @@ async def zerto_create_tagged_checkpoint(
the Zerto API accepts, so this is the only place that context can live. the Zerto API accepts, so this is the only place that context can live.
Name format: ai:<agent> | <action> | vm=<vm> | change=<change_id> | <utc> Name format: ai:<agent> | <action> | vm=<vm> | change=<change_id> | <utc>
Docs: tagged checkpoints are not supported when the protected site is Azure or AWS; Works on Azure and AWS protected VPGs despite what the 9.0 API reference says,
run this against the vSphere protected ZVM. but the tag appears late there (~34s Azure, ~128s AWS) and is stamped after the
request, so it may sit after a prompt mutation.
""" """
try: try:
result = await _find(query) result = await _find(query)
@@ -446,16 +463,20 @@ async def zerto_recover_file(
checkpoint_identifier: str, checkpoint_identifier: str,
guest_path: str, guest_path: str,
confirmed: bool = False, confirmed: bool = False,
dest_dir: str | None = None,
) -> str: ) -> str:
"""File-level recovery from a journal checkpoint. The VM stays up. """File-level recovery from a journal checkpoint. The VM stays up.
Returns the file CONTENT, not a path. The caller may be on another machine,
so a path on this host is of no use to them, and letting a caller choose
where bytes land is an arbitrary write once this server is shared.
Requires confirmed=true (human yes). Cannot run during clone/test/live/EJC. Requires confirmed=true (human yes). Cannot run during clone/test/live/EJC.
10.9 FLR Operator role fails; use an Administrator account. 10.9 FLR Operator role fails; use an Administrator account.
Locally replicated VPGs only. FLR is performed at the VPG's recovery site, Locally replicated VPGs only. FLR is performed at the VPG's recovery site,
so a VPG replicating to a cloud ZCA must be recovered from that ZCA's API. so a VPG replicating to a cloud ZCA must be recovered from that ZCA's API.
guest_path is the path on the guest: /home/x/f.conf or C:\\Users\\x\\f.txt. guest_path is the path on the guest: /home/x/f.conf or C:\\Users\\x\\f.txt.
Files larger than max_recover_bytes are refused; use the whole-VM ladder.
""" """
if not confirmed: if not confirmed:
return _dump( return _dump(
@@ -471,7 +492,10 @@ async def zerto_recover_file(
gate = await _flr_site_gate(client, vpg_identifier) gate = await _flr_site_gate(client, vpg_identifier)
if not gate.get("ok"): if not gate.get("ok"):
return _dump(gate) return _dump(gate)
dest = Path(dest_dir or _settings.get("recovery_dir") or "./recovered") # Server-owned, from config only. This string is also handed to the
# appliance as initialDownloadPath, so a caller-supplied value would let a
# caller point the ZVM at a path of their choosing.
dest = Path(_settings.get("recovery_dir") or "./recovered")
dest.mkdir(parents=True, exist_ok=True) dest.mkdir(parents=True, exist_ok=True)
session_id: str | None = None session_id: str | None = None
before = await _live_session_ids(client) before = await _live_session_ids(client)
@@ -490,16 +514,30 @@ async def zerto_recover_file(
token = download_token_from(token_payload) token = download_token_from(token_payload)
blob = await client.fetch_download(token) blob = await client.fetch_download(token)
name = Path(guest_path.replace("\\", "/")).name or "recovered.bin" name = Path(guest_path.replace("\\", "/")).name or "recovered.bin"
out_path = dest / name limit = int(_settings.get("max_recover_bytes") or MAX_RECOVER_BYTES)
out_path.write_bytes(blob) if len(blob) > limit:
payload = { payload = {
"ok": True, "ok": False,
"path": str(out_path.resolve()), "too_large": True,
"bytes": len(blob), "bytes": len(blob),
"session_id": session_id, "limit": limit,
"flr_path": flr_path, "message": (
"message": f"Wrote {len(blob)} bytes to {out_path}", f"{name} is {len(blob)} bytes, over the {limit} byte limit for "
} "file level recovery. Use a bounded whole-VM operation instead: "
"offsite clone or failover test."
),
}
else:
payload = {
"ok": True,
"name": name,
"bytes": len(blob),
"sha256": hashlib.sha256(blob).hexdigest(),
"session_id": session_id,
"flr_path": flr_path,
**_encode_recovered(blob),
"message": f"Recovered {len(blob)} bytes of {name} from the journal.",
}
except ZertoError as exc: except ZertoError as exc:
payload = {"ok": False, "session_id": session_id, "message": str(exc)} payload = {"ok": False, "session_id": session_id, "message": str(exc)}
finally: finally:
+1 -2
View File
@@ -127,7 +127,6 @@ def can_tag(status: int | None, substatus: int | None) -> tuple[bool, str | None
return False, f"VPG status is {status_name(status)}; not MeetingSLA" return False, f"VPG status is {status_name(status)}; not MeetingSLA"
if substatus in SYNCING: if substatus in SYNCING:
return False, ( return False, (
f"VPG is {substatus_name(substatus)}; " f"VPG is {substatus_name(substatus)}; checkpoints are not durable until sync ends"
"checkpoints are not durable until sync ends"
) )
return True, None return True, None
+70
View File
@@ -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"
+41
View File
@@ -256,3 +256,44 @@ def test_flr_gate_refuses_remote_recovery_site_and_names_it():
# must tell the operator where the operation actually lives # must tell the operator where the operation actually lives
assert out["recovery_site"] == "aws-zca" assert out["recovery_site"] == "aws-zca"
assert "aws-zca" in out["message"] assert "aws-zca" in out["message"]
def test_encode_recovered_text_and_binary():
from zerto_rewind_mcp.server import _encode_recovered
text = _encode_recovered(b"listen: 0.0.0.0:8443\n")
assert text["encoding"] == "text"
assert text["content"] == "listen: 0.0.0.0:8443\n"
binary = _encode_recovered(b"\x89PNG\r\n\x1a\n\xff\xfe")
assert binary["encoding"] == "base64"
import base64 as b64
assert b64.b64decode(binary["content"]) == b"\x89PNG\r\n\x1a\n\xff\xfe"
def test_recover_file_takes_no_caller_destination():
"""The caller must not choose where bytes land, nor where the ZVM mounts.
dest_dir used to be a tool parameter whose value was also passed to the
appliance as initialDownloadPath.
"""
import inspect
from zerto_rewind_mcp.server import zerto_recover_file
params = set(inspect.signature(zerto_recover_file).parameters)
assert "dest_dir" not in params
assert params == {
"vpg_identifier",
"vm_identifier",
"checkpoint_identifier",
"guest_path",
"confirmed",
}
def test_recover_byte_cap_is_configurable_and_has_a_default():
from zerto_rewind_mcp.server import MAX_RECOVER_BYTES
assert MAX_RECOVER_BYTES == 1_048_576