email-connect/tests/test_sender_release.py

"""Sealed release and complete native-adapter assembly, with synthetic API effects."""

import copy
import importlib.util
import json
import os
import subprocess
import sys
import threading
from pathlib import Path
from wsgiref.simple_server import make_server

import pytest
from test_callers import application
from test_sender_native import API, obj

SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
sys.path.insert(0, str(SCRIPTS))
import sender_native as native
import sender_release as release
from sender_transition import Refused, digest

spec = importlib.util.spec_from_file_location("sender_launcher", SCRIPTS / "release-sender.py")
launcher = importlib.util.module_from_spec(spec)
spec.loader.exec_module(launcher)


def successful_receipt():
    value = {"schema": "platform.email-connect-smtp-foundation.v1", "proposal_digest": native.FOUNDATION,
             "status": "foundation_admitted_projection_ready", "application_sender_changed": False,
             "service_account_uid": "sa-uid"}
    for key in ("native_binding_verified", "native_consumer_auth_verified", "exact_read_verified",
                "mutation_sibling_metadata_denied", "bounded_ttl", "other_service_account_denied",
                "wrong_audience_denied", "workload_token_revocation_verified", "projection_ready"):
        value[key] = True
    return value


class Assembly(API, release.Native):
    """Only the transport is replaced; all Native operations run unchanged.

    API objects model Kubernetes. Helper Python runs in a subprocess against a
    private synthetic ledger. Host UID/path substitutions are explicit here;
    pinned-image isolation is separately covered by rehearse-sender-native.py.
    """

    def __init__(self, root):
        super().__init__()
        self.root = root
        self.crash_config = False
        self.port = None
        self.projection_meta = {"uid": "new-secret", "owners": [{"uid": "projection-uid", "controller": True}]}
        self.objects[("deployment", native.NS)]["spec"]["template"]["spec"]["volumes"].append(
            {"name": "data", "persistentVolumeClaim": {"claimName": native.PVC}})

    def metadata(self, name):
        return copy.deepcopy(self.projection_meta) if name == "email-connect-platform-smtp" else super().metadata(name)

    def k(self, *args, body=None, timeout=180):
        if args[0] == "exec":
            code = args[-1]
            code = code.replace("file:/data/email-connect.db", "file:" + str(self.root / "ledger.db"))
            code = code.replace("/tmp/ledger.sqlite3", str(self.root / "helper-snapshot.sqlite3"))
            code = code.replace("/tmp/recovered.sqlite3", str(self.root / "helper-recovered.sqlite3"))
            code = code.replace("assert os.getuid()==10001", "assert os.getuid()==" + str(os.getuid()))
            code = code.replace("127.0.0.1:8080", f"127.0.0.1:{self.port}")
            sender = self.objects[("configmap", "email-connect-config")]["data"]["EMAIL_CONNECT_SENDER"]
            result = subprocess.run([sys.executable, "-c", code], input=body, capture_output=True,
                timeout=30, check=False, env={**os.environ, "PYTHONPATH": str(SCRIPTS.parent / "src"),
                    "EMAIL_CONNECT_SENDER": sender, "EMAIL_CONNECT_SMTP_USERNAME": sender,
                    "EMAIL_CONNECT_SMTP_PASSWORD": "synthetic-password",
                    "EMAIL_CONNECT_CALLERS_FILE": "/etc/email-connect-callers/callers.json"})
            assert result.returncode == 0, result.stderr.decode()
            return result.stdout
        result = super().k(*args, body=body, timeout=timeout)
        if args[0] == "patch" and "--dry-run=server" not in args:
            if args[1] == "configmap" and self.crash_config:
                self.crash_config = False
                raise OSError("lost config response")
            if args[1] == "deployment":
                self.pods.pop("receiver", None)
                dep = self.objects[("deployment", native.NS)]
                if dep["spec"]["replicas"] == 1:
                    pod = obj("Pod", "receiver", uid="resumed-receiver-uid", spec=copy.deepcopy(dep["spec"]["template"]["spec"]))
                    pod["metadata"].update(labels={"app.kubernetes.io/name": native.NS},
                        ownerReferences=[{"kind": "ReplicaSet", "name": "rs", "uid": "rs-uid", "controller": True}])
                    pod["status"] = {"phase": "Running", "containerStatuses": [{"ready": True, "imageID": native.IMAGE}]}
                    self.pods["receiver"] = pod
                    rs = obj("ReplicaSet", "rs", uid="rs-uid")
                    rs["metadata"]["ownerReferences"] = [{"uid": native.DEPLOYMENT, "controller": True}]
                    self.objects[("replicaset", "rs")] = rs
        return result


