timmy-talking-turd/scripts/operator_backup.py

#!/usr/bin/env python3
"""Operator backup and restore for the records Timmy must not lose.

The app keeps the journal in the person's own browser, so the server side of Timmy holds no diary.
What the operator does hold, and what cannot be rebuilt from anywhere else, is the corpus: the
append-only ``manifest.jsonl`` with its provenance records and its withdrawal tombstones, plus the
derived images those records point at. Losing that file loses the proof that a participant
withdrew, which is the one receipt the project has promised to keep.

This tool therefore:

- archives only records, and names every path it refused to archive with the reason
  (``model-weights``, ``transient-inference``, ``secret``, ``log``, ``symlink``, ``oversize``), so a
  missing file is an auditable decision instead of a silent omission;
- writes one SHA-256 per archived member and a reproducible archive (sorted members, zeroed
  timestamps and ownership), so two backups of the same records are byte-identical;
- verifies every digest, member name, and record count before a restore writes anything, and
  refuses absolute, traversal, link, and unlisted members.

Usage:

    python3 scripts/operator_backup.py backup --source /srv/timmy-corpus --archive /srv/backups/timmy.tar.gz
    python3 scripts/operator_backup.py verify --archive /srv/backups/timmy.tar.gz --manifest /srv/backups/timmy.tar.gz.manifest.json
    python3 scripts/operator_backup.py restore --archive ... --manifest ... --target /srv/timmy-corpus.restored
"""
from __future__ import annotations

import argparse
import gzip
import hashlib
import io
import json
import os
import sys
import tarfile
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath

SCHEMA_VERSION = 1
TOOL = "operator_backup.py"
PAYLOAD_PREFIX = "payload/"
DEFAULT_MAX_BYTES = 64 * 1024 * 1024

WEIGHT_SUFFIXES = (".gguf", ".safetensors", ".onnx", ".bin", ".pt", ".ckpt", ".pth")
WEIGHT_DIRS = ("weights", "models")
TRANSIENT_DIRS = ("transient", "inference-tmp", "inference-tmp-images", "uploads", "tmp", "cache")
LOG_DIRS = ("logs",)
LOG_SUFFIXES = (".log",)
SECRET_NAMES = (".env", "id_rsa", "id_ed25519", "id_ecdsa", "tokens.json", "credentials.json")
SECRET_SUFFIXES = (".key", ".pem", ".p12", ".pfx", ".env")


class BackupError(Exception):
    """A refusal that is reported on stderr and exits non-zero."""


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 classify(relative: PurePosixPath) -> str | None:
    """Return the reason a path is not archived, or ``None`` when it is a record."""
    parts = relative.parts[:-1]
    name = relative.name
    lowered = [part.lower() for part in parts]
    suffix = PurePosixPath(name.lower()).suffix

    if name.lower().startswith(".env") or name.lower() in SECRET_NAMES or suffix in SECRET_SUFFIXES:
        return "secret"
    if suffix in WEIGHT_SUFFIXES or any(part in WEIGHT_DIRS for part in lowered):
        return "model-weights"
    if any(part in TRANSIENT_DIRS for part in lowered):
        return "transient-inference"
    if any(part in LOG_DIRS for part in lowered) or suffix in LOG_SUFFIXES:
        return "log"
    return None


def record_counts(manifest_path: Path) -> dict[str, int]:
    """Count the append-only manifest the way a trainer would read it."""
    counts = {"samples": 0, "tombstones": 0, "activeSamples": 0}
    if not manifest_path.is_file():
        return counts
    revoked: set[str] = set()
    samples: list[dict] = []
    for line in manifest_path.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        try:
            entry = json.loads(line)
        except json.JSONDecodeError as error:
            raise BackupError(f"manifest {manifest_path} is not valid JSON Lines: {error}") from error
        kind = entry.get("recordType")
        if kind == "sample":
            samples.append(entry)
            counts["samples"] += 1
        elif kind == "tombstone":
            counts["tombstones"] += 1
            if entry.get("contributionId"):
                revoked.add(str(entry["contributionId"]))
    counts["activeSamples"] = sum(
        1 for entry in samples
        if str(entry.get("contributionId")) not in revoked
        and str((entry.get("lifecycle") or {}).get("state")) == "active"
    )
    return counts


