"""Real stored hierarchy and policy; provider calls are replaced, never live.""" from __future__ import annotations import json from types import SimpleNamespace from unittest.mock import Mock, patch import pytest from sqlalchemy import Column, String, Table, create_engine from sqlalchemy.orm import Session from govoplan_core.admin.models import SystemSettings from govoplan_core.core.campaigns import CampaignMailPolicyContext from govoplan_core.db.base import Base from govoplan_core.security.credential_envelopes import CredentialEnvelope from govoplan_core.security.secrets import encrypt_secret from govoplan_core.tenancy.scope import Tenant from govoplan_mail.backend import capabilities, mail_profiles, server_hierarchy from govoplan_mail.backend.db.models import MailProfilePolicy, MailServerCredentialBinding, MailServerEndpoint, MailServerProfile from govoplan_mail.backend.sending import smtp as smtp_module @pytest.fixture def hierarchy(tmp_path): engine = create_engine(f"sqlite+pysqlite:///{tmp_path / 'smtp-policy.db'}") if "access_users" not in Base.metadata.tables: Table("access_users", Base.metadata, Column("id", String(36), primary_key=True)) for table in (Base.metadata.tables["access_users"], SystemSettings.__table__, Tenant.__table__, CredentialEnvelope.__table__, MailServerProfile.__table__, MailServerEndpoint.__table__, MailServerCredentialBinding.__table__, MailProfilePolicy.__table__): table.create(engine) with Session(engine) as session: session.add(SystemSettings(id="global", settings={})) session.add(Tenant(id="tenant-1", slug="test", name="Test", settings={})) profile = MailServerProfile(id="profile-1", tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", name="Test mail", slug="test", smtp_config={"host": "smtp.example.test", "port": 587, "security": "starttls"}, imap_config={"host": "imap.example.test", "port": 993, "security": "tls"}, smtp_transport_revision="smtp-current", imap_transport_revision="imap-current", inherit_to_lower_scopes=True) session.add(profile) for protocol, port in (("smtp", 587), ("imap", 993)): server = MailServerEndpoint(id=f"{protocol}-server", profile_id=profile.id, tenant_id="tenant-1", protocol=protocol, name=protocol, scope_type="tenant", scope_id="tenant-1", inherit_to_lower_scopes=True, is_default=True, is_active=True, transport_revision=f"{protocol}-current", config={"host": f"{protocol}.example.test", "port": port, "security": "starttls" if protocol == "smtp" else "tls"}) credential = CredentialEnvelope(id=f"{protocol}-credential", tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", name=protocol, credential_kind="username_password", public_data={"username": f"{protocol}-user"}, secret_data_encrypted=encrypt_secret(json.dumps({"password": f"fake-{protocol}-password"})), secret_keys=["password"], allowed_modules=["mail"], allowed_server_refs=[f"mail:{protocol}-server"], inherit_to_lower_scopes=True, is_active=True) session.add_all([server, credential, MailServerCredentialBinding(id=f"{protocol}-binding", server_id=server.id, credential_id=credential.id, is_default=True)]) for scope in ("system", "tenant"): session.add(MailProfilePolicy(id=f"{scope}-policy", tenant_id=None if scope == "system" else "tenant-1", scope_type=scope, scope_id=None if scope == "system" else "tenant-1", policy={"smtp_credentials": {"inherit": False}, "imap_credentials": {"inherit": False}})) session.commit() context = CampaignMailPolicyContext(id="campaign-1", tenant_id="tenant-1") provider = SimpleNamespace(get_campaign_mail_policy_context=lambda *_args, **_kwargs: context) # Only the optional Campaign context provider and external I/O are mocked. # Profiles, policy inheritance, endpoint/credential ACLs and decryption are real. with patch.object(mail_profiles, "_campaign_policy_provider", return_value=provider), \ patch("socket.create_connection", side_effect=AssertionError("No live network in regression tests")): yield SimpleNamespace(session=session, profile=profile, context=context) engine.dispose() def smtp_args(**overrides): return {"tenant_id": "tenant-1", "campaign_id": "campaign-1", "profile_id": "profile-1", "envelope_from": "sender@example.test", "envelope_recipients": ["recipient@example.test"], "from_header": "sender@example.test", "expected_smtp_transport_revision": "smtp-current", "smtp_server_id": "smtp-server", "smtp_credential_id": "smtp-credential", **overrides} def selection(): return {"smtp_server_id": "smtp-server", "smtp_credential_id": "smtp-credential", "imap_server_id": "imap-server", "imap_credential_id": "imap-credential"} def test_smtp_batch_accepts_explicit_smtp_when_other_protocol_requires_explicit_selection(hierarchy): # The full frozen selection is valid; runtime SMTP deliberately carries only SMTP. summary = capabilities.campaign_profile_delivery_summary(hierarchy.session, tenant_id="tenant-1", campaign_id="campaign-1", profile_id="profile-1", **selection()) assert summary["smtp_available"] and summary["imap_available"] fake_connection = Mock() with patch.object(smtp_module, "_open_smtp", return_value=fake_connection) as opener, \ patch.object(server_hierarchy, "resolve_credential_envelope", wraps=server_hierarchy.resolve_credential_envelope) as decrypt: with capabilities.campaign_smtp_batch(hierarchy.session, **smtp_args()) as batch: assert batch.status == "ready" assert batch.connection_count == 1 assert [call.kwargs["credential_id"] for call in decrypt.call_args_list] == ["smtp-credential"] assert opener.call_args.args[0].username == "smtp-user" fake_connection.sendmail.assert_not_called() fake_connection.send_message.assert_not_called() def test_smtp_single_uses_same_selected_protocol_authorization(hierarchy): result = SimpleNamespace(envelope_recipients=["recipient@example.test"], refused_recipients={}) with patch.object(capabilities, "send_email_bytes", return_value=result) as send, \ patch.object(server_hierarchy, "resolve_credential_envelope", wraps=server_hierarchy.resolve_credential_envelope) as decrypt: sent = capabilities.send_campaign_email_bytes(hierarchy.session, message_bytes=b"frozen test message", **smtp_args()) assert sent.accepted_count == 1 assert [call.kwargs["credential_id"] for call in decrypt.call_args_list] == ["smtp-credential"] assert send.call_args.kwargs["smtp_config"].username == "smtp-user" def imap_args(**overrides): return {"tenant_id": "tenant-1", "campaign_id": "campaign-1", "profile_id": "profile-1", "message_bytes": b"frozen test message", "folder": "Sent", "expected_smtp_transport_revision": "smtp-current", "expected_imap_transport_revision": "imap-current", # An IMAP-only call need not submit an unrelated SMTP credential. "smtp_server_id": "smtp-server", "imap_server_id": "imap-server", "imap_credential_id": "imap-credential", **overrides} def test_imap_append_checks_and_decrypts_only_selected_protocol(hierarchy): with patch.object(capabilities, "append_message_to_sent", return_value=SimpleNamespace(folder="Sent")) as append, \ patch.object(server_hierarchy, "resolve_credential_envelope", wraps=server_hierarchy.resolve_credential_envelope) as decrypt: result = capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args()) assert result.folder == "Sent" assert [call.kwargs["credential_id"] for call in decrypt.call_args_list] == ["imap-credential"] assert append.call_args.kwargs["imap_config"].username == "imap-user" @pytest.mark.parametrize("operation", ["smtp_batch", "smtp_single", "imap_append", "imap_endpoint_only"]) def test_selected_protocol_still_requires_explicit_credentials_before_decryption_or_provider(hierarchy, operation): if operation == "imap_endpoint_only": hierarchy.profile.imap_config = None # Current endpoint exists without the legacy mirror. hierarchy.session.commit() with patch.object(server_hierarchy, "resolve_credential_envelope", side_effect=AssertionError("Policy must reject before decryption")) as decrypt, \ patch.object(smtp_module, "_open_smtp", side_effect=AssertionError("No network")) as opener, \ patch.object(capabilities, "send_email_bytes", side_effect=AssertionError("No SMTP")) as send, \ patch.object(capabilities, "append_message_to_sent", side_effect=AssertionError("No IMAP")) as append: with pytest.raises(mail_profiles.MailProfileError, match=f"effective {'SMTP' if operation.startswith('smtp') else 'IMAP'}"): if operation == "smtp_batch": with capabilities.campaign_smtp_batch(hierarchy.session, **smtp_args(smtp_credential_id=None)): pass elif operation == "smtp_single": capabilities.send_campaign_email_bytes(hierarchy.session, message_bytes=b"test", **smtp_args(smtp_credential_id=None)) else: capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(imap_credential_id=None)) decrypt.assert_not_called(); opener.assert_not_called(); send.assert_not_called(); append.assert_not_called() @pytest.mark.parametrize("operation", ["smtp_batch", "smtp_single", "imap_append"]) @pytest.mark.parametrize("mutation", ["stale_revision", "inactive_credential", "wrong_server", "wrong_tenant"]) def test_selected_transport_revision_and_credential_authority_remain_fail_closed(hierarchy, operation, mutation): protocol = "smtp" if operation.startswith("smtp") else "imap" overrides = {} if mutation == "stale_revision": overrides[f"expected_{protocol}_transport_revision"] = "stale-build-revision" elif mutation == "wrong_server": overrides[f"{protocol}_credential_id"] = "imap-credential" if protocol == "smtp" else "smtp-credential" else: credential = hierarchy.session.get(CredentialEnvelope, f"{protocol}-credential") if mutation == "inactive_credential": credential.is_active = False else: credential.tenant_id = "other-tenant" hierarchy.session.commit() with patch.object(server_hierarchy, "resolve_credential_envelope", side_effect=AssertionError("Reject stale/unauthorized before decrypt")) as decrypt, \ patch.object(smtp_module, "_open_smtp", side_effect=AssertionError("No network")) as opener, \ patch.object(capabilities, "send_email_bytes", side_effect=AssertionError("No SMTP")) as send, \ patch.object(capabilities, "append_message_to_sent", side_effect=AssertionError("No IMAP")) as append: with pytest.raises(mail_profiles.MailProfileError): if operation == "smtp_batch": with capabilities.campaign_smtp_batch(hierarchy.session, **smtp_args(**overrides)): pass elif operation == "smtp_single": capabilities.send_campaign_email_bytes(hierarchy.session, message_bytes=b"test", **smtp_args(**overrides)) else: capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(**overrides)) decrypt.assert_not_called(); opener.assert_not_called(); send.assert_not_called(); append.assert_not_called() @pytest.mark.parametrize("missing", ["smtp_credential_id", "imap_credential_id"]) def test_full_authoring_and_summary_still_check_both_protocols_without_decryption(hierarchy, missing): complete = selection() raw = {"server": {"mail_profile_id": "profile-1", **complete}} mail_profiles.assert_campaign_mail_policy_allows_json(hierarchy.session, tenant_id="tenant-1", campaign_id="campaign-1", raw_json=raw) incomplete = {**complete, missing: None} with patch.object(server_hierarchy, "resolve_credential_envelope", side_effect=AssertionError("Summary must not decrypt")) as decrypt: with pytest.raises(mail_profiles.MailProfileError, match="explicit credential selection"): capabilities.campaign_profile_delivery_summary(hierarchy.session, tenant_id="tenant-1", campaign_id="campaign-1", profile_id="profile-1", **incomplete) with pytest.raises(mail_profiles.MailProfileError, match="explicit credential selection"): mail_profiles.assert_campaign_mail_policy_allows_json(hierarchy.session, tenant_id="tenant-1", campaign_id="campaign-1", raw_json={"server": {"mail_profile_id": "profile-1", **incomplete}}) decrypt.assert_not_called() def test_batch_still_enforces_all_recipient_domains_before_connection(hierarchy): row = hierarchy.session.get(MailProfilePolicy, "system-policy") row.policy = {**row.policy, "blacklist": {"recipient_domains": ["blocked.example"]}} hierarchy.session.commit() with patch.object(smtp_module, "_open_smtp", side_effect=AssertionError("Forbidden recipient must never connect")) as opener: with pytest.raises(mail_profiles.MailProfileError, match="effective Mail policy"): with capabilities.campaign_smtp_batch(hierarchy.session, **smtp_args(envelope_recipients=["ok@example.test", "no@blocked.example"])): pass opener.assert_not_called()