awoooi/scripts/backup/verify-host188-products-archive.py

#!/usr/bin/env python3
"""Validate an isolated Host188 product backup restore without data writes."""

from __future__ import annotations

import argparse
import hashlib
import json
import sqlite3
import tarfile
import tempfile
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"}


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:
    parser = argparse.ArgumentParser()
    parser.add_argument("--root", type=Path, required=True)
    args = parser.parse_args()
    print(json.dumps(validate_restore(args.root), separators=(",", ":")))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())