def collect(source: Path, max_bytes: int) -> tuple[list[tuple[PurePosixPath, Path]], list[dict]]:
    included: list[tuple[PurePosixPath, Path]] = []
    excluded: list[dict] = []
    for dirpath, dirnames, filenames in os.walk(source, followlinks=False):
        dirnames.sort()
        base = Path(dirpath)
        for name in sorted(filenames):
            path = base / name
            relative = PurePosixPath(path.relative_to(source).as_posix())
            if path.is_symlink():
                excluded.append({"path": str(relative), "reason": "symlink", "bytes": 0})
                continue
            reason = classify(relative)
            if reason:
                excluded.append({"path": str(relative), "reason": reason, "bytes": _size(path)})
                continue
            size = _size(path)
            if size > max_bytes:
                excluded.append({"path": str(relative), "reason": "oversize", "bytes": size})
                continue
            included.append((relative, path))
    included.sort(key=lambda item: str(item[0]))
    excluded.sort(key=lambda entry: entry["path"])
    return included, excluded


def _size(path: Path) -> int:
    try:
        return path.stat().st_size
    except OSError:
        return 0


def backup(*, source: Path, archive: Path, manifest_path: Path, label: str, max_bytes: int) -> dict:
    if not source.is_dir():
        raise BackupError(f"source {source} is not a directory")
    if archive.exists():
        raise BackupError(f"archive {archive} already exists; move it aside instead of overwriting a backup")

    included, excluded = collect(source, max_bytes)
    files = [{"path": str(relative), "sha256": sha256_file(path), "bytes": _size(path)} for relative, path in included]
    counts = record_counts(source / "manifest.jsonl")

    archive.parent.mkdir(parents=True, exist_ok=True)
    partial = archive.with_name(archive.name + ".partial")
    with partial.open("wb") as raw:
        # gzip mtime=0 and zeroed tar metadata keep two backups of the same records identical.
        with gzip.GzipFile(fileobj=raw, mode="wb", mtime=0) as compressed:
            with tarfile.open(fileobj=compressed, mode="w", format=tarfile.PAX_FORMAT) as tar:
                for (relative, path), entry in zip(included, files):
                    payload = path.read_bytes()
                    info = tarfile.TarInfo(PAYLOAD_PREFIX + str(relative))
                    info.size = len(payload)
                    info.mode = 0o644
                    info.mtime = 0
                    info.uid = info.gid = 0
                    info.uname = info.gname = ""
                    tar.addfile(info, io.BytesIO(payload))
                    if hashlib.sha256(payload).hexdigest() != entry["sha256"]:
                        raise BackupError(f"{relative} changed while it was being archived")
    os.replace(partial, archive)

    manifest = {
        "schemaVersion": SCHEMA_VERSION,
        "tool": TOOL,
        "createdAt": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "label": label,
        "sourceLabel": source.name,
        "archive": {"name": archive.name, "sha256": sha256_file(archive), "bytes": _size(archive)},
        "records": counts,
        "files": files,
        "excluded": excluded,
    }
    manifest_path.parent.mkdir(parents=True, exist_ok=True)
    manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return {
        "archive": str(archive),
        "manifest": str(manifest_path),
        "files": len(files),
        "bytes": manifest["archive"]["bytes"],
        "records": counts,
        "excluded": {entry["reason"]: sum(1 for other in excluded if other["reason"] == entry["reason"]) for entry in excluded},
    }


def load_manifest(manifest_path: Path) -> dict:
    if not manifest_path.is_file():
        raise BackupError(f"manifest {manifest_path} does not exist")
    try:
        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as error:
        raise BackupError(f"manifest {manifest_path} is not valid JSON: {error}") from error
    if manifest.get("schemaVersion") != SCHEMA_VERSION:
        raise BackupError(f"manifest schemaVersion {manifest.get('schemaVersion')!r} is not {SCHEMA_VERSION}")
    return manifest


