email-connect/scripts/ledger_archive.py

#!/usr/bin/env python3
"""Private transactional-ledger archive primitives; no mail/provider code."""

from __future__ import annotations

import hashlib
import json
import os
import re
import sqlite3
import stat
import subprocess
import tarfile
import tempfile
from contextlib import closing
from pathlib import Path

MAX_BYTES = 128 * 1024**2
TABLES = {"deliveries", "verifications", "suppressions", "send_attempts", "verification_attempts"}
PREFIX = "s3://railiance-platform-pg-backup/platform-pg/application-archives/email-connect/"


class Refused(RuntimeError):
    pass


def digest(path):
    with path.open("rb") as stream:
        return hashlib.file_digest(stream, "sha256").hexdigest()


def private_directory(path):
    info = path.lstat()
    if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077:
        raise Refused("private_owner_directory_required")


def private_file(path):
    info = path.lstat()
    if not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077:
        raise Refused("private_owner_file_required")


def write_private(path, value):
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
    with os.fdopen(fd, "wb") as stream:
        stream.write(value)
        stream.flush()
        os.fsync(stream.fileno())


def checkpoint(path, value):
    fd, name = tempfile.mkstemp(prefix="receipt-", dir=path.parent)
    with os.fdopen(fd, "w") as stream:
        json.dump(value, stream, indent=2)
        stream.flush()
        os.fsync(stream.fileno())
    Path(name).replace(path)


def fingerprint(path):
    private_file(path)
    if not 0 < path.stat().st_size <= MAX_BYTES:
        raise Refused("ledger_size_outside_contract")
    with closing(sqlite3.connect(path.absolute().as_uri() + "?mode=ro", uri=True)) as database:
        database.execute("PRAGMA trusted_schema=OFF")
        if database.execute("PRAGMA integrity_check").fetchall() != [("ok",)]:
            raise Refused("ledger_integrity_failed")
        tables = {
            row[0]
            for row in database.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'"
            )
        }
        if tables not in (TABLES, TABLES | {"caller_buckets"}):
            raise Refused("unexpected_ledger_schema")
        schema = list(
            database.execute("SELECT type,name,tbl_name,sql FROM sqlite_master ORDER BY type,name")
        )
        result = {
            "schema_sha256": hashlib.sha256(json.dumps(schema).encode()).hexdigest(),
            "user_version": database.execute("PRAGMA user_version").fetchone()[0],
            "tables": {},
        }
        for name in sorted(tables):
            rows = [
                json.dumps(
                    row, separators=(",", ":"), default=lambda value: {"bytes_hex": value.hex()}
                ).encode()
                for row in database.execute('SELECT * FROM "' + name + '"')
            ]
            result["tables"][name] = {
                "rows": len(rows),
                "sha256": hashlib.sha256(b"\n".join(sorted(rows))).hexdigest(),
            }
        return result


def sqlite_backup(source, output):
    """Use SQLite backup rather than copying a file that may still have WAL."""
    private_directory(output.parent)
    if source.is_symlink() or not source.is_file() or output.exists() or output.is_symlink():
        raise Refused("invalid_sqlite_backup_paths")
    # No source write or checkpoint; a read-only connection includes committed WAL.
    with closing(sqlite3.connect(source.absolute().as_uri() + "?mode=ro", uri=True)) as src:
        if (
            src.execute("PRAGMA page_count").fetchone()[0]
            * src.execute("PRAGMA page_size").fetchone()[0]
            > MAX_BYTES
        ):
            raise Refused("ledger_size_outside_contract")
        write_private(output, b"")
        with closing(sqlite3.connect(output)) as dst:
            src.backup(dst, pages=256)
    return fingerprint(output)


def validate_capture(directory):
    private_directory(directory)
    private_file(directory / "receipt.json")
    receipt = json.loads((directory / "receipt.json").read_text())
    if (
        receipt.get("status") != "captured"
        or receipt.get("application") != "email-connect"
        or receipt.get("writers_quiesced") is not True
        or receipt.get("receiver_resumed") is not True
        or receipt.get("source_pvc_read_only") is not True
    ):
        raise Refused("capture_incomplete")
    path = directory / "database.sqlite3"
    actual = fingerprint(path)
    if (
        digest(path) != receipt.get("database_sha256")
        or path.stat().st_size != receipt.get("database_bytes")
        or actual != receipt.get("ledger_fingerprint")
    ):
        raise Refused("capture_binding_mismatch")
    return receipt


def command(argv, timeout=900):
    # Keep private child diagnostics out of both memory pipes and the journal.
    with tempfile.TemporaryFile() as output:
        child = subprocess.Popen(argv, stdout=output, stderr=output)
        try:
            result = child.wait(timeout=timeout)
        except BaseException:
            child.terminate()
            try:
                child.wait(timeout=800)
            except subprocess.TimeoutExpired:
                child.kill()
                child.wait()
            raise
        if result:
            raise Refused("archive_command_failed")


