feat: enforce Mail-owned campaign transport boundary
This commit is contained in:
@@ -2,20 +2,144 @@ from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
from govoplan_core.security.secrets import decrypt_secret, encrypt_secret
|
||||
from govoplan_mail.backend.config import ImapConfig, SmtpConfig
|
||||
from govoplan_mail.backend.mail_profiles import (
|
||||
EffectiveCredentialPolicy,
|
||||
EffectiveMailProfilePolicy,
|
||||
MailProfileError,
|
||||
_assert_campaign_inherits_profile_credentials,
|
||||
_campaign_mail_profile_reference_id,
|
||||
_apply_profile_transport_update,
|
||||
campaign_profile_transport_revisions,
|
||||
_merge_policy,
|
||||
_next_profile_transport_state,
|
||||
_policy_parent_lock_message,
|
||||
_policy_parent_lock_violations,
|
||||
delete_mail_profile_credentials,
|
||||
)
|
||||
|
||||
|
||||
class MailProfileTransportHelperTests(unittest.TestCase):
|
||||
def test_campaign_transport_revisions_are_random_opaque_values(self):
|
||||
profile = SimpleNamespace(
|
||||
smtp_config={"host": "smtp.example.org", "port": 587, "security": "starttls", "timeout_seconds": 30},
|
||||
smtp_username="first-user",
|
||||
smtp_password_encrypted="first-secret-ciphertext",
|
||||
smtp_transport_revision="032837ce-fd97-401a-8a9f-8df67f3ab41d",
|
||||
imap_config=None,
|
||||
imap_username=None,
|
||||
imap_password_encrypted=None,
|
||||
imap_transport_revision="9ca09521-c543-42ed-a8f2-f06783f1ebd3",
|
||||
)
|
||||
|
||||
original = campaign_profile_transport_revisions(profile)
|
||||
profile.smtp_password_encrypted = "second-secret-ciphertext"
|
||||
|
||||
self.assertEqual(campaign_profile_transport_revisions(profile), original)
|
||||
self.assertNotIn("smtp.example.org", repr(original))
|
||||
self.assertEqual(original["smtp"], profile.smtp_transport_revision)
|
||||
|
||||
def test_profile_deletion_scrubs_both_passwords_and_writes_non_secret_audit(self):
|
||||
profile = SimpleNamespace(
|
||||
id="profile-1",
|
||||
tenant_id="tenant-1",
|
||||
scope_type="tenant",
|
||||
scope_id="tenant-1",
|
||||
smtp_password_encrypted="smtp-ciphertext",
|
||||
imap_password_encrypted="imap-ciphertext",
|
||||
)
|
||||
session = SimpleNamespace(add=lambda _value: None, flush=lambda: None)
|
||||
|
||||
with patch("govoplan_mail.backend.mail_profiles.audit_event") as audit:
|
||||
deleted = delete_mail_profile_credentials(
|
||||
session, # type: ignore[arg-type]
|
||||
profile=profile, # type: ignore[arg-type]
|
||||
deletion_reason="profile_deactivated",
|
||||
user_id="user-1",
|
||||
)
|
||||
|
||||
self.assertEqual(deleted, ("smtp", "imap"))
|
||||
self.assertIsNone(profile.smtp_password_encrypted)
|
||||
self.assertIsNone(profile.imap_password_encrypted)
|
||||
self.assertEqual(audit.call_count, 1)
|
||||
self.assertEqual(audit.call_args.kwargs["action"], "mail.profile_credentials_deleted")
|
||||
self.assertEqual(audit.call_args.kwargs["scope"], "tenant")
|
||||
self.assertEqual(audit.call_args.kwargs["details"]["deletion_reason"], "profile_deactivated")
|
||||
self.assertNotIn("ciphertext", repr(audit.call_args.kwargs))
|
||||
|
||||
def test_system_profile_secret_deletion_uses_system_audit_scope(self):
|
||||
profile = SimpleNamespace(
|
||||
id="profile-system",
|
||||
tenant_id=None,
|
||||
scope_type="system",
|
||||
scope_id=None,
|
||||
smtp_password_encrypted="smtp-ciphertext",
|
||||
imap_password_encrypted=None,
|
||||
)
|
||||
session = SimpleNamespace(add=lambda _value: None, flush=lambda: None)
|
||||
|
||||
with patch("govoplan_mail.backend.mail_profiles.audit_event") as audit:
|
||||
delete_mail_profile_credentials(
|
||||
session, # type: ignore[arg-type]
|
||||
profile=profile, # type: ignore[arg-type]
|
||||
deletion_reason="module_data_retired",
|
||||
)
|
||||
|
||||
self.assertEqual(audit.call_args.kwargs["scope"], "system")
|
||||
self.assertIsNone(audit.call_args.kwargs["tenant_id"])
|
||||
self.assertNotIn("ciphertext", repr(audit.call_args.kwargs))
|
||||
|
||||
def test_repeated_profile_secret_deletion_is_an_idempotent_no_op(self):
|
||||
profile = SimpleNamespace(
|
||||
id="profile-1",
|
||||
tenant_id="tenant-1",
|
||||
scope_type="tenant",
|
||||
scope_id="tenant-1",
|
||||
smtp_password_encrypted=None,
|
||||
imap_password_encrypted=None,
|
||||
)
|
||||
session = SimpleNamespace(add=lambda _value: self.fail("no-op must not add"), flush=lambda: self.fail("no-op must not flush"))
|
||||
|
||||
with patch("govoplan_mail.backend.mail_profiles.audit_event") as audit:
|
||||
deleted = delete_mail_profile_credentials(
|
||||
session, # type: ignore[arg-type]
|
||||
profile=profile, # type: ignore[arg-type]
|
||||
deletion_reason="profile_deactivated",
|
||||
)
|
||||
|
||||
self.assertEqual(deleted, ())
|
||||
audit.assert_not_called()
|
||||
|
||||
def test_profile_secret_deletion_does_not_mutate_ciphertext_when_audit_fails(self):
|
||||
profile = SimpleNamespace(
|
||||
id="profile-1",
|
||||
tenant_id="tenant-1",
|
||||
scope_type="tenant",
|
||||
scope_id="tenant-1",
|
||||
smtp_password_encrypted="smtp-ciphertext",
|
||||
imap_password_encrypted="imap-ciphertext",
|
||||
)
|
||||
session = SimpleNamespace(add=lambda _value: None, flush=lambda: None)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"govoplan_mail.backend.mail_profiles.audit_event",
|
||||
side_effect=RuntimeError("audit unavailable"),
|
||||
),
|
||||
self.assertRaisesRegex(RuntimeError, "audit unavailable"),
|
||||
):
|
||||
delete_mail_profile_credentials(
|
||||
session, # type: ignore[arg-type]
|
||||
profile=profile, # type: ignore[arg-type]
|
||||
deletion_reason="profile_deactivated",
|
||||
)
|
||||
|
||||
self.assertEqual(profile.smtp_password_encrypted, "smtp-ciphertext")
|
||||
self.assertEqual(profile.imap_password_encrypted, "imap-ciphertext")
|
||||
|
||||
def test_next_transport_state_uses_saved_profile_credentials(self):
|
||||
profile = SimpleNamespace(
|
||||
tenant_id="tenant-1",
|
||||
@@ -41,16 +165,26 @@ class MailProfileTransportHelperTests(unittest.TestCase):
|
||||
|
||||
def test_apply_transport_update_preserves_unsupplied_passwords(self):
|
||||
profile = SimpleNamespace(
|
||||
id="profile-1",
|
||||
tenant_id="tenant-1",
|
||||
scope_type="tenant",
|
||||
scope_id="tenant-1",
|
||||
smtp_config={"host": "smtp.example.org", "port": 587, "security": "starttls", "timeout_seconds": 30},
|
||||
smtp_username="saved-smtp",
|
||||
smtp_password_encrypted=encrypt_secret("smtp-secret"),
|
||||
smtp_transport_revision="smtp-before",
|
||||
imap_config={"host": "imap.example.org", "port": 993, "security": "tls", "sent_folder": "Sent", "timeout_seconds": 30},
|
||||
imap_username="saved-imap",
|
||||
imap_password_encrypted=encrypt_secret("imap-secret"),
|
||||
imap_transport_revision="imap-before",
|
||||
)
|
||||
session = SimpleNamespace(flush=lambda: None)
|
||||
|
||||
_apply_profile_transport_update(
|
||||
session, # type: ignore[arg-type]
|
||||
profile,
|
||||
user_id="user-1",
|
||||
api_key_id=None,
|
||||
smtp=SmtpConfig(host="smtp2.example.org", username="new-smtp"),
|
||||
imap=ImapConfig(host="imap2.example.org", username="new-imap"),
|
||||
clear_imap=False,
|
||||
@@ -62,14 +196,158 @@ class MailProfileTransportHelperTests(unittest.TestCase):
|
||||
self.assertEqual(profile.imap_config["host"], "imap2.example.org")
|
||||
self.assertEqual(profile.imap_username, "new-imap")
|
||||
self.assertEqual(decrypt_secret(profile.imap_password_encrypted), "imap-secret")
|
||||
self.assertNotEqual(profile.smtp_transport_revision, "smtp-before")
|
||||
self.assertNotEqual(profile.imap_transport_revision, "imap-before")
|
||||
|
||||
_apply_profile_transport_update(profile, smtp=None, imap=None, clear_imap=True)
|
||||
with patch("govoplan_mail.backend.mail_profiles.audit_event") as audit:
|
||||
_apply_profile_transport_update(
|
||||
session, # type: ignore[arg-type]
|
||||
profile,
|
||||
user_id="user-1",
|
||||
api_key_id=None,
|
||||
smtp=None,
|
||||
imap=None,
|
||||
clear_imap=True,
|
||||
)
|
||||
self.assertIsNone(profile.imap_config)
|
||||
self.assertIsNone(profile.imap_username)
|
||||
self.assertIsNone(profile.imap_password_encrypted)
|
||||
self.assertEqual(audit.call_args.kwargs["action"], "mail.profile_credentials_deleted")
|
||||
self.assertEqual(audit.call_args.kwargs["details"]["protocol"], "imap")
|
||||
self.assertNotIn("imap-secret", repr(audit.call_args.kwargs))
|
||||
|
||||
def test_password_replacement_preserves_revision_and_is_audited_without_secret(self):
|
||||
profile = SimpleNamespace(
|
||||
id="profile-1",
|
||||
tenant_id="tenant-1",
|
||||
scope_type="tenant",
|
||||
scope_id="tenant-1",
|
||||
smtp_config={"host": "smtp.example.org", "port": 587, "security": "starttls", "timeout_seconds": 30},
|
||||
smtp_username="saved-smtp",
|
||||
smtp_password_encrypted=encrypt_secret("old-secret"),
|
||||
smtp_transport_revision="smtp-before",
|
||||
imap_config=None,
|
||||
imap_username=None,
|
||||
imap_password_encrypted=None,
|
||||
imap_transport_revision="imap-before",
|
||||
)
|
||||
session = SimpleNamespace(flush=lambda: None)
|
||||
|
||||
with patch("govoplan_mail.backend.mail_profiles.audit_event") as audit:
|
||||
_apply_profile_transport_update(
|
||||
session, # type: ignore[arg-type]
|
||||
profile,
|
||||
user_id="user-1",
|
||||
api_key_id="key-1",
|
||||
smtp=SmtpConfig(
|
||||
host="smtp.example.org",
|
||||
port=587,
|
||||
security="starttls",
|
||||
timeout_seconds=30,
|
||||
password="new-secret",
|
||||
),
|
||||
imap=None,
|
||||
clear_imap=False,
|
||||
)
|
||||
|
||||
self.assertEqual(decrypt_secret(profile.smtp_password_encrypted), "new-secret")
|
||||
self.assertEqual(profile.smtp_transport_revision, "smtp-before")
|
||||
self.assertEqual(audit.call_args.kwargs["action"], "mail.profile_credentials_replaced")
|
||||
self.assertEqual(audit.call_args.kwargs["details"]["protocol"], "smtp")
|
||||
self.assertNotIn("old-secret", repr(audit.call_args.kwargs))
|
||||
self.assertNotIn("new-secret", repr(audit.call_args.kwargs))
|
||||
|
||||
def test_audit_failure_leaves_existing_ciphertext_and_revision_unchanged(self):
|
||||
encrypted = encrypt_secret("old-secret")
|
||||
profile = SimpleNamespace(
|
||||
id="profile-1",
|
||||
tenant_id="tenant-1",
|
||||
scope_type="tenant",
|
||||
scope_id="tenant-1",
|
||||
smtp_config={"host": "smtp.example.org", "port": 587, "security": "starttls", "timeout_seconds": 30},
|
||||
smtp_username="saved-smtp",
|
||||
smtp_password_encrypted=encrypted,
|
||||
smtp_transport_revision="smtp-before",
|
||||
imap_config=None,
|
||||
imap_username=None,
|
||||
imap_password_encrypted=None,
|
||||
imap_transport_revision="imap-before",
|
||||
)
|
||||
session = SimpleNamespace(flush=lambda: None)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"govoplan_mail.backend.mail_profiles.audit_event",
|
||||
side_effect=RuntimeError("audit unavailable"),
|
||||
),
|
||||
self.assertRaisesRegex(RuntimeError, "audit unavailable"),
|
||||
):
|
||||
_apply_profile_transport_update(
|
||||
session, # type: ignore[arg-type]
|
||||
profile,
|
||||
user_id="user-1",
|
||||
api_key_id=None,
|
||||
smtp=SmtpConfig(host="smtp.example.org", password="new-secret"),
|
||||
imap=None,
|
||||
clear_imap=False,
|
||||
)
|
||||
|
||||
self.assertEqual(profile.smtp_password_encrypted, encrypted)
|
||||
self.assertEqual(profile.smtp_transport_revision, "smtp-before")
|
||||
|
||||
def test_exact_password_update_is_an_idempotent_no_op(self):
|
||||
encrypted = encrypt_secret("same-secret")
|
||||
profile = SimpleNamespace(
|
||||
id="profile-1",
|
||||
tenant_id="tenant-1",
|
||||
scope_type="tenant",
|
||||
scope_id="tenant-1",
|
||||
smtp_config={"host": "smtp.example.org", "port": 587, "security": "starttls", "timeout_seconds": 30},
|
||||
smtp_username="saved-smtp",
|
||||
smtp_password_encrypted=encrypted,
|
||||
smtp_transport_revision="smtp-before",
|
||||
imap_config=None,
|
||||
imap_username=None,
|
||||
imap_password_encrypted=None,
|
||||
imap_transport_revision="imap-before",
|
||||
)
|
||||
session = SimpleNamespace(flush=lambda: None)
|
||||
|
||||
with patch("govoplan_mail.backend.mail_profiles.audit_event") as audit:
|
||||
_apply_profile_transport_update(
|
||||
session, # type: ignore[arg-type]
|
||||
profile,
|
||||
user_id="user-1",
|
||||
api_key_id=None,
|
||||
smtp=SmtpConfig(host="smtp.example.org", password="same-secret"),
|
||||
imap=None,
|
||||
clear_imap=False,
|
||||
)
|
||||
|
||||
self.assertEqual(profile.smtp_password_encrypted, encrypted)
|
||||
self.assertEqual(profile.smtp_transport_revision, "smtp-before")
|
||||
audit.assert_not_called()
|
||||
|
||||
|
||||
class MailProfilePolicyHelperTests(unittest.TestCase):
|
||||
def test_campaign_contract_accepts_only_mail_profile_reference(self):
|
||||
self.assertEqual(
|
||||
_campaign_mail_profile_reference_id({"mail_profile_id": " profile-1 "}),
|
||||
"profile-1",
|
||||
)
|
||||
for legacy in ("smtp", "imap", "credentials", "inherit_smtp_credentials"):
|
||||
with self.subTest(legacy=legacy), self.assertRaisesRegex(MailProfileError, "select a Mail profile"):
|
||||
_campaign_mail_profile_reference_id({"mail_profile_id": "profile-1", legacy: {}})
|
||||
|
||||
def test_campaign_delivery_fails_when_policy_requires_local_credentials(self):
|
||||
profile = SimpleNamespace(imap_config=None)
|
||||
policy = EffectiveMailProfilePolicy(
|
||||
smtp_credentials=EffectiveCredentialPolicy(inherit=False),
|
||||
)
|
||||
|
||||
with self.assertRaisesRegex(MailProfileError, "Store the credentials on a Mail profile"):
|
||||
_assert_campaign_inherits_profile_credentials(profile, policy)
|
||||
|
||||
def test_merge_policy_respects_locked_lower_level_limits(self):
|
||||
policy = EffectiveMailProfilePolicy()
|
||||
_merge_policy(
|
||||
|
||||
Reference in New Issue
Block a user