email-connect/scripts/backup-ledger.py

#!/usr/bin/env python3
"""Node ledger backup or attended recovery; default performs preview only."""

from __future__ import annotations

import argparse
import fcntl
import json
import os
import re
import shutil
import signal
import sys
import tempfile
from datetime import UTC, datetime
from pathlib import Path

from ledger_archive import (
    PREFIX,
    Refused,
    checkpoint,
    command,
    digest,
    private_directory,
    private_file,
    recover,
    seal,
)

NAME = re.compile(r"\d{8}T\d{6}Z-[a-f0-9]{24}\.tar\.age")
PRIMARY_TOOL = Path("/opt/railiance-platform/scripts/primary_application_archive.py")
PRIMARY_PYTHON = Path("/opt/railiance-platform/backup-venv/bin/python")
MINIMUM_FREE_BYTES = 8 * 1024**3
LOCAL_BUDGET_BYTES = 8 * 1024**3
RECIPIENT = "age1zvryunvjhvpkmasskauga2heeg0ztnte9ymgppvjge36ekumk50syr3tsz"


def capacity(state, minimum_free=MINIMUM_FREE_BYTES, budget=LOCAL_BUDGET_BYTES):
    ancestor = state
    while not ancestor.exists():
        ancestor = ancestor.parent
    if shutil.disk_usage(ancestor).free < minimum_free:
        raise Refused("backup_disk_headroom_required")
    used = 0
    if state.exists():
        private_directory(state)
        for path in state.rglob("*"):
            if path.is_symlink():
                raise Refused("backup_state_symlink_requires_review")
            if path.is_file():
                used += path.stat().st_size
        # Reserve room for capture, ciphertext, readback and a failure packet.
        if used + 512 * 1024**2 > budget:
            raise Refused("backup_local_retention_review_required")
    return {"status": "headroom_verified", "local_bytes": used,
            "minimum_free_bytes": minimum_free, "local_budget_bytes": budget}


def publish(encrypted, fetched, receipt_path, captured, tool, python):
    command(
        [
            str(python),
            str(tool),
            "transfer",
            "--application",
            "email-connect",
            "--source",
            str(encrypted),
            "--output",
            str(fetched),
            "--receipt",
            str(receipt_path),
            "--captured-at",
            captured["captured_at"],
            "--execute",
        ]
    )
    private_file(fetched)
    receipt = json.loads(receipt_path.read_text())
    if (
        receipt.get("schema") != "platform.application-primary-archive.v1"
        or receipt.get("status") != "offsite_fetched_verified"
        or receipt.get("application") != "email-connect"
        or receipt.get("captured_at") != captured["captured_at"]
        or receipt.get("remote_retrieval_verified") is not True
        or receipt.get("immutable_create") is not True
        or not receipt.get("version_id")
        or receipt["version_id"] == "null"
        or receipt.get("ciphertext_sha256") != digest(encrypted)
        or receipt.get("ciphertext_sha256") != digest(fetched)
        or receipt.get("ciphertext_bytes") != encrypted.stat().st_size
        or not receipt.get("destination", "").startswith(PREFIX)
    ):
        raise Refused("primary_receipt_not_verified")
    return receipt


