geointel/scripts/release_backup_guard.py

#!/usr/bin/env python3
"""Verification guard shared by destructive GeoIntel operator commands."""

from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

from release_backup_snapshot import verify_backup as verify_byte_snapshots


@dataclass(frozen=True)
class VerifiedBackup:
    backup_dir: Path
    release_id: str
    created_at: datetime
    age_hours: float
    backup_tool_revision: str


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 _created_at(value: object) -> datetime:
    if not isinstance(value, str):
        raise RuntimeError("Backup manifest does not contain created_at")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc)


def verify_current_backup(
    backup_dir: str | Path,
    *,
    max_age_hours: float = 24.0,
    now: datetime | None = None,
) -> VerifiedBackup:
    """Verify checksums and release metadata without mutating the backup."""
    if max_age_hours <= 0:
        raise ValueError("max_age_hours must be greater than zero")

    root = Path(backup_dir).expanduser().resolve()
    if not root.is_dir():
        raise RuntimeError(f"Backup directory does not exist: {root}")

    required = {
        "manifest.json",
        "CHECKSUMS.sha256",
        "database.dump",
        "database.list",
        "database-metadata.tsv",
        "table-counts.tsv",
        "storage-manifest.tsv",
    }
    missing = sorted(name for name in required if not (root / name).is_file())
    if missing:
        raise RuntimeError(f"Backup is incomplete; missing: {', '.join(missing)}")
    if not (root / "storage-snapshot").is_dir():
        raise RuntimeError("Backup is incomplete; missing: storage-snapshot")

    checksum_lines = (root / "CHECKSUMS.sha256").read_text(encoding="utf-8").splitlines()
    checked: set[str] = set()
    for line in checksum_lines:
        if not line.strip():
            continue
        try:
            expected, name = line.split(maxsplit=1)
        except ValueError as exc:
            raise RuntimeError("Backup checksum file has an invalid line") from exc
        name = name.lstrip("*")
        if "/" in name or "\\" in name or name in {".", ".."}:
            raise RuntimeError(f"Backup checksum contains an unsafe path: {name}")
        target = root / name
        if not target.is_file():
            raise RuntimeError(f"Backup checksum target is missing: {name}")
        if _sha256(target) != expected.lower():
            raise RuntimeError(f"Backup checksum mismatch: {name}")
        checked.add(name)

    unchecked = sorted((required - {"CHECKSUMS.sha256"}) - checked)
    if unchecked:
        raise RuntimeError(f"Backup checksum coverage is incomplete: {', '.join(unchecked)}")

    manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
    if manifest.get("schema_version") != 1 or manifest.get("read_only_source") is not True:
        raise RuntimeError("Backup manifest schema or read-only marker is invalid")
    if manifest.get("database_password_secure") is not True:
        raise RuntimeError("Backup was made from an insecure database configuration")
    if manifest.get("inventory_mode") != "sha256" or manifest.get("storage_inventory_requested") is not True:
        raise RuntimeError("Destructive maintenance requires a SHA-256 storage inventory backup")
    if manifest.get("storage_snapshot_requested") is not True:
        raise RuntimeError("Destructive maintenance requires a byte-complete storage snapshot")
    verify_byte_snapshots(root)

    created = _created_at(manifest.get("created_at"))
    current = now or datetime.now(timezone.utc)
    if current.tzinfo is None:
        current = current.replace(tzinfo=timezone.utc)
    age_hours = (current.astimezone(timezone.utc) - created).total_seconds() / 3600
    if age_hours < -0.1:
        raise RuntimeError("Backup timestamp is in the future")
    if age_hours > max_age_hours:
        raise RuntimeError(
            f"Backup is {age_hours:.1f} hours old; maximum allowed age is {max_age_hours:.1f} hours"
        )

    release_id = manifest.get("release_id")
    backup_tool_revision = manifest.get("backup_tool_revision", manifest.get("git_commit"))
    if not isinstance(release_id, str) or not release_id:
        raise RuntimeError("Backup release id is missing")
    if not isinstance(backup_tool_revision, str) or len(backup_tool_revision) < 7:
        raise RuntimeError("Backup tool revision is missing")
    return VerifiedBackup(
        backup_dir=root,
        release_id=release_id,
        created_at=created,
        age_hours=age_hours,
        backup_tool_revision=backup_tool_revision,
    )


def require_confirmation(actual: str | None, expected: str) -> None:
    if actual != expected:
        raise RuntimeError(f"Refusing destructive maintenance; pass --confirm {expected}")