email-connect/tests/test_ledger_backup.py

"""Only synthetic ledger records; no production URLs, custody or SMTP sends."""

import copy
import hashlib
import importlib.util
import io
import json
import sqlite3
import subprocess
import sys
import tarfile
from datetime import UTC, datetime, timedelta
from pathlib import Path

import pytest

from email_connect.transactional import SQLiteDeliveryStore, TransactionalApplication


@pytest.fixture
def modules(monkeypatch):
    scripts = Path(__file__).resolve().parents[1] / "scripts"
    monkeypatch.syspath_prepend(str(scripts))
    result = []
    for name in ["ledger_archive", "capture-ledger", "backup-ledger"]:
        spec = importlib.util.spec_from_file_location(
            name.replace("-", "_"), scripts / (name + ".py")
        )
        module = importlib.util.module_from_spec(spec)
        monkeypatch.setitem(sys.modules, name.replace("-", "_"), module)
        spec.loader.exec_module(module)
        result.append(module)
    return result


def synthetic_store(path):
    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    store = SQLiteDeliveryStore(str(path))
    path.chmod(0o600)
    accepted = {
        "registration_id": "accepted",
        "normalized_email": "accepted@example.test",
        "preferred_username": "synthetic",
        "client_id": "synthetic-client",
        "tenant": "synthetic-tenant",
        "correlation_id": "trace-1",
    }
    expires = (datetime.now(UTC) + timedelta(days=1)).isoformat()

    def reserve(payload, handle):
        return store.reserve_verification(
            payload, hashlib.sha256(handle.encode()).hexdigest(), expires
        )[0]

    request = reserve(accepted, "accepted-handle")
    store.verification_accepted(request, "synthetic-provider-reference")
    reserve(dict(accepted, registration_id="unknown"), "unknown-handle")
    consumed = reserve(dict(accepted, registration_id="consumed"), "consumed-handle")
    store.verification_accepted(consumed, "synthetic-consumed-reference")
    store.consume_verification(hashlib.sha256(b"consumed-handle").hexdigest())
    canceled = reserve(dict(accepted, registration_id="canceled"), "canceled-handle")
    store.verification_accepted(canceled, "synthetic-canceled-reference")
    store.cancel_verification(hashlib.sha256(b"canceled-handle").hexdigest())
    store.reserve_send("accepted-invitation", "synthetic-accepted-fingerprint")
    store.record("accepted-invitation", "synthetic-invitation-reference")
    store.reserve_send("unknown-invitation", "synthetic-unknown-fingerprint")
    store.suppress("suppressed@example.test", "synthetic complaint")
    store.db.close()
    return accepted


def captured(directory, archive, image="synthetic@sha256:fixture"):
    directory.mkdir(mode=0o700)
    synthetic_store(directory / "database.sqlite3")
    receipt = {
        "status": "captured",
        "application": "email-connect",
        "image": image,
        "captured_at": "2026-10-07T20:00:00Z",
        "elapsed_seconds": 30,
        "writers_quiesced": True,
        "receiver_resumed": True,
        "source_pvc_read_only": True,
        "database_sha256": archive.digest(directory / "database.sqlite3"),
        "database_bytes": (directory / "database.sqlite3").stat().st_size,
        "ledger_fingerprint": archive.fingerprint(directory / "database.sqlite3"),
    }
    archive.write_private(directory / "receipt.json", json.dumps(receipt).encode())
    return receipt


def invoke(app, payload, path):
    result = {}
    raw = json.dumps(payload).encode()
    environ = {
        "PATH_INFO": path,
        "REQUEST_METHOD": "POST",
        "CONTENT_LENGTH": str(len(raw)),
        "wsgi.input": io.BytesIO(raw),
        "HTTP_AUTHORIZATION": "Bearer synthetic-token",
    }
    body = json.loads(b"".join(app(environ, lambda status, headers: result.update(status=status))))
    return result["status"], body


