security: harden Campaign delivery effects and reconciliation

tests/test_report_email_security.py

@@ -0,0 +1,169 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+from fastapi import HTTPException
+from pydantic import ValidationError
+
+from govoplan_campaign.backend import router
+from govoplan_campaign.backend.reports.emailing import CampaignReportEmailError, send_campaign_report_email
+from govoplan_campaign.backend.schemas import ReportEmailRequest
+
+
+def test_report_email_recipient_schema_normalizes_and_has_safe_attachment_default() -> None:
+    request = ReportEmailRequest.model_validate({
+        "to": ["  First@Example.test ", "first@example.test", "second@example.test"],
+    })
+
+    assert request.to == ["First@Example.test", "second@example.test"]
+    assert request.attach_jobs_csv is False
+
+
+@pytest.mark.parametrize(
+    "recipients",
+    [
+        [],
+        [f"recipient-{index}@example.test" for index in range(51)],
+        ["a" * 310 + "@example.test"],
+        ["victim@example.test\r\nBcc: attacker@example.test"],
+        ["missing-at.example.test"],
+        ["@example.test"],
+        ["local@"],
+        ["local @example.test"],
+        ["one@two@example.test"],
+        ["victim@example.test,Bcc:attacker"],
+        ["Display<a@example.test>"],
+    ],
+)
+def test_report_email_recipient_schema_rejects_unsafe_addresses(recipients: list[str]) -> None:
+    with pytest.raises(ValidationError):
+        ReportEmailRequest.model_validate({"to": recipients})
+
+
+def test_report_jobs_csv_attachment_requires_recipient_export_permission() -> None:
+    payload = ReportEmailRequest(to=["recipient@example.test"], attach_jobs_csv=True)
+    principal = SimpleNamespace(tenant_id="tenant-1")
+
+    with (
+        patch.object(
+            router,
+            "_get_campaign_for_principal",
+            return_value=SimpleNamespace(id="campaign-1", current_version_id=None),
+        ),
+        patch.object(
+            router,
+            "_require_permission",
+            side_effect=HTTPException(status_code=403, detail="missing export"),
+        ) as require_permission,
+        patch.object(router, "send_campaign_report_email") as send_report,
+    ):
+        with pytest.raises(HTTPException) as captured:
+            router.email_campaign_report(
+                "campaign-1",
+                payload,
+                session=object(),  # type: ignore[arg-type]
+                principal=principal,  # type: ignore[arg-type]
+            )
+
+    assert captured.value.status_code == 403
+    require_permission.assert_called_once_with(principal, "campaigns:recipient:export")
+    send_report.assert_not_called()
+
+
+def test_report_email_requires_permission_to_use_selected_mail_profile() -> None:
+    payload = ReportEmailRequest(to=["recipient@example.test"])
+    principal = SimpleNamespace(tenant_id="tenant-1")
+    campaign = SimpleNamespace(id="campaign-1", current_version_id="version-1")
+    version = SimpleNamespace(
+        id="version-1",
+        campaign_id="campaign-1",
+        raw_json={"server": {"mail_profile_id": "profile-1"}},
+    )
+    session = SimpleNamespace(get=lambda _model, _id: version)
+    with (
+        patch.object(router, "_get_campaign_for_principal", return_value=campaign),
+        patch.object(
+            router,
+            "_require_mail_profile_use_if_needed",
+            side_effect=HTTPException(status_code=403, detail="missing profile use"),
+        ) as require_profile_use,
+        patch.object(router, "send_campaign_report_email") as send_report,
+    ):
+        with pytest.raises(HTTPException) as captured:
+            router.email_campaign_report(
+                "campaign-1",
+                payload,
+                session=session,  # type: ignore[arg-type]
+                principal=principal,  # type: ignore[arg-type]
+            )
+
+    assert captured.value.status_code == 403
+    require_profile_use.assert_called_once_with(principal, version.raw_json)
+    send_report.assert_not_called()
+
+
+def test_unexpected_report_failure_does_not_leak_internal_details() -> None:
+    payload = ReportEmailRequest(to=["recipient@example.test"])
+    principal = SimpleNamespace(tenant_id="tenant-1")
+    with (
+        patch.object(
+            router,
+            "_get_campaign_for_principal",
+            return_value=SimpleNamespace(id="campaign-1", current_version_id=None),
+        ),
+        patch.object(
+            router,
+            "send_campaign_report_email",
+            side_effect=RuntimeError("smtp.internal.example provider-secret"),
+        ),
+    ):
+        with pytest.raises(HTTPException) as captured:
+            router.email_campaign_report(
+                "campaign-1",
+                payload,
+                session=object(),  # type: ignore[arg-type]
+                principal=principal,  # type: ignore[arg-type]
+            )
+
+    assert captured.value.status_code == 500
+    assert captured.value.detail == "Campaign report email could not be completed."
+
+
+def test_report_send_fails_closed_until_durable_mail_outbox_exists() -> None:
+    campaign = SimpleNamespace(
+        id="campaign-1",
+        tenant_id="tenant-1",
+        name="Campaign",
+        external_id="campaign-1",
+    )
+    version = SimpleNamespace(
+        id="version-1",
+        campaign_id="campaign-1",
+        raw_json={"server": {"mail_profile_id": "profile-1"}},
+        execution_snapshot={"snapshot_version": "5"},
+    )
+    config = SimpleNamespace(
+        server=SimpleNamespace(profile_capabilities=SimpleNamespace(smtp_available=True)),
+    )
+    snapshot = SimpleNamespace(smtp_transport_revision="frozen-build-revision")
+    class Session:
+        def get(self, _model, _id):
+            return campaign
+
+    with (
+        patch("govoplan_campaign.backend.reports.emailing._selected_version", return_value=version),
+        patch("govoplan_campaign.backend.reports.emailing._load_config", return_value=config),
+        patch("govoplan_campaign.backend.reports.emailing.ensure_execution_snapshot", return_value=snapshot),
+        patch("govoplan_campaign.backend.reports.emailing.generate_campaign_report") as generate_report,
+    ):
+        with pytest.raises(CampaignReportEmailError, match="durable, idempotent Mail-owned outbox"):
+            send_campaign_report_email(
+                Session(),  # type: ignore[arg-type]
+                tenant_id="tenant-1",
+                campaign_id="campaign-1",
+                to=["recipient@example.test"],
+            )
+
+    generate_report.assert_not_called()