@pytest.fixture
def assembled(tmp_path, monkeypatch):
    root = tmp_path / "fixture"
    root.mkdir(mode=0o700)
    api = Assembly(root)
    foundation = {"resources": [
        {"apiVersion": "v1", "kind": "ServiceAccount", "metadata": {"name": "sa"}},
        {"apiVersion": "external-secrets.io/v1", "kind": "SecretStore", "metadata": {"name": "store"}, "spec": {"fixture": True}},
        {"apiVersion": "external-secrets.io/v1", "kind": "ExternalSecret", "metadata": {"name": "projection"}, "spec": {"fixture": True}},
    ]}
    for wanted in foundation["resources"]:
        value = obj(wanted["kind"], wanted["metadata"]["name"], spec=wanted.get("spec"))
        value["apiVersion"] = wanted["apiVersion"]
        value["automountServiceAccountToken"] = False
        value["status"] = {"conditions": [{"type": "Ready", "status": "True"}], "refreshTime": "fixture"}
        api.objects[(wanted["kind"], wanted["metadata"]["name"])] = value
    monkeypatch.setattr(native, "FOUNDATION_SPEC", digest(foundation))
    monkeypatch.setattr(release, "FOUNDATION_SPEC", digest(foundation))
    monkeypatch.setattr(release, "STATE", root / "attempts")
    lock = root / "capture.lock"
    monkeypatch.setattr(native, "CAPTURE_LOCK", lock)
    def factory(**kwargs):
        api.execute = kwargs.get("execute", False)
        api.journal = kwargs.get("journal")
        api.foundation_pins = None
        api.receiver_uid = None
        api.lock_path = lock
        return api
    monkeypatch.setattr(release, "Native", factory)
    owner = {"id": release.TASK, "workplan_id": release.WORKPLAN, "status": "done"}
    monkeypatch.setattr(release, "owner_return", lambda base: owner)
    sources = {name: (SCRIPTS / name).read_text() for name in release.FILES}
    packet = release.prepare(sources, foundation, successful_receipt())
    app, store, provider = application(root)
    store.reserve_send("before", "fixed-fingerprint")
    store.record("before", "accepted-before")
    server = make_server("127.0.0.1", 0, app)
    api.port = server.server_port
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        yield api, packet, store, provider
    finally:
        server.shutdown()
        thread.join(timeout=5)
        server.server_close()
        store.db.close()


def test_complete_native_assembly_forward_and_current_ledger_recovery(assembled):
    api, packet, store, provider = assembled
    result = release.execute(packet)
    assert result["status"] == "sender_deployed_verified"
    assert result["continuity"]["all_six_tables_checked"] and not result["ledger_restored"]
    assert list(api.pods) == ["receiver"] and not provider.calls
    with pytest.raises((FileExistsError, Refused)):
        release.execute(packet)
    store.reserve_send("after", "later-fingerprint")
    store.record("after", "accepted-after")
    store.reserve_send("uncertain", "unknown-fingerprint")
    before = api.observer.inspect(api.root / "ledger.db")["ledger_fingerprint"]
    recovered = release.execute(packet, recover=True)
    assert recovered["status"] == "configuration_recovered_verified"
    assert recovered["sender"] == "forgejo@coulomb.social"
    assert api.observer.inspect(api.root / "ledger.db")["ledger_fingerprint"] == before
    assert list(api.pods) == ["receiver"] and not provider.calls
    with pytest.raises(Refused, match="already_completed"):
        release.execute(packet, recover=True)


