email-connect/tests/test_sender_native.py

"""Adapter transport/fault tests use synthetic API objects, never a live cluster."""

import copy
import io
import json
import os
import subprocess
import sys
import threading
from pathlib import Path
from types import SimpleNamespace
from wsgiref.simple_server import make_server

import pytest
from test_callers import application
from test_sender_transition import initial

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


def obj(kind, name, *, spec=None, data=None, uid=None):
    value = {"apiVersion": "v1", "kind": kind,
             "metadata": {"name": name, "uid": uid or name + "-uid", "resourceVersion": "123"}}
    if spec is not None:
        value["spec"] = spec
    if data is not None:
        value["data"] = data
    return value


class API(native.Native):
    def __init__(self):
        super().__init__()
        before = initial()
        before["deployment_spec"]["template"]["spec"]["volumes"] = [
            {"name": "caller-policy", "configMap": {"name": "email-connect-callers-user-engine-276257849bf2"}}]
        self.objects = {
            ("namespace", "kube-system"): obj("Namespace", "kube-system", uid=native.CLUSTER),
            ("namespace", native.NS): obj("Namespace", native.NS, uid=native.NAMESPACE),
            ("deployment", native.NS): obj("Deployment", native.NS, spec=before["deployment_spec"], uid=native.DEPLOYMENT),
            ("configmap", "email-connect-config"): obj("ConfigMap", "email-connect-config", data=before["config_data"], uid=native.CONFIG),
            ("pvc", native.PVC): obj("PersistentVolumeClaim", native.PVC, spec={"volumeName": "fixture-volume"}),
            ("pv", "fixture-volume"): obj("PersistentVolume", "fixture-volume", spec={
                "claimRef": {"uid": native.PVC + "-uid"}, "nodeAffinity": {"required": {
                    "nodeSelectorTerms": [{"matchExpressions": [{"key": "kubernetes.io/hostname",
                        "operator": "In", "values": ["fixture-host"]}]}]}}}),
            ("service", native.NS): obj("Service", native.NS, spec={"selector": {"app": native.NS}}),
            ("externalsecret", "email-connect-runtime"): obj("ExternalSecret", "email-connect-runtime", spec={"retained": True}),
            ("configmap", "email-connect-callers-user-engine-276257849bf2"):
                obj("ConfigMap", "policy", data={"callers.json": "synthetic-policy"}),
        }
        dep = self.objects[("deployment", native.NS)]
        dep["metadata"]["generation"] = 74
        dep["status"] = {"readyReplicas": 1, "observedGeneration": 74}
        self.objects[("pvc", native.PVC)]["status"] = {"phase": "Bound"}
        self.pods = {}
        self.policies = [obj("NetworkPolicy", "deny-all", spec={"podSelector": {}, "policyTypes": ["Egress"]})]
        self.calls = []
        self.loss = None
        self.runtime_proof = None

    def metadata(self, name):
        return {"uid": name + "-uid", "owners": []}

    def get(self, kind, name=None, *, namespace=native.NS):
        if kind == "nodes":
            node = obj("Node", "fixture-node")
            node["metadata"]["labels"] = {"kubernetes.io/hostname": "fixture-host"}
            return {"items": [node]}
        if kind == "networkpolicies":
            return {"items": copy.deepcopy(self.policies)}
        if kind == "pods":
            return {"items": copy.deepcopy(list(self.pods.values()))}
        if kind == "pod":
            return copy.deepcopy(self.pods.get(name))
        return copy.deepcopy(self.objects.get((kind, name)))

    def k(self, *args, body=None, timeout=180):
        self.calls.append((args, body))
        if args[0] == "patch":
            kind, name = args[1:3]
            patch = json.loads(body)
            value = copy.deepcopy(self.objects[(kind, name)])
            assert patch[0]["value"] == value["metadata"]["uid"]
            assert patch[1]["value"] == value["metadata"]["resourceVersion"]
            field = patch[2]["path"][1:]
            assert value[field] == patch[2]["value"]
            value[field] = patch[3]["value"]
            if "--dry-run=server" not in args:
                self.objects[(kind, name)] = value
            return json.dumps(value).encode()
        if args[0] == "create":
            value = json.loads(body)
            value["metadata"]["uid"] = "created-helper-uid"
            value["metadata"]["resourceVersion"] = "456"
            if "--dry-run=server" not in args:
                self.pods[value["metadata"]["name"]] = value
                if self.loss == "create":
                    raise OSError("lost create response")
            return json.dumps(value).encode()
        if args[0] == "delete":
            name = next(a for a in args if a.startswith("--raw=")).rsplit("/", 1)[1]
            assert json.loads(body)["preconditions"]["uid"] == self.pods[name]["metadata"]["uid"]
            self.pods.pop(name)
            return b"{}"
        if args[0] == "exec":
            code = args[-1]
            compile(code, "runtime-probe", "exec")
            self.runtime_proof = code
            return b'{"mail_sent":false,"private_values_printed":false,"unauthenticated_denial":401}'
        if args[0] in {"wait", "rollout"}:
            return b""
        raise AssertionError(args)


