GovOPlaN/govoplan-mail
1
0
Fork 0
Code Issues 11 Pull Requests Actions Packages Projects Releases Wiki Activity

security: enforce actor-aware Mail profile access

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 17:51:25 +02:00
parent 9d1d9bfb58
commit 06e0fbd3c3
7 changed files with 1336 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files

206
tests/test_mail_profile_helpers.py
View File

@@ -14,15 +14,75 @@ from govoplan_mail.backend.mail_profiles import (
_campaign_mail_profile_reference_id,
_apply_profile_transport_update,
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},
@@ -140,7 +200,7 @@ class MailProfileTransportHelperTests(unittest.TestCase):
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):
def test_next_transport_state_uses_only_saved_non_secret_metadata(self):
profile = SimpleNamespace(
tenant_id="tenant-1",
scope_type="tenant",
@@ -153,12 +213,19 @@ class MailProfileTransportHelperTests(unittest.TestCase):
imap_password_encrypted=encrypt_secret("imap-secret"),
)
smtp, imap = _next_profile_transport_state(profile, smtp=None, imap=None, clear_imap=False)
self.assertEqual(smtp.username, "saved-smtp")
self.assertEqual(smtp.password, "smtp-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)
self.assertEqual(imap.username, "saved-imap")
self.assertEqual(imap.password, "imap-secret")
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)
@@ -180,15 +247,17 @@ class MailProfileTransportHelperTests(unittest.TestCase):
)
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,
)
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")
@@ -199,7 +268,10 @@ class MailProfileTransportHelperTests(unittest.TestCase):
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:
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,
@@ -209,6 +281,7 @@ class MailProfileTransportHelperTests(unittest.TestCase):
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)
@@ -216,6 +289,53 @@ class MailProfileTransportHelperTests(unittest.TestCase):
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",
@@ -295,7 +415,7 @@ class MailProfileTransportHelperTests(unittest.TestCase):
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):
def test_supplied_password_is_replaced_without_decrypting_old_ciphertext(self):
encrypted = encrypt_secret("same-secret")
profile = SimpleNamespace(
id="profile-1",
@@ -313,7 +433,13 @@ class MailProfileTransportHelperTests(unittest.TestCase):
)
session = SimpleNamespace(flush=lambda: None)
with patch("govoplan_mail.backend.mail_profiles.audit_event") as audit:
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,
@@ -324,9 +450,49 @@ class MailProfileTransportHelperTests(unittest.TestCase):
clear_imap=False,
)
self.assertEqual(profile.smtp_password_encrypted, encrypted)
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")
audit.assert_not_called()
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):

49
tests/test_mailbox_index.py Normal file
View File

@@ -0,0 +1,49 @@
from __future__ import annotations
import unittest
from govoplan_mail.backend.db.models import MailMailboxFolderIndex, MailMailboxMessageIndex
from govoplan_mail.backend.mailbox_index import clear_mailbox_index
class _DeleteQuery:
def __init__(self, model: type, deleted_models: list[type]) -> None:
self.model = model
self.deleted_models = deleted_models
def filter(self, *_criteria):
return self
def delete(self, *, synchronize_session: bool) -> int:
if synchronize_session is not False:
raise AssertionError("bulk invalidation must not synchronize loaded cache rows")
self.deleted_models.append(self.model)
return 2 if self.model is MailMailboxMessageIndex else 1
class _Session:
def __init__(self) -> None:
self.deleted_models: list[type] = []
def query(self, model: type) -> _DeleteQuery:
return _DeleteQuery(model, self.deleted_models)
class MailboxIndexInvalidationTests(unittest.TestCase):
def test_profile_wide_invalidation_deletes_messages_before_folders(self) -> None:
session = _Session()
deleted_folders, deleted_messages = clear_mailbox_index(
session, # type: ignore[arg-type]
profile_id="profile-1",
)
self.assertEqual(
session.deleted_models,
[MailMailboxMessageIndex, MailMailboxFolderIndex],
)
self.assertEqual((deleted_folders, deleted_messages), (1, 2))
if __name__ == "__main__":
unittest.main()

419
tests/test_profile_actor_authorization.py Normal file
View File

