188 lines
13 KiB
Python
188 lines
13 KiB
Python
"""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()
|