def test_lost_native_config_reply_retains_attempt_and_recovers(assembled):
    api, packet, store, provider = assembled
    api.crash_config = True
    with pytest.raises(OSError, match="lost config"):
        release.execute(packet)
    directory = release.attempt(packet)
    entries = [json.loads(line) for line in (directory / "journal.jsonl").read_text().splitlines()]
    assert entries[-1]["phase"] == "requires_inspection"
    assert api.objects[("deployment", native.NS)]["spec"]["replicas"] == 0
    assert api.objects[("configmap", "email-connect-config")]["data"]["EMAIL_CONNECT_SENDER"] == "platform@coulomb.social"
    store.reserve_send("late-unknown", "unknown-fingerprint")
    result = release.execute(packet, recover=True)
    assert result["status"] == "configuration_recovered_verified" and not provider.calls
    assert store.delivery_status("late-unknown")["state"] == "unknown"


@pytest.mark.parametrize("change", ["source", "delta", "owner", "receipt", "foundation", "digest"])
def test_seal_refuses_tampering_before_native_effects(assembled, change):
    api, original, _, _ = assembled
    packet = copy.deepcopy(original)
    if change == "source": packet["sources"]["sender_native.py"] += "\n# changed\n"
    if change == "delta": packet["proposal"]["after"]["deployment_spec"]["replicas"] = 2
    if change == "owner": packet["proposal"]["owner_return"]["status"] = "todo"
    if change == "receipt": packet["proposal"]["foundation_receipt"]["workload_token_revocation_verified"] = False
    if change == "foundation": packet["proposal"]["foundation"]["resources"].pop()
    if change != "digest": packet["proposal_digest"] = "sha256:" + digest(packet["proposal"])
    else: packet["proposal_digest"] = "sha256:wrong"
    calls = len(api.calls)
    with pytest.raises(Refused): release.execute(packet)
    assert len(api.calls) == calls and not release.STATE.exists()


def test_replaced_projection_refuses_before_pause_but_does_not_block_recovery(assembled):
    api, packet, _, _ = assembled
    api.projection_meta["uid"] = "replacement-secret"
    with pytest.raises(Refused, match="projection_identity"):
        release.execute(packet)
    assert not release.STATE.exists()
    api.projection_meta["uid"] = packet["proposal"]["foundation_pins"]["Secret"]
    release.execute(packet)
    api.projection_meta["uid"] = "replacement-secret"
    assert release.execute(packet, recover=True)["status"] == "configuration_recovered_verified"


def test_execute_and_recover_require_founder_terminal_before_transport(assembled, monkeypatch):
    api, packet, _, _ = assembled
    path = api.root / "packet.json"
    release.write(path, packet)
    monkeypatch.setattr(launcher.sys.stdin, "isatty", lambda: False)
    monkeypatch.setattr(launcher, "remote", lambda request: pytest.fail("must not contact node"))
    for action in ("execute", "recover"):
        with pytest.raises(Refused, match="founder_terminal"):
            launcher.main(["--" + action, "--packet", str(path), "--expect-digest", packet["proposal_digest"]])


def test_owner_task_must_be_exact_completed_handoff(monkeypatch):
    import io
    for value in ({"id": release.TASK, "workplan_id": release.WORKPLAN, "status": "todo"},
                  {"id": "different", "workplan_id": release.WORKPLAN, "status": "done"}):
        monkeypatch.setattr(release.urllib.request, "urlopen", lambda *a, value=value, **kw: io.BytesIO(json.dumps(value).encode()))
        with pytest.raises(Refused, match="owner_return"):
            release.owner_return("http://fixture.invalid")


def test_inspection_checks_live_recovery_process_identity(assembled, monkeypatch):
    _api, packet, _, _ = assembled
    release.execute(packet)
    directory = release.attempt(packet)
    recorded = {"pid": 987654321, "start_ticks": "fixture-ticks"}
    release.write(directory / "recovery-process-1000.json", recorded)
    monkeypatch.setattr(release, "process_identity", lambda pid: recorded if pid == recorded["pid"] else None)
    assert release.inspect(packet)["status"] == "attempt_process_running"
    monkeypatch.setattr(release, "process_identity", lambda pid: None)
    assert release.inspect(packet)["last_phase"] == "completed"


