From 06e0fbd3c3a606320f2d60f671845939cf398f83 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 21 Jul 2026 17:51:25 +0200 Subject: [PATCH] security: enforce actor-aware Mail profile access --- src/govoplan_mail/backend/mail_profiles.py | 346 ++++++++++++++++- src/govoplan_mail/backend/mailbox_index.py | 21 + src/govoplan_mail/backend/router.py | 423 +++++++++++++++------ tests/test_mail_profile_helpers.py | 206 +++++++++- tests/test_mailbox_index.py | 49 +++ tests/test_profile_actor_authorization.py | 419 ++++++++++++++++++++ tests/test_router_profile_deletion.py | 30 ++ 7 files changed, 1336 insertions(+), 158 deletions(-) create mode 100644 tests/test_mailbox_index.py create mode 100644 tests/test_profile_actor_authorization.py diff --git a/src/govoplan_mail/backend/mail_profiles.py b/src/govoplan_mail/backend/mail_profiles.py index 31ce9fc..a412075 100644 --- a/src/govoplan_mail/backend/mail_profiles.py +++ b/src/govoplan_mail/backend/mail_profiles.py @@ -13,7 +13,9 @@ from govoplan_core.admin.settings import get_system_settings from govoplan_core.audit.logging import audit_event from govoplan_core.core.access import CAPABILITY_ACCESS_DIRECTORY, AccessDirectory from govoplan_core.core.campaigns import ( + CAPABILITY_CAMPAIGNS_ACCESS, CAPABILITY_CAMPAIGNS_MAIL_POLICY_CONTEXT, + CampaignAccessProvider, CampaignMailPolicyContext, CampaignMailPolicyContextProvider, ) @@ -21,6 +23,7 @@ from govoplan_core.core.policy import policy_source_step from govoplan_core.core.runtime import get_registry from govoplan_mail.backend.db.models import MailProfilePolicy, MailServerProfile, new_uuid from govoplan_mail.backend.config import ImapConfig, SmtpConfig +from govoplan_mail.backend.mailbox_index import clear_mailbox_index from govoplan_core.security.secrets import decrypt_secret, encrypt_secret @@ -270,6 +273,16 @@ def _campaign_policy_provider() -> CampaignMailPolicyContextProvider | None: return capability +def _campaign_access_provider() -> CampaignAccessProvider | None: + registry = get_registry() + if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(CAPABILITY_CAMPAIGNS_ACCESS): + return None + capability = registry.require_capability(CAPABILITY_CAMPAIGNS_ACCESS) + if not isinstance(capability, CampaignAccessProvider): + raise MailProfileError("Campaign access capability is invalid") + return capability + + def _campaign_policy_context(session: Session, *, tenant_id: str, campaign_id: str) -> CampaignMailPolicyContext: provider = _campaign_policy_provider() if provider is None: @@ -855,10 +868,11 @@ def _profile_allowed_for_context( campaign: CampaignMailPolicyContext | None = None, owner_user_id: str | None = None, owner_group_id: str | None = None, + require_active: bool = True, ) -> bool: scope_type = _profile_scope_type(profile) scope_id = _profile_scope_id(profile) - if not profile.is_active: + if require_active and not profile.is_active: return False if not policy.profile_id_allowed(profile.id): return False @@ -881,15 +895,229 @@ def _profile_allowed_for_context( return _profile_transport_allowed(profile, policy) +def campaign_mail_context_visible_to_actor( + session: Session, + *, + tenant_id: str, + campaign_id: str, + user_id: str, + group_ids: Iterable[str] = (), + tenant_admin: bool = False, +) -> bool: + provider = _campaign_access_provider() + if provider is None: + return False + return bool( + provider.can_read_campaign( + session, + tenant_id=tenant_id, + campaign_id=campaign_id, + user_id=user_id, + group_ids=group_ids, + tenant_admin=tenant_admin, + ) + ) + + +def mail_profile_scope_visible_to_actor( + session: Session, + *, + scope_type: str, + scope_id: str | None, + profile_tenant_id: str | None, + tenant_id: str, + user_id: str, + group_ids: Iterable[str] = (), + can_manage_tenant_profiles: bool = False, + tenant_admin: bool = False, +) -> bool: + """Apply stable ownership/ACL discovery without activity or policy state.""" + + clean_scope_type = _normalize_scope_type(scope_type) + if clean_scope_type == "system": + return True + if profile_tenant_id != tenant_id: + return False + if clean_scope_type == "tenant": + return True + if clean_scope_type == "user": + return bool(can_manage_tenant_profiles or (scope_id and scope_id == user_id)) + if clean_scope_type == "group": + return bool( + can_manage_tenant_profiles + or (scope_id and scope_id in {str(group_id) for group_id in group_ids}) + ) + if clean_scope_type != "campaign" or not scope_id: + return False + # Mail administration is not a Campaign ACL override. Only Campaign's own + # access capability decides this, including its exact tenant-admin rule. + return campaign_mail_context_visible_to_actor( + session, + tenant_id=tenant_id, + campaign_id=scope_id, + user_id=user_id, + group_ids=group_ids, + tenant_admin=tenant_admin, + ) + + +def mail_profile_visible_to_actor( + session: Session, + *, + profile: MailServerProfile, + tenant_id: str, + user_id: str, + group_ids: Iterable[str] = (), + can_manage_tenant_profiles: bool = False, + can_manage_system_profiles: bool = False, + tenant_admin: bool = False, + administrative_visibility: bool = False, + require_active: bool = True, +) -> bool: + """Return whether an actor may discover or directly use a profile. + + System and tenant profiles are shared definitions. Narrower profiles are + visible only to their owner context, unless the actor is a tenant-wide Mail + profile administrator. Campaign visibility is delegated to Campaign's ACL + capability so Mail never guesses another module's object permissions. + """ + + normalized_group_ids = tuple(sorted({str(group_id) for group_id in group_ids})) + scope_type = _profile_scope_type(profile) + scope_id = _profile_scope_id(profile) + if not mail_profile_scope_visible_to_actor( + session, + scope_type=scope_type, + scope_id=scope_id, + profile_tenant_id=profile.tenant_id, + tenant_id=tenant_id, + user_id=user_id, + group_ids=normalized_group_ids, + can_manage_tenant_profiles=( + administrative_visibility and can_manage_tenant_profiles + ), + tenant_admin=tenant_admin, + ): + return False + if administrative_visibility and scope_type == "system" and can_manage_system_profiles: + return True + if administrative_visibility and scope_type != "system" and can_manage_tenant_profiles: + return True + if scope_type in {"system", "tenant"}: + return _profile_allowed_for_actor_policy( + session, + profile=profile, + tenant_id=tenant_id, + user_id=user_id, + group_ids=normalized_group_ids, + require_active=require_active, + ) + if scope_type == "user": + return _profile_allowed_for_actor_policy( + session, + profile=profile, + tenant_id=tenant_id, + user_id=user_id, + group_ids=(), + require_active=require_active, + ) + if scope_type == "group": + return _profile_allowed_for_actor_policy( + session, + profile=profile, + tenant_id=tenant_id, + user_id=user_id, + group_ids=(scope_id,), + require_active=require_active, + ) + if scope_type != "campaign" or not scope_id: + return False + campaign = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=scope_id) + policy = effective_mail_profile_policy(session, tenant_id=tenant_id, campaign_id=scope_id) + return _profile_allowed_for_context( + profile, + tenant_id=tenant_id, + policy=policy, + campaign=campaign, + owner_user_id=campaign.owner_user_id, + owner_group_id=campaign.owner_group_id, + require_active=require_active, + ) + + +def _profile_allowed_for_actor_policy( + session: Session, + *, + profile: MailServerProfile, + tenant_id: str, + user_id: str, + group_ids: Iterable[str], + require_active: bool = True, +) -> bool: + """Apply every unambiguous actor policy context conservatively. + + Direct Mail screens have no selected task/group context. The actor's user + policy always applies; for shared system/tenant profiles each group policy + also has to allow the profile. A caller that needs campaign-specific policy + must use the explicit campaign context path instead. + """ + + normalized_group_ids = tuple(sorted({str(group_id) for group_id in group_ids})) + scope_type = _profile_scope_type(profile) + if scope_type == "group": + contexts: tuple[str | None, ...] = normalized_group_ids + elif scope_type == "user": + contexts = (None,) + else: + contexts = (None, *normalized_group_ids) + if not contexts: + return False + for group_id in contexts: + policy = effective_mail_profile_policy( + session, + tenant_id=tenant_id, + owner_user_id=user_id, + owner_group_id=group_id, + ) + if not _profile_allowed_for_context( + profile, + tenant_id=tenant_id, + policy=policy, + owner_user_id=user_id, + owner_group_id=group_id, + require_active=require_active, + ): + return False + return True + + def list_mail_server_profiles( session: Session, *, tenant_id: str, include_inactive: bool = False, campaign_id: str | None = None, + actor_user_id: str | None = None, + actor_group_ids: Iterable[str] = (), + actor_can_manage_tenant_profiles: bool = False, + actor_can_manage_system_profiles: bool = False, + actor_tenant_admin: bool = False, + actor_administrative_visibility: bool = False, ) -> list[MailServerProfile]: + normalized_actor_group_ids = tuple( + sorted({str(group_id) for group_id in actor_group_ids}) + ) if campaign_id: campaign = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=campaign_id) + if actor_user_id and not campaign_mail_context_visible_to_actor( + session, + tenant_id=tenant_id, + campaign_id=campaign.id, + user_id=actor_user_id, + group_ids=normalized_actor_group_ids, + tenant_admin=actor_tenant_admin, + ): + raise MailProfileError("Campaign not found for mail-server profile policy") policy = effective_mail_profile_policy(session, tenant_id=tenant_id, campaign_id=campaign.id) query = session.query(MailServerProfile).filter(or_(*_context_profile_clauses(campaign, policy))) if not include_inactive: @@ -905,6 +1133,7 @@ def list_mail_server_profiles( campaign=campaign, owner_user_id=campaign.owner_user_id, owner_group_id=campaign.owner_group_id, + require_active=not include_inactive, ) ] else: @@ -912,9 +1141,63 @@ def list_mail_server_profiles( if not include_inactive: query = query.filter(MailServerProfile.is_active.is_(True)) profiles = query.all() + if actor_user_id: + profiles = [ + profile + for profile in profiles + if mail_profile_visible_to_actor( + session, + profile=profile, + tenant_id=tenant_id, + user_id=actor_user_id, + group_ids=normalized_actor_group_ids, + can_manage_tenant_profiles=actor_can_manage_tenant_profiles, + can_manage_system_profiles=actor_can_manage_system_profiles, + tenant_admin=actor_tenant_admin, + administrative_visibility=actor_administrative_visibility, + require_active=not include_inactive, + ) + ] return sorted(profiles, key=lambda item: (PROFILE_SCOPE_ORDER.get(_profile_scope_type(item), 99), (item.name or "").casefold())) +def get_mail_server_profile_for_actor( + session: Session, + *, + tenant_id: str, + profile_id: str, + user_id: str, + group_ids: Iterable[str] = (), + can_manage_tenant_profiles: bool = False, + can_manage_system_profiles: bool = False, + tenant_admin: bool = False, + administrative_visibility: bool = False, + require_active: bool = False, +) -> MailServerProfile: + profile = get_mail_server_profile( + session, + tenant_id=tenant_id, + profile_id=profile_id, + require_active=require_active, + ) + if not mail_profile_visible_to_actor( + session, + profile=profile, + tenant_id=tenant_id, + user_id=user_id, + group_ids=group_ids, + can_manage_tenant_profiles=can_manage_tenant_profiles, + can_manage_system_profiles=can_manage_system_profiles, + tenant_admin=tenant_admin, + administrative_visibility=administrative_visibility, + require_active=require_active, + ): + # Deliberately indistinguishable from an absent profile to prevent id + # probing across user, group, and campaign scopes. + raise MailProfileError("Mail-server profile not found") + return profile + + def get_mail_server_profile( session: Session, *, @@ -948,6 +1231,7 @@ def ensure_mail_profile_allowed_for_campaign( campaign=campaign, owner_user_id=campaign.owner_user_id, owner_group_id=campaign.owner_group_id, + require_active=require_active, ): raise MailProfileError("Mail-server profile is not allowed for this campaign owner or policy") return profile @@ -1086,6 +1370,10 @@ def create_mail_server_profile( imap_password = None if imap is not None: imap_payload, imap_username, imap_password, _, _ = _transport_payload(imap) + if not is_active and (smtp_password not in (None, "") or imap_password not in (None, "")): + raise MailProfileError( + "Inactive mail-server profiles cannot retain credentials; create the active profile or omit credentials" + ) profile = MailServerProfile( tenant_id=profile_tenant_id, scope_type=clean_scope_type, @@ -1123,6 +1411,30 @@ def update_mail_server_profile( imap: ImapConfig | None = None, clear_imap: bool = False, ) -> MailServerProfile: + if is_active is False: + raise MailProfileError( + "Deactivate profiles through DELETE so credentials are scrubbed and the deletion is audited" + ) + if not profile.is_active and ( + profile.smtp_password_encrypted or profile.imap_password_encrypted + ): + raise MailProfileError( + "Inactive profile still contains legacy credentials; use DELETE to scrub them before editing or reactivating" + ) + inactive_password_update = ( + not profile.is_active + and is_active is not True + and any( + config is not None + and "password" in config.model_fields_set + and config.password not in (None, "") + for config in (smtp, imap) + ) + ) + if inactive_password_update: + raise MailProfileError( + "Credentials may only be added to an inactive profile when it is activated in the same update" + ) if name is not None: clean_name = name.strip() if not clean_name: @@ -1166,13 +1478,17 @@ def _next_profile_transport_state( smtp: SmtpConfig | None, imap: ImapConfig | None, clear_imap: bool, -) -> tuple[SmtpConfig, ImapConfig | None]: - next_smtp = smtp or smtp_config_from_profile(profile) +) -> tuple[SmtpConfig | dict[str, Any], ImapConfig | dict[str, Any] | None]: + # Policy validation only needs non-secret transport metadata. Never decrypt + # an unchanged protocol merely because profile metadata or the other + # protocol is being updated; corrupt/retired ciphertext must not block the + # credential replacement and deletion recovery paths. + next_smtp: SmtpConfig | dict[str, Any] = smtp or dict(profile.smtp_config or {}) if clear_imap: return next_smtp, None if imap is not None: return next_smtp, imap - return next_smtp, imap_config_from_profile(profile) + return next_smtp, dict(profile.imap_config or {}) if profile.imap_config else None def _assert_profile_transport_allowed( @@ -1180,8 +1496,8 @@ def _assert_profile_transport_allowed( *, tenant_id: str, profile: MailServerProfile, - smtp: SmtpConfig, - imap: ImapConfig | None, + smtp: SmtpConfig | dict[str, Any], + imap: ImapConfig | dict[str, Any] | None, ) -> None: scope_type = _profile_scope_type(profile) scope_id = _profile_scope_id(profile) @@ -1229,6 +1545,7 @@ def _apply_profile_transport_update( user_id=user_id, api_key_id=api_key_id, ) + clear_mailbox_index(session, profile_id=profile.id) profile.imap_config = None profile.imap_username = None profile.imap_password_encrypted = None @@ -1259,12 +1576,6 @@ def _apply_profile_transport_payload( for field_name in ("username", "password", "enabled"): current_payload.pop(field_name, None) current_username = _profile_username(profile, protocol) - current_password_encrypted = ( - profile.smtp_password_encrypted - if protocol == "smtp" - else profile.imap_password_encrypted - ) - current_password = decrypt_secret(current_password_encrypted) if password_supplied else None identity_changed = ( current_payload != payload or (username_supplied and current_username != username) @@ -1273,7 +1584,7 @@ def _apply_profile_transport_payload( session, profile=profile, protocol=protocol, - password_supplied=password_supplied and current_password != password, + password_supplied=password_supplied, next_password=password, user_id=user_id, api_key_id=api_key_id, @@ -1282,15 +1593,20 @@ def _apply_profile_transport_payload( profile.smtp_config = payload if username_supplied: profile.smtp_username = username - if password_supplied and current_password != password: + if password_supplied: profile.smtp_password_encrypted = encrypt_secret(password) if identity_changed: profile.smtp_transport_revision = new_uuid() return + # Cached mailbox metadata and message summaries belong to the exact IMAP + # account identity. Clear them in the same transaction before changing an + # identity or credential so another account can never inherit them. + if identity_changed or password_supplied: + clear_mailbox_index(session, profile_id=profile.id) profile.imap_config = payload if username_supplied: profile.imap_username = username - if password_supplied and current_password != password: + if password_supplied: profile.imap_password_encrypted = encrypt_secret(password) if identity_changed: profile.imap_transport_revision = new_uuid() diff --git a/src/govoplan_mail/backend/mailbox_index.py b/src/govoplan_mail/backend/mailbox_index.py index 2ba416a..8df0967 100644 --- a/src/govoplan_mail/backend/mailbox_index.py +++ b/src/govoplan_mail/backend/mailbox_index.py @@ -36,6 +36,27 @@ class CachedMessagePage: stale: bool +def clear_mailbox_index(session: Session, *, profile_id: str) -> tuple[int, int]: + """Remove every cached mailbox row for a profile in the caller's transaction. + + System profiles can be used by more than one tenant, so invalidation is + intentionally profile-wide rather than limited to the tenant making the + transport change. Message rows are removed before their folder metadata. + """ + + deleted_messages = ( + session.query(MailMailboxMessageIndex) + .filter(MailMailboxMessageIndex.profile_id == profile_id) + .delete(synchronize_session=False) + ) + deleted_folders = ( + session.query(MailMailboxFolderIndex) + .filter(MailMailboxFolderIndex.profile_id == profile_id) + .delete(synchronize_session=False) + ) + return int(deleted_folders or 0), int(deleted_messages or 0) + + def begin_mailbox_refresh(tenant_id: str, profile_id: str, folder: str) -> bool: key = (tenant_id, profile_id, folder) with _refresh_lock: diff --git a/src/govoplan_mail/backend/router.py b/src/govoplan_mail/backend/router.py index b83c708..7793248 100644 --- a/src/govoplan_mail/backend/router.py +++ b/src/govoplan_mail/backend/router.py @@ -1,11 +1,10 @@ from __future__ import annotations import dataclasses -import logging from types import SimpleNamespace from typing import Any -from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status +from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy import func, or_ from sqlalchemy.orm import Session @@ -41,24 +40,27 @@ from govoplan_core.core.change_sequence import ( sequence_watermark_is_expired, ) from govoplan_core.core.pagination import KeysetCursorError, decode_keyset_cursor, encode_keyset_cursor, keyset_query_fingerprint -from govoplan_core.db.session import get_database, get_session +from govoplan_core.db.session import get_session from govoplan_mail.backend.mailbox_index import ( - begin_mailbox_refresh, cache_mailbox_folders, cache_mailbox_messages, cached_mailbox_folders, cached_mailbox_message_page, - finish_mailbox_refresh, + clear_mailbox_index, ) from govoplan_mail.backend.mail_profiles import ( MailProfileError, + campaign_mail_context_visible_to_actor, create_mail_server_profile, delete_mail_profile_credentials, effective_mail_profile_policy_for_scope, get_mail_profile_policy, get_mail_server_profile, + get_mail_server_profile_for_actor, imap_config_from_profile, list_mail_server_profiles, + mail_profile_visible_to_actor, + mail_profile_scope_visible_to_actor, parent_mail_profile_policy, profile_response_payload, set_mail_profile_policy, @@ -71,7 +73,6 @@ from govoplan_mail.backend.sending.imap import ImapAppendError, ImapConfiguratio from govoplan_mail.backend.sending.smtp import test_smtp_login router = APIRouter(prefix="/mail", tags=["mail"]) -logger = logging.getLogger(__name__) MAIL_MODULE_ID = "mail" MAIL_PROFILES_COLLECTION = "mail.profiles" @@ -145,6 +146,55 @@ def _require_mailbox_read_scope(principal: ApiPrincipal) -> None: _require_scope(principal, "mail:profile:use") +def _principal_is_tenant_admin(principal: ApiPrincipal) -> bool: + return has_scope(principal, "tenant:*") + + +def _principal_can_manage_tenant_profiles(principal: ApiPrincipal) -> bool: + return has_scope(principal, "mail:profile:write") + + +def _principal_can_manage_system_profiles(principal: ApiPrincipal) -> bool: + return has_scope(principal, "system:settings:write") + + +def _profile_actor_kwargs( + principal: ApiPrincipal, + *, + administrative_visibility: bool, +) -> dict[str, object]: + return { + "actor_user_id": principal.user.id, + "actor_group_ids": principal.group_ids, + "actor_can_manage_tenant_profiles": _principal_can_manage_tenant_profiles(principal), + "actor_can_manage_system_profiles": _principal_can_manage_system_profiles(principal), + "actor_tenant_admin": _principal_is_tenant_admin(principal), + "actor_administrative_visibility": administrative_visibility, + } + + +def _get_profile_for_principal( + session: Session, + *, + principal: ApiPrincipal, + profile_id: str, + require_active: bool = False, + administrative_visibility: bool = False, +): + return get_mail_server_profile_for_actor( + session, + tenant_id=principal.tenant_id, + profile_id=profile_id, + user_id=principal.user.id, + group_ids=principal.group_ids, + can_manage_tenant_profiles=_principal_can_manage_tenant_profiles(principal), + can_manage_system_profiles=_principal_can_manage_system_profiles(principal), + tenant_admin=_principal_is_tenant_admin(principal), + administrative_visibility=administrative_visibility, + require_active=require_active, + ) + + def _require_policy_read_scope(principal: ApiPrincipal, scope_type: str) -> None: if scope_type == "system": _require_scope(principal, "system:settings:read") @@ -163,6 +213,86 @@ def _require_policy_write_scope(principal: ApiPrincipal, scope_type: str) -> Non _require_any_scope(principal, "admin:policies:write", "mail:profile:write") +def _require_policy_scope_visibility( + session: Session, + *, + principal: ApiPrincipal, + scope_type: str, + scope_id: str | None, +) -> None: + if scope_type in {"system", "tenant"}: + return + if has_scope(principal, "admin:policies:read"): + return + if not mail_profile_scope_visible_to_actor( + session, + scope_type=scope_type, + scope_id=scope_id, + profile_tenant_id=principal.tenant_id, + tenant_id=principal.tenant_id, + user_id=principal.user.id, + group_ids=principal.group_ids, + can_manage_tenant_profiles=_principal_can_manage_tenant_profiles(principal), + tenant_admin=_principal_is_tenant_admin(principal), + ): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Mail profile policy not found") + + +def _require_campaign_context_visibility( + session: Session, + *, + principal: ApiPrincipal, + campaign_id: str | None, +) -> None: + if not campaign_id: + return + if not campaign_mail_context_visible_to_actor( + session, + tenant_id=principal.tenant_id, + campaign_id=campaign_id, + user_id=principal.user.id, + group_ids=principal.group_ids, + tenant_admin=_principal_is_tenant_admin(principal), + ): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Campaign not found") + + +def _require_policy_campaign_context( + session: Session, + *, + principal: ApiPrincipal, + scope_type: str, + scope_id: str | None, + campaign_id: str | None, +) -> None: + if scope_type == "campaign" and campaign_id and campaign_id != scope_id: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail="campaign_id must match the campaign policy scope_id", + ) + _require_campaign_context_visibility( + session, + principal=principal, + campaign_id=campaign_id or (scope_id if scope_type == "campaign" else None), + ) + + +def _require_campaign_profile_mutation_access( + session: Session, + *, + principal: ApiPrincipal, + scope_type: str, + scope_id: str | None, +) -> None: + if scope_type != "campaign": + return + _require_campaign_context_visibility( + session, + principal=principal, + campaign_id=scope_id, + ) + + def _profile_response(profile) -> MailServerProfileResponse: return MailServerProfileResponse.model_validate(profile_response_payload(profile)) @@ -283,6 +413,45 @@ def _mail_deleted_item(entry) -> DeltaDeletedItem: ) +def _profile_change_visible_to_principal( + session: Session, + *, + entry: ChangeSequenceEntry, + principal: ApiPrincipal, +) -> bool: + try: + profile = get_mail_server_profile( + session, + tenant_id=principal.tenant_id, + profile_id=entry.resource_id, + ) + except MailProfileError: + payload = entry.payload if isinstance(entry.payload, dict) else {} + return mail_profile_scope_visible_to_actor( + session, + scope_type=str(payload.get("scope_type") or "tenant"), + scope_id=str(payload["scope_id"]) if payload.get("scope_id") else None, + profile_tenant_id=entry.tenant_id, + tenant_id=principal.tenant_id, + user_id=principal.user.id, + group_ids=principal.group_ids, + can_manage_tenant_profiles=_principal_can_manage_tenant_profiles(principal), + tenant_admin=_principal_is_tenant_admin(principal), + ) + return mail_profile_visible_to_actor( + session, + profile=profile, + tenant_id=principal.tenant_id, + user_id=principal.user.id, + group_ids=principal.group_ids, + can_manage_tenant_profiles=_principal_can_manage_tenant_profiles(principal), + can_manage_system_profiles=_principal_can_manage_system_profiles(principal), + tenant_admin=_principal_is_tenant_admin(principal), + administrative_visibility=True, + require_active=False, + ) + + def _mailbox_summary_response(message) -> MailMailboxMessageSummaryResponse: return MailMailboxMessageSummaryResponse( uid=message.uid, @@ -451,65 +620,6 @@ def _mailbox_messages_response( ) -def _schedule_mailbox_refresh( - background_tasks: BackgroundTasks, - *, - tenant_id: str, - profile_id: str, - folder: str, - limit: int, - offset: int = 0, - include_folder_status: bool = False, -) -> bool: - if not begin_mailbox_refresh(tenant_id, profile_id, folder): - return False - background_tasks.add_task( - _refresh_mailbox_index_task, - tenant_id=tenant_id, - profile_id=profile_id, - folder=folder, - limit=limit, - offset=offset, - include_folder_status=include_folder_status, - ) - return True - - -def _refresh_mailbox_index_task( - *, - tenant_id: str, - profile_id: str, - folder: str, - limit: int, - offset: int, - include_folder_status: bool, -) -> None: - try: - with get_database().session() as session: - imap = _imap_config_for_profile(session, tenant_id=tenant_id, profile_id=profile_id) - result = load_imap_mailbox_bootstrap( - imap_config=imap, - folder=folder, - limit=limit, - offset=offset, - include_folder_status=include_folder_status, - ) - cache_mailbox_folders(session, tenant_id=tenant_id, profile_id=profile_id, result=result.folders) - cache_mailbox_messages(session, tenant_id=tenant_id, profile_id=profile_id, result=result.messages) - session.commit() - except Exception: - # Background refresh is opportunistic. Foreground requests will surface - # concrete IMAP errors when no usable cache is available. - logger.debug( - "Background mailbox index refresh failed (profile_id=%r, folder=%r)", - profile_id, - folder, - exc_info=True, - ) - finally: - finish_mailbox_refresh(tenant_id, profile_id, folder) - - def _cached_folder_selection(folders, requested: str, detected_sent_folder: str | None = None) -> str: names = {folder.name for folder in folders} if requested in names: @@ -521,8 +631,13 @@ def _cached_folder_selection(folders, requested: str, detected_sent_folder: str return folders[0].name if folders else requested -def _imap_config_for_profile(session: Session, *, tenant_id: str, profile_id: str): - profile = get_mail_server_profile(session, tenant_id=tenant_id, profile_id=profile_id, require_active=True) +def _imap_config_for_principal(session: Session, *, principal: ApiPrincipal, profile_id: str): + profile = _get_profile_for_principal( + session, + principal=principal, + profile_id=profile_id, + require_active=True, + ) imap = imap_config_from_profile(profile) if imap is None: raise MailProfileError("Mail-server profile has no IMAP configuration") @@ -538,7 +653,7 @@ def _parent_mail_profile_policy_payload(session: Session, *, tenant_id: str, sco def _full_mail_settings_delta_response( session: Session, *, - tenant_id: str, + principal: ApiPrincipal, scope_type: str, scope_id: str | None, include_inactive: bool, @@ -546,16 +661,17 @@ def _full_mail_settings_delta_response( ) -> MailSettingsDeltaResponse: profiles = list_mail_server_profiles( session, - tenant_id=tenant_id, + tenant_id=principal.tenant_id, include_inactive=include_inactive, campaign_id=campaign_id, + **_profile_actor_kwargs(principal, administrative_visibility=True), ) return MailSettingsDeltaResponse( profiles=[_profile_response(profile) for profile in profiles], - policy=_policy_response(session, tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id, campaign_id=campaign_id), + policy=_policy_response(session, tenant_id=principal.tenant_id, scope_type=scope_type, scope_id=scope_id, campaign_id=campaign_id), changed_sections=["profiles", "policy"], deleted=[], - watermark=_mail_settings_watermark(session, tenant_id=tenant_id), + watermark=_mail_settings_watermark(session, tenant_id=principal.tenant_id), has_more=False, full=True, ) @@ -593,10 +709,23 @@ def mail_settings_delta( clean_scope_type = scope_type.strip().casefold() _require_any_scope(principal, "mail:profile:read", "system:settings:read") _require_policy_read_scope(principal, clean_scope_type) + _require_policy_scope_visibility( + session, + principal=principal, + scope_type=clean_scope_type, + scope_id=scope_id, + ) + _require_policy_campaign_context( + session, + principal=principal, + scope_type=clean_scope_type, + scope_id=scope_id, + campaign_id=campaign_id, + ) if since is None: return _full_mail_settings_delta_response( session, - tenant_id=principal.tenant_id, + principal=principal, scope_type=clean_scope_type, scope_id=scope_id, include_inactive=include_inactive, @@ -606,16 +735,22 @@ def mail_settings_delta( if entries is None: return _full_mail_settings_delta_response( session, - tenant_id=principal.tenant_id, + principal=principal, scope_type=clean_scope_type, scope_id=scope_id, include_inactive=include_inactive, campaign_id=campaign_id, ) + visible_profile_entries = [ + entry + for entry in entries + if entry.collection == MAIL_PROFILES_COLLECTION + and entry.resource_type == MAIL_PROFILE_RESOURCE + and _profile_change_visible_to_principal(session, entry=entry, principal=principal) + ] changed_profile_ids = { entry.resource_id - for entry in entries - if entry.collection == MAIL_PROFILES_COLLECTION and entry.resource_type == MAIL_PROFILE_RESOURCE + for entry in visible_profile_entries } visible_profiles = { profile.id: profile @@ -624,6 +759,7 @@ def mail_settings_delta( tenant_id=principal.tenant_id, include_inactive=include_inactive, campaign_id=campaign_id, + **_profile_actor_kwargs(principal, administrative_visibility=True), ) if profile.id in changed_profile_ids } @@ -635,10 +771,8 @@ def mail_settings_delta( changed_sections.append("policy") deleted = [ _mail_deleted_item(entry) - for entry in entries - if entry.collection == MAIL_PROFILES_COLLECTION - and entry.resource_type == MAIL_PROFILE_RESOURCE - and entry.resource_id not in visible_profiles + for entry in visible_profile_entries + if entry.resource_id not in visible_profiles ] return MailSettingsDeltaResponse( profiles=[_profile_response(profile) for profile in visible_profiles.values()], @@ -665,6 +799,7 @@ def list_profiles( tenant_id=principal.tenant_id, include_inactive=include_inactive, campaign_id=campaign_id, + **_profile_actor_kwargs(principal, administrative_visibility=True), ) return MailServerProfileListResponse(profiles=[_profile_response(profile) for profile in profiles]) except MailProfileError as exc: @@ -678,6 +813,12 @@ def create_profile( session: Session = Depends(get_session), ): _require_profile_write_scope(principal, payload.scope_type) + _require_campaign_profile_mutation_access( + session, + principal=principal, + scope_type=payload.scope_type, + scope_id=payload.scope_id, + ) if payload.credentials_supplied(): _require_profile_credentials_scope(principal, payload.scope_type) try: @@ -720,7 +861,14 @@ def get_profile( ): _require_any_scope(principal, "mail:profile:read", "system:settings:read") try: - return _profile_response(get_mail_server_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id)) + return _profile_response( + _get_profile_for_principal( + session, + principal=principal, + profile_id=profile_id, + administrative_visibility=True, + ) + ) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @@ -735,6 +883,12 @@ def update_profile( try: profile = get_mail_server_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id) _require_profile_write_scope(principal, profile.scope_type or "tenant") + _require_campaign_profile_mutation_access( + session, + principal=principal, + scope_type=profile.scope_type or "tenant", + scope_id=profile.scope_id, + ) if payload.credentials_supplied() or payload.clear_imap: _require_profile_credentials_scope(principal, profile.scope_type or "tenant") smtp_config = payload.smtp_config() @@ -791,6 +945,12 @@ def deactivate_profile( try: profile = get_mail_server_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id) _require_profile_write_scope(principal, profile.scope_type or "tenant") + _require_campaign_profile_mutation_access( + session, + principal=principal, + scope_type=profile.scope_type or "tenant", + scope_id=profile.scope_id, + ) _require_profile_credentials_scope(principal, profile.scope_type or "tenant") was_active = bool(profile.is_active) profile.is_active = False @@ -801,6 +961,10 @@ def deactivate_profile( user_id=principal.user.id, api_key_id=principal.api_key_id, ) + # Deactivation invalidates unauthenticated IMAP profiles too. Keep the + # cache deletion in this transaction so a failed audit/change record + # cannot leave profile state and cached mailbox data out of sync. + clear_mailbox_index(session, profile_id=profile.id) session.add(profile) if was_active or deleted_protocols: _record_mail_change( @@ -839,6 +1003,19 @@ def read_mail_profile_policy( ): scope_type = scope_type.strip().casefold() _require_policy_read_scope(principal, scope_type) + _require_policy_scope_visibility( + session, + principal=principal, + scope_type=scope_type, + scope_id=scope_id, + ) + _require_policy_campaign_context( + session, + principal=principal, + scope_type=scope_type, + scope_id=scope_id, + campaign_id=campaign_id, + ) try: return _policy_response(session, tenant_id=principal.tenant_id, scope_type=scope_type, scope_id=scope_id, campaign_id=campaign_id) except MailProfileError as exc: @@ -855,6 +1032,19 @@ def write_mail_profile_policy( ): scope_type = scope_type.strip().casefold() _require_policy_write_scope(principal, scope_type) + _require_policy_scope_visibility( + session, + principal=principal, + scope_type=scope_type, + scope_id=scope_id, + ) + _require_policy_campaign_context( + session, + principal=principal, + scope_type=scope_type, + scope_id=scope_id, + campaign_id=None, + ) try: set_mail_profile_policy( session, @@ -889,7 +1079,12 @@ def test_profile_smtp( _require_scope(principal, "mail:profile:test") _require_scope(principal, "mail:profile:use") try: - profile = get_mail_server_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id, require_active=True) + profile = _get_profile_for_principal( + session, + principal=principal, + profile_id=profile_id, + require_active=True, + ) smtp = smtp_config_from_profile(profile) result = test_smtp_login(smtp_config=smtp) return MailConnectionTestResponse(ok=True, protocol="smtp", host=result.host, port=result.port, security=result.security, message="SMTP connection successful.", details={"authenticated": result.authenticated}) @@ -908,7 +1103,12 @@ def test_profile_imap( _require_scope(principal, "mail:profile:test") _require_scope(principal, "mail:profile:use") try: - profile = get_mail_server_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id, require_active=True) + profile = _get_profile_for_principal( + session, + principal=principal, + profile_id=profile_id, + require_active=True, + ) imap = imap_config_from_profile(profile) if imap is None: raise MailProfileError("Mail-server profile has no IMAP configuration") @@ -929,7 +1129,12 @@ def list_profile_imap_folders( _require_scope(principal, "mail:profile:test") _require_scope(principal, "mail:profile:use") try: - profile = get_mail_server_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id, require_active=True) + profile = _get_profile_for_principal( + session, + principal=principal, + profile_id=profile_id, + require_active=True, + ) imap = imap_config_from_profile(profile) if imap is None: raise MailProfileError("Mail-server profile has no IMAP configuration") @@ -944,7 +1149,6 @@ def list_profile_imap_folders( @router.get("/profiles/{profile_id}/mailbox/folders", response_model=MailImapFolderListResponse) def list_profile_mailbox_folders( profile_id: str, - background_tasks: BackgroundTasks, include_status: bool = Query(default=False), refresh: bool = False, principal: ApiPrincipal = Depends(get_api_principal), @@ -952,18 +1156,10 @@ def list_profile_mailbox_folders( ): _require_mailbox_read_scope(principal) try: - imap = _imap_config_for_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id) + imap = _imap_config_for_principal(session, principal=principal, profile_id=profile_id) if not include_status and not refresh: cached = cached_mailbox_folders(session, tenant_id=principal.tenant_id, profile_id=profile_id) - if cached is not None: - refreshing = cached.stale and _schedule_mailbox_refresh( - background_tasks, - tenant_id=principal.tenant_id, - profile_id=profile_id, - folder="INBOX", - limit=DEFAULT_MAILBOX_MESSAGE_LIMIT, - include_folder_status=include_status, - ) + if cached is not None and not cached.stale: result = SimpleNamespace( host=imap.host, port=imap.port, @@ -971,7 +1167,7 @@ def list_profile_mailbox_folders( folders=cached.folders, detected_sent_folder=None, ) - return _mailbox_folder_response(result, from_cache=True, refreshing=refreshing, indexed_at=cached.indexed_at) + return _mailbox_folder_response(result, from_cache=True, refreshing=False, indexed_at=cached.indexed_at) result = list_imap_folders(imap_config=imap, include_status=include_status) cache_mailbox_folders(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result) session.commit() @@ -987,7 +1183,6 @@ def list_profile_mailbox_folders( @router.get("/profiles/{profile_id}/mailbox/bootstrap", response_model=MailMailboxBootstrapResponse) def bootstrap_profile_mailbox( profile_id: str, - background_tasks: BackgroundTasks, folder: str = Query(default="INBOX", min_length=1, max_length=255), limit: int = Query(default=DEFAULT_MAILBOX_MESSAGE_LIMIT, ge=1, le=100), offset: int = Query(default=0, ge=0, le=100000), @@ -998,10 +1193,10 @@ def bootstrap_profile_mailbox( ) -> MailMailboxBootstrapResponse: _require_mailbox_read_scope(principal) try: - imap = _imap_config_for_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id) + imap = _imap_config_for_principal(session, principal=principal, profile_id=profile_id) if not refresh: cached_folders = cached_mailbox_folders(session, tenant_id=principal.tenant_id, profile_id=profile_id) - if cached_folders is not None: + if cached_folders is not None and not cached_folders.stale: selected_folder = _cached_folder_selection(cached_folders.folders, folder) cached_messages = cached_mailbox_message_page( session, @@ -1011,16 +1206,7 @@ def bootstrap_profile_mailbox( limit=limit, offset=offset, ) - if cached_messages is not None: - refreshing = (cached_folders.stale or cached_messages.stale) and _schedule_mailbox_refresh( - background_tasks, - tenant_id=principal.tenant_id, - profile_id=profile_id, - folder=selected_folder, - limit=limit, - offset=offset, - include_folder_status=include_status, - ) + if cached_messages is not None and not cached_messages.stale: folder_result = SimpleNamespace( host=imap.host, port=imap.port, @@ -1044,7 +1230,7 @@ def bootstrap_profile_mailbox( folders=_mailbox_folder_response( folder_result, from_cache=True, - refreshing=refreshing, + refreshing=False, indexed_at=cached_folders.indexed_at, ), messages=_mailbox_messages_response( @@ -1062,7 +1248,7 @@ def bootstrap_profile_mailbox( full=not cursor_stable, messages=cached_messages.messages, from_cache=True, - refreshing=refreshing, + refreshing=False, indexed_at=cached_messages.indexed_at, ), ) @@ -1118,7 +1304,6 @@ def bootstrap_profile_mailbox( @router.get("/profiles/{profile_id}/mailbox/messages", response_model=MailMailboxMessageListResponse) def list_profile_mailbox_messages( profile_id: str, - background_tasks: BackgroundTasks, folder: str = Query(default="INBOX", min_length=1, max_length=255), limit: int | None = Query(default=None, ge=1, le=100), offset: int = Query(default=0, ge=0, le=100000), @@ -1129,7 +1314,7 @@ def list_profile_mailbox_messages( ): _require_mailbox_read_scope(principal) try: - imap = _imap_config_for_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id) + imap = _imap_config_for_principal(session, principal=principal, profile_id=profile_id) effective_limit = _mailbox_cursor_limit(cursor, limit) fingerprint = _mailbox_cursor_fingerprint(tenant_id=principal.tenant_id, profile_id=profile_id, folder=folder, limit=effective_limit) effective_offset, after_uid, expected_uidvalidity, cursor_values = _mailbox_cursor_position(cursor, fingerprint=fingerprint, offset=offset) @@ -1145,15 +1330,7 @@ def list_profile_mailbox_messages( cache_matches_cursor = cached is not None and ( cursor is None or (expected_uidvalidity is not None and cached.uidvalidity == expected_uidvalidity) ) - if cached is not None and cache_matches_cursor: - refreshing = cached.stale and _schedule_mailbox_refresh( - background_tasks, - tenant_id=principal.tenant_id, - profile_id=profile_id, - folder=folder, - limit=effective_limit, - offset=effective_offset, - ) + if cached is not None and cache_matches_cursor and not cached.stale: next_cursor, cursor_stable = _next_mailbox_cursor( tenant_id=principal.tenant_id, profile_id=profile_id, @@ -1179,7 +1356,7 @@ def list_profile_mailbox_messages( full=not cursor_stable, messages=cached.messages, from_cache=True, - refreshing=refreshing, + refreshing=False, indexed_at=cached.indexed_at, ) result = list_imap_messages( @@ -1236,7 +1413,7 @@ def get_profile_mailbox_message( ): _require_mailbox_read_scope(principal) try: - imap = _imap_config_for_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id) + imap = _imap_config_for_principal(session, principal=principal, profile_id=profile_id) result = get_imap_message(imap_config=imap, folder=folder, uid=message_uid) return MailMailboxMessageResponse( profile_id=profile_id, diff --git a/tests/test_mail_profile_helpers.py b/tests/test_mail_profile_helpers.py index 9cbb40a..4be6243 100644 --- a/tests/test_mail_profile_helpers.py +++ b/tests/test_mail_profile_helpers.py @@ -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): diff --git a/tests/test_mailbox_index.py b/tests/test_mailbox_index.py new file mode 100644 index 0000000..650298c --- /dev/null +++ b/tests/test_mailbox_index.py @@ -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() diff --git a/tests/test_profile_actor_authorization.py b/tests/test_profile_actor_authorization.py new file mode 100644 index 0000000..c192635 --- /dev/null +++ b/tests/test_profile_actor_authorization.py @@ -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() diff --git a/tests/test_router_profile_deletion.py b/tests/test_router_profile_deletion.py index 67fc964..96b7c5d 100644 --- a/tests/test_router_profile_deletion.py +++ b/tests/test_router_profile_deletion.py @@ -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), ):