scripts/secure_backup.py

#!/usr/bin/env python3
"""Secure, deterministic backup and staged restore for Hermes persistent state.

Only explicitly approved state is copied. SQLite is captured through its online
backup API, every member is checksummed, archives are extracted without
``extractall``, and restore replaces managed roots so deleted files cannot
silently reappear.
"""
from __future__ import annotations

import hashlib
import json
import os
import shutil
import sqlite3
import tarfile
import tempfile
import time
from pathlib import Path, PurePosixPath
from typing import Any, Iterable

try:
    from backup_allowlist import (
        ALLOWED_DIR_NAMES,
        ALLOWED_FILE_NAMES,
        BackupAllowlistError,
        is_path_allowed,
        scan_file_for_secrets,
    )
except ImportError:  # pragma: no cover - package-style import
    from scripts.backup_allowlist import (
        ALLOWED_DIR_NAMES,
        ALLOWED_FILE_NAMES,
        BackupAllowlistError,
        is_path_allowed,
        scan_file_for_secrets,
    )

FORMAT_VERSION = 1
MANIFEST_NAME = "backup_manifest.json"
DATABASE_NAME = "hermes_domain.sqlite3"
MAX_ARCHIVE_FILES = 100_000
MAX_ARCHIVE_BYTES = 4 * 1024 * 1024 * 1024


class BackupIntegrityError(RuntimeError):
    pass


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 _table_exists(conn: sqlite3.Connection, table: str) -> bool:
    row = conn.execute(
        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,)
    ).fetchone()
    return row is not None


def _sanitize_sqlite_snapshot(conn: sqlite3.Connection) -> None:
    """Remove runtime credentials/session material from a backup copy.

    The live database is never modified.  Session token hashes are still
    authentication material, and exchange credential references can reveal
    secret-store structure, so both are excluded from portable backups.  The
    snapshot is vacuumed after deletion so removed rows do not remain in free
    pages inside the archive.
    """
    if _table_exists(conn, "sessions"):
        conn.execute("DELETE FROM sessions")
    if _table_exists(conn, "exchange_accounts"):
        columns = {
            str(row[1]) for row in conn.execute("PRAGMA table_info(exchange_accounts)")
        }
        assignments = []
        if "credential_ref" in columns:
            assignments.append("credential_ref=NULL")
        if "enabled" in columns:
            assignments.append("enabled=0")
        if "close_only" in columns:
            assignments.append("close_only=1")
        if assignments:
            conn.execute(f"UPDATE exchange_accounts SET {','.join(assignments)}")
    conn.commit()
    conn.execute("VACUUM")


def _copy_sqlite_consistently(source: Path, destination: Path) -> None:
    destination.parent.mkdir(parents=True, exist_ok=True)
    src = sqlite3.connect(f"file:{source}?mode=ro", uri=True, timeout=30)
    dst = sqlite3.connect(str(destination), timeout=30)
    try:
        src.backup(dst)
        _sanitize_sqlite_snapshot(dst)
        issues = dst.execute("PRAGMA integrity_check").fetchall()
        if issues != [("ok",)]:
            raise BackupIntegrityError(f"sqlite integrity check failed: {issues[:3]}")
    finally:
        dst.close()
        src.close()


def _iter_approved_files(source: Path) -> Iterable[tuple[Path, Path]]:
    """Yield allowlisted regular files and fail closed on every source symlink."""
    for root_file in sorted(ALLOWED_FILE_NAMES - {DATABASE_NAME}):
        path = source / root_file
        if path.is_symlink():
            raise BackupAllowlistError(f"Refusing backup symlink: {root_file}")
        if path.is_file():
            yield path, Path(root_file)

    for dirname in sorted(ALLOWED_DIR_NAMES):
        root = source / dirname
        if root.is_symlink():
            raise BackupAllowlistError(f"Refusing backup symlink: {dirname}")
        if not root.is_dir():
            continue
        for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
            current = Path(dirpath)
            for name in sorted(dirnames):
                candidate = current / name
                if candidate.is_symlink():
                    relative = candidate.relative_to(source).as_posix()
                    raise BackupAllowlistError(f"Refusing backup symlink: {relative}")
            for name in sorted(filenames):
                path = current / name
                relative = path.relative_to(source)
                if path.is_symlink():
                    raise BackupAllowlistError(
                        f"Refusing backup symlink: {relative.as_posix()}"
                    )
                resolved = path.resolve()
                if source != resolved and source not in resolved.parents:
                    raise BackupAllowlistError(
                        f"Refusing backup path escape: {relative.as_posix()}"
                    )
                if path.is_file() and is_path_allowed(relative.as_posix()):
                    yield path, relative


