from __future__ import annotations import unittest from types import SimpleNamespace from unittest.mock import ANY, patch from fastapi import HTTPException from sqlalchemy.dialects import postgresql from govoplan_mail.backend import router from govoplan_mail.backend import mail_profiles from govoplan_mail.backend.mail_profiles import ( EffectiveMailProfilePolicy, MailProfileError, get_mail_server_profile_for_actor, 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"}, smtp_password_encrypted=None, imap_password_encrypted=None, ) 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 def refresh(self, _value): return None class _Principal: tenant_id = "tenant-1" user = SimpleNamespace(id="user-1") group_ids = frozenset({"group-1"}) api_key_id = None 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_profile_mutation_selector_is_owner_bounded_and_row_locked(self) -> None: statement = mail_profiles._mail_server_profile_mutation_statement( # noqa: SLF001 tenant_id="tenant-1", profile_id="profile-1", owner_user_id="user-1", can_manage_tenant_profiles=False, can_manage_system_profiles=False, ) sql = str(statement.compile(dialect=postgresql.dialect())) self.assertIn("FOR UPDATE", sql) self.assertIn("mail_server_profiles.tenant_id =", sql) self.assertIn("mail_server_profiles.scope_type =", sql) self.assertIn("mail_server_profiles.scope_id =", sql) self.assertTrue(statement.get_execution_options()["populate_existing"]) def test_profile_mutation_selector_without_authority_cannot_lock_a_row(self) -> None: statement = mail_profiles._mail_server_profile_mutation_statement( # noqa: SLF001 tenant_id="tenant-1", profile_id="profile-1", owner_user_id=None, can_manage_tenant_profiles=False, can_manage_system_profiles=False, ) sql = str(statement.compile(dialect=postgresql.dialect())) self.assertIn("mail_server_profiles.id IS NULL", sql) self.assertIn("FOR UPDATE", sql) def test_self_service_profile_permissions_are_limited_to_own_user_scope(self) -> None: principal = _Principal( {"mail:profile:write_own", "mail:secret:manage_own"} ) router._require_profile_write_scope( # noqa: SLF001 - authorization seam principal, # type: ignore[arg-type] "user", "user-1", ) router._require_profile_credentials_scope( # noqa: SLF001 - authorization seam principal, # type: ignore[arg-type] "user", "user-1", ) for scope_type, scope_id in ( ("user", "user-2"), ("tenant", "tenant-1"), ("group", "group-1"), ("campaign", "campaign-1"), ("system", None), ): with self.subTest(scope_type=scope_type, scope_id=scope_id): with self.assertRaises(HTTPException) as write_denied: router._require_profile_write_scope( # noqa: SLF001 principal, # type: ignore[arg-type] scope_type, scope_id, ) self.assertEqual(write_denied.exception.status_code, 403) with self.assertRaises(HTTPException) as secret_denied: router._require_profile_credentials_scope( # noqa: SLF001 principal, # type: ignore[arg-type] scope_type, scope_id, ) self.assertEqual(secret_denied.exception.status_code, 403) def test_self_service_create_rejects_another_user_before_persistence(self) -> None: principal = _Principal( { "mail:profile:read", "mail:profile:write_own", "mail:secret:manage_own", } ) payload = MailServerProfileCreateRequest.model_validate( { "name": "Other user's profile", "scope_type": "user", "scope_id": "user-2", "smtp": { "host": "smtp.example.test", "password": "not-persisted", }, } ) with ( patch("govoplan_mail.backend.router.create_mail_server_profile") as create, self.assertRaises(HTTPException) as denied, ): router.create_profile( payload, principal=principal, # type: ignore[arg-type] session=_Session(), # type: ignore[arg-type] ) self.assertEqual(denied.exception.status_code, 403) create.assert_not_called() def test_self_service_update_hides_another_user_before_mutation(self) -> None: principal = _Principal( {"mail:profile:write_own", "mail:secret:manage_own"} ) profile = _profile( "other-user-profile", scope_type="user", scope_id="user-2", ) with ( patch( "govoplan_mail.backend.router.get_mail_server_profile", return_value=profile, ), patch("govoplan_mail.backend.router.update_mail_server_profile") as update, self.assertRaises(HTTPException) as denied, ): router.update_profile( profile.id, MailServerProfileUpdateRequest(name="Must not change"), principal=principal, # type: ignore[arg-type] session=_Session(), # type: ignore[arg-type] ) self.assertEqual(denied.exception.status_code, 404) update.assert_not_called() def test_self_service_delete_hides_another_user_before_mutation(self) -> None: principal = _Principal( {"mail:profile:write_own", "mail:secret:manage_own"} ) profile = _profile( "other-user-profile", scope_type="user", scope_id="user-2", ) with ( patch( "govoplan_mail.backend.router.get_mail_server_profile", return_value=profile, ), patch("govoplan_mail.backend.router.delete_mail_profile_credentials") as delete, self.assertRaises(HTTPException) as denied, ): router.deactivate_profile( profile.id, principal=principal, # type: ignore[arg-type] session=_Session(), # type: ignore[arg-type] ) self.assertEqual(denied.exception.status_code, 404) delete.assert_not_called() def test_mutation_lookup_returns_the_same_not_found_for_missing_and_non_owned(self) -> None: principal = _Principal({"mail:profile:write_own"}) other_profile = _profile( "other-user-profile", scope_type="user", scope_id="user-2", ) cases = ( (other_profile, None), (None, MailProfileError("Mail-server profile not found")), ) for returned, failure in cases: with self.subTest(failure=failure is not None): kwargs = {"side_effect": failure} if failure else {"return_value": returned} with ( patch( "govoplan_mail.backend.router.get_mail_server_profile", **kwargs, ), self.assertRaises(HTTPException) as denied, ): router._profile_for_mutation( # noqa: SLF001 - authorization seam _Session(), # type: ignore[arg-type] principal=principal, # type: ignore[arg-type] profile_id="candidate-id", ) self.assertEqual(denied.exception.status_code, 404) self.assertEqual(denied.exception.detail, "Mail-server profile not found") def test_broad_profile_admin_retains_cross_owner_mutation_access(self) -> None: principal = _Principal({"mail:profile:write"}) profile = _profile( "other-user-profile", scope_type="user", scope_id="user-2", ) with patch( "govoplan_mail.backend.router.get_mail_server_profile", return_value=profile, ) as get_profile: resolved = router._profile_for_mutation( # noqa: SLF001 - authorization seam _Session(), # type: ignore[arg-type] principal=principal, # type: ignore[arg-type] profile_id=profile.id, ) self.assertIs(resolved, profile) get_profile.assert_called_once_with( ANY, tenant_id="tenant-1", profile_id=profile.id, for_update=True, mutation_owner_user_id=None, can_manage_tenant_profiles=True, can_manage_system_profiles=False, ) def test_self_service_transport_rebind_requires_own_secret_authority(self) -> None: principal = _Principal({"mail:profile:write_own"}) profile = _profile( "own-user-profile", scope_type="user", scope_id="user-1", ) profile.smtp_config = { "host": "smtp.example.test", "port": 587, "security": "starttls", "timeout_seconds": 30, } profile.smtp_password_encrypted = "existing-ciphertext" payload = MailServerProfileUpdateRequest.model_validate( {"smtp": {"host": "smtp.attacker.test"}} ) session = _Session() with ( patch( "govoplan_mail.backend.router.get_mail_server_profile", return_value=profile, ), patch("govoplan_mail.backend.router.update_mail_server_profile") as update, self.assertRaises(HTTPException) as denied, ): router.update_profile( profile.id, payload, principal=principal, # type: ignore[arg-type] session=session, # type: ignore[arg-type] ) self.assertEqual(denied.exception.status_code, 403) self.assertEqual(session.rollbacks, 1) update.assert_not_called() def test_self_service_imap_rebind_requires_own_secret_authority(self) -> None: principal = _Principal({"mail:profile:write_own"}) profile = _profile( "own-user-profile", scope_type="user", scope_id="user-1", ) profile.imap_config = { "host": "imap.example.test", "port": 993, "security": "tls", "sent_folder": "auto", "timeout_seconds": 30, } profile.imap_password_encrypted = "existing-ciphertext" payload = MailServerProfileUpdateRequest.model_validate( {"imap": {"host": "imap.attacker.test"}} ) with ( patch( "govoplan_mail.backend.router.get_mail_server_profile", return_value=profile, ), patch("govoplan_mail.backend.router.update_mail_server_profile") as update, self.assertRaises(HTTPException) as denied, ): router.update_profile( profile.id, payload, principal=principal, # type: ignore[arg-type] session=_Session(), # type: ignore[arg-type] ) self.assertEqual(denied.exception.status_code, 403) update.assert_not_called() def test_self_service_transport_rebind_is_allowed_with_own_secret_authority(self) -> None: principal = _Principal( {"mail:profile:write_own", "mail:secret:manage_own"} ) profile = _profile( "own-user-profile", scope_type="user", scope_id="user-1", ) profile.smtp_config = { "host": "smtp.example.test", "port": 587, "security": "starttls", "timeout_seconds": 30, } profile.smtp_password_encrypted = "existing-ciphertext" payload = MailServerProfileUpdateRequest.model_validate( {"smtp": {"host": "smtp.allowed.test"}} ) session = _Session() with ( patch( "govoplan_mail.backend.router.get_mail_server_profile", return_value=profile, ), patch( "govoplan_mail.backend.router.update_mail_server_profile", return_value=profile, ) as update, patch("govoplan_mail.backend.router._record_mail_change"), patch( "govoplan_mail.backend.router._profile_response", return_value=profile, ), ): result = router.update_profile( profile.id, payload, principal=principal, # type: ignore[arg-type] session=session, # type: ignore[arg-type] ) self.assertIs(result, profile) self.assertEqual(session.commits, 1) update.assert_called_once() def test_transport_endpoint_change_without_stored_secret_needs_only_write_authority(self) -> None: principal = _Principal({"mail:profile:write_own"}) profile = _profile( "own-user-profile", scope_type="user", scope_id="user-1", ) payload = MailServerProfileUpdateRequest.model_validate( {"smtp": {"host": "smtp.allowed.test"}} ) session = _Session() with ( patch( "govoplan_mail.backend.router.get_mail_server_profile", return_value=profile, ), patch( "govoplan_mail.backend.router.update_mail_server_profile", return_value=profile, ) as update, patch("govoplan_mail.backend.router._record_mail_change"), patch( "govoplan_mail.backend.router._profile_response", return_value=profile, ), ): router.update_profile( profile.id, payload, principal=principal, # type: ignore[arg-type] session=session, # type: ignore[arg-type] ) self.assertEqual(session.commits, 1) update.assert_called_once() 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_self_service_repair_visibility_is_limited_to_the_exact_owner(self) -> None: own_profile = _profile( "own-policy-invalid", scope_type="user", scope_id="user-1", ) other_profile = _profile( "other-policy-invalid", scope_type="user", scope_id="user-2", ) denied = EffectiveMailProfilePolicy( allowed_profile_id_sets=[{"different-profile"}] ) session = _Session([own_profile, other_profile]) with patch( "govoplan_mail.backend.mail_profiles.effective_mail_profile_policy", return_value=denied, ): ordinary_visible = list_mail_server_profiles( session, # type: ignore[arg-type] tenant_id="tenant-1", include_inactive=True, actor_user_id="user-1", actor_administrative_visibility=True, ) repair_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_own_profiles=True, actor_administrative_visibility=True, ) resolved = get_mail_server_profile_for_actor( session, # type: ignore[arg-type] tenant_id="tenant-1", profile_id=own_profile.id, user_id="user-1", can_manage_own_profiles=True, administrative_visibility=True, ) with self.assertRaises(MailProfileError): get_mail_server_profile_for_actor( session, # type: ignore[arg-type] tenant_id="tenant-1", profile_id=other_profile.id, user_id="user-1", can_manage_own_profiles=True, administrative_visibility=True, ) self.assertEqual(ordinary_visible, []) self.assertEqual([item.id for item in repair_visible], [own_profile.id]) self.assertIs(resolved, own_profile) 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()