govoplan-campaign/tests/test_report_email_security.py

from __future__ import annotations

from types import SimpleNamespace
from unittest.mock import patch
from email.message import EmailMessage

import pytest
from fastapi import HTTPException
from pydantic import ValidationError

from govoplan_campaign.backend.routes import reports as 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_email_recipient_error_identifies_the_invalid_list_item() -> None:
    with pytest.raises(ValidationError) as captured:
        ReportEmailRequest.model_validate({
            "to": ["first@example.test", "invalid recipient", "third@example.test"],
        })

    assert captured.value.errors()[0]["loc"] == ("to", 1)
    assert "email addresses" in captured.value.errors()[0]["msg"]


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",
        has=lambda scope: scope == "campaigns:recipient:export",
    )
    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",
        has=lambda scope: scope == "campaigns:recipient:export",
    )
    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()


def test_live_report_submits_exact_message_to_durable_mail_command() -> 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": "7"},
    )
    config = SimpleNamespace(
        server=SimpleNamespace(
            profile_capabilities=SimpleNamespace(smtp_available=True)
        ),
    )
    snapshot = SimpleNamespace(
        mail_profile_id="profile-1",
        smtp_server_id="smtp-1",
        smtp_credential_id="credential-1",
        smtp_transport_revision="revision-1",
    )
    message = EmailMessage()
    message["From"] = "Sender <sender@example.test>"
    message["To"] = "recipient@example.test"
    message["Subject"] = "Report"
    message.set_content("content")
    mail = SimpleNamespace(
        durable_delivery_available=True,
        submit_delivery_command=lambda _session, **_kwargs: {
            "id": "command-1",
            "status": "pending",
            "duplicate": False,
        },
    )

    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",
            return_value={},
        ),
        patch(
            "govoplan_campaign.backend.reports.emailing.build_report_message",
            return_value=message,
        ),
        patch(
            "govoplan_campaign.backend.reports.emailing._effective_from",
            return_value=("sender@example.test", "Sender"),
        ),
        patch(
            "govoplan_campaign.backend.reports.emailing.mail_integration",
            return_value=mail,
        ),
    ):
        result = send_campaign_report_email(
            Session(),  # type: ignore[arg-type]
            tenant_id="tenant-1",
            campaign_id="campaign-1",
            to=["recipient@example.test"],
            idempotency_key="request-1",
            created_by_user_id="user-1",
        )

    assert result.command_id == "command-1"
    assert result.delivery_status == "pending"
    assert result.sent is False
    assert result.audit_dict()["recipient_count"] == 1
    assert "recipient@example.test" not in repr(result.audit_dict())