def test_sqlite_backup_includes_committed_wal_without_source_checkpoint(modules, tmp_path):
    archive, _, _ = modules
    source = tmp_path / "source.sqlite3"
    synthetic_store(source)
    with sqlite3.connect(source) as writer:
        writer.execute("PRAGMA journal_mode=WAL")
        writer.execute("INSERT INTO deliveries VALUES ('wal-event','wal-reference')")
        writer.commit()
        original_bytes = source.read_bytes()
        before = archive.fingerprint(source)
        output = tmp_path / "private" / "database.sqlite3"
        output.parent.mkdir(mode=0o700)
        assert archive.sqlite_backup(source, output) == before
        assert source.read_bytes() == original_bytes
        with sqlite3.connect(output) as restored:
            assert restored.execute(
                "SELECT provider_ref FROM deliveries WHERE event_id='wal-event'"
            ).fetchone() == ("wal-reference",)
        writer.execute("INSERT INTO deliveries VALUES ('uncommitted','private')")
        output2 = output.parent / "another.sqlite3"
        archive.sqlite_backup(source, output2)
        with sqlite3.connect(output2) as restored:
            assert (
                restored.execute("SELECT 1 FROM deliveries WHERE event_id='uncommitted'").fetchone()
                is None
            )
        writer.rollback()


def test_real_encryption_recovery_preserves_challenge_and_delivery_behavior(modules, tmp_path):
    archive, _, _ = modules
    stage = tmp_path / "stage"
    stage.mkdir(mode=0o700)
    receipt = captured(stage / "capture", archive)
    identity = tmp_path / "synthetic-identity"
    subprocess.run(["age-keygen", "-o", str(identity)], capture_output=True, check=True)
    public = (
        subprocess.run(["age-keygen", "-y", str(identity)], capture_output=True, check=True)
        .stdout.decode()
        .strip()
    )
    encrypted, _ = archive.seal(stage / "capture", stage, public)
    first_ciphertext = encrypted.read_bytes()
    assert (
        archive.seal(stage / "capture", stage, "unused-retry-recipient")[0].read_bytes()
        == first_ciphertext
    )
    transfer = dict(
        receipt,
        schema="platform.application-primary-archive.v1",
        status="offsite_fetched_verified",
        remote_retrieval_verified=True,
        version_id="synthetic-version",
        object_key="platform-pg/application-archives/email-connect/2026/10/07/20261007T200000Z-"
        + "a" * 24
        + ".tar.age",
        destination=archive.PREFIX + "2026/10/07/20261007T200000Z-" + "a" * 24 + ".tar.age",
        ciphertext_sha256=archive.digest(encrypted),
        ciphertext_bytes=encrypted.stat().st_size,
    )
    transfer_path = tmp_path / "transfer.json"
    archive.write_private(transfer_path, json.dumps(transfer).encode())
    destination = tmp_path / "recovered"
    result = archive.recover(encrypted, transfer_path, identity, destination)
    assert result["status"] == "decrypted_ledger_verified" and not result["production_restored"]
    assert not (destination / "packet.tar").exists()
    assert all(p.stat().st_mode & 0o777 == 0o600 for p in destination.iterdir())
    path = destination / "database.sqlite3"
    before = archive.fingerprint(path)
    store = SQLiteDeliveryStore(str(path))
    assert archive.fingerprint(path) == before

    class ForbiddenProvider:
        def send(self, *args):
            raise AssertionError("recovery must not resend")

    app = TransactionalApplication(
        store, ForbiddenProvider(), "synthetic-token", "https://issuer.example.test"
    )
    payload = {
        "registration_id": "accepted",
        "normalized_email": "accepted@example.test",
        "preferred_username": "synthetic",
        "client_id": "synthetic-client",
        "tenant": "synthetic-tenant",
        "correlation_id": "another-trace",
    }
    status, response = invoke(app, payload, "/v1/registration-verifications")
    assert status.startswith("202") and response["accepted"]
    status, response = invoke(
        app, dict(payload, registration_id="unknown"), "/v1/registration-verifications"
    )
    assert status.startswith("422") and response["error"] == "delivery_outcome_unknown"
    for handle in ["consumed-handle", "canceled-handle"]:
        _, response = invoke(app, {"handle": handle}, "/v1/registration-verifications/consume")
        assert response["error"] == "verification_invalid"
    assert store.reference("accepted-invitation") == "synthetic-invitation-reference"
    assert store.is_suppressed("suppressed@example.test")
    store.db.close()
    assert archive.fingerprint(path) == before