def verified_members(archive: Path, manifest: dict) -> list[tuple[str, bytes]]:
    """Check every member of the archive before anything is written to disk."""
    if not archive.is_file():
        raise BackupError(f"archive {archive} does not exist")
    expected = {entry["path"]: entry for entry in manifest.get("files") or []}
    extracted: list[tuple[str, bytes]] = []
    seen: set[str] = set()
    with tarfile.open(archive, "r:gz") as tar:
        for info in tar.getmembers():
            if info.isdir():
                continue
            if info.issym() or info.islnk():
                raise BackupError(f"archive member {info.name!r} is a link, not a regular file")
            if not info.isfile():
                raise BackupError(f"archive member {info.name!r} is not a regular file")
            if os.path.isabs(info.name):
                raise BackupError(f"archive member {info.name!r} is an absolute path")
            if not info.name.startswith(PAYLOAD_PREFIX):
                raise BackupError(f"archive member {info.name!r} is not under {PAYLOAD_PREFIX}")
            relative = info.name[len(PAYLOAD_PREFIX):]
            parts = PurePosixPath(relative).parts
            if not parts or any(part in ("..", ".", "") for part in parts):
                raise BackupError(f"archive member {info.name!r} resolves outside the restore root")
            if relative not in expected:
                raise BackupError(f"archive member {info.name!r} is not listed in the manifest")
            handle = tar.extractfile(info)
            if handle is None:
                raise BackupError(f"archive member {info.name!r} cannot be read")
            payload = handle.read()
            if hashlib.sha256(payload).hexdigest() != expected[relative]["sha256"]:
                raise BackupError(
                    f"archive member {info.name!r} does not match the manifest digest "
                    "(the archive changed after it was recorded)"
                )
            seen.add(relative)
            extracted.append((relative, payload))
    missing = sorted(set(expected) - seen)
    if missing:
        raise BackupError(f"archive is missing members the manifest lists: {', '.join(missing)}")
    return extracted


def verify(*, archive: Path, manifest_path: Path) -> dict:
    manifest = load_manifest(manifest_path)
    members_list = verified_members(archive, manifest)
    return {"archive": str(archive), "files": len(members_list), "records": manifest.get("records", {})}


def restore(*, archive: Path, manifest_path: Path, target: Path) -> dict:
    manifest = load_manifest(manifest_path)
    if target.exists():
        if not target.is_dir():
            raise BackupError(f"restore target {target} is not a directory")
        if any(target.iterdir()):
            raise BackupError(f"restore target {target} is not empty; restore into a new directory")
    # Digests and member names are checked before the first byte is written, so a hostile or
    # tampered archive cannot leave a half-restored record set behind.
    members_list = verified_members(archive, manifest)
    target.mkdir(parents=True, exist_ok=True)
    for relative, payload in members_list:
        destination = target / relative
        destination.parent.mkdir(parents=True, exist_ok=True)
        destination.write_bytes(payload)
        os.chmod(destination, 0o644)

    counts = record_counts(target / "manifest.jsonl")
    if manifest.get("records") and counts != manifest["records"]:
        raise BackupError(
            f"restored record counts {counts} do not match the backup manifest {manifest['records']}"
        )
    return {"target": str(target), "files": len(members_list), "records": counts}


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    subparsers = parser.add_subparsers(dest="command", required=True)

    backup_parser = subparsers.add_parser("backup", help="archive the records and write a manifest")
    backup_parser.add_argument("--source", required=True, type=Path)
    backup_parser.add_argument("--archive", required=True, type=Path)
    backup_parser.add_argument("--manifest", type=Path)
    backup_parser.add_argument("--label", default="manual")
    backup_parser.add_argument("--max-bytes", type=int, default=DEFAULT_MAX_BYTES)

    verify_parser = subparsers.add_parser("verify", help="recompute every digest in an archive")
    verify_parser.add_argument("--archive", required=True, type=Path)
    verify_parser.add_argument("--manifest", required=True, type=Path)

    restore_parser = subparsers.add_parser("restore", help="verify, then write the records into a new directory")
    restore_parser.add_argument("--archive", required=True, type=Path)
    restore_parser.add_argument("--manifest", required=True, type=Path)
    restore_parser.add_argument("--target", required=True, type=Path)
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        if args.command == "backup":
            manifest_path = args.manifest or args.archive.with_name(args.archive.name + ".manifest.json")
            receipt = backup(
                source=args.source, archive=args.archive, manifest_path=manifest_path,
                label=args.label, max_bytes=args.max_bytes,
            )
        elif args.command == "verify":
            receipt = verify(archive=args.archive, manifest_path=args.manifest)
        else:
            receipt = restore(archive=args.archive, manifest_path=args.manifest, target=args.target)
    except BackupError as error:
        print(f"refused: {error}", file=sys.stderr)
        return 2
    print(json.dumps(receipt, indent=2, sort_keys=True))
    return 0


if __name__ == "__main__":
    sys.exit(main())