def build_backup(source_dir: Path, archive_path: Path) -> dict[str, Any]:
    source = source_dir.resolve()
    archive_path.parent.mkdir(parents=True, exist_ok=True)
    if not source.is_dir():
        raise FileNotFoundError(source)

    with tempfile.TemporaryDirectory(prefix="hermes-backup-") as temp:
        stage = Path(temp) / "state"
        stage.mkdir()
        files: list[dict[str, Any]] = []
        managed_roots = sorted(set(ALLOWED_DIR_NAMES) | set(ALLOWED_FILE_NAMES) | {DATABASE_NAME})

        rejected: list[dict[str, str]] = []
        for src, relative in _iter_approved_files(source):
            reason = scan_file_for_secrets(src)
            if reason:
                # Exclude just this file rather than aborting the whole backup.
                # A single vendored/bundled file (e.g. a skill doc containing
                # an example "api_key: ..." placeholder) must not block
                # persistence of unrelated real user state (memories, plans,
                # cron). The file is still refused -- never copied into the
                # archive -- so no secret-like content actually leaves the
                # host either way.
                rejected.append({"path": relative.as_posix(), "reason": reason})
                continue
            dst = stage / relative
            dst.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(src, dst)

        database = source / DATABASE_NAME
        if database.is_symlink():
            raise BackupAllowlistError(f"Refusing backup symlink: {DATABASE_NAME}")
        if database.is_file():
            _copy_sqlite_consistently(database, stage / DATABASE_NAME)

        for path in sorted(stage.rglob("*")):
            if path.is_file():
                relative = path.relative_to(stage).as_posix()
                files.append({"path": relative, "size": path.stat().st_size, "sha256": _sha256(path)})

        manifest = {
            "format_version": FORMAT_VERSION,
            "created_at_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "managed_roots": managed_roots,
            "files": files,
            "rejected_files": rejected,
        }
        (stage / MANIFEST_NAME).write_text(
            json.dumps(manifest, sort_keys=True, separators=(",", ":")), encoding="utf-8"
        )

        temporary_archive = archive_path.with_suffix(archive_path.suffix + ".tmp")
        with tarfile.open(temporary_archive, "w:gz", format=tarfile.PAX_FORMAT) as tar:
            for path in sorted(stage.rglob("*")):
                relative = path.relative_to(stage).as_posix()
                info = tar.gettarinfo(str(path), arcname=relative)
                info.uid = info.gid = 0
                info.uname = info.gname = ""
                info.mtime = 0
                if path.is_file():
                    with path.open("rb") as handle:
                        tar.addfile(info, handle)
                elif path.is_dir():
                    tar.addfile(info)
        os.replace(temporary_archive, archive_path)
        return {**manifest, "archive_sha256": _sha256(archive_path), "archive": str(archive_path)}


def _safe_member_path(root: Path, name: str) -> Path:
    pure = PurePosixPath(name)
    if pure.is_absolute() or not pure.parts or any(part in {"", ".", ".."} for part in pure.parts):
        raise BackupIntegrityError(f"unsafe archive path: {name!r}")
    target = (root / Path(*pure.parts)).resolve()
    if target != root and root not in target.parents:
        raise BackupIntegrityError(f"archive member escapes destination: {name!r}")
    return target


