853 lines
33 KiB
Python
853 lines
33 KiB
Python
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()
|