@pytest.fixture
def api():
    return API()


@pytest.fixture
def armed(api, tmp_path):
    private = tmp_path / "private"
    private.mkdir(mode=0o700)
    before = api.observe()
    journal = Journal(private / "journal.jsonl", proposal={"before": before, "after": native.desired(before)})
    api.execute = True
    api.journal = journal
    api.lock_fd = 999  # Synthetic adapter only; real lock behavior has its own test.
    try:
        yield api, journal
    finally:
        journal.close()


def test_preparation_is_five_dry_runs_and_preserves_whole_spec(api):
    before = copy.deepcopy(api.objects)
    plan = api.prepare()
    assert plan["native_server_dry_runs"] == 5
    assert api.objects == before and not api.pods
    assert all("--dry-run=server" in args for args, _ in api.calls)
    assert plan["before"]["protected"]["node"]["name"] == "fixture-node"
    assert plan["after"]["deployment_spec"]["template"]["spec"]["containers"][0]["image"] == native.IMAGE


@pytest.mark.parametrize("drift", ["cluster", "namespace", "deployment", "config", "volume", "node"])
def test_native_identity_or_volume_scope_drift_refuses(api, drift):
    keys = {"cluster": ("namespace", "kube-system"), "namespace": ("namespace", native.NS),
            "deployment": ("deployment", native.NS), "config": ("configmap", "email-connect-config")}
    if drift in keys:
        api.objects[keys[drift]]["metadata"]["uid"] = "foreign"
    elif drift == "volume":
        api.objects[("pv", "fixture-volume")]["spec"]["claimRef"]["uid"] = "other-pvc"
    else:
        api.objects[("pv", "fixture-volume")]["spec"]["nodeAffinity"]["required"]["nodeSelectorTerms"] = []
    with pytest.raises(Refused):
        api.prepare()
    assert not api.calls


def test_mutation_needs_armed_lock_and_admission_preparation_cannot_mutate(api):
    current = api.observe()
    api.execute = True
    with pytest.raises(Refused, match="locked_execution"):
        api.patch_deployment(current["deployment_spec"], dict(current["deployment_spec"], replicas=0))
    with pytest.raises(Refused, match="read_only"):
        api.prepare()
    assert not api.calls


def test_capture_lock_is_exclusive_and_private(api, tmp_path, monkeypatch):
    private = tmp_path / "private"
    private.mkdir(mode=0o700)
    path = private / "capture.lock"
    monkeypatch.setattr(native, "CAPTURE_LOCK", path)
    api.execute = True
    api.journal = SimpleNamespace()
    api.lock_path = path
    api.acquire()
    other = API()
    other.execute, other.journal, other.lock_path = True, SimpleNamespace(), path
    try:
        with pytest.raises(BlockingIOError):
            other.acquire()
        assert other.lock_fd is None
        assert path.stat().st_mode & 0o777 == 0o600
    finally:
        api.close()


def test_lost_helper_create_does_not_authorize_adoption_or_deletion(armed):
    api, journal = armed
    api.loss = "create"
    with pytest.raises(OSError):
        api.ensure_helper(journal)
    assert not api.helpers() and len(api.pods) == 1
    with pytest.raises(Refused, match="unconfirmed_helper"):
        api.ensure_helper(journal)
    api.cleanup(journal)
    assert len(api.pods) == 1
    assert not any(args[0] == "delete" for args, _ in api.calls)