def extract_validated(archive_path: Path, destination: Path) -> dict[str, Any]:
    destination.mkdir(parents=True, exist_ok=True)
    root = destination.resolve()
    file_count = 0
    total_bytes = 0
    with tarfile.open(archive_path, "r:*") as tar:
        for member in tar:
            target = _safe_member_path(root, member.name)
            if member.issym() or member.islnk() or member.isdev() or member.isfifo():
                raise BackupIntegrityError(f"unsafe archive member type: {member.name!r}")
            if member.isdir():
                target.mkdir(parents=True, exist_ok=True)
                continue
            if not member.isfile():
                raise BackupIntegrityError(f"unsupported archive member: {member.name!r}")
            file_count += 1
            total_bytes += int(member.size)
            if file_count > MAX_ARCHIVE_FILES or total_bytes > MAX_ARCHIVE_BYTES:
                raise BackupIntegrityError("archive resource limits exceeded")
            target.parent.mkdir(parents=True, exist_ok=True)
            source = tar.extractfile(member)
            if source is None:
                raise BackupIntegrityError(f"cannot read archive member: {member.name!r}")
            with source, target.open("wb") as output:
                shutil.copyfileobj(source, output, length=1024 * 1024)

    manifest_path = root / MANIFEST_NAME
    if not manifest_path.is_file():
        raise BackupIntegrityError("backup manifest missing")
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    if manifest.get("format_version") != FORMAT_VERSION:
        raise BackupIntegrityError("unsupported backup format version")
    allowed_roots = set(ALLOWED_DIR_NAMES) | set(ALLOWED_FILE_NAMES) | {DATABASE_NAME}
    managed_roots = manifest.get("managed_roots", [])
    if not isinstance(managed_roots, list) or any(
        not isinstance(name, str) or name not in allowed_roots for name in managed_roots
    ):
        raise BackupIntegrityError("backup manifest contains a non-allowlisted managed root")
    entries = manifest.get("files", [])
    if not isinstance(entries, list):
        raise BackupIntegrityError("backup manifest file list is invalid")
    expected_paths: set[str] = set()
    for entry in entries:
        if not isinstance(entry, dict) or not isinstance(entry.get("path"), str):
            raise BackupIntegrityError("backup manifest file entry is invalid")
        relative = str(entry["path"])
        if not is_path_allowed(relative):
            raise BackupIntegrityError(f"backup manifest path is not allowlisted: {relative}")
        if relative in expected_paths:
            raise BackupIntegrityError(f"duplicate backup manifest path: {relative}")
        expected_paths.add(relative)
    actual_paths = {
        path.relative_to(root).as_posix()
        for path in root.rglob("*")
        if path.is_file() and path.name != MANIFEST_NAME
    }
    if actual_paths != expected_paths:
        raise BackupIntegrityError("archive file set does not match manifest")
    for entry in manifest.get("files", []):
        path = _safe_member_path(root, str(entry["path"]))
        if path.stat().st_size != int(entry["size"]) or _sha256(path) != entry["sha256"]:
            raise BackupIntegrityError(f"checksum mismatch: {entry['path']}")
    database = root / DATABASE_NAME
    if database.is_file():
        conn = sqlite3.connect(str(database))
        try:
            if conn.execute("PRAGMA integrity_check").fetchall() != [("ok",)]:
                raise BackupIntegrityError("restored sqlite database failed integrity check")
        finally:
            conn.close()
    return manifest


def restore_backup(archive_path: Path, target_dir: Path) -> dict[str, Any]:
    target = target_dir.resolve()
    target.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory(prefix="hermes-restore-", dir=str(target.parent)) as temp:
        stage = Path(temp) / "stage"
        manifest = extract_validated(archive_path, stage)
        roots = [str(root) for root in manifest.get("managed_roots", [])]
        rollback = Path(temp) / "rollback"
        rollback.mkdir()
        moved_existing: list[str] = []
        installed: list[str] = []
        try:
            for name in roots:
                if "/" in name or "\\" in name or name in {"", ".", "..", MANIFEST_NAME}:
                    raise BackupIntegrityError(f"invalid managed root: {name!r}")
                current = target / name
                if current.exists() or current.is_symlink():
                    os.replace(current, rollback / name)
                    moved_existing.append(name)
                incoming = stage / name
                if incoming.exists():
                    os.replace(incoming, current)
                    installed.append(name)
        except Exception:
            for name in reversed(installed):
                current = target / name
                if current.is_dir():
                    shutil.rmtree(current)
                elif current.exists() or current.is_symlink():
                    current.unlink()
            for name in reversed(moved_existing):
                os.replace(rollback / name, target / name)
            raise
        return {"restored": installed, "removed_stale_roots": sorted(set(roots) - set(installed)), "manifest": manifest}