#!/usr/bin/env python3 """Validate an isolated Host188 product backup restore without data writes.""" from __future__ import annotations import argparse import fcntl import hashlib import json import os import sqlite3 import tarfile import tempfile import time from pathlib import Path, PurePosixPath from typing import Any EXPECTED_PREFIXES = { "vibework-postgres-": ".dump", "fifa-postgres-": ".dump", "fifa-redis-": ".rdb", "n8n-online-": ".tar.gz", } FIXED_ASSETS = {"n8n-validation.json"} RUNTIME_LOCK = Path("/tmp/agent99-host110-backup-runtime.lock") def acquire_runtime_lock(): if Path(__file__).resolve().parent != Path("/backup/scripts"): return None if RUNTIME_LOCK.is_symlink(): raise RuntimeError("backup_runtime_gate_unsafe") inherited_fd_path: Path | None = None proc_fd_root = Path(f"/proc/{os.getpid()}/fd") if proc_fd_root.is_dir(): candidate = proc_fd_root / "197" if candidate.exists() or candidate.is_symlink(): inherited_fd_path = candidate else: candidate = Path("/dev/fd/197") if candidate.exists() or candidate.is_symlink(): inherited_fd_path = candidate if inherited_fd_path is None: handle = RUNTIME_LOCK.open("a+") else: if not RUNTIME_LOCK.is_file(): raise RuntimeError("backup_runtime_gate_unavailable") if proc_fd_root.is_dir(): try: inherited_target = inherited_fd_path.resolve(strict=True) expected_target = RUNTIME_LOCK.resolve(strict=True) except OSError as exc: raise RuntimeError("backup_runtime_inherited_fd_unresolved") from exc else: try: raw_target = fcntl.fcntl(197, getattr(fcntl, "F_GETPATH", 50), b"\0" * 1024) inherited_target = Path(os.fsdecode(raw_target.split(b"\0", 1)[0])).resolve(strict=True) expected_target = RUNTIME_LOCK.resolve(strict=True) except OSError as exc: raise RuntimeError("backup_runtime_inherited_fd_unresolved") from exc if inherited_target != expected_target: raise RuntimeError("backup_runtime_inherited_fd_mismatch") try: handle = os.fdopen(os.dup(197), "a+") except OSError as exc: raise RuntimeError("backup_runtime_inherited_fd_unavailable") from exc if RUNTIME_LOCK.stat().st_uid != os.getuid(): handle.close() raise RuntimeError("backup_runtime_gate_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() raise RuntimeError("backup_runtime_deployment_active") time.sleep(0.2) def sha256_file(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 read_magic(path: Path) -> bytes: with path.open("rb") as handle: return handle.read(5) def _single_prefixed_asset(names: set[str], prefix: str, suffix: str) -> str: matches = sorted(name for name in names if name.startswith(prefix) and name.endswith(suffix)) if len(matches) != 1: raise ValueError(f"asset_cardinality_invalid:{prefix}") return matches[0] def _extract_n8n_archive(archive_path: Path, target: Path) -> None: with tarfile.open(archive_path, "r:gz") as archive: members = archive.getmembers() for member in members: path = PurePosixPath(member.name) if path.is_absolute() or ".." in path.parts: raise ValueError("n8n_archive_unsafe_path") if not (member.isfile() or member.isdir()): raise ValueError("n8n_archive_unsafe_member") archive.extractall(target, members=members) def validate_restore(root: Path) -> dict[str, Any]: manifest_path = root / "backup-manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) if manifest.get("schemaVersion") != "host188_product_backup_manifest_v1": raise ValueError("manifest_schema_invalid") if manifest.get("sourceHost") != "192.168.0.188": raise ValueError("manifest_source_invalid") if manifest.get("controllerHost") != "192.168.0.110": raise ValueError("manifest_controller_invalid") if manifest.get("credentialsDecrypted") is not False: raise ValueError("credential_boundary_invalid") if manifest.get("serviceStopped") is not False: raise ValueError("service_stop_boundary_invalid") if manifest.get("offsiteWritePerformed") is not False: raise ValueError("offsite_boundary_invalid") if manifest.get("recoveryReadinessClaimed") is not False: raise ValueError("recovery_readiness_false_green") if manifest.get("n8nEncryptionKeyEscrowRequired") is not True: raise ValueError("n8n_escrow_requirement_missing") assets = manifest.get("assets") if not isinstance(assets, list): raise ValueError("manifest_assets_invalid") by_name: dict[str, dict[str, Any]] = {} for item in assets: if not isinstance(item, dict): raise ValueError("manifest_asset_invalid") name = str(item.get("name") or "") if not name or name in by_name or Path(name).name != name: raise ValueError("manifest_asset_name_invalid") by_name[name] = item names = set(by_name) if not FIXED_ASSETS.issubset(names): raise ValueError("fixed_asset_missing") selected = { prefix: _single_prefixed_asset(names, prefix, suffix) for prefix, suffix in EXPECTED_PREFIXES.items() } expected_names = set(selected.values()) | FIXED_ASSETS if names != expected_names: raise ValueError("unexpected_manifest_asset") for name, item in by_name.items(): path = root / name if not path.is_file(): raise ValueError(f"asset_missing:{name}") if path.stat().st_size != int(item.get("sizeBytes") or -1): raise ValueError(f"asset_size_mismatch:{name}") if sha256_file(path) != str(item.get("sha256") or ""): raise ValueError(f"asset_hash_mismatch:{name}") for prefix in ("vibework-postgres-", "fifa-postgres-"): if read_magic(root / selected[prefix]) != b"PGDMP": raise ValueError(f"postgres_magic_invalid:{prefix}") if read_magic(root / selected["fifa-redis-"]) != b"REDIS": raise ValueError("redis_magic_invalid") validation = json.loads((root / "n8n-validation.json").read_text(encoding="utf-8")) if validation.get("sqliteIntegrity") != "ok": raise ValueError("n8n_prior_validation_invalid") if validation.get("credentialsDecrypted") is not False: raise ValueError("n8n_prior_credential_boundary_invalid") with tempfile.TemporaryDirectory( prefix="host188-n8n-restore-", dir=root.parent ) as temp_dir: extracted = Path(temp_dir) _extract_n8n_archive(root / selected["n8n-online-"], extracted) database = extracted / "database.sqlite" with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection: integrity = connection.execute("PRAGMA integrity_check").fetchone()[0] if integrity != "ok": raise ValueError("n8n_sqlite_integrity_invalid") counts: dict[str, int] = {} for category in ("workflows", "credentials"): directory = extracted / category rows = sorted(directory.rglob("*.json")) if directory.is_dir() else [] for path in rows: json.loads(path.read_text(encoding="utf-8")) counts[category] = len(rows) if counts["workflows"] != int(validation.get("workflowFiles") or 0): raise ValueError("n8n_workflow_count_mismatch") if counts["credentials"] != int(validation.get("credentialFiles") or 0): raise ValueError("n8n_credential_count_mismatch") return { "schemaVersion": "host188_product_restore_validation_v1", "dataVerified": True, "assetCount": len(by_name), "postgresArchiveCount": 2, "redisArchiveCount": 1, "n8nSqliteIntegrity": "ok", "credentialsDecrypted": False, "recoveryReadinessClaimed": False, } def main() -> int: runtime_lock = acquire_runtime_lock() parser = argparse.ArgumentParser() parser.add_argument("--root", type=Path, required=True) args = parser.parse_args() print(json.dumps(validate_restore(args.root), separators=(",", ":"))) if runtime_lock is not None: runtime_lock.close() return 0 if __name__ == "__main__": raise SystemExit(main())