From 50c509d16154a3bb0d0041c1fa98eb2503f7fbde Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 21 Jul 2026 17:53:43 +0200 Subject: [PATCH] security: seal Campaign delivery inputs --- src/govoplan_campaign/backend/db/models.py | 1 + .../4d5e6f7a9203_execution_input_seal.py | 26 ++++ .../4d5e6f7a9203_v019_execution_input_seal.py | 26 ++++ src/govoplan_campaign/backend/schemas.py | 9 +- .../backend/sending/execution.py | 129 +++++++++++++--- src/govoplan_campaign/backend/sending/jobs.py | 22 ++- tests/test_execution_snapshot_integrity.py | 139 ++++++++++++++++++ tests/test_imap_append_idempotency.py | 42 +++++- 8 files changed, 350 insertions(+), 44 deletions(-) create mode 100644 src/govoplan_campaign/backend/migrations/dev_versions/4d5e6f7a9203_execution_input_seal.py create mode 100644 src/govoplan_campaign/backend/migrations/versions/4d5e6f7a9203_v019_execution_input_seal.py create mode 100644 tests/test_execution_snapshot_integrity.py diff --git a/src/govoplan_campaign/backend/db/models.py b/src/govoplan_campaign/backend/db/models.py index 4cc23e6..52a9b6b 100644 --- a/src/govoplan_campaign/backend/db/models.py +++ b/src/govoplan_campaign/backend/db/models.py @@ -240,6 +240,7 @@ class CampaignJob(Base, TimestampMixin): eml_local_path: Mapped[str | None] = mapped_column(String(1000)) eml_size_bytes: Mapped[int | None] = mapped_column(Integer) eml_sha256: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) + execution_input_sha256: Mapped[str | None] = mapped_column(String(64), nullable=True) build_status: Mapped[str] = mapped_column(String(50), default=JobBuildStatus.PENDING.value, nullable=False, index=True) validation_status: Mapped[str] = mapped_column(String(50), default=JobValidationStatus.NEEDS_REVIEW.value, nullable=False, index=True) diff --git a/src/govoplan_campaign/backend/migrations/dev_versions/4d5e6f7a9203_execution_input_seal.py b/src/govoplan_campaign/backend/migrations/dev_versions/4d5e6f7a9203_execution_input_seal.py new file mode 100644 index 0000000..e6779e0 --- /dev/null +++ b/src/govoplan_campaign/backend/migrations/dev_versions/4d5e6f7a9203_execution_input_seal.py @@ -0,0 +1,26 @@ +"""seal each campaign delivery job's immutable execution input + +Revision ID: 4d5e6f7a9203 +Revises: 3c4d5e6f8192 +Create Date: 2026-07-21 00:00:01.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "4d5e6f7a9203" +down_revision = "3c4d5e6f8192" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("campaign_jobs") as batch: + batch.add_column(sa.Column("execution_input_sha256", sa.String(length=64), nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table("campaign_jobs") as batch: + batch.drop_column("execution_input_sha256") diff --git a/src/govoplan_campaign/backend/migrations/versions/4d5e6f7a9203_v019_execution_input_seal.py b/src/govoplan_campaign/backend/migrations/versions/4d5e6f7a9203_v019_execution_input_seal.py new file mode 100644 index 0000000..e6779e0 --- /dev/null +++ b/src/govoplan_campaign/backend/migrations/versions/4d5e6f7a9203_v019_execution_input_seal.py @@ -0,0 +1,26 @@ +"""seal each campaign delivery job's immutable execution input + +Revision ID: 4d5e6f7a9203 +Revises: 3c4d5e6f8192 +Create Date: 2026-07-21 00:00:01.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "4d5e6f7a9203" +down_revision = "3c4d5e6f8192" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("campaign_jobs") as batch: + batch.add_column(sa.Column("execution_input_sha256", sa.String(length=64), nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table("campaign_jobs") as batch: + batch.drop_column("execution_input_sha256") diff --git a/src/govoplan_campaign/backend/schemas.py b/src/govoplan_campaign/backend/schemas.py index dbca868..e14abcc 100644 --- a/src/govoplan_campaign/backend/schemas.py +++ b/src/govoplan_campaign/backend/schemas.py @@ -421,11 +421,10 @@ class CampaignResolveOutcomeRequest(BaseModel): note: str | None = Field(default=None, max_length=2000) @model_validator(mode="after") - def require_imap_reconciliation_evidence(self) -> "CampaignResolveOutcomeRequest": - if self.decision.startswith("imap_"): - self.note = (self.note or "").strip() - if not self.note: - raise ValueError("IMAP reconciliation requires an evidence note") + def require_reconciliation_evidence(self) -> "CampaignResolveOutcomeRequest": + self.note = (self.note or "").strip() + if not self.note: + raise ValueError("Reconciliation requires an evidence note") return self diff --git a/src/govoplan_campaign/backend/sending/execution.py b/src/govoplan_campaign/backend/sending/execution.py index 84a8bd8..6e605f6 100644 --- a/src/govoplan_campaign/backend/sending/execution.py +++ b/src/govoplan_campaign/backend/sending/execution.py @@ -124,28 +124,35 @@ def _policy_fingerprint(raw_json: dict[str, Any], delivery: DeliveryConfig) -> s ) +def _job_execution_input_payload(job: CampaignJob) -> dict[str, Any]: + return { + "job_id": job.id, + "entry_index": job.entry_index, + "entry_id": job.entry_id, + "recipient_email": job.recipient_email, + "subject": job.subject, + "message_id_header": job.message_id_header, + "eml_size_bytes": job.eml_size_bytes, + "eml_sha256": job.eml_sha256, + "build_status": job.build_status, + "validation_status": job.validation_status, + "resolved_recipients_sha256": _sha256(job.resolved_recipients or {}), + "resolved_attachments_sha256": _sha256(job.resolved_attachments or []), + "issues_sha256": _sha256(job.issues_snapshot or []), + } + + +def job_execution_input_hash(job: CampaignJob) -> str: + return _sha256(_job_execution_input_payload(job)) + + def job_manifest_hash(jobs: Iterable[CampaignJob]) -> str: """Hash the immutable per-message execution records in stable order.""" - payload: list[dict[str, Any]] = [] - for job in sorted(jobs, key=lambda item: (item.entry_index, item.id)): - payload.append( - { - "job_id": job.id, - "entry_index": job.entry_index, - "entry_id": job.entry_id, - "recipient_email": job.recipient_email, - "subject": job.subject, - "message_id_header": job.message_id_header, - "eml_size_bytes": job.eml_size_bytes, - "eml_sha256": job.eml_sha256, - "build_status": job.build_status, - "validation_status": job.validation_status, - "resolved_recipients_sha256": _sha256(job.resolved_recipients or {}), - "resolved_attachments_sha256": _sha256(job.resolved_attachments or []), - "issues_sha256": _sha256(job.issues_snapshot or []), - } - ) + payload = [ + _job_execution_input_payload(job) + for job in sorted(jobs, key=lambda item: (item.entry_index, item.id)) + ] return _sha256(payload) @@ -161,6 +168,8 @@ def create_execution_snapshot( ) -> tuple[dict[str, Any], str]: raw_json = version.raw_json if isinstance(version.raw_json, dict) else {} job_list = list(jobs) + for job in job_list: + job.execution_input_sha256 = job_execution_input_hash(job) summary = build_summary if isinstance(build_summary, dict) else {} queueable_statuses = {JobValidationStatus.READY.value, JobValidationStatus.WARNING.value} payload = ExecutionSnapshot( @@ -181,7 +190,77 @@ def create_execution_snapshot( return payload, snapshot_hash(payload) -def ensure_execution_snapshot(session: Session, version: CampaignVersion) -> ExecutionSnapshot: +def _assert_snapshot_matches_persisted_inputs( + session: Session, + version: CampaignVersion, + snapshot: ExecutionSnapshot, + *, + effect_job: CampaignJob | None = None, +) -> None: + """Fail closed when any build-bound input drifted after snapshot creation.""" + + raw_json = version.raw_json if isinstance(version.raw_json, dict) else {} + if snapshot.campaign_version_id != version.id: + raise ExecutionSnapshotError("Execution snapshot campaign version mismatch") + if snapshot.campaign_json_sha256 != _sha256(raw_json): + raise ExecutionSnapshotError( + "Campaign inputs changed after this execution snapshot was built. " + "Revalidate and rebuild the campaign before delivery." + ) + if not snapshot.smtp_transport_revision: + raise ExecutionSnapshotError("Execution snapshot has no SMTP transport revision") + if not snapshot.job_manifest_sha256: + raise ExecutionSnapshotError("Execution snapshot has no built-job manifest checksum") + if not snapshot.effective_policy_sha256: + raise ExecutionSnapshotError("Execution snapshot has no effective-policy checksum") + if snapshot.effective_policy_sha256 != _policy_fingerprint(raw_json, snapshot.delivery): + raise ExecutionSnapshotError( + "Campaign delivery policy changed after the execution snapshot was created. " + "Revalidate and rebuild the campaign before delivery." + ) + + if effect_job is not None: + if effect_job.campaign_version_id != version.id: + raise ExecutionSnapshotError("Campaign job does not belong to the snapshotted version") + if not getattr(effect_job, "execution_input_sha256", None): + raise ExecutionSnapshotError("Campaign job has no execution-input checksum; rebuild before delivery") + if effect_job.execution_input_sha256 != job_execution_input_hash(effect_job): + raise ExecutionSnapshotError( + "Built campaign job inputs changed after the execution snapshot was created. " + "Revalidate and rebuild the campaign before delivery." + ) + return + + jobs = ( + session.query(CampaignJob) + .filter(CampaignJob.campaign_version_id == version.id) + .order_by(CampaignJob.entry_index.asc(), CampaignJob.id.asc()) + .all() + ) + queueable_statuses = {JobValidationStatus.READY.value, JobValidationStatus.WARNING.value} + if snapshot.job_count != len(jobs): + raise ExecutionSnapshotError( + "Built campaign jobs changed after the execution snapshot was created. " + "Revalidate and rebuild the campaign before delivery." + ) + queueable_count = sum(1 for job in jobs if job.validation_status in queueable_statuses) + if ( + snapshot.queueable_job_count != queueable_count + or snapshot.job_manifest_sha256 != job_manifest_hash(jobs) + or any(getattr(job, "execution_input_sha256", None) != job_execution_input_hash(job) for job in jobs) + ): + raise ExecutionSnapshotError( + "Built campaign job inputs changed after the execution snapshot was created. " + "Revalidate and rebuild the campaign before delivery." + ) + + +def ensure_execution_snapshot( + session: Session, + version: CampaignVersion, + *, + effect_job: CampaignJob | None = None, +) -> ExecutionSnapshot: """Return a validated snapshot, creating one for pre-migration builds. New builds create the snapshot after persisting their jobs. The fallback is @@ -208,9 +287,17 @@ def ensure_execution_snapshot(session: Session, version: CampaignVersion) -> Exe ) snapshot = ExecutionSnapshot.model_validate(version.execution_snapshot) expected = snapshot_hash(snapshot.model_dump(mode="json")) - if version.execution_snapshot_hash and version.execution_snapshot_hash != expected: + if not version.execution_snapshot_hash: + raise ExecutionSnapshotError("Execution snapshot checksum is missing") + if version.execution_snapshot_hash != expected: raise ExecutionSnapshotError("Execution snapshot checksum mismatch") _assert_snapshot_profile_matches_version(version, snapshot) + _assert_snapshot_matches_persisted_inputs( + session, + version, + snapshot, + effect_job=effect_job, + ) return snapshot from govoplan_campaign.backend.persistence.campaigns import load_version_config diff --git a/src/govoplan_campaign/backend/sending/jobs.py b/src/govoplan_campaign/backend/sending/jobs.py index e7f3023..b91784e 100644 --- a/src/govoplan_campaign/backend/sending/jobs.py +++ b/src/govoplan_campaign/backend/sending/jobs.py @@ -1063,11 +1063,10 @@ def reconcile_job_outcome( note=note, commit=commit, ) - if job.send_status not in { - JobSendStatus.OUTCOME_UNKNOWN.value, - JobSendStatus.SENDING.value, - JobSendStatus.CLAIMED.value, - }: + evidence_note = (note or "").strip() + if not evidence_note: + raise QueueingError("SMTP reconciliation requires an evidence note") + if job.send_status != JobSendStatus.OUTCOME_UNKNOWN.value: raise QueueingError(f"Job status {job.send_status} does not require reconciliation") version = _get_version_for_campaign(session, campaign, version_id=job.campaign_version_id) snapshot = ensure_execution_snapshot(session, version) @@ -1079,7 +1078,7 @@ def reconcile_job_outcome( job.sent_at = job.sent_at or now job.outcome_unknown_at = None job.claim_token = None - job.last_error = note or "Operator confirmed SMTP acceptance after an uncertain outcome." + job.last_error = evidence_note job.imap_status = JobImapStatus.PENDING.value if snapshot.delivery.imap_append_sent.enabled else JobImapStatus.NOT_REQUESTED.value files_integration().mark_job_attachment_uses_sent(session, job) attempt_status = "reconciled_smtp_accepted" @@ -1088,7 +1087,7 @@ def reconcile_job_outcome( job.queue_status = JobQueueStatus.DRAFT.value job.outcome_unknown_at = None job.claim_token = None - job.last_error = note or "Operator confirmed that SMTP did not accept this message." + job.last_error = evidence_note attempt_status = "reconciled_not_sent" else: raise QueueingError("decision must be 'smtp_accepted' or 'not_sent'") @@ -1131,10 +1130,7 @@ def _reconcile_imap_append_outcome( raise QueueingError("IMAP reconciliation requires an evidence note") if job.send_status not in SMTP_ACCEPTED_STATUSES: raise QueueingError("Only an SMTP-accepted job can reconcile a Sent-folder append") - if job.imap_status not in { - JobImapStatus.OUTCOME_UNKNOWN.value, - JobImapStatus.APPENDING.value, - }: + if job.imap_status != JobImapStatus.OUTCOME_UNKNOWN.value: raise QueueingError(f"IMAP status {job.imap_status} does not require reconciliation") attempt = ( @@ -1516,7 +1512,7 @@ def _send_job_delivery_context(session: Session, job: CampaignJob) -> _SendJobDe if not version: raise SendJobError("Campaign version not found") try: - snapshot = ensure_execution_snapshot(session, version) + snapshot = ensure_execution_snapshot(session, version, effect_job=job) except ExecutionSnapshotError as exc: raise SendJobError(str(exc)) from exc @@ -2044,7 +2040,7 @@ def append_sent_for_job(session: Session, *, job_id: str, dry_run: bool = False) if not version: raise SendJobError("Campaign version not found") try: - snapshot = ensure_execution_snapshot(session, version) + snapshot = ensure_execution_snapshot(session, version, effect_job=job) except ExecutionSnapshotError as exc: raise SendJobError(str(exc)) from exc if not snapshot.delivery.imap_append_sent.enabled: diff --git a/tests/test_execution_snapshot_integrity.py b/tests/test_execution_snapshot_integrity.py new file mode 100644 index 0000000..cc56208 --- /dev/null +++ b/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="", + 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] + ) diff --git a/tests/test_imap_append_idempotency.py b/tests/test_imap_append_idempotency.py index 45e8654..1355664 100644 --- a/tests/test_imap_append_idempotency.py +++ b/tests/test_imap_append_idempotency.py @@ -218,10 +218,11 @@ def test_imap_claim_migration_preserves_and_renumbers_duplicate_attempts() -> No ] -def test_imap_reconciliation_requires_an_evidence_note() -> None: +@pytest.mark.parametrize("decision", ["smtp_accepted", "not_sent", "imap_appended", "imap_not_appended"]) +def test_reconciliation_requires_an_evidence_note(decision: str) -> None: with pytest.raises(ValueError, match="evidence note"): CampaignResolveOutcomeRequest( - decision="imap_not_appended", + decision=decision, # type: ignore[arg-type] note=" ", ) @@ -299,7 +300,11 @@ def test_imap_reconciliation_preserves_attempt_and_only_not_appended_is_retryabl files().mark_job_attachment_uses_sent.assert_not_called() -def test_imap_reconciliation_rejects_a_retryable_or_completed_state() -> None: +@pytest.mark.parametrize( + "imap_status", + [JobImapStatus.APPENDING.value, JobImapStatus.FAILED.value, JobImapStatus.APPENDED.value], +) +def test_imap_reconciliation_rejects_live_retryable_or_completed_state(imap_status: str) -> None: campaign = SimpleNamespace(id="campaign-1") job = SimpleNamespace( id="job-1", @@ -307,7 +312,7 @@ def test_imap_reconciliation_rejects_a_retryable_or_completed_state() -> None: campaign_id="campaign-1", campaign_version_id="version-1", send_status=JobSendStatus.SMTP_ACCEPTED.value, - imap_status=JobImapStatus.FAILED.value, + imap_status=imap_status, ) session = MagicMock() session.get.return_value = job @@ -326,10 +331,37 @@ def test_imap_reconciliation_rejects_a_retryable_or_completed_state() -> None: ) +@pytest.mark.parametrize("send_status", [JobSendStatus.CLAIMED.value, JobSendStatus.SENDING.value]) +def test_smtp_reconciliation_rejects_a_live_worker_state(send_status: str) -> None: + campaign = SimpleNamespace(id="campaign-1") + job = SimpleNamespace( + id="job-1", + tenant_id="tenant-1", + campaign_id="campaign-1", + campaign_version_id="version-1", + send_status=send_status, + ) + session = MagicMock() + session.get.return_value = job + with patch( + "govoplan_campaign.backend.sending.jobs._get_campaign_for_tenant", + return_value=campaign, + ): + with pytest.raises(QueueingError, match="does not require reconciliation"): + reconcile_job_outcome( + session, + tenant_id="tenant-1", + campaign_id="campaign-1", + job_id="job-1", + decision="not_sent", + note="Checked provider logs.", + ) + + @pytest.mark.parametrize( ("decision", "note"), [ - ("smtp_accepted", None), + ("smtp_accepted", "SMTP provider evidence checked."), ("imap_appended", "Mailbox evidence checked."), ], )