def test_live_capture_lock_prevents_effects_and_retains_exclusive_attempt(assembled):
    api, packet, _, _ = assembled
    import fcntl
    fd = os.open(api.lock_path, os.O_CREAT | os.O_RDWR, 0o600)
    fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
    before = copy.deepcopy(api.objects)
    try:
        with pytest.raises(BlockingIOError): release.execute(packet)
        assert api.objects == before
        directory = release.attempt(packet)
        assert (directory / "capsule.json").exists()
        with pytest.raises(FileExistsError): release.execute(packet)
    finally:
        os.close(fd)
    assert release.execute(packet, recover=True)["status"] == "configuration_recovered_verified"


@pytest.mark.parametrize("problem", ["truncated", "private", "earlier_success"])
def test_foundation_journal_requires_private_complete_final_success(tmp_path, problem):
    path = tmp_path / "foundation.jsonl"
    data = json.dumps(successful_receipt()) + "\n"
    if problem == "truncated": data += '{"status":'
    if problem == "earlier_success": data += json.dumps({"status": "requires_inspection"}) + "\n"
    path.write_text(data)
    path.chmod(0o644 if problem == "private" else 0o600)
    with pytest.raises(Refused): launcher.read_foundation_receipt(path)


def test_current_owner_wait_blocks_sealing_before_remote_call(tmp_path, monkeypatch):
    tmp_path.chmod(0o700)
    def waiting(base):
        raise Refused("verified_credential_owner_return_required")
    monkeypatch.setattr(release, "owner_return", waiting)
    monkeypatch.setattr(launcher, "remote", lambda request: pytest.fail("no node call before owner return"))
    with pytest.raises(Refused, match="owner_return"):
        launcher.main(["--prepare", "--packet", str(tmp_path / "packet.json"),
                       "--foundation-receipt", str(tmp_path / "absent.jsonl")])
    assert not (tmp_path / "packet.json").exists()


def test_wrong_confirmation_cannot_reach_remote_mutation(assembled, monkeypatch):
    api, packet, _, _ = assembled
    path = api.root / "packet.json"
    release.write(path, packet)
    monkeypatch.setattr(launcher.sys.stdin, "isatty", lambda: True)
    monkeypatch.setattr("builtins.input", lambda _: "DEPLOY sha256:wrong")
    monkeypatch.setattr(launcher, "remote", lambda request: pytest.fail("wrong confirmation executed"))
    with pytest.raises(Refused, match="not_confirmed"):
        launcher.main(["--execute", "--packet", str(path), "--expect-digest", packet["proposal_digest"]])


def test_bootstrap_checks_source_bytes_before_executing_them(tmp_path):
    sentinel = tmp_path / "must-not-exist"
    sources = {"sender_release.py": "from pathlib import Path\nPath(" + repr(str(sentinel)) + ").touch()\n"}
    request = {"sources": sources, "source_hashes": {"sender_release.py": "0" * 64}}
    result = subprocess.run([sys.executable, "-I", "-c", launcher.BOOTSTRAP],
                            input=json.dumps(request).encode(), capture_output=True, check=False, timeout=15)
    assert result.returncode != 0 and not sentinel.exists()


def test_remote_timeout_is_unconfirmed_not_stopped_or_retried(monkeypatch):
    calls = []
    def timeout(*args, **kwargs):
        calls.append(args)
        raise subprocess.TimeoutExpired("fixture-ssh", 900)
    monkeypatch.setattr(launcher.subprocess, "run", timeout)
    result, code = launcher.remote({"operation": "execute"})
    assert code == 1 and result["status"] == "remote_outcome_unconfirmed"
    assert not result["retry_automatically"] and len(calls) == 1


def test_process_identity_follows_real_process_lifetime():
    child = subprocess.Popen([sys.executable, "-c", "import time;time.sleep(30)"])
    try:
        first = release.process_identity(child.pid)
        assert first["pid"] == child.pid and first["start_ticks"].isdigit()
        assert release.process_identity(child.pid) == first
    finally:
        child.terminate()
        child.wait(timeout=10)
    assert release.process_identity(child.pid) is None