362 lines
12 KiB
Python
362 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Claude Stop hook that completes a pending operator-triggered promotion."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import fcntl
|
|
import json
|
|
import os
|
|
import secrets
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from collections.abc import Callable, Mapping
|
|
from pathlib import Path
|
|
from typing import Final, NamedTuple, TextIO
|
|
|
|
MAX_FRAME: Final = 64 * 1024
|
|
PROMOTER_TIMEOUT_SECONDS: Final = 10.0
|
|
LEASE_TTL_SECONDS: Final = 60 * 60
|
|
PROMOTER: Final = Path(__file__).resolve().with_name("lease_promote.py")
|
|
PENDING_DIRECTORY: Final = "mosaic-lease"
|
|
LOCK_FILE: Final = "promotion.lock"
|
|
RESULT_FILE: Final = "last-result.json"
|
|
TERMINAL_FAILURE_CODES: Final = frozenset(
|
|
{
|
|
"RECEIPT_REPLAY",
|
|
"RECEIPT_MISMATCH",
|
|
"INVALID_LEASE_TRANSITION",
|
|
"PROMOTION_TOKEN_INVALID",
|
|
}
|
|
)
|
|
|
|
|
|
class PendingChallenge(NamedTuple):
|
|
value: str
|
|
device: int
|
|
inode: int
|
|
|
|
|
|
def reject_duplicate_json_keys(pairs: list[tuple[str, object]]) -> dict[str, object]:
|
|
value: dict[str, object] = {}
|
|
for key, item in pairs:
|
|
if key in value:
|
|
raise ValueError("duplicate promoter JSON key")
|
|
value[key] = item
|
|
return value
|
|
|
|
|
|
def session_pending_name(environ: Mapping[str, str]) -> tuple[Path, str]:
|
|
runtime_dir = Path(environ["XDG_RUNTIME_DIR"])
|
|
session_id = environ["MOSAIC_LEASE_SESSION_ID"]
|
|
if not runtime_dir.is_absolute():
|
|
raise ValueError("XDG_RUNTIME_DIR must be absolute")
|
|
if len(session_id) != 64 or any(character not in "0123456789abcdef" for character in session_id):
|
|
raise ValueError("invalid lease session id")
|
|
return runtime_dir, f"pending-{session_id}"
|
|
|
|
|
|
def open_pending_directory(runtime_dir: Path) -> int | None:
|
|
directory_flags = (
|
|
os.O_RDONLY
|
|
| getattr(os, "O_CLOEXEC", 0)
|
|
| getattr(os, "O_DIRECTORY", 0)
|
|
| getattr(os, "O_NOFOLLOW", 0)
|
|
)
|
|
try:
|
|
runtime_descriptor = os.open(runtime_dir, directory_flags)
|
|
except FileNotFoundError:
|
|
return None
|
|
try:
|
|
runtime_metadata = os.fstat(runtime_descriptor)
|
|
if (
|
|
not stat.S_ISDIR(runtime_metadata.st_mode)
|
|
or runtime_metadata.st_uid != os.getuid()
|
|
or stat.S_IMODE(runtime_metadata.st_mode) != 0o700
|
|
):
|
|
raise ValueError("unsafe XDG runtime directory")
|
|
try:
|
|
descriptor = os.open(PENDING_DIRECTORY, directory_flags, dir_fd=runtime_descriptor)
|
|
except FileNotFoundError:
|
|
return None
|
|
finally:
|
|
os.close(runtime_descriptor)
|
|
|
|
metadata = os.fstat(descriptor)
|
|
if (
|
|
not stat.S_ISDIR(metadata.st_mode)
|
|
or metadata.st_uid != os.getuid()
|
|
or stat.S_IMODE(metadata.st_mode) != 0o700
|
|
):
|
|
os.close(descriptor)
|
|
raise ValueError("unsafe promotion pending directory")
|
|
return descriptor
|
|
|
|
|
|
def acquire_lock(directory_descriptor: int) -> int:
|
|
flags = (
|
|
os.O_RDWR
|
|
| os.O_CREAT
|
|
| getattr(os, "O_CLOEXEC", 0)
|
|
| getattr(os, "O_NOFOLLOW", 0)
|
|
)
|
|
descriptor = os.open(LOCK_FILE, flags, 0o600, dir_fd=directory_descriptor)
|
|
metadata = os.fstat(descriptor)
|
|
if (
|
|
not stat.S_ISREG(metadata.st_mode)
|
|
or metadata.st_uid != os.getuid()
|
|
or stat.S_IMODE(metadata.st_mode) != 0o600
|
|
):
|
|
os.close(descriptor)
|
|
raise ValueError("unsafe promotion lock file")
|
|
try:
|
|
fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError:
|
|
os.close(descriptor)
|
|
raise
|
|
return descriptor
|
|
|
|
|
|
def read_pending(directory_descriptor: int, name: str) -> PendingChallenge | None:
|
|
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
|
|
try:
|
|
descriptor = os.open(name, flags, dir_fd=directory_descriptor)
|
|
except FileNotFoundError:
|
|
return None
|
|
try:
|
|
metadata = os.fstat(descriptor)
|
|
if (
|
|
not stat.S_ISREG(metadata.st_mode)
|
|
or metadata.st_uid != os.getuid()
|
|
or stat.S_IMODE(metadata.st_mode) != 0o600
|
|
or metadata.st_size <= 0
|
|
or metadata.st_size > MAX_FRAME
|
|
):
|
|
raise ValueError("unsafe promotion pending file")
|
|
raw = os.read(descriptor, MAX_FRAME + 1)
|
|
finally:
|
|
os.close(descriptor)
|
|
if len(raw) > MAX_FRAME:
|
|
raise ValueError("oversized promotion challenge")
|
|
challenge = raw.decode("utf-8")
|
|
if (
|
|
len(challenge) != 64
|
|
or any(character not in "0123456789abcdef" for character in challenge)
|
|
):
|
|
raise ValueError("invalid promotion challenge")
|
|
return PendingChallenge(challenge, metadata.st_dev, metadata.st_ino)
|
|
|
|
|
|
def write_result(
|
|
directory_descriptor: int,
|
|
attempt_id: str,
|
|
verified: bool,
|
|
reason: str | None,
|
|
session_id: str,
|
|
wall_clock: float,
|
|
) -> None:
|
|
result = {
|
|
"attempt_id": attempt_id,
|
|
"expires_at_wallclock": wall_clock + LEASE_TTL_SECONDS if verified else None,
|
|
"reason": reason,
|
|
"session_id": session_id,
|
|
"ts": wall_clock,
|
|
"verified": verified,
|
|
}
|
|
temporary = f".{RESULT_FILE}.tmp-{secrets.token_hex(8)}"
|
|
flags = (
|
|
os.O_WRONLY
|
|
| os.O_CREAT
|
|
| os.O_EXCL
|
|
| getattr(os, "O_CLOEXEC", 0)
|
|
| getattr(os, "O_NOFOLLOW", 0)
|
|
)
|
|
descriptor = os.open(temporary, flags, 0o600, dir_fd=directory_descriptor)
|
|
try:
|
|
os.fchmod(descriptor, 0o600)
|
|
with os.fdopen(descriptor, "w", encoding="utf-8", closefd=False) as stream:
|
|
json.dump(result, stream, separators=(",", ":"), sort_keys=True)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(
|
|
temporary,
|
|
RESULT_FILE,
|
|
src_dir_fd=directory_descriptor,
|
|
dst_dir_fd=directory_descriptor,
|
|
)
|
|
os.fsync(directory_descriptor)
|
|
except Exception:
|
|
try:
|
|
os.unlink(temporary, dir_fd=directory_descriptor)
|
|
except FileNotFoundError:
|
|
pass
|
|
raise
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
|
|
def delete_pending_if_unchanged(
|
|
directory_descriptor: int,
|
|
name: str,
|
|
pending: PendingChallenge,
|
|
error_stream: TextIO,
|
|
) -> None:
|
|
quarantine = f".{name}.delete-{secrets.token_hex(8)}"
|
|
try:
|
|
os.rename(
|
|
name,
|
|
quarantine,
|
|
src_dir_fd=directory_descriptor,
|
|
dst_dir_fd=directory_descriptor,
|
|
)
|
|
except FileNotFoundError:
|
|
return
|
|
except OSError as error:
|
|
print(f"Mosaic promotion could not quarantine pending file: {error}", file=error_stream)
|
|
return
|
|
|
|
try:
|
|
moved = os.stat(
|
|
quarantine,
|
|
dir_fd=directory_descriptor,
|
|
follow_symlinks=False,
|
|
)
|
|
if (moved.st_dev, moved.st_ino) == (pending.device, pending.inode):
|
|
os.unlink(quarantine, dir_fd=directory_descriptor)
|
|
os.fsync(directory_descriptor)
|
|
return
|
|
|
|
print("Mosaic promotion pending file changed; preserving replacement.", file=error_stream)
|
|
try:
|
|
os.link(
|
|
quarantine,
|
|
name,
|
|
src_dir_fd=directory_descriptor,
|
|
dst_dir_fd=directory_descriptor,
|
|
follow_symlinks=False,
|
|
)
|
|
except FileExistsError:
|
|
print(
|
|
f"Mosaic promotion preserved replacement as {quarantine}.",
|
|
file=error_stream,
|
|
)
|
|
else:
|
|
os.unlink(quarantine, dir_fd=directory_descriptor)
|
|
os.fsync(directory_descriptor)
|
|
except OSError as error:
|
|
print(f"Mosaic promotion could not resolve pending file: {error}", file=error_stream)
|
|
|
|
|
|
def parse_reply(completed: subprocess.CompletedProcess[str]) -> dict[str, object] | None:
|
|
if completed.returncode != 0:
|
|
return None
|
|
try:
|
|
value = json.loads(
|
|
completed.stdout,
|
|
object_pairs_hook=reject_duplicate_json_keys,
|
|
)
|
|
except (json.JSONDecodeError, RecursionError, TypeError, ValueError):
|
|
return None
|
|
if not isinstance(value, dict):
|
|
return None
|
|
if set(value) == {"stage", "ok", "state"}:
|
|
if (
|
|
value.get("stage") == "promote_lease"
|
|
and value.get("ok") is True
|
|
and value.get("state") == "VERIFIED"
|
|
):
|
|
return value
|
|
return None
|
|
if set(value) == {"stage", "ok", "code"}:
|
|
if (
|
|
value.get("stage") in {"observe_receipt", "promote_lease"}
|
|
and value.get("ok") is False
|
|
and isinstance(value.get("code"), str)
|
|
and value.get("code")
|
|
):
|
|
return value
|
|
return None
|
|
|
|
|
|
def main(
|
|
*,
|
|
environ: Mapping[str, str] | None = None,
|
|
stderr: TextIO | None = None,
|
|
run: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
|
|
now: Callable[[], float] = time.time,
|
|
) -> int:
|
|
source_environment = os.environ if environ is None else environ
|
|
error_stream = sys.stderr if stderr is None else stderr
|
|
directory_descriptor: int | None = None
|
|
lock_descriptor: int | None = None
|
|
|
|
try:
|
|
runtime_dir, pending_name = session_pending_name(source_environment)
|
|
session_id = source_environment["MOSAIC_LEASE_SESSION_ID"]
|
|
directory_descriptor = open_pending_directory(runtime_dir)
|
|
if directory_descriptor is None:
|
|
return 0
|
|
try:
|
|
lock_descriptor = acquire_lock(directory_descriptor)
|
|
except (BlockingIOError, FileNotFoundError):
|
|
print("Mosaic promotion completion deferred: promotion is in progress.", file=error_stream)
|
|
return 0
|
|
pending = read_pending(directory_descriptor, pending_name)
|
|
if pending is None:
|
|
return 0
|
|
completed = run(
|
|
[
|
|
sys.executable,
|
|
"-I",
|
|
"-S",
|
|
"-B",
|
|
str(PROMOTER),
|
|
"--complete",
|
|
pending.value,
|
|
],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
env=dict(source_environment),
|
|
timeout=PROMOTER_TIMEOUT_SECONDS,
|
|
)
|
|
reply = parse_reply(completed)
|
|
if reply is not None and reply.get("ok") is True:
|
|
write_result(directory_descriptor, pending.value, True, None, session_id, now())
|
|
delete_pending_if_unchanged(
|
|
directory_descriptor,
|
|
pending_name,
|
|
pending,
|
|
error_stream,
|
|
)
|
|
print("Mosaic lease promotion completed.", file=error_stream)
|
|
return 0
|
|
|
|
if reply is not None:
|
|
code = str(reply["code"])
|
|
print(f"Mosaic promotion incomplete: {code}.", file=error_stream)
|
|
if code in TERMINAL_FAILURE_CODES:
|
|
write_result(directory_descriptor, pending.value, False, code, session_id, now())
|
|
delete_pending_if_unchanged(
|
|
directory_descriptor,
|
|
pending_name,
|
|
pending,
|
|
error_stream,
|
|
)
|
|
else:
|
|
diagnostic = completed.stderr.strip() or f"promoter exit {completed.returncode}"
|
|
print(f"Mosaic promotion retryable failure: {diagnostic}.", file=error_stream)
|
|
except (KeyError, OSError, RecursionError, UnicodeError, ValueError, subprocess.SubprocessError) as error:
|
|
print(f"Mosaic promotion completion deferred: {type(error).__name__}: {error}", file=error_stream)
|
|
finally:
|
|
if lock_descriptor is not None:
|
|
os.close(lock_descriptor)
|
|
if directory_descriptor is not None:
|
|
os.close(directory_descriptor)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|