@pytest.mark.parametrize("bad", ["duplicate", "traversal", "link", "extra"])
def test_packet_refuses_all_unsafe_members_before_extract(modules, tmp_path, bad):
    archive, _, _ = modules
    packet = tmp_path / "packet.tar"
    with tarfile.open(packet, "w") as output:
        for name in ["database.sqlite3", "receipt.json"]:
            entry = tarfile.TarInfo(name)
            entry.size = 1
            if bad == "link" and name == "database.sqlite3":
                entry.type, entry.linkname, entry.size = tarfile.SYMTYPE, "/tmp/private", 0
            output.addfile(entry, io.BytesIO(b"x"))
        if bad != "link":
            entry = tarfile.TarInfo(
                {"duplicate": "database.sqlite3", "traversal": "../escape", "extra": "smtp.env"}[
                    bad
                ]
            )
            entry.size = 1
            output.addfile(entry, io.BytesIO(b"x"))
    destination = tmp_path / "out"
    destination.mkdir(mode=0o700)
    with pytest.raises(archive.Refused, match="invalid_packet_members"):
        archive.unpack(packet, destination)
    assert not list(destination.iterdir())


def deployment(capture):
    return {
        "spec": {"replicas": 1, "template": {"spec": {"containers": [{"image": capture.IMAGE}]}}},
        "status": {"readyReplicas": 1},
    }


@pytest.mark.parametrize(
    "failure", [None, "capture", "interrupt", "helper-cleanup", "foreign-image"]
)
def test_capture_resumes_unchanged_receiver_after_failures(modules, monkeypatch, tmp_path, failure):
    archive, capture, _ = modules
    before = deployment(capture)
    current = copy.deepcopy(before)
    calls = []

    def k(*args, **kwargs):
        calls.append(args)
        if args[0] == "patch":
            patches = json.loads(kwargs["body"])
            assert patches[0]["value"] == before["spec"]["template"]
            assert patches[1]["value"] == current["spec"]["replicas"]
            current["spec"]["replicas"] = patches[2]["value"]
        if args[0] == "delete" and failure == "helper-cleanup":
            raise capture.Refused("cleanup failed")
        return b""

    def document(*args, **kwargs):
        return {"items": []} if args[1] == "pods" else copy.deepcopy(current)

    def capture_database(name, destination):
        if failure == "capture":
            raise capture.Refused("capture failed")
        if failure == "interrupt":
            raise KeyboardInterrupt()
        if failure == "foreign-image":
            current["spec"]["template"]["spec"]["containers"][0]["image"] = "foreign-image"
        synthetic_store(destination / "database.sqlite3")
        path = destination / "database.sqlite3"
        return {
            "database_sha256": archive.digest(path),
            "database_bytes": path.stat().st_size,
            "ledger_fingerprint": archive.fingerprint(path),
            "writers_quiesced": True,
            "source_pvc_read_only": True,
        }

    monkeypatch.setattr(capture, "k", k)
    monkeypatch.setattr(capture, "document", document)
    monkeypatch.setattr(capture, "capture_database", capture_database)
    destination = tmp_path / "capture"
    body = json.dumps(capture.helper("synthetic-capture")).encode()
    if failure:
        with pytest.raises((archive.Refused, KeyboardInterrupt)):
            capture.capture(before, body, destination)
    else:
        assert capture.capture(before, body, destination)["status"] == "captured"
    receipt = json.loads((destination / "receipt.json").read_text())
    assert any(args[0] == "delete" for args in calls)
    if failure == "foreign-image":
        assert current["spec"]["replicas"] == 0 and receipt["receiver_state_requires_check"]
        assert not receipt.get("receiver_resumed")
    else:
        assert current == before and receipt["receiver_resumed"]
    assert receipt["status"] == ("captured" if failure is None else "failed")