def test_helper_cleanup_uses_recorded_uid_and_refuses_replacement(armed):
    api, journal = armed
    name = api.ensure_helper(journal)
    assert api.helpers()[name]["uid"] == "created-helper-uid"
    api.pods[name]["metadata"]["uid"] = "replacement"
    with pytest.raises(Refused, match="helper_identity"):
        api.cleanup(journal)
    assert name in api.pods
    api.pods[name]["metadata"]["uid"] = "created-helper-uid"
    api.cleanup(journal)
    assert not api.pods
    assert journal.entries[-1]["phase"] == "native_helper_deleted"


@pytest.mark.parametrize("change", ["foreign_mount", "writable_helper", "egress"])
def test_fence_refuses_extra_writer_or_changed_helper(armed, change):
    api, journal = armed
    api.objects[("deployment", native.NS)]["spec"]["replicas"] = 0
    name = api.ensure_helper(journal)
    if change == "foreign_mount":
        api.pods["other"] = obj("Pod", "other", spec={"volumes": [{"persistentVolumeClaim": {"claimName": native.PVC}}]})
    elif change == "writable_helper":
        api.pods[name]["spec"]["containers"][0]["volumeMounts"][0]["readOnly"] = False
    else:
        api.policies[0]["spec"]["egress"] = [{}]
    with pytest.raises((Refused, api.archive.Refused, api.capture.Refused)):
        api.fence(journal)


def test_unrecorded_receiver_cannot_use_the_pvc_during_postflight(armed):
    api, _ = armed
    api.pods["receiver"] = obj("Pod", "receiver", uid="other-uid", spec={
        "volumes": [{"persistentVolumeClaim": {"claimName": native.PVC}}]})
    api.receiver_uid = "verified-uid"
    with pytest.raises(Refused, match="unowned_ledger"):
        api.check_consumers(allow_receiver=True)
    api.pods["receiver"]["metadata"]["uid"] = "verified-uid"
    api.check_consumers(allow_receiver=True)


def test_runtime_probe_checks_actual_owner_app_without_provider_send(armed, tmp_path):
    api, journal = armed
    target = journal.entries[0]["proposal"]["before"]
    pod = obj("Pod", "receiver", uid="receiver-uid", spec={"containers": copy.deepcopy(
        target["deployment_spec"]["template"]["spec"]["containers"])})
    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}]}
    rs = obj("ReplicaSet", "rs", uid="rs-uid")
    rs["metadata"]["ownerReferences"] = [{"uid": native.DEPLOYMENT, "controller": True}]
    api.objects[("replicaset", "rs")] = rs
    api.pods["receiver"] = pod
    api.verify(target, journal)
    app, store, provider = application(tmp_path)
    server = make_server("127.0.0.1", 0, app)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        code = api.runtime_proof.replace("127.0.0.1:8080", f"127.0.0.1:{server.server_port}")
        result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True,
            env={**os.environ, "EMAIL_CONNECT_SENDER": "forgejo@coulomb.social",
                 "EMAIL_CONNECT_SMTP_USERNAME": "forgejo@coulomb.social",
                 "EMAIL_CONNECT_SMTP_PASSWORD": "synthetic-private-password",
                 "EMAIL_CONNECT_CALLERS_FILE": "/etc/email-connect-callers/callers.json"},
            timeout=15, check=False)
        assert result.returncode == 0, result.stderr
        assert json.loads(result.stdout)["unauthenticated_denial"] == 401
        assert "synthetic-private-password" not in result.stdout + result.stderr
        assert not provider.calls
        assert not store.db.execute("SELECT * FROM deliveries").fetchall()
    finally:
        server.shutdown()
        thread.join(timeout=5)
        server.server_close()
        store.db.close()


@pytest.mark.parametrize("response_kind", ["metadata", "full_secret", "missing"])
def test_metadata_content_negotiation_never_falls_back_to_secret_values(tmp_path, monkeypatch, response_kind):
    calls, cleanup = [], []
    class Process:
        def poll(self): return None
        def terminate(self): cleanup.append("terminated")
        def wait(self, timeout): return 0
    class Socket:
        def settimeout(self, value): pass
        def connect(self, path): pass
        def close(self): pass
    class Connection:
        def __init__(self, *args, **kwargs): pass
        def request(self, method, path, headers): calls.append((method, path, headers))
        def getresponse(self):
            data = {"kind": "PartialObjectMetadata", "apiVersion": "meta.k8s.io/v1", "metadata": {"uid": "fixture"}}
            if response_kind == "full_secret": data = {"kind": "Secret", "data": {"password": "private"}}
            response = io.BytesIO(json.dumps(data).encode())
            response.status = 404 if response_kind == "missing" else 200
            return response
        def close(self): pass
    monkeypatch.setattr(native.subprocess, "Popen", lambda *a, **kw: Process())
    monkeypatch.setattr(native.socket, "socket", lambda *a: Socket())
    monkeypatch.setattr(native.http.client, "HTTPConnection", Connection)
    monkeypatch.setattr(native.time, "sleep", lambda _: None)
    api = native.Native()
    if response_kind == "full_secret":
        with pytest.raises(Refused, match="metadata_only_response"):
            api.metadata("email-connect-platform-smtp")
    else:
        value = api.metadata("email-connect-platform-smtp")
        assert value == (None if response_kind == "missing" else {"uid": "fixture", "owners": []})
    assert len(calls) == 1 and "PartialObjectMetadata" in calls[0][2]["Accept"]
    assert cleanup == ["terminated"]