def backup(state, primary_tool, primary_python, recipient, resume=False, mirror_required=False):
    state.mkdir(mode=0o700, parents=True, exist_ok=True)
    private_directory(state)
    fd = os.open(state / "backup.lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
    with os.fdopen(fd, "w") as lock:
        try:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            raise Refused("backup_already_running") from None
        pending = list(state.glob("pending-*"))
        if resume:
            if len(pending) != 1:
                raise Refused("one_pending_capture_required")
            stage = pending[0]
            private_directory(stage)
        else:
            if pending:
                raise Refused("pending_backup_requires_review")
            stage = Path(tempfile.mkdtemp(prefix="pending-", dir=state))
            command(
                [
                    sys.executable,
                    str(Path(__file__).with_name("capture-ledger.py")),
                    "--execute",
                    "--destination",
                    str(stage / "capture"),
                ],
                timeout=1800,
            )
        encrypted, captured = seal(stage / "capture", stage, recipient)
        fd, name = tempfile.mkstemp(prefix="readback-", dir=stage)
        os.close(fd)
        fetched = Path(name)
        fetched.unlink()
        transfer = publish(
            encrypted,
            fetched,
            stage / "primary-transfer.json",
            captured,
            primary_tool,
            primary_python,
        )
        if mirror_required:
            transfer.update(image=captured["image"], database_sha256=captured["database_sha256"],
                            ledger_fingerprint=captured["ledger_fingerprint"])
            checkpoint(stage / "primary-transfer.json", transfer)
            command([str(primary_python), str(Path(__file__).with_name("mirror-ledger.py")),
                     "--archive", str(encrypted), "--primary-receipt", str(stage / "primary-transfer.json"),
                     "--output", str(stage / "mirror.json")])
            mirrored = json.loads((stage / "mirror.json").read_text())
            if mirrored.get("hash_readback_verified") is not True:
                raise Refused("recovery_mirror_required")
            transfer["recovery_mirror"] = mirrored
        filename = transfer["object_key"].rsplit("/", 1)[-1]
        if not NAME.fullmatch(filename):
            raise Refused("invalid_archive_name")
        receipt = dict(
            transfer,
            image=captured["image"],
            database_sha256=captured["database_sha256"],
            database_bytes=captured["database_bytes"],
            ledger_fingerprint=captured["ledger_fingerprint"],
            pause_and_capture_seconds=captured["elapsed_seconds"],
            local_archive=filename,
            application_restore_verified=False,
            completed_at=datetime.now(UTC).isoformat(),
        )
        output = state / filename
        if output.exists() or output.is_symlink():
            private_file(output)
            if digest(output) != receipt["ciphertext_sha256"]:
                raise Refused("local_archive_collision")
        else:
            # Create private file from the start, including manual invocations.
            fd = os.open(output, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
            with os.fdopen(fd, "wb") as dst, fetched.open("rb") as src:
                shutil.copyfileobj(src, dst)
                dst.flush()
                os.fsync(dst.fileno())
        inventory_path = state / "inventory.json"
        inventory = json.loads(inventory_path.read_text()) if inventory_path.exists() else []
        if not any(item["object_key"] == receipt["object_key"] for item in inventory):
            inventory.append(receipt)
        checkpoint(inventory_path, inventory)
        checkpoint(state / "last-success.json", receipt)
        shutil.rmtree(stage)
        return receipt


def freshness(state, now=None, hours=26, mirror_required=False):
    now = now or datetime.now(UTC)
    private_directory(state)
    private_file(state / "last-success.json")
    receipt = json.loads((state / "last-success.json").read_text())
    stamp = datetime.fromisoformat(receipt["captured_at"].replace("Z", "+00:00"))
    age = (now - stamp).total_seconds()
    if not 0 <= age <= hours * 3600:
        raise Refused("backup_stale")
    name = receipt.get("local_archive", "")
    if (
        receipt.get("status") != "offsite_fetched_verified"
        or receipt.get("application") != "email-connect"
        or receipt.get("remote_retrieval_verified") is not True
        or not receipt.get("version_id")
        or receipt["version_id"] == "null"
        or not NAME.fullmatch(name)
    ):
        raise Refused("verified_backup_required")
    private_file(state / name)
    if digest(state / name) != receipt["ciphertext_sha256"]:
        raise Refused("local_ciphertext_mismatch")
    if mirror_required:
        mirror = receipt.get("recovery_mirror", {})
        if (mirror.get("hash_readback_verified") is not True
                or mirror.get("bucket") != "railiance-email-connect-archives"
                or mirror.get("project_id") != "1d37ea63-8c8f-44e1-80e1-2de862579ba7"
                or not mirror.get("version_id") or mirror["version_id"] == "null"
                or not mirror.get("receipt_version_id") or mirror["receipt_version_id"] == "null"):
            raise Refused("verified_recovery_mirror_required")
    return {"status": "fresh", "age_hours": round(age / 3600, 2), "remote_retrieval_verified": True}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("action", choices=["backup", "recover", "freshness"])
    parser.add_argument("--execute", action="store_true")
    parser.add_argument("--resume-pending", action="store_true")
    parser.add_argument("--mirror-required", action="store_true")
    parser.add_argument("--state-dir", type=Path, default=Path("/var/lib/email-connect-backup"))
    parser.add_argument("--primary-tool", type=Path, default=PRIMARY_TOOL)
    parser.add_argument("--primary-python", type=Path, default=PRIMARY_PYTHON)
    parser.add_argument("--age-recipient", default=RECIPIENT)
    parser.add_argument("--archive", type=Path)
    parser.add_argument("--transfer-receipt", type=Path)
    parser.add_argument("--identity", type=Path)
    parser.add_argument("--destination", type=Path)
    args = parser.parse_args()
    if args.action == "freshness":
        print(json.dumps(freshness(args.state_dir, mirror_required=args.mirror_required)))
    elif args.action == "recover":
        if not args.execute:
            print(json.dumps({"status": "reviewed", "decryption_performed": False}))
            return 0
        if not all([args.archive, args.transfer_receipt, args.identity, args.destination]):
            raise Refused("recovery_arguments_required")
        print(
            json.dumps(
                recover(args.archive, args.transfer_receipt, args.identity, args.destination)
            )
        )
    elif args.execute:
        if (
            not args.state_dir.is_absolute()
            or not args.primary_tool.is_file()
            or not args.primary_python.is_file()
        ):
            raise Refused("invalid_backup_configuration")
        capacity(args.state_dir)
        command(["age", "--version"])
        command(
            [
                str(args.primary_python),
                str(args.primary_tool),
                "check",
                "--application",
                "email-connect",
            ]
        )
        print(
            json.dumps(
                backup(
                    args.state_dir,
                    args.primary_tool,
                    args.primary_python,
                    args.age_recipient,
                    args.resume_pending,
                    args.mirror_required,
                )
            )
        )
    else:
        if args.resume_pending:
            raise Refused("resume_requires_execute")
        command(
            [
                sys.executable,
                str(Path(__file__).with_name("capture-ledger.py")),
                "--destination",
                "/tmp/email-ledger-capture-review",
            ]
        )
        print(
            json.dumps({"status": "reviewed", "live_changes": False, "provider_contacted": False})
        )
    return 0


if __name__ == "__main__":

    def terminate(signum, frame):
        raise Refused("backup_interrupted")

    signal.signal(signal.SIGTERM, terminate)
    try:
        raise SystemExit(main())
    except Exception as error:
        print(json.dumps({"status": "failed", "failure_type": type(error).__name__}))
        raise SystemExit(1)