railiance-apps/tools/vergabe_pilot_export.py

#!/usr/bin/env python3
"""Seal the accepted recovery point for founder-only workstation transfer.

Run by the founder through the remote wrapper. Reads only this completed
capture; writes an age-encrypted packet and private founder handoff files.
No Kubernetes, database, OpenBao or provider access.
"""

import json
import os
import pwd
import shutil
import stat
import subprocess
import tempfile
import zipfile
from pathlib import Path

from vergabe_pilot_recovery_store import Refused, digest, private_directory, require

RUN_ID = "20261105T233543Z-89bfbe04dece"
BASE = Path("/var/lib/vergabe-pilot-recovery")
NAME = "vergabe-pilot-" + RUN_ID
# Public recipient declared by railiance-platform/lib/railiance-backup-common.sh.
RECIPIENT = "age1zvryunvjhvpkmasskauga2heeg0ztnte9ymgppvjge36ekumk50syr3tsz"
EXPECTED = {
    "database.dump": (169232, "8758ebd2e2b6c9e75742c42f2d32f71596522f5aade3a7212ed5a04c3cf3476c"),
    "stores.tar.gz": (3489, "6affa22da26badb4daceb87acf62d4eba57bfbfc2d60da0823b607c449596d66"),
}
IMAGE = ("forgejo.coulomb.social/coulomb/vergabe-teilnahme@sha256:"
         "a26444f59c259698159c69ccb96f73dc648a261ece4c86bb2037a9d977870d91")


def regular(path, owner):
    info = path.lstat()
    require(stat.S_ISREG(info.st_mode) and info.st_uid == owner
            and not info.st_mode & 0o077, "private_regular_file_required")
    return info


def capture_files(directory):
    private_directory(directory)
    files = {}
    for name in [*EXPECTED, "manifest.private.json", "receipt.json"]:
        path = directory / name
        info = regular(path, os.geteuid())
        require(0 < info.st_size <= 1024 * 1024, "capture_file_bound")
        files[name] = {"bytes": info.st_size, "sha256": digest(path)}
        if name in EXPECTED:
            require((info.st_size, files[name]["sha256"]) == EXPECTED[name], "capture_hash_mismatch")
    receipt = json.loads((directory / "receipt.json").read_text())
    require(receipt["run_id"] == RUN_ID and receipt["status"] == "captured_and_resumed"
            and receipt["helm_revision"] == 4 and receipt["database"] == "vergabe_demo_company"
            and all(receipt.get(key) is True for key in ["app_resumed", "writers_quiesced",
                    "helper_cleanup_confirmed", "restart_fixture_verified"]), "capture_not_accepted")
    for name, prefix in [("database.dump", "database"), ("stores.tar.gz", "stores")]:
        require(receipt[prefix + "_sha256"] == files[name]["sha256"]
                and receipt[prefix + "_bytes"] == files[name]["bytes"], "receipt_hash_mismatch")
    manifest = json.loads((directory / "manifest.private.json").read_text())
    require(manifest["image"] == IMAGE and manifest["database"] == "vergabe_demo_company",
            "capture_binding_mismatch")
    return files