@pytest.mark.parametrize("failure", [None, "publish", "encrypt", "mirror"])
def test_backup_success_requires_readback_and_preserves_one_pending_capture(
    modules, monkeypatch, tmp_path, failure
):
    archive, _, runner = modules
    captures, publications = [], []
    original_seal = runner.seal
    state = tmp_path / "state"

    def command(argv, **kwargs):
        if "--primary-receipt" in argv:
            if failure == "mirror":
                raise archive.Refused("synthetic mirror failure")
            archive.checkpoint(Path(argv[-1]), {"hash_readback_verified": True})
            return
        captures.append(argv)
        captured(Path(argv[-1]), archive)

    def seal(directory, stage, recipient):
        if failure == "encrypt":
            raise archive.Refused("synthetic encryption failure")
        return original_seal(directory, stage, recipient)

    def publish(encrypted, fetched, receipt_path, capture, tool, python):
        publications.append(encrypted.read_bytes())
        if failure == "publish":
            raise archive.Refused("synthetic readback failure")
        archive.write_private(fetched, encrypted.read_bytes())
        key = (
            "platform-pg/application-archives/email-connect/2026/10/07/20261007T200000Z-"
            + "b" * 24
            + ".tar.age"
        )
        return {
            "schema": "platform.application-primary-archive.v1",
            "application": "email-connect",
            "status": "offsite_fetched_verified",
            "remote_retrieval_verified": True,
            "version_id": "synthetic-version",
            "object_key": key,
            "destination": archive.PREFIX + key.rsplit("/", 1)[-1],
            "captured_at": capture["captured_at"],
            "ciphertext_sha256": archive.digest(encrypted),
            "ciphertext_bytes": encrypted.stat().st_size,
            "immutable_create": True,
        }

    identity = tmp_path / "synthetic-identity"
    subprocess.run(["age-keygen", "-o", str(identity)], capture_output=True, check=True)
    public = (
        subprocess.run(["age-keygen", "-y", str(identity)], capture_output=True, check=True)
        .stdout.decode()
        .strip()
    )
    monkeypatch.setattr(runner, "command", command)
    monkeypatch.setattr(runner, "seal", seal)
    monkeypatch.setattr(runner, "publish", publish)
    if failure:
        with pytest.raises(archive.Refused):
            runner.backup(state, Path("synthetic-tool"), Path("synthetic-python"), public,
                          mirror_required=failure == "mirror")
        assert not (state / "last-success.json").exists()
        assert len(list(state.glob("pending-*"))) == 1
        with pytest.raises(archive.Refused, match="pending_backup_requires_review"):
            runner.backup(state, Path("synthetic-tool"), Path("synthetic-python"), public)
        if failure in {"publish", "mirror"}:
            mirror_required = failure == "mirror"
            ciphertext = publications[0]
            failure = None
            result = runner.backup(
                state, Path("synthetic-tool"), Path("synthetic-python"), public, resume=True,
                mirror_required=mirror_required,
            )
            assert len(captures) == 1 and publications == [ciphertext, ciphertext]
            assert result["application_restore_verified"] is False
    else:
        result = runner.backup(state, Path("synthetic-tool"), Path("synthetic-python"), public)
        assert not list(state.glob("pending-*"))
        assert not list(state.rglob("database.sqlite3"))
        assert json.loads((state / "inventory.json").read_text()) == [result]
        assert runner.freshness(state, datetime(2026, 10, 8, 20, tzinfo=UTC))["status"] == "fresh"
        with pytest.raises(archive.Refused, match="backup_stale"):
            runner.freshness(state, datetime(2026, 10, 9, 20, tzinfo=UTC))
        (state / result["local_archive"]).write_bytes(b"corruption")
        with pytest.raises(archive.Refused, match="local_ciphertext_mismatch"):
            runner.freshness(state, datetime(2026, 10, 8, 20, tzinfo=UTC))


@pytest.mark.parametrize(
    "field,value",
    [
        ("receiver_resumed", False),
        ("database_sha256", "bad"),
        ("ledger_fingerprint", {}),
        ("source_pvc_read_only", False),
    ],
)
def test_invalid_capture_cannot_be_sealed(modules, tmp_path, field, value):
    archive, _, _ = modules
    stage = tmp_path / "stage"
    stage.mkdir(mode=0o700)
    receipt = captured(stage / "capture", archive)
    receipt[field] = value
    (stage / "capture" / "receipt.json").write_text(json.dumps(receipt))
    with pytest.raises(archive.Refused):
        archive.seal(stage / "capture", stage, "unused public recipient")
    assert not (stage / "packet.tar").exists()


