feat(agent99): bind host110 backup runtime deploy

scripts/backup/verify-host110-backup-runtime.py

@@ -0,0 +1,164 @@
+#!/usr/bin/env python3
+"""Independent no-write verifier for the Host110 backup runtime transaction."""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import fcntl
+import hashlib
+import json
+import os
+import re
+import sys
+import time
+from pathlib import Path
+
+
+EXPECTED_DESTINATION = Path("/backup/scripts")
+RUNTIME_LOCK = Path("/tmp/agent99-host110-backup-runtime.lock")
+EXPECTED_FILES = (
+    "common.sh",
+    "backup-all.sh",
+    "backup-host188-products.sh",
+    "verify-host188-products-backup.sh",
+    "verify-host188-products-archive.py",
+    "backup-gitea.sh",
+    "gitea-full-backup-restore-drill.sh",
+    "backup-configs.sh",
+    "backup-awoooi.sh",
+    "backup-awoooi-frequent.sh",
+    "backup-clawbot.sh",
+    "backup-sentry.sh",
+    "check-backup-integrity.sh",
+    "sync-offsite-backups.sh",
+    "backup-offsite-readiness-gate.sh",
+    "verify-offsite-full-sync.sh",
+    "enforce-latest-only-retention.sh",
+)
+
+
+def fail(reason: str) -> None:
+    print(f"HOST110_BACKUP_RUNTIME_VERIFY_OK=0\nERROR={reason}", file=sys.stderr)
+    raise SystemExit(1)
+
+
+def sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def acquire_runtime_read_lock(destination: Path):
+    if destination != Path("/backup/scripts") or os.environ.get("BACKUP_RUNTIME_DEPLOY_CONTEXT") == "1":
+        return None
+    if RUNTIME_LOCK.is_symlink() or not RUNTIME_LOCK.is_file():
+        fail("runtime_lock_unavailable")
+    handle = RUNTIME_LOCK.open("rb")
+    if RUNTIME_LOCK.stat().st_uid != os.getuid():
+        handle.close()
+        fail("runtime_lock_owner_invalid")
+    deadline = time.monotonic() + 60
+    while True:
+        try:
+            fcntl.flock(handle.fileno(), fcntl.LOCK_SH | fcntl.LOCK_NB)
+            return handle
+        except BlockingIOError:
+            if time.monotonic() >= deadline:
+                handle.close()
+                fail("runtime_deployment_active")
+            time.sleep(0.2)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    manifest_source = parser.add_mutually_exclusive_group(required=True)
+    manifest_source.add_argument("--manifest")
+    manifest_source.add_argument("--manifest-base64")
+    parser.add_argument("--destination", required=True)
+    parser.add_argument("--source-revision", required=True)
+    parser.add_argument("--run-id", required=True)
+    parser.add_argument("--expected-verifier-sha256", required=True)
+    args = parser.parse_args()
+
+    if not re.fullmatch(r"[0-9a-f]{40}", args.source_revision):
+        fail("invalid_source_revision")
+    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,95}", args.run_id):
+        fail("invalid_run_id")
+    destination = Path(args.destination)
+    if destination != EXPECTED_DESTINATION or destination.is_symlink() or not destination.is_dir():
+        fail("invalid_destination")
+    runtime_lock = acquire_runtime_read_lock(destination)
+
+    if not re.fullmatch(r"[0-9a-f]{64}", args.expected_verifier_sha256):
+        fail("invalid_verifier_identity")
+    try:
+        if args.manifest:
+            manifest_text = Path(args.manifest).read_text(encoding="utf-8")
+        else:
+            manifest_text = base64.b64decode(args.manifest_base64, validate=True).decode("utf-8")
+        document = json.loads(manifest_text)
+    except (OSError, UnicodeDecodeError, ValueError, json.JSONDecodeError):
+        fail("manifest_unavailable")
+    rows = document.get("files")
+    if (
+        document.get("schemaVersion") != "agent99_host110_backup_runtime_source_manifest_v1"
+        or document.get("sourceRevision") != args.source_revision
+        or document.get("runId") != args.run_id
+        or not re.fullmatch(r"[0-9a-f]{40}", str(document.get("sourceHead", "")))
+        or document.get("verifierSha256") != args.expected_verifier_sha256
+        or not isinstance(rows, list)
+        or len(rows) != len(EXPECTED_FILES)
+    ):
+        fail("manifest_contract_failed")
+    by_name: dict[str, str] = {}
+    for row in rows:
+        if not isinstance(row, dict):
+            fail("manifest_file_set_failed")
+        name = str(row.get("name", ""))
+        if name in by_name:
+            fail("manifest_file_set_failed")
+        by_name[name] = str(row.get("sha256", ""))
+    if set(by_name) != set(EXPECTED_FILES) or not all(re.fullmatch(r"[0-9a-f]{64}", value) for value in by_name.values()):
+        fail("manifest_file_set_failed")
+
+    verifier_path = Path(__file__)
+    self_identity_verified = verifier_path.is_file() and sha256(verifier_path) == args.expected_verifier_sha256
+    if args.manifest and not self_identity_verified:
+        fail("verifier_identity_failed")
+
+    root = destination.resolve(strict=True)
+    for name in EXPECTED_FILES:
+        path = destination / name
+        if path.is_symlink() or not path.is_file():
+            fail("runtime_file_type_failed")
+        resolved = path.resolve(strict=True)
+        if resolved.parent != root:
+            fail("runtime_file_boundary_failed")
+        stat = resolved.stat()
+        if stat.st_uid != os.getuid() or stat.st_gid != os.getgid() or stat.st_mode & 0o777 != 0o755:
+            fail("runtime_file_metadata_failed")
+        if sha256(resolved) != by_name[name]:
+            fail("runtime_file_identity_failed")
+
+    result = {
+        "schemaVersion": "agent99_host110_backup_runtime_verifier_v1",
+        "ok": True,
+        "sourceRevision": args.source_revision,
+        "runId": args.run_id,
+        "fileCount": len(EXPECTED_FILES),
+        "destination": str(EXPECTED_DESTINATION),
+        "remoteWritePerformed": False,
+        "secretValuesRead": False,
+        "selfIdentityVerified": self_identity_verified,
+        "verifierSha256": args.expected_verifier_sha256,
+    }
+    print(json.dumps(result, ensure_ascii=True, sort_keys=True))
+    if runtime_lock is not None:
+        runtime_lock.close()
+
+
+if __name__ == "__main__":
+    main()