def test_secret_full_queries_and_unknown_metadata_names_refuse():
    api = native.Native()
    with pytest.raises(Refused, match="outside_reviewed_scope"):
        api.get("secrets", "anything")
    with pytest.raises(Refused, match="metadata_scope_drift"):
        api.metadata("unrelated")


@pytest.mark.parametrize("failure", [None, "status", "digest", "scope", "identity"])
def test_admission_requires_complete_owner_receipt_and_matching_native_identity(armed, monkeypatch, failure):
    api, journal = armed
    receipt = {"schema": "platform.email-connect-smtp-foundation.v1",
               "proposal_digest": native.FOUNDATION,
               "status": "foundation_admitted_projection_ready",
               "application_sender_changed": False, "service_account_uid": "admitted-sa"}
    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"):
        receipt[key] = True
    if failure == "status": receipt["status"] = "started"
    if failure == "digest": receipt["proposal_digest"] = "other"
    if failure == "scope": receipt["exact_read_verified"] = False
    if failure == "identity": receipt["service_account_uid"] = "different-sa"
    pins = {"ServiceAccount": "admitted-sa", "Secret": "new-secret"}
    monkeypatch.setattr(api, "projection", lambda proposal: pins)
    if failure:
        with pytest.raises(Refused):
            api.admission({"foundation_receipt": receipt})
        assert not api.calls and api.foundation_pins is None
    else:
        api.admission({"foundation_receipt": receipt})
        assert api.foundation_pins == pins
        assert journal.entries[-1]["phase"] == "native_foundation_verified"


@pytest.mark.parametrize("failure", [None, "spec", "ready", "owner", "account"])
def test_projection_checks_exact_specs_readiness_and_metadata_owner(api, monkeypatch, failure):
    foundation = {"resources": [
        {"kind": "ServiceAccount", "apiVersion": "v1", "metadata": {"name": "sa"}},
        {"kind": "SecretStore", "apiVersion": "external-secrets.io/v1", "metadata": {"name": "store"}, "spec": {"scope": "fixed"}},
        {"kind": "ExternalSecret", "apiVersion": "external-secrets.io/v1", "metadata": {"name": "projection"}, "spec": {"scope": "fixed"}},
    ]}
    monkeypatch.setattr(native, "FOUNDATION_SPEC", native.digest(foundation))
    for wanted in foundation["resources"]:
        value = obj(wanted["kind"], wanted["metadata"]["name"], spec=wanted.get("spec"))
        value["apiVersion"] = wanted["apiVersion"]
        if wanted["kind"] == "ServiceAccount": value["automountServiceAccountToken"] = False
        else: value["status"] = {"conditions": [{"type": "Ready", "status": "True"}], "refreshTime": "fixture"}
        api.objects[(wanted["kind"], wanted["metadata"]["name"])] = value
    metadata = {"uid": "secret-uid", "owners": [{"uid": "projection-uid", "controller": True}]}
    monkeypatch.setattr(api, "metadata", lambda name: metadata)
    if failure == "spec": api.objects[("SecretStore", "store")]["spec"] = {"scope": "expanded"}
    if failure == "ready": api.objects[("ExternalSecret", "projection")]["status"]["conditions"] = []
    if failure == "owner": metadata["owners"][0]["uid"] = "someone-else"
    if failure == "account": api.objects[("ServiceAccount", "sa")]["automountServiceAccountToken"] = True
    if failure:
        with pytest.raises(Refused): api.projection({"foundation": foundation})
    else:
        assert api.projection({"foundation": foundation})["Secret"] == "secret-uid"