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_mail_selection_from_json, campaign_profile_transport_revisions, create_mail_server_profile, _merge_policy, _next_profile_transport_state, _policy_parent_lock_message, _policy_parent_lock_violations, delete_mail_profile_credentials, update_mail_server_profile, ) class MailProfileTransportHelperTests(unittest.TestCase): def test_inactive_profile_creation_rejects_dormant_credentials(self): session = SimpleNamespace() with ( patch( "govoplan_mail.backend.mail_profiles._scope_tuple_for_create", return_value=("tenant-1", "tenant", "tenant-1"), ), patch("govoplan_mail.backend.mail_profiles._ensure_scope_allows_profile_creation"), patch("govoplan_mail.backend.mail_profiles.assert_mail_policy_allows_transport"), patch("govoplan_mail.backend.mail_profiles._ensure_unique_slug"), self.assertRaisesRegex(MailProfileError, "cannot retain credentials"), ): create_mail_server_profile( session, # type: ignore[arg-type] tenant_id="tenant-1", user_id="user-1", name="Dormant secret", slug=None, description=None, smtp=SmtpConfig(host="smtp.example.test", password="secret"), imap=None, is_active=False, ) def test_inactive_profile_cannot_retain_or_recreate_dormant_credentials(self): session = SimpleNamespace() legacy = SimpleNamespace( is_active=False, smtp_password_encrypted="legacy-ciphertext", imap_password_encrypted=None, ) with self.assertRaisesRegex(MailProfileError, "use DELETE to scrub"): update_mail_server_profile( session, # type: ignore[arg-type] legacy, # type: ignore[arg-type] tenant_id="tenant-1", user_id="user-1", is_active=True, ) self.assertFalse(legacy.is_active) self.assertEqual(legacy.smtp_password_encrypted, "legacy-ciphertext") scrubbed = SimpleNamespace( is_active=False, smtp_password_encrypted=None, imap_password_encrypted=None, ) with self.assertRaisesRegex(MailProfileError, "activated in the same update"): update_mail_server_profile( session, # type: ignore[arg-type] scrubbed, # type: ignore[arg-type] tenant_id="tenant-1", user_id="user-1", smtp=SmtpConfig(host="smtp.example.test", password="new-secret"), ) self.assertFalse(scrubbed.is_active) self.assertIsNone(scrubbed.smtp_password_encrypted) 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_only_saved_non_secret_metadata(self): profile = SimpleNamespace( tenant_id="tenant-1", scope_type="tenant", scope_id=None, smtp_config={"host": "smtp.example.org", "port": 587, "security": "starttls", "timeout_seconds": 30}, smtp_username="saved-smtp", smtp_password_encrypted=encrypt_secret("smtp-secret"), 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"), ) with patch( "govoplan_mail.backend.mail_profiles.decrypt_secret", side_effect=AssertionError("policy validation must not decrypt credentials"), ): smtp, imap = _next_profile_transport_state(profile, smtp=None, imap=None, clear_imap=False) self.assertEqual(smtp["host"], "smtp.example.org") self.assertNotIn("username", smtp) self.assertNotIn("password", smtp) self.assertIsNotNone(imap) assert imap is not None self.assertEqual(imap["host"], "imap.example.org") self.assertNotIn("username", imap) self.assertNotIn("password", imap) _, cleared_imap = _next_profile_transport_state(profile, smtp=None, imap=None, clear_imap=True) self.assertIsNone(cleared_imap) 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) with patch("govoplan_mail.backend.mail_profiles.clear_mailbox_index") as clear_index: _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, ) clear_index.assert_called_once_with(session, profile_id="profile-1") self.assertEqual(profile.smtp_config["host"], "smtp2.example.org") self.assertEqual(profile.smtp_username, "new-smtp") self.assertEqual(decrypt_secret(profile.smtp_password_encrypted), "smtp-secret") 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") with ( patch("govoplan_mail.backend.mail_profiles.audit_event") as audit, patch("govoplan_mail.backend.mail_profiles.clear_mailbox_index") as clear_index, ): _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, ) clear_index.assert_called_once_with(session, profile_id="profile-1") 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_imap_password_replacement_clears_cache_without_rotating_identity_revision(self): profile = SimpleNamespace( id="profile-1", tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", smtp_config={"host": "smtp.example.org"}, smtp_username=None, smtp_password_encrypted=None, smtp_transport_revision="smtp-before", imap_config={ "host": "imap.example.org", "port": 993, "security": "tls", "sent_folder": "auto", "timeout_seconds": 30, }, imap_username="saved-imap", imap_password_encrypted=encrypt_secret("old-secret"), imap_transport_revision="imap-before", ) session = SimpleNamespace(flush=lambda: None) with ( patch("govoplan_mail.backend.mail_profiles.audit_event"), patch("govoplan_mail.backend.mail_profiles.clear_mailbox_index") as clear_index, ): _apply_profile_transport_update( session, # type: ignore[arg-type] profile, user_id="user-1", api_key_id=None, smtp=None, imap=ImapConfig( host="imap.example.org", port=993, security="tls", timeout_seconds=30, password="new-secret", ), clear_imap=False, ) clear_index.assert_called_once_with(session, profile_id="profile-1") self.assertEqual(profile.imap_transport_revision, "imap-before") self.assertEqual(decrypt_secret(profile.imap_password_encrypted), "new-secret") 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_supplied_password_is_replaced_without_decrypting_old_ciphertext(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, patch( "govoplan_mail.backend.mail_profiles.decrypt_secret", side_effect=AssertionError("replacement must not decrypt old ciphertext"), ), ): _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.assertNotEqual(profile.smtp_password_encrypted, encrypted) self.assertEqual(decrypt_secret(profile.smtp_password_encrypted), "same-secret") self.assertEqual(profile.smtp_transport_revision, "smtp-before") self.assertEqual(audit.call_args.kwargs["action"], "mail.profile_credentials_replaced") def test_metadata_update_does_not_decrypt_unrelated_transport_credentials(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="corrupt-smtp-ciphertext", 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="corrupt-imap-ciphertext", imap_transport_revision="imap-before", name="Profile", slug="profile", description=None, is_active=True, updated_by_user_id=None, ) session = SimpleNamespace(add=lambda _value: None, flush=lambda: None) with ( patch("govoplan_mail.backend.mail_profiles.decrypt_secret", side_effect=AssertionError("must not decrypt")), patch("govoplan_mail.backend.mail_profiles._assert_profile_transport_allowed") as policy_check, ): update_mail_server_profile( session, # type: ignore[arg-type] profile, tenant_id="tenant-1", user_id="user-1", description="Updated", ) policy_check.assert_called_once() self.assertEqual(profile.description, "Updated") self.assertEqual(profile.smtp_password_encrypted, "corrupt-smtp-ciphertext") self.assertEqual(profile.imap_password_encrypted, "corrupt-imap-ciphertext") class MailProfilePolicyHelperTests(unittest.TestCase): def test_campaign_contract_accepts_only_mail_owned_references(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 Mail resources"): _campaign_mail_profile_reference_id({"mail_profile_id": "profile-1", legacy: {}}) self.assertEqual( campaign_mail_selection_from_json( { "server": { "mail_profile_id": "profile-1", "smtp_server_id": "smtp-1", "smtp_credential_id": "credential-1", } } )["smtp_credential_id"], "credential-1", ) 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, "explicit credential selection"): _assert_campaign_inherits_profile_credentials(profile, policy) def test_merge_policy_respects_locked_lower_level_limits(self): policy = EffectiveMailProfilePolicy() _merge_policy( policy, { "blacklist": {"smtp_hosts": ["*.blocked.example"]}, "allow_lower_level_limits": {"blacklist.smtp_hosts": False}, }, source="system", ) _merge_policy( policy, {"blacklist": {"smtp_hosts": ["*.tenant.example"]}}, source="tenant", source_id="tenant-1", ) self.assertEqual(policy.blacklist_patterns["smtp_hosts"], ["*.blocked.example"]) self.assertFalse(policy.allow_lower_level_limits["blacklist.smtp_hosts"]) def test_parent_lock_violations_are_reported_per_field(self): violations = _policy_parent_lock_violations( { "allow_user_profiles": False, "smtp_credentials.inherit": False, "blacklist.smtp_hosts": False, }, { "allow_user_profiles": True, "smtp_credentials": {"inherit": False}, "blacklist": {"smtp_hosts": ["*.blocked.example"]}, "allow_lower_level_limits": {"allow_user_profiles": True}, }, ) self.assertEqual( violations, [ "allow_user_profiles", "blacklist.smtp_hosts", "smtp_credentials.inherit", "allow_lower_level_limits.allow_user_profiles", ], ) self.assertEqual( _policy_parent_lock_message("smtp_credentials.inherit"), "SMTP credential inheritance is locked by an ancestor policy", ) if __name__ == "__main__": unittest.main()