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

#!/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")
EXPECTED_EXPORTER_DESTINATION = Path("/home/wooo/scripts/backup-health-textfile-exporter.py")
RUNTIME_LOCK = Path("/tmp/agent99-host110-backup-runtime.lock")
EXPECTED_BACKUP_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",
)
EXPORTER_FILE = "backup-health-textfile-exporter.py"
EXPECTED_FILES = EXPECTED_BACKUP_FILES + (EXPORTER_FILE,)


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("--exporter-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")
    exporter_destination = Path(args.exporter_destination)
    if exporter_destination != EXPECTED_EXPORTER_DESTINATION:
        fail("invalid_exporter_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 = exporter_destination if name == EXPORTER_FILE else destination / name
        if path.is_symlink() or not path.is_file():
            fail("runtime_file_type_failed")
        resolved = path.resolve(strict=True)
        if name == EXPORTER_FILE:
            if resolved != exporter_destination:
                fail("runtime_file_boundary_failed")
        elif 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),
        "backupScriptFileCount": len(EXPECTED_BACKUP_FILES),
        "backupHealthExporterIncluded": True,
        "destination": str(EXPECTED_DESTINATION),
        "exporterDestination": str(EXPECTED_EXPORTER_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()