def test_helper_has_no_secret_or_token_mount_and_requires_no_egress(modules):
    _, capture, _ = modules
    pod = capture.helper("synthetic-helper")
    container = pod["spec"]["containers"][0]
    assert not pod["spec"]["automountServiceAccountToken"]
    assert "env" not in container and "envFrom" not in container
    assert container["volumeMounts"][0]["readOnly"]
    labels = pod["metadata"]["labels"]
    deny = {"spec": {"policyTypes": ["Ingress", "Egress"], "podSelector": {}}}
    scoped = {
        "spec": {
            "policyTypes": ["Egress"],
            "podSelector": {"matchLabels": {"app.kubernetes.io/name": "email-connect"}},
            "egress": [{}],
        }
    }
    capture.require_no_helper_egress([deny, scoped], labels)
    with pytest.raises(capture.Refused, match="helper_egress_not_denied"):
        capture.require_no_helper_egress([], labels)
    broad = copy.deepcopy(scoped)
    broad["spec"]["podSelector"] = {}
    with pytest.raises(capture.Refused, match="helper_egress_not_denied"):
        capture.require_no_helper_egress([deny, broad], labels)
    assert capture.selected(
        {"matchExpressions": [{"key": "email-ledger-capture", "operator": "Exists"}]}, labels
    )
    assert not capture.selected(
        {
            "matchExpressions": [
                {"key": "email-ledger-capture", "operator": "In", "values": ["other"]}
            ]
        },
        labels,
    )


def test_scoped_ledger_rehearsal_preserves_ownership_and_budgets(modules, tmp_path, monkeypatch, capsys):
    archive, _, _ = modules
    path = tmp_path / "source.sqlite3"
    synthetic_store(path)
    store = SQLiteDeliveryStore(str(path))
    payload = {
        "registration_id": "scoped", "normalized_email": "scoped@example.test",
        "preferred_username": "synthetic", "client_id": "synthetic-client",
        "tenant": "synthetic-tenant", "correlation_id": "trace-scoped",
    }
    expires = (datetime.now(UTC) + timedelta(days=1)).isoformat()
    request, _ = store.reserve_verification(
        payload, hashlib.sha256(b"scoped-handle").hexdigest(), expires, caller_id="caller-a"
    )
    store.verification_accepted(request, "synthetic-scoped-reference")
    store.reserve_verification(
        dict(payload, registration_id="scoped-unknown"),
        hashlib.sha256(b"scoped-unknown-handle").hexdigest(), expires, caller_id="caller-b"
    )
    assert store.request_budget("caller-a", rate=2, burst=1, now=100)[0]
    store.db.close()
    before = archive.fingerprint(path)
    assert before["tables"]["caller_buckets"]["rows"] == 1
    script = Path(__file__).resolve().parents[1] / "scripts/rehearse-ledger.py"
    spec = importlib.util.spec_from_file_location("rehearsal", script)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    recovered = tmp_path / "recovered.sqlite3"
    code = module.CODE.replace("Path('/tmp/recovered.sqlite3')", "Path(" + repr(str(recovered)) + ")")
    monkeypatch.setattr(sys, "stdin", io.TextIOWrapper(io.BytesIO(path.read_bytes())))
    monkeypatch.setattr(__import__("os"), "getuid", lambda: 10001)
    exec(compile(code, str(script), "exec"), {})
    receipt = json.loads(capsys.readouterr().out)
    assert receipt["checks"]["accepted_verifications"] == 4
    assert receipt["checks"]["unknown_verifications"] == 2
    assert receipt["mail_sent"] is False
    assert archive.fingerprint(recovered) == before
    restarted = SQLiteDeliveryStore(str(recovered))
    assert not restarted.request_budget("caller-a", rate=2, burst=1, now=100)[0]
    restarted.db.close()