def seal(directory, stage, recipient):
    captured = validate_capture(directory)
    encrypted, sealed_path = stage / "packet.tar.age", stage / "sealed.json"
    binding = {
        key: captured[key]
        for key in ["database_sha256", "captured_at", "image", "ledger_fingerprint"]
    }
    if sealed_path.exists():
        private_file(sealed_path)
        private_file(encrypted)
        receipt = json.loads(sealed_path.read_text())
        if receipt.get("binding") != binding or receipt.get("ciphertext_sha256") != digest(
            encrypted
        ):
            raise Refused("sealed_packet_binding_mismatch")
        return encrypted, captured
    if encrypted.exists() or encrypted.is_symlink():
        raise Refused("partial_encryption_requires_review")
    archive = stage / "packet.tar"
    with archive.open("xb") as stream:
        os.chmod(archive, 0o600)
        with tarfile.open(fileobj=stream, mode="w") as packet:
            for name in ["database.sqlite3", "receipt.json"]:
                packet.add(directory / name, arcname=name, recursive=False)
    command(["age", "-r", recipient, "-o", str(encrypted), str(archive)])
    encrypted.chmod(0o600)
    private_file(encrypted)
    if not 0 < encrypted.stat().st_size <= MAX_BYTES + 1024**2:
        raise Refused("encrypted_packet_size_outside_contract")
    checkpoint(sealed_path, {"binding": binding, "ciphertext_sha256": digest(encrypted)})
    return encrypted, captured


def unpack(packet, destination):
    if packet.stat().st_size > MAX_BYTES + 1024**2:
        raise Refused("packet_too_large")
    with tarfile.open(packet, "r:") as archive:
        entries = archive.getmembers()
        if (
            len(entries) != 2
            or {entry.name for entry in entries} != {"database.sqlite3", "receipt.json"}
            or any(
                not entry.isfile() or entry.sparse is not None or entry.size <= 0
                for entry in entries
            )
            or sum(entry.size for entry in entries) > MAX_BYTES + 1024**2
            or any(entry.size > 1024**2 for entry in entries if entry.name == "receipt.json")
        ):
            raise Refused("invalid_packet_members")
        for entry in entries:
            with (
                archive.extractfile(entry) as source,
                (destination / entry.name).open("xb") as target,
            ):
                os.chmod(target.name, 0o600)
                remaining = entry.size
                while remaining:
                    block = source.read(min(remaining, 8 * 1024**2))
                    if not block:
                        raise Refused("truncated_packet_member")
                    target.write(block)
                    remaining -= len(block)
    return validate_capture(destination)


def recover(encrypted, transfer_path, identity, destination):
    for path in [encrypted, transfer_path, identity]:
        private_file(path)
    transfer = json.loads(transfer_path.read_text())
    key = transfer.get("object_key", "")
    suffix = key.removeprefix("platform-pg/application-archives/email-connect/")
    version = transfer.get("version_id")
    if (
        not isinstance(version, str)
        or version in {"", "null"}
        or any(char in version for char in "\r\n")
        or not key.startswith("platform-pg/application-archives/email-connect/")
        or not re.fullmatch(r"\d{4}/\d{2}/\d{2}/\d{8}T\d{6}Z-[a-f0-9]{24}\.tar\.age", suffix)
        or transfer.get("destination") != "s3://railiance-platform-pg-backup/" + key
    ):
        raise Refused("verified_transfer_required")
    if (
        transfer.get("schema") != "platform.application-primary-archive.v1"
        or transfer.get("application") != "email-connect"
        or transfer.get("status") != "offsite_fetched_verified"
        or transfer.get("remote_retrieval_verified") is not True
        or not transfer.get("version_id")
        or transfer["version_id"] == "null"
        or not transfer.get("destination", "").startswith(PREFIX)
        or transfer.get("ciphertext_sha256") != digest(encrypted)
        or transfer.get("ciphertext_bytes") != encrypted.stat().st_size
    ):
        raise Refused("verified_transfer_required")
    if not destination.is_absolute() or destination.exists() or destination.is_symlink():
        raise Refused("fresh_absolute_destination_required")
    destination.mkdir(mode=0o700)
    packet = destination / "packet.tar"
    # age must create plaintext with private modes even in an interactive shell.
    result = subprocess.run(
        ["age", "-d", "-i", str(identity), "-o", str(packet), str(encrypted)],
        capture_output=True,
        timeout=900,
        umask=0o077,
    )
    if result.returncode:
        raise Refused("decryption_failed")
    private_file(packet)
    captured = unpack(packet, destination)
    for key in ["captured_at", "image", "database_sha256", "ledger_fingerprint"]:
        if transfer.get(key) != captured.get(key):
            raise Refused("recovery_capture_binding_mismatch")
    packet.unlink()
    result = {
        "status": "decrypted_ledger_verified",
        "application": "email-connect",
        "object_key": transfer["object_key"],
        "version_id": transfer["version_id"],
        "ciphertext_sha256": transfer["ciphertext_sha256"],
        "database_sha256": captured["database_sha256"],
        "ledger_fingerprint": captured["ledger_fingerprint"],
        "mail_sent": False,
        "production_restored": False,
        "application_restore_verified": False,
    }
    checkpoint(destination / "unpack-receipt.json", result)
    return result