def seal(directory, age):
    files = capture_files(directory)
    metadata = {"schema": "railiance.vergabe-pilot-recovery-packet.v1", "run_id": RUN_ID,
                "database": "vergabe_demo_company", "image": IMAGE, "files": files}
    ciphertext = directory / "recovery-point.zip.age"
    seal_receipt = directory / "seal.json"
    if ciphertext.exists() or seal_receipt.exists():
        regular(ciphertext, os.geteuid())
        regular(seal_receipt, os.geteuid())
        result = json.loads(seal_receipt.read_text())
        require(result["packet"] == metadata and result["recipient"] == RECIPIENT
                and result["ciphertext_sha256"] == digest(ciphertext)
                and result["ciphertext_bytes"] == ciphertext.stat().st_size,
                "existing_seal_mismatch")
        return ciphertext, result
    with tempfile.TemporaryDirectory(prefix="seal-", dir=directory) as temporary:
        packet = Path(temporary) / "recovery-point.zip"
        with zipfile.ZipFile(packet, "x", compression=zipfile.ZIP_STORED) as archive:
            for name in files:
                archive.write(directory / name, name)
            archive.writestr("packet.json", json.dumps(metadata, sort_keys=True))
        packet.chmod(0o600)
        # Recheck frozen source hashes after packing, before encryption/publication.
        require(capture_files(directory) == files, "capture_changed_during_export")
        encrypted = Path(temporary) / "packet.age"
        fd = os.open(encrypted, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
        with os.fdopen(fd, "wb") as output:
            result = subprocess.run([age, "-r", RECIPIENT, str(packet)], stdout=output,
                                    stderr=subprocess.DEVNULL, timeout=60)
            require(result.returncode == 0, "encryption_failed")
            output.flush()
            os.fsync(output.fileno())
        require(encrypted.stat().st_size <= 4 * 1024 * 1024, "encrypted_packet_bound")
        with encrypted.open("rb") as source:
            require(source.read(22) == b"age-encryption.org/v1\n", "age_header_required")
        result = {"schema": "railiance.vergabe-pilot-sealed-export.v1", "run_id": RUN_ID,
                  "recipient": RECIPIENT, "ciphertext_sha256": digest(encrypted),
                  "ciphertext_bytes": encrypted.stat().st_size, "packet": metadata,
                  "provider_publication_verified": False, "restore_verified": False}
        fd = os.open(seal_receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
        with os.fdopen(fd, "w") as output:
            json.dump(result, output, indent=2)
            output.flush()
            os.fsync(output.fileno())
        # No overwrite, including after an interrupted export. A partial pair
        # requires review rather than silently replacing an earlier ciphertext.
        os.link(encrypted, ciphertext)
    return ciphertext, result


def handoff(source, directory, name, uid, gid):
    target = directory / name
    if target.exists() or target.is_symlink():
        regular(target, uid)
        require(digest(target) == digest(source), "existing_handoff_mismatch")
        return
    fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
    with os.fdopen(fd, "wb") as output, source.open("rb") as stream:
        shutil.copyfileobj(stream, output)
        output.flush()
        os.fsync(output.fileno())
        os.fchown(output.fileno(), uid, gid)


def main():
    require(os.geteuid() == 0, "founder_node_sudo_required")
    account = pwd.getpwnam("tegwick")
    require(os.environ.get("SUDO_USER") == account.pw_name
            and os.environ.get("SUDO_UID") == str(account.pw_uid) and account.pw_uid > 0,
            "founder_ssh_account_required")
    home = Path(account.pw_dir)
    info = home.lstat()
    require(str(home) == "/home/tegwick" and stat.S_ISDIR(info.st_mode)
            and info.st_uid == account.pw_uid and not info.st_mode & 0o022,
            "founder_home_required")
    age = next((str(p) for p in [Path("/usr/bin/age"), Path("/usr/local/bin/age"),
                Path("/opt/railiance-platform/tools/vendor/bin/age"),
                home / "railiance-platform/tools/vendor/bin/age"] if p.is_file() and os.access(p, os.X_OK)), None)
    require(age, "age_unavailable")
    private_directory(BASE)
    encrypted, result = seal(BASE / RUN_ID, age)
    handoff(encrypted, home, NAME + ".zip.age", account.pw_uid, account.pw_gid)
    handoff(encrypted.parent / "seal.json", home, NAME + ".transfer.json", account.pw_uid, account.pw_gid)
    print(json.dumps({"status": "encrypted_handoff_ready", "run_id": RUN_ID,
                      "ciphertext_sha256": result["ciphertext_sha256"],
                      "ciphertext_bytes": result["ciphertext_bytes"], "application_changed": False}))


if __name__ == "__main__":
    os.umask(0o077)
    try:
        main()
    except Exception as error:
        print(json.dumps({"status": "export_failed", "failure": str(error) if isinstance(error, Refused)
                          else "export_command_failed", "details_suppressed": True}))
        raise SystemExit(1)