security: seal Campaign delivery inputs

tests/test_execution_snapshot_integrity.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from govoplan_campaign.backend.campaign.models import DeliveryConfig
+from govoplan_campaign.backend.sending.execution import (
+    ExecutionSnapshotError,
+    create_execution_snapshot,
+    ensure_execution_snapshot,
+)
+
+
+def _raw_campaign() -> dict[str, object]:
+    return {
+        "version": "1.0",
+        "campaign": {"id": "campaign-1", "name": "Campaign", "mode": "send"},
+        "server": {"mail_profile_id": "profile-1"},
+        "recipients": {"from": [{"email": "sender@example.test"}]},
+        "template": {"subject": "Subject", "text": "Body", "body_mode": "text"},
+        "entries": {"inline": []},
+    }
+
+
+def _job() -> SimpleNamespace:
+    return SimpleNamespace(
+        id="job-1",
+        campaign_version_id="version-1",
+        entry_index=0,
+        entry_id="entry-1",
+        recipient_email="recipient@example.test",
+        subject="Subject",
+        message_id_header="<message@example.test>",
+        eml_size_bytes=123,
+        eml_sha256="eml-digest",
+        build_status="built",
+        validation_status="ready",
+        resolved_recipients={"to": ["recipient@example.test"]},
+        resolved_attachments=[],
+        issues_snapshot=[],
+    )
+
+
+class _Query:
+    def __init__(self, jobs: list[SimpleNamespace]):
+        self.jobs = jobs
+
+    def filter(self, *_args):
+        return self
+
+    def order_by(self, *_args):
+        return self
+
+    def all(self):
+        return list(self.jobs)
+
+
+class _Session:
+    def __init__(self, jobs: list[SimpleNamespace]):
+        self.jobs = jobs
+
+    def query(self, _model):
+        return _Query(self.jobs)
+
+
+def _snapshotted_version(job: SimpleNamespace):
+    version = SimpleNamespace(
+        id="version-1",
+        campaign_id="campaign-1",
+        raw_json=_raw_campaign(),
+        build_summary={"build_token": "build-1", "built_at": "2026-07-21T00:00:00+00:00"},
+    )
+    payload, digest = create_execution_snapshot(
+        version,  # type: ignore[arg-type]
+        mail_profile_id="profile-1",
+        smtp_transport_revision="smtp-revision",
+        imap_transport_revision="imap-revision",
+        delivery=DeliveryConfig(),
+        jobs=[job],  # type: ignore[list-item]
+        build_summary=version.build_summary,
+    )
+    version.execution_snapshot = payload
+    version.execution_snapshot_hash = digest
+    return version
+
+
+def _ensure(session: _Session, version) -> None:
+    with patch(
+        "govoplan_campaign.backend.sending.execution.files_integration",
+        return_value=SimpleNamespace(available=False),
+    ):
+        ensure_execution_snapshot(session, version)  # type: ignore[arg-type]
+
+
+def test_v5_snapshot_requires_its_persisted_checksum() -> None:
+    job = _job()
+    version = _snapshotted_version(job)
+    version.execution_snapshot_hash = None
+
+    with pytest.raises(ExecutionSnapshotError, match="checksum is missing"):
+        _ensure(_Session([job]), version)
+
+
+def test_campaign_json_drift_is_rejected_before_delivery() -> None:
+    job = _job()
+    version = _snapshotted_version(job)
+    version.raw_json["template"] = {"subject": "Changed", "text": "Body"}
+
+    with pytest.raises(ExecutionSnapshotError, match="Campaign inputs changed"):
+        _ensure(_Session([job]), version)
+
+
+def test_post_build_recipient_drift_cannot_redirect_delivery() -> None:
+    job = _job()
+    version = _snapshotted_version(job)
+    job.resolved_recipients = {"to": ["attacker@example.test"]}
+
+    with pytest.raises(ExecutionSnapshotError, match="job inputs changed"):
+        _ensure(_Session([job]), version)
+
+
+def test_effect_check_verifies_only_the_claimed_job_in_constant_time() -> None:
+    job = _job()
+    version = _snapshotted_version(job)
+    session = SimpleNamespace(
+        query=lambda *_args: pytest.fail("per-effect validation must not rescan every campaign job")
+    )
+
+    with patch(
+        "govoplan_campaign.backend.sending.execution.files_integration",
+        return_value=SimpleNamespace(available=False),
+    ):
+        ensure_execution_snapshot(
+            session,  # type: ignore[arg-type]
+            version,
+            effect_job=job,  # type: ignore[arg-type]
+        )