@@ -0,0 +1,419 @@
from __future__ import annotations
import unittest
from types import SimpleNamespace
from unittest.mock import patch
from fastapi import HTTPException
from govoplan_mail.backend import router
from govoplan_mail.backend.mail_profiles import (
EffectiveMailProfilePolicy,
MailProfileError,
list_mail_server_profiles,
mail_profile_visible_to_actor,
)
from govoplan_mail.backend.schemas import (
MailServerProfileCreateRequest,
MailServerProfileUpdateRequest,
)
def _profile(
profile_id: str,
*,
scope_type: str,
scope_id: str | None,
tenant_id: str | None = "tenant-1",
):
return SimpleNamespace(
id=profile_id,
tenant_id=tenant_id,
scope_type=scope_type,
scope_id=scope_id,
is_active=True,
name=profile_id,
smtp_config={"host": "smtp.example.test"},
imap_config={"host": "imap.example.test"},
)
class _Query:
def __init__(self, profiles):
self._profiles = profiles
def filter(self, *_args):
return self
def all(self):
return list(self._profiles)
class _Session:
def __init__(self, profiles=()):
self.profiles = list(profiles)
self.commits = 0
self.rollbacks = 0
def query(self, _model):
return _Query(self.profiles)
def get(self, _model, profile_id):
return next((profile for profile in self.profiles if profile.id == profile_id), None)
def commit(self):
self.commits += 1
def rollback(self):
self.rollbacks += 1
class _Principal:
tenant_id = "tenant-1"
user = SimpleNamespace(id="user-1")
group_ids = frozenset({"group-1"})
def __init__(self, scopes):
self._scopes = frozenset(scopes)
def has(self, scope: str) -> bool:
return scope in self._scopes
class ProfileActorAuthorizationTests(unittest.TestCase):
def test_general_profile_list_hides_other_owner_contexts(self) -> None:
profiles = [
_profile("system", scope_type="system", scope_id=None, tenant_id=None),
_profile("tenant", scope_type="tenant", scope_id="tenant-1"),
_profile("own-user", scope_type="user", scope_id="user-1"),
_profile("other-user", scope_type="user", scope_id="user-2"),
_profile("own-group", scope_type="group", scope_id="group-1"),
_profile("other-group", scope_type="group", scope_id="group-2"),
]
session = _Session(profiles)
with patch(
"govoplan_mail.backend.mail_profiles.effective_mail_profile_policy",
return_value=EffectiveMailProfilePolicy(),
):
visible = list_mail_server_profiles(
session, # type: ignore[arg-type]
tenant_id="tenant-1",
actor_user_id="user-1",
actor_group_ids=(group_id for group_id in ("group-1",)),
)
self.assertEqual(
{profile.id for profile in visible},
{"system", "tenant", "own-user", "own-group"},
)
def test_general_access_denies_shared_profile_excluded_by_effective_policy(self) -> None:
denied = EffectiveMailProfilePolicy(allowed_profile_id_sets=[{"approved"}])
cases = (("system", None), ("tenant", "tenant-1"))
for scope_type, tenant_id in cases:
with self.subTest(scope_type=scope_type):
profile = _profile("denied", scope_type=scope_type, scope_id=tenant_id, tenant_id=tenant_id)
with patch("govoplan_mail.backend.mail_profiles.effective_mail_profile_policy", return_value=denied):
visible = mail_profile_visible_to_actor(
_Session(), # type: ignore[arg-type]
profile=profile,
tenant_id="tenant-1",
user_id="user-1",
group_ids={"group-1"},
)
self.assertFalse(visible)
def test_administrative_list_can_include_inactive_visible_profiles(self) -> None:
profile = _profile("inactive", scope_type="tenant", scope_id="tenant-1")
profile.is_active = False
session = _Session([profile])
with patch(
"govoplan_mail.backend.mail_profiles.effective_mail_profile_policy",
return_value=EffectiveMailProfilePolicy(),
):
visible = list_mail_server_profiles(
session, # type: ignore[arg-type]
tenant_id="tenant-1",
include_inactive=True,
actor_user_id="user-1",
actor_can_manage_tenant_profiles=True,
actor_administrative_visibility=True,
)
self.assertEqual([item.id for item in visible], ["inactive"])
def test_profile_admin_can_list_but_not_use_a_policy_denied_profile(self) -> None:
profile = _profile("denied", scope_type="tenant", scope_id="tenant-1")
session = _Session([profile])
denied = EffectiveMailProfilePolicy(allowed_profile_id_sets=[{"approved"}])
with patch("govoplan_mail.backend.mail_profiles.effective_mail_profile_policy", return_value=denied):
listed = list_mail_server_profiles(
session, # type: ignore[arg-type]
tenant_id="tenant-1",
actor_user_id="user-1",
actor_group_ids={"group-1"},
actor_can_manage_tenant_profiles=True,
actor_administrative_visibility=True,
)
self.assertEqual([item.id for item in listed], ["denied"])
principal = _Principal(
{"mail:profile:write", "mail:profile:test", "mail:profile:use"}
)
with (
patch("govoplan_mail.backend.router.smtp_config_from_profile") as materialize,
patch("govoplan_mail.backend.router.test_smtp_login") as provider,
self.assertRaises(HTTPException) as captured,
):
router.test_profile_smtp(
"denied",
principal=principal, # type: ignore[arg-type]
session=session, # type: ignore[arg-type]
)
self.assertEqual(captured.exception.status_code, 404)
materialize.assert_not_called()
provider.assert_not_called()
def test_campaign_scoped_profile_requires_campaign_acl_before_policy_resolution(self) -> None:
profile = _profile("campaign-profile", scope_type="campaign", scope_id="campaign-1")
access = SimpleNamespace(can_read_campaign=lambda *_args, **_kwargs: False)
with (
patch("govoplan_mail.backend.mail_profiles._campaign_access_provider", return_value=access),
patch("govoplan_mail.backend.mail_profiles._campaign_policy_context") as policy_context,
):
visible = mail_profile_visible_to_actor(
_Session(), # type: ignore[arg-type]
profile=profile,
tenant_id="tenant-1",
user_id="user-1",
)
self.assertFalse(visible)
policy_context.assert_not_called()
def test_mail_profile_admin_does_not_bypass_campaign_acl_but_tenant_admin_does(self) -> None:
profile = _profile("campaign-profile", scope_type="campaign", scope_id="campaign-1")
tenant_admin_values: list[bool] = []
def can_read(*_args, **kwargs):
tenant_admin_values.append(bool(kwargs["tenant_admin"]))
return bool(kwargs["tenant_admin"])
access = SimpleNamespace(can_read_campaign=can_read)
with (
patch("govoplan_mail.backend.mail_profiles._campaign_access_provider", return_value=access),
patch("govoplan_mail.backend.mail_profiles._campaign_policy_context") as policy_context,
):
mail_admin_visible = mail_profile_visible_to_actor(
_Session(), # type: ignore[arg-type]
profile=profile,
tenant_id="tenant-1",
user_id="user-1",
can_manage_tenant_profiles=True,
administrative_visibility=True,
)
tenant_admin_visible = mail_profile_visible_to_actor(
_Session(), # type: ignore[arg-type]
profile=profile,
tenant_id="tenant-1",
user_id="user-1",
can_manage_tenant_profiles=True,
tenant_admin=True,
administrative_visibility=True,
)
self.assertFalse(mail_admin_visible)
self.assertTrue(tenant_admin_visible)
self.assertEqual(tenant_admin_values, [False, True])
policy_context.assert_not_called()
def test_policy_read_rejects_unshared_and_mismatched_campaign_contexts(self) -> None:
principal = _Principal({"mail:profile:read"})
with (
patch(
"govoplan_mail.backend.router.campaign_mail_context_visible_to_actor",
return_value=False,
),
patch("govoplan_mail.backend.router._policy_response") as policy_response,
self.assertRaises(HTTPException) as unshared,
):
router.read_mail_profile_policy(
"tenant",
scope_id="tenant-1",
campaign_id="campaign-1",
principal=principal, # type: ignore[arg-type]
session=_Session(), # type: ignore[arg-type]
)
self.assertEqual(unshared.exception.status_code, 404)
policy_response.assert_not_called()
with self.assertRaises(HTTPException) as mismatched:
router._require_policy_campaign_context( # noqa: SLF001 - authorization seam regression
_Session(), # type: ignore[arg-type]
principal=principal, # type: ignore[arg-type]
scope_type="campaign",
scope_id="campaign-1",
campaign_id="campaign-2",
)
self.assertEqual(mismatched.exception.status_code, 422)
def test_campaign_profile_crud_requires_campaign_acl(self) -> None:
principal = _Principal({"mail:profile:write", "mail:secret:manage"})
profile = _profile("campaign-profile", scope_type="campaign", scope_id="campaign-1")
create_payload = MailServerProfileCreateRequest.model_validate(
{
"name": "Campaign profile",
"scope_type": "campaign",
"scope_id": "campaign-1",
"smtp": {"host": "smtp.example.test"},
}
)
with (
patch(
"govoplan_mail.backend.router.campaign_mail_context_visible_to_actor",
return_value=False,
),
patch("govoplan_mail.backend.router.create_mail_server_profile") as create,
self.assertRaises(HTTPException) as create_denied,
):
router.create_profile(
create_payload,
principal=principal, # type: ignore[arg-type]
session=_Session(), # type: ignore[arg-type]
)
self.assertEqual(create_denied.exception.status_code, 404)
create.assert_not_called()
for operation in ("update", "delete"):
with (
self.subTest(operation=operation),
patch("govoplan_mail.backend.router.get_mail_server_profile", return_value=profile),
patch(
"govoplan_mail.backend.router.campaign_mail_context_visible_to_actor",
return_value=False,
),
patch("govoplan_mail.backend.router.update_mail_server_profile") as update,
patch("govoplan_mail.backend.router.delete_mail_profile_credentials") as delete,
self.assertRaises(HTTPException) as denied,
):
if operation == "update":
router.update_profile(
profile.id,
MailServerProfileUpdateRequest(name="Renamed"),
principal=principal, # type: ignore[arg-type]
session=_Session(), # type: ignore[arg-type]
)
else:
router.deactivate_profile(
profile.id,
principal=principal, # type: ignore[arg-type]
session=_Session(), # type: ignore[arg-type]
)
self.assertEqual(denied.exception.status_code, 404)
update.assert_not_called()
delete.assert_not_called()
def test_inactive_visible_profile_change_remains_a_tombstone_candidate(self) -> None:
profile = _profile("deactivated", scope_type="tenant", scope_id="tenant-1")
profile.is_active = False
entry = SimpleNamespace(
resource_id=profile.id,
tenant_id="tenant-1",
payload={"scope_type": "tenant", "scope_id": "tenant-1"},
)
principal = _Principal({"mail:profile:read"})
with patch(
"govoplan_mail.backend.mail_profiles.effective_mail_profile_policy",
return_value=EffectiveMailProfilePolicy(),
):
visible = router._profile_change_visible_to_principal( # noqa: SLF001
_Session([profile]), # type: ignore[arg-type]
entry=entry, # type: ignore[arg-type]
principal=principal, # type: ignore[arg-type]
)
self.assertTrue(visible)
def test_stale_mailbox_cache_refreshes_synchronously_under_request_authorization(self) -> None:
principal = _Principal({"mail:mailbox:read", "mail:profile:use"})
session = _Session()
imap = SimpleNamespace(
host="imap.example.test",
port=993,
security=SimpleNamespace(value="tls"),
)
stale = SimpleNamespace(stale=True)
provider_result = SimpleNamespace(
host="imap.example.test",
port=993,
security="tls",
folders=[
SimpleNamespace(
name="INBOX",
flags=[],
message_count=0,
unseen_count=0,
)
],
detected_sent_folder=None,
)
with (
patch("govoplan_mail.backend.router._imap_config_for_principal", return_value=imap),
patch("govoplan_mail.backend.router.cached_mailbox_folders", return_value=stale),
patch("govoplan_mail.backend.router.list_imap_folders", return_value=provider_result) as provider,
patch("govoplan_mail.backend.router.cache_mailbox_folders") as cache,
):
response = router.list_profile_mailbox_folders(
"profile-1",
include_status=False,
refresh=False,
principal=principal, # type: ignore[arg-type]
session=session, # type: ignore[arg-type]
)
provider.assert_called_once_with(imap_config=imap, include_status=False)
cache.assert_called_once()
self.assertFalse(response.from_cache)
self.assertFalse(response.refreshing)
self.assertEqual(session.commits, 1)
def test_unauthorized_profile_test_fails_before_credentials_or_provider_effect(self) -> None:
principal = _Principal({"mail:profile:test", "mail:profile:use"})
with (
patch(
"govoplan_mail.backend.router.get_mail_server_profile_for_actor",
side_effect=MailProfileError("Mail-server profile not found"),
),
patch("govoplan_mail.backend.router.smtp_config_from_profile") as materialize,
patch("govoplan_mail.backend.router.test_smtp_login") as provider,
):
with self.assertRaises(HTTPException) as captured:
router.test_profile_smtp(
"other-user-profile",
principal=principal, # type: ignore[arg-type]
session=_Session(), # type: ignore[arg-type]
)
self.assertEqual(captured.exception.status_code, 404)
materialize.assert_not_called()
provider.assert_not_called()
def test_unauthorized_mailbox_read_fails_before_credential_decryption(self) -> None:
principal = _Principal({"mail:mailbox:read", "mail:profile:use"})
with (
patch(
"govoplan_mail.backend.router.get_mail_server_profile_for_actor",
side_effect=MailProfileError("Mail-server profile not found"),
),
patch("govoplan_mail.backend.router.imap_config_from_profile") as materialize,
):
with self.assertRaisesRegex(MailProfileError, "not found"):
router._imap_config_for_principal( # noqa: SLF001 - authorization seam regression
_Session(), # type: ignore[arg-type]
principal=principal, # type: ignore[arg-type]
profile_id="other-group-profile",
)
materialize.assert_not_called()
if __name__ == "__main__":
unittest.main()

