email-connect/scripts/rehearse-sender-native.py

#!/usr/bin/env python3
"""Explicit disposable-image rehearsal; synthetic ledger, no Kubernetes or SMTP."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import secrets
import subprocess
import tempfile
from pathlib import Path

from sender_native import IMAGE, Native
from sender_transition import Journal, continuity, require

from email_connect.transactional import SQLiteDeliveryStore


def docker(*args, body=None):
    result = subprocess.run(["docker", *args], input=body, capture_output=True,
                            timeout=120, check=False)
    require(result.returncode == 0, "isolated_container_command_failed")
    return result.stdout


def synthetic(path):
    store = SQLiteDeliveryStore(str(path))
    payload = {"tenant": "fixture", "client_id": "fixture", "registration_id": "accepted",
               "normalized_email": "fixture@example.test", "preferred_username": "fixture"}
    for kind in ("accepted", "unknown", "consumed", "canceled"):
        payload["registration_id"] = kind
        request, _ = store.reserve_verification(payload, "hash-" + kind, "2030-01-01T00:00:00+00:00")
        if kind != "unknown":
            store.verification_accepted(request, "fixture-reference-" + kind)
        if kind == "consumed":
            store.consume_verification("hash-consumed")
        if kind == "canceled":
            store.cancel_verification("hash-canceled")
    store.reserve_send("accepted-invitation", "accepted-fingerprint")
    store.record("accepted-invitation", "fixture-receipt")
    store.reserve_send("unknown-invitation", "unknown-fingerprint")
    store.suppress("suppressed@example.test", "fixture")
    store.db.close()
    path.chmod(0o444)  # Synthetic fixture is mounted read-only for UID10001.


class Isolated(Native):
    def __init__(self, journal, container):
        super().__init__(execute=True, journal=journal)
        self.container = container
        self.lock_fd = -1  # No host capture lock: transport is Docker-only below.

    def k(self, *args, body=None, timeout=180):
        require(args[0] == "exec" and self.container in args and "--" in args,
                "fixture_transport_scope_changed")
        command = list(args[args.index("--") + 1:])
        require(command[:2] == ["python", "-c"], "fixture_command_scope_changed")
        return docker("exec", "-i", self.container, *command, body=body)

    def fence(self, journal):
        self.check_consumers(allow_receiver=False)

    def check_consumers(self, *, allow_receiver):
        data = json.loads(docker("inspect", self.container))[0]
        require(data["HostConfig"]["NetworkMode"] == "none"
                and data["HostConfig"]["ReadonlyRootfs"] is True
                and not data["HostConfig"].get("PortBindings")
                and data["Config"]["User"] == "10001:10001", "fixture_isolation_changed")
        binds = [m for m in data["Mounts"] if m["Type"] == "bind"]
        require(len(binds) == 1 and binds[0]["Destination"] == "/data" and not binds[0]["RW"],
                "fixture_mount_scope_changed")

    def ensure_helper(self, journal):
        return self.container


def rehearse(output):
    require(not output.exists(), "fresh_output_required")
    image = json.loads(docker("image", "inspect", IMAGE))[0]
    require(image["Id"] == IMAGE.split("@", 1)[1], "cached_image_mismatch")
    container = None
    with tempfile.TemporaryDirectory(prefix="sender-native-rehearsal-") as directory:
        root = Path(directory)
        source = root / "source"
        source.mkdir(mode=0o755)
        synthetic(source / "email-connect.db")
        journal = Journal(root / "journal.jsonl", proposal={"synthetic_fixture": True})
        try:
            name = "sender-rehearsal-" + secrets.token_hex(6)
            container = docker("run", "--detach", "--rm", "--pull=never", "--name", name,
                "--network=none", "--read-only", "--user=10001:10001",
                "--tmpfs=/tmp:rw,noexec,nosuid,size=160m,mode=1777",
                "--mount", "type=bind,src=" + str(source) + ",dst=/data,readonly",
                IMAGE, "python", "-c", "import time;time.sleep(600)").decode().strip()
            require(len(container) == 64 and all(c in "0123456789abcdef" for c in container),
                    "fixture_container_identity_missing")
            adapter = Isolated(journal, container)
            first, outcomes = adapter.snapshot(journal, quiesced=True)
            second, _ = adapter.snapshot(journal, quiesced=False)
            compared = continuity(first, second)
            proof = {"schema": "email-connect.sender-native-snapshot-rehearsal.v1",
                "image": IMAGE, "network": "none", "source_mount_read_only": True,
                "native_capture_and_rehearsal_code_used": True,
                "kubernetes_api_used": False, "production_data_used": False,
                "smtp_provider_called": False, "outcomes": outcomes["outcomes"],
                "continuity": compared,
                "rehearsal": next(e["proof"] for e in journal.entries
                                  if e["phase"] == "native_snapshot_rehearsed")}
        finally:
            journal.close()
            if container:
                docker("rm", "--force", container)
    proof["container_removed"] = True
    files = [Path(__file__), Path(__file__).with_name("sender_native.py"),
             Path(__file__).with_name("sender_transition.py"),
             Path(__file__).with_name("capture-ledger.py"),
             Path(__file__).with_name("ledger_archive.py"),
             Path(__file__).with_name("rehearse-ledger.py"),
             Path(__file__).with_name("inspect-sender-ledger.py")]
    proof["sources"] = [{"path": "scripts/" + p.name, "sha256": hashlib.sha256(p.read_bytes()).hexdigest()}
                        for p in files]
    fd = os.open(output, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
    with os.fdopen(fd, "w") as stream:
        json.dump(proof, stream, indent=2)
        stream.write("\n")
    return proof


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    result = rehearse(args.output)
    print(json.dumps({"status": "isolated_snapshot_adapter_verified",
                      "container_removed": result["container_removed"], "image": result["image"]}))