30
tests/test_router_profile_deletion.py
View File

@@ -4,6 +4,8 @@ import unittest
from types import SimpleNamespace
from unittest.mock import patch
from fastapi import HTTPException
from govoplan_mail.backend.router import create_profile, deactivate_profile, update_profile
from govoplan_mail.backend.schemas import MailServerProfileCreateRequest, MailServerProfileUpdateRequest
@@ -48,6 +50,7 @@ class MailProfileDeletionRouteTests(unittest.TestCase):
patch("govoplan_mail.backend.router._require_profile_write_scope"),
patch("govoplan_mail.backend.router._require_profile_credentials_scope"),
patch("govoplan_mail.backend.router.delete_mail_profile_credentials", return_value=()),
patch("govoplan_mail.backend.router.clear_mailbox_index") as clear_index,
patch("govoplan_mail.backend.router._record_mail_change") as record_change,
patch("govoplan_mail.backend.router._profile_response", return_value=self.profile),
):
@@ -56,6 +59,7 @@ class MailProfileDeletionRouteTests(unittest.TestCase):
self.assertIs(result, self.profile)
self.assertEqual(session.commits, 1)
self.assertEqual(session.rollbacks, 0)
clear_index.assert_called_once_with(session, profile_id="profile-1")
record_change.assert_not_called()
def test_change_feed_reports_whether_credentials_were_deleted(self) -> None:
@@ -66,6 +70,7 @@ class MailProfileDeletionRouteTests(unittest.TestCase):
patch("govoplan_mail.backend.router._require_profile_write_scope"),
patch("govoplan_mail.backend.router._require_profile_credentials_scope"),
patch("govoplan_mail.backend.router.delete_mail_profile_credentials", return_value=()),
patch("govoplan_mail.backend.router.clear_mailbox_index") as clear_index,
patch("govoplan_mail.backend.router._record_mail_change") as record_change,
patch("govoplan_mail.backend.router._profile_response", return_value=self.profile),
):
@@ -73,6 +78,7 @@ class MailProfileDeletionRouteTests(unittest.TestCase):
self.assertFalse(record_change.call_args.kwargs["payload"]["credentials_deleted"])
self.assertEqual(record_change.call_args.kwargs["payload"]["deleted_credential_protocols"], [])
clear_index.assert_called_once_with(session, profile_id="profile-1")
def test_generic_secret_deletion_failure_rolls_back(self) -> None:
self.profile.is_active = True
@@ -89,6 +95,29 @@ class MailProfileDeletionRouteTests(unittest.TestCase):
self.assertEqual(session.commits, 0)
self.assertEqual(session.rollbacks, 1)
def test_patch_deactivation_is_rejected_in_favor_of_audited_delete(self) -> None:
self.profile.is_active = True
self.profile.smtp_password_encrypted = "existing-ciphertext"
session = _Session()
payload = MailServerProfileUpdateRequest(is_active=False)
with (
patch("govoplan_mail.backend.router.get_mail_server_profile", return_value=self.profile),
patch("govoplan_mail.backend.router._require_profile_write_scope"),
self.assertRaises(HTTPException) as captured,
):
update_profile(
self.profile.id,
payload,
principal=self.principal,
session=session,
)
self.assertEqual(captured.exception.status_code, 422)
self.assertTrue(self.profile.is_active)
self.assertEqual(self.profile.smtp_password_encrypted, "existing-ciphertext")
self.assertEqual(session.commits, 0)
self.assertEqual(session.rollbacks, 1)
def test_system_profile_changes_are_instance_wide_in_the_change_feed(self) -> None:
system_profile = SimpleNamespace(
id="profile-system",
@@ -132,6 +161,7 @@ class MailProfileDeletionRouteTests(unittest.TestCase):
patch("govoplan_mail.backend.router._require_profile_write_scope"),
patch("govoplan_mail.backend.router._require_profile_credentials_scope"),
patch("govoplan_mail.backend.router.delete_mail_profile_credentials", return_value=()),
patch("govoplan_mail.backend.router.clear_mailbox_index"),
patch("govoplan_mail.backend.router._record_mail_change") as delete_change,
patch("govoplan_mail.backend.router._profile_response", return_value=system_profile),
):