from __future__ import annotations import dataclasses from types import SimpleNamespace from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy import func, or_ from sqlalchemy.orm import Session from govoplan_mail.backend.schemas import ( MailAddressLookupCandidate, MailAddressLookupResponse, MailConnectionTestResponse, MailCampaignCredentialCreateRequest, MailCredentialBindRequest, MailCredentialCreateRequest, MailCredentialEnvelopeResponse, MailCredentialListResponse, MailCredentialUpdateRequest, MailImapFolderListResponse, MailImapFolderResponse, MailMailboxBootstrapResponse, MailImapTestRequest, MailMailboxAttachmentResponse, MailMailboxMessageDetailResponse, MailMailboxMessageListResponse, MailMailboxMessageResponse, MailMailboxMessageSummaryResponse, MailProfilePolicyResponse, MailProfilePolicyUpdateRequest, MailSettingsDeltaResponse, MailServerProfileCreateRequest, MailServerEndpointCreateRequest, MailServerEndpointResponse, MailServerEndpointUpdateRequest, MailServerProfileListResponse, MailServerProfileResponse, MailServerProfileUpdateRequest, MailSmtpTestRequest, ) from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope from govoplan_core.api.v1.schemas import DeltaDeletedItem from govoplan_core.core.change_sequence import ( ChangeSequenceEntry, decode_sequence_watermark, encode_sequence_watermark, record_change, 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_session from govoplan_mail.backend.mailbox_index import ( cache_mailbox_folders, cache_mailbox_messages, cached_mailbox_folders, cached_mailbox_message_page, clear_mailbox_index, ) from govoplan_mail.backend.mail_profiles import ( MailProfileError, campaign_mail_owner_context, 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, smtp_config_from_profile, update_mail_server_profile, ) from govoplan_mail.backend.config import ImapConfig, SmtpConfig from govoplan_mail.backend.runtime import get_registry from govoplan_mail.backend.server_hierarchy import ( MailHierarchyContext, MailServerHierarchyError, bind_mail_credential, create_and_bind_mail_credential, create_mail_server_endpoint, get_available_mail_credential, get_bound_mail_credential, get_mail_server_endpoint, hierarchy_context_for_profile, initialize_profile_hierarchy, list_available_mail_credentials, mail_credential_payload, mail_server_ref, mail_server_endpoint_payload, profile_hierarchy_payload, resolve_mail_transport, sync_default_profile_credential, sync_default_profile_server, unlink_mail_credential, update_bound_mail_credential, update_mail_server_endpoint, ) from govoplan_mail.backend.sending.imap import ImapAppendError, ImapConfigurationError, get_imap_message, list_imap_folders, list_imap_messages, load_imap_mailbox_bootstrap, test_imap_login from govoplan_mail.backend.sending.smtp import test_smtp_login router = APIRouter(prefix="/mail", tags=["mail"]) MAIL_MODULE_ID = "mail" MAIL_PROFILES_COLLECTION = "mail.profiles" MAIL_POLICIES_COLLECTION = "mail.profile_policies" MAIL_SETTINGS_COLLECTIONS = (MAIL_PROFILES_COLLECTION, MAIL_POLICIES_COLLECTION) MAIL_PROFILE_RESOURCE = "mail_profile" MAIL_POLICY_RESOURCE = "mail_profile_policy" MAIL_SERVER_RESOURCE = "mail_server" MAIL_CREDENTIAL_RESOURCE = "mail_credential" MAILBOX_MESSAGES_CURSOR_SCOPE = "mail.mailbox.messages.v1" DEFAULT_MAILBOX_MESSAGE_LIMIT = 50 CAPABILITY_ADDRESSES_LOOKUP = "addresses.lookup" def _capability_payload(value: object) -> dict[str, Any]: if dataclasses.is_dataclass(value): return dataclasses.asdict(value) if isinstance(value, dict): return dict(value) payload: dict[str, Any] = {} for key in ( "contact_id", "address_book_id", "display_name", "email", "email_label", "organization", "role_title", "tags", "source_kind", "source_ref", "source_revision", "provenance", ): if hasattr(value, key): payload[key] = getattr(value, key) return payload def _registry_capability(name: str) -> object | None: registry = get_registry() if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(name): return None return registry.capability(name) def _require_scope(principal: ApiPrincipal, scope: str) -> None: if not has_scope(principal, scope): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {scope}") def _require_any_scope(principal: ApiPrincipal, *scopes: str) -> None: if not any(has_scope(principal, scope) for scope in scopes): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Requires one of: " + ", ".join(scopes)) def _is_own_user_scope(principal: ApiPrincipal, scope_type: str, scope_id: str | None) -> bool: return scope_type == "user" and scope_id == principal.user.id def _require_profile_write_scope( principal: ApiPrincipal, scope_type: str, scope_id: str | None = None, ) -> None: if scope_type == "system": _require_scope(principal, "system:settings:write") return if has_scope(principal, "mail:profile:write"): return if _is_own_user_scope(principal, scope_type, scope_id) and has_scope( principal, "mail:profile:write_own" ): return raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: mail:profile:write", ) def _require_profile_credentials_scope( principal: ApiPrincipal, scope_type: str, scope_id: str | None = None, ) -> None: if scope_type == "system": _require_scope(principal, "system:settings:write") return if has_scope(principal, "mail:secret:manage"): return if _is_own_user_scope(principal, scope_type, scope_id) and has_scope( principal, "mail:secret:manage_own" ): return raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: mail:secret:manage", ) def _profile_for_mutation( session: Session, *, principal: ApiPrincipal, profile_id: str, ): """Resolve a profile through the actor's mutation boundary. Mutation routes deliberately do not apply effective transport policy here: an owner must still be able to repair or deactivate a profile after policy becomes more restrictive. Missing and non-owned identifiers are kept indistinguishable for self-service actors. """ try: profile = get_mail_server_profile( session, tenant_id=principal.tenant_id, profile_id=profile_id, for_update=True, mutation_owner_user_id=( principal.user.id if has_scope(principal, "mail:profile:write_own") else None ), can_manage_tenant_profiles=_principal_can_manage_tenant_profiles(principal), can_manage_system_profiles=_principal_can_manage_system_profiles(principal), ) _require_profile_write_scope( principal, profile.scope_type or "tenant", profile.scope_id, ) except (MailProfileError, HTTPException) as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Mail-server profile not found", ) from exc return profile def _server_for_mutation( session: Session, *, principal: ApiPrincipal, profile_id: str, server_id: str, ): profile = _profile_for_mutation( session, principal=principal, profile_id=profile_id, ) try: server = get_mail_server_endpoint( session, profile=profile, server_id=server_id, context=hierarchy_context_for_profile( profile, user_id=principal.user.id, group_ids=principal.group_ids, administrative=True, ), require_active=False, for_update=True, ) except MailServerHierarchyError as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Mail server not found", ) from exc return profile, server def _server_response( session: Session, *, principal: ApiPrincipal, profile, server, ) -> MailServerEndpointResponse: context = hierarchy_context_for_profile( profile, user_id=principal.user.id, group_ids=principal.group_ids, administrative=True, ) return MailServerEndpointResponse.model_validate( mail_server_endpoint_payload( session, server=server, context=context, include_inactive_credentials=True, ) ) def _record_profile_child_change( session: Session, *, principal: ApiPrincipal, profile, operation: str, resource_type: str, resource_id: str, ) -> None: _record_mail_change( session, collection=MAIL_PROFILES_COLLECTION, resource_type=MAIL_PROFILE_RESOURCE, resource_id=profile.id, operation="updated", principal=principal, tenant_id=profile.tenant_id, payload={ "scope_type": profile.scope_type, "scope_id": profile.scope_id, "child_operation": operation, "child_resource_type": resource_type, "child_resource_id": resource_id, }, ) _TRANSPORT_ENDPOINT_FIELDS = ("host", "port", "security") def _transport_endpoint_changed( current: dict[str, Any] | None, updated: SmtpConfig | ImapConfig, ) -> bool: current_payload = current or {} updated_payload = updated.model_dump(mode="json") return any( current_payload.get(field) != updated_payload.get(field) for field in _TRANSPORT_ENDPOINT_FIELDS ) def _profile_update_requires_credentials_scope( profile, *, payload: MailServerProfileUpdateRequest, smtp: SmtpConfig | None, imap: ImapConfig | None, ) -> bool: if payload.credentials_supplied() or payload.clear_imap: return True if ( smtp is not None and getattr(profile, "smtp_password_encrypted", None) and _transport_endpoint_changed(profile.smtp_config, smtp) ): return True return bool( imap is not None and getattr(profile, "imap_password_encrypted", None) and _transport_endpoint_changed(profile.imap_config, imap) ) def _profile_has_stored_credentials(profile) -> bool: return bool( getattr(profile, "smtp_password_encrypted", None) or getattr(profile, "imap_password_encrypted", None) ) def _require_mailbox_read_scope(principal: ApiPrincipal) -> None: _require_scope(principal, "mail:mailbox:read") _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_own_profiles(principal: ApiPrincipal) -> bool: return has_scope(principal, "mail:profile:write_own") 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_own_profiles": _principal_can_manage_own_profiles(principal), "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_own_profiles=_principal_can_manage_own_profiles(principal), 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") elif scope_type == "tenant": _require_any_scope(principal, "admin:policies:read", "mail:profile:read") else: _require_any_scope(principal, "admin:policies:read", "mail:profile:read", "campaigns:campaign:read") def _require_policy_write_scope(principal: ApiPrincipal, scope_type: str) -> None: if scope_type == "system": _require_scope(principal, "system:settings:write") elif scope_type == "tenant": _require_scope(principal, "admin:policies:write") else: _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 _mail_hierarchy_context( principal: ApiPrincipal, *, target_scope_type: str | None = None, target_scope_id: str | None = None, administrative: bool = False, ) -> MailHierarchyContext: resolved_scope_type = target_scope_type or "user" resolved_scope_id = ( target_scope_id if target_scope_type is not None else principal.user.id ) return MailHierarchyContext( tenant_id=principal.tenant_id, user_id=principal.user.id, group_ids=frozenset(str(item) for item in principal.group_ids), target_scope_type=resolved_scope_type, target_scope_id=resolved_scope_id, administrative=administrative, ) def _profile_response( profile, *, session: Session | None = None, principal: ApiPrincipal | None = None, target_scope_type: str | None = None, target_scope_id: str | None = None, include_inactive: bool = False, administrative: bool = False, ) -> MailServerProfileResponse: payload = profile_response_payload(profile) if session is None or principal is None: return MailServerProfileResponse.model_validate(payload) context = _mail_hierarchy_context( principal, target_scope_type=target_scope_type, target_scope_id=target_scope_id, administrative=administrative, ) servers = profile_hierarchy_payload( session, profile=profile, context=context, include_inactive=include_inactive, ) payload["servers"] = servers credentials: dict[str, dict[str, Any]] = { "smtp": {"username": None}, "imap": {"username": None}, } for protocol in ("smtp", "imap"): candidates = [server for server in servers if server["protocol"] == protocol] selected = next( ( server for server in candidates if server["is_default"] and server["is_active"] ), candidates[0] if candidates else None, ) if selected is None: continue payload[protocol] = dict(selected["config"]) bound_credentials = list(selected.get("credentials") or []) selected_credential = next( ( credential for credential in bound_credentials if credential.get("is_default") and credential.get("is_active") ), bound_credentials[0] if bound_credentials else None, ) if selected_credential is not None: credentials[protocol] = { "username": ( selected_credential.get("public_data") or {} ).get("username") } payload[f"{protocol}_password_configured"] = any( credential.get("secret_configured") for server in candidates for credential in (server.get("credentials") or []) ) payload["credentials"] = credentials return MailServerProfileResponse.model_validate(payload) def _policy_response( session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None, campaign_id: str | None = None, ) -> MailProfilePolicyResponse: policy = get_mail_profile_policy(session, tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id) effective_policy = effective_mail_profile_policy_for_scope( session, tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id, campaign_id=campaign_id, ) effective = effective_policy.as_dict() effective_sources = effective_policy.source_policies parent_policy = parent_mail_profile_policy(session, tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id) if scope_type != "system" else None parent = parent_policy.as_dict() if parent_policy else None parent_sources = parent_policy.source_policies if parent_policy else [] return MailProfilePolicyResponse( scope_type=scope_type, scope_id=scope_id, policy=policy, effective_policy=effective, parent_policy=parent, effective_policy_sources=effective_sources, parent_policy_sources=parent_sources, ) def _record_mail_change( session: Session, *, collection: str, resource_type: str, resource_id: str | None, operation: str, principal: ApiPrincipal, tenant_id: str | None, payload: dict[str, object] | None = None, ) -> None: if not resource_id: return record_change( session, module_id=MAIL_MODULE_ID, collection=collection, resource_type=resource_type, resource_id=resource_id, operation=operation, tenant_id=tenant_id, actor_type="user", actor_id=principal.user.id, payload=payload or {}, ) def _mail_delta_query(session: Session, *, tenant_id: str, since_sequence: int): return session.query(ChangeSequenceEntry).filter( ChangeSequenceEntry.id > since_sequence, ChangeSequenceEntry.module_id == MAIL_MODULE_ID, ChangeSequenceEntry.collection.in_(MAIL_SETTINGS_COLLECTIONS), or_(ChangeSequenceEntry.tenant_id == tenant_id, ChangeSequenceEntry.tenant_id.is_(None)), ) def _mail_settings_watermark(session: Session, *, tenant_id: str) -> str: sequence_id = _mail_delta_query(session, tenant_id=tenant_id, since_sequence=0).with_entities(func.max(ChangeSequenceEntry.id)).scalar() return encode_sequence_watermark(int(sequence_id or 0)) def _mail_settings_entries(session: Session, *, tenant_id: str, since: str, limit: int): try: since_sequence = decode_sequence_watermark(since) except ValueError as exc: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc expired_for_tenant = sequence_watermark_is_expired( session, since=since_sequence, tenant_id=tenant_id, module_id=MAIL_MODULE_ID, collections=MAIL_SETTINGS_COLLECTIONS, ) expired_for_system = sequence_watermark_is_expired( session, since=since_sequence, tenant_id=None, module_id=MAIL_MODULE_ID, collections=MAIL_SETTINGS_COLLECTIONS, ) if expired_for_tenant or expired_for_system: return None, False entries_plus_one = _mail_delta_query( session, tenant_id=tenant_id, since_sequence=since_sequence, ).order_by(ChangeSequenceEntry.id.asc()).limit(limit + 1).all() has_more = len(entries_plus_one) > limit return entries_plus_one[:limit], has_more def _mail_settings_response_watermark(session: Session, *, tenant_id: str, entries, has_more: bool) -> str: return encode_sequence_watermark(entries[-1].id) if has_more and entries else _mail_settings_watermark(session, tenant_id=tenant_id) def _mail_deleted_item(entry) -> DeltaDeletedItem: return DeltaDeletedItem( id=entry.resource_id, resource_type=entry.resource_type, revision=encode_sequence_watermark(entry.id), deleted_at=entry.created_at if entry.operation == "deleted" else None, ) 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_own_profiles=_principal_can_manage_own_profiles(principal), 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, folder=message.folder, subject=message.subject, from_header=message.from_header, to_header=message.to_header, cc_header=message.cc_header, date=message.date, message_id=message.message_id, flags=message.flags, size_bytes=message.size_bytes, body_preview=message.body_preview, attachment_count=message.attachment_count, ) def _mailbox_detail_response(message) -> MailMailboxMessageDetailResponse: return MailMailboxMessageDetailResponse( uid=message.uid, folder=message.folder, subject=message.subject, from_header=message.from_header, to_header=message.to_header, cc_header=message.cc_header, date=message.date, message_id=message.message_id, flags=message.flags, size_bytes=message.size_bytes, body_preview=message.body_preview, body_text=message.body_text, body_html=message.body_html, headers=message.headers, attachments=[MailMailboxAttachmentResponse(filename=item.filename, content_type=item.content_type, size_bytes=item.size_bytes) for item in message.attachments], ) def _mailbox_cursor_fingerprint(*, tenant_id: str, profile_id: str, folder: str, limit: int) -> str: return keyset_query_fingerprint( MAILBOX_MESSAGES_CURSOR_SCOPE, {"tenant_id": tenant_id, "profile_id": profile_id, "folder": folder, "limit": limit, "sort": "imap.uid.desc"}, ) def _mailbox_cursor_values(cursor: str | None, *, fingerprint: str | None = None) -> dict[str, object] | None: try: return decode_keyset_cursor(MAILBOX_MESSAGES_CURSOR_SCOPE, cursor, fingerprint=fingerprint) except KeysetCursorError as exc: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc def _mailbox_cursor_limit(cursor: str | None, explicit_limit: int | None) -> int: if explicit_limit is not None: return explicit_limit values = _mailbox_cursor_values(cursor) if cursor else None raw_limit = values.get("limit") if values else DEFAULT_MAILBOX_MESSAGE_LIMIT if not isinstance(raw_limit, int) or raw_limit < 1 or raw_limit > 100: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid pagination cursor") return raw_limit def _mailbox_cursor_position(cursor: str | None, *, fingerprint: str, offset: int) -> tuple[int, str | None, str | None, dict[str, object] | None]: values = _mailbox_cursor_values(cursor, fingerprint=fingerprint) if cursor else None if values is None: return offset, None, None, None raw_offset = values.get("offset") if not isinstance(raw_offset, int) or raw_offset < 0: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid pagination cursor") after_uid = values.get("after_uid") or values.get("anchor_uid") uidvalidity = values.get("uidvalidity") if not isinstance(after_uid, str) or not after_uid: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid pagination cursor") if uidvalidity is not None and not isinstance(uidvalidity, str): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid pagination cursor") return raw_offset, after_uid, uidvalidity, values def _next_mailbox_cursor( *, tenant_id: str, profile_id: str, folder: str, limit: int, offset: int, total_count: int, uidvalidity: str | None, messages, ) -> tuple[str | None, bool]: cursor_stable = bool(uidvalidity) next_offset = offset + len(messages) if not cursor_stable or next_offset >= total_count or not messages: return None, cursor_stable return ( encode_keyset_cursor( MAILBOX_MESSAGES_CURSOR_SCOPE, fingerprint=_mailbox_cursor_fingerprint(tenant_id=tenant_id, profile_id=profile_id, folder=folder, limit=limit), values={ "offset": next_offset, "limit": limit, "uidvalidity": uidvalidity, "after_uid": messages[-1].uid, }, ), cursor_stable, ) def _mailbox_folder_response( result, *, from_cache: bool = False, refreshing: bool = False, indexed_at=None, ) -> MailImapFolderListResponse: folders = [MailImapFolderResponse(name=item.name, flags=item.flags, message_count=item.message_count, unseen_count=item.unseen_count) for item in result.folders] return MailImapFolderListResponse( ok=True, host=result.host, port=result.port, security=result.security, message=f"Found {len(folders)} IMAP folder(s).", folders=folders, detected_sent_folder=result.detected_sent_folder, from_cache=from_cache, refreshing=refreshing, indexed_at=indexed_at, ) def _mailbox_messages_response( *, profile_id: str, folder: str, host: str | None, port: int | None, security: str | None, total_count: int, offset: int, limit: int, cursor: str | None, next_cursor: str | None, cursor_stable: bool, full: bool, messages, from_cache: bool = False, refreshing: bool = False, indexed_at=None, ) -> MailMailboxMessageListResponse: return MailMailboxMessageListResponse( profile_id=profile_id, folder=folder, host=host, port=port, security=security, total_count=total_count, offset=offset, limit=limit, cursor=cursor, next_cursor=next_cursor, cursor_stable=cursor_stable, full=full, from_cache=from_cache, refreshing=refreshing, indexed_at=indexed_at, messages=[_mailbox_summary_response(message) for message in messages], ) 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: return requested if "INBOX" in names: return "INBOX" if detected_sent_folder and detected_sent_folder in names: return detected_sent_folder return folders[0].name if folders else requested def _transport_context_for_principal( session: Session, *, principal: ApiPrincipal, campaign_id: str | None = None, ) -> MailHierarchyContext: if not campaign_id: return _mail_hierarchy_context(principal) _require_campaign_context_visibility( session, principal=principal, campaign_id=campaign_id, ) campaign = campaign_mail_owner_context( session, tenant_id=principal.tenant_id, campaign_id=campaign_id, ) return MailHierarchyContext( tenant_id=principal.tenant_id, user_id=campaign.owner_user_id, group_ids=( frozenset({campaign.owner_group_id}) if campaign.owner_group_id else frozenset() ), target_scope_type="campaign", target_scope_id=campaign.id, ) def _imap_config_for_principal( session: Session, *, principal: ApiPrincipal, profile_id: str, server_id: str | None = None, credential_id: str | None = None, campaign_id: str | None = None, ): profile = _get_profile_for_principal( session, principal=principal, profile_id=profile_id, require_active=True, ) try: return resolve_mail_transport( session, profile=profile, protocol="imap", context=_transport_context_for_principal( session, principal=principal, campaign_id=campaign_id, ), server_id=server_id, credential_id=credential_id, ).config except MailServerHierarchyError as exc: raise MailProfileError(str(exc)) from exc def _parent_mail_profile_policy_payload(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None): if scope_type == "system": return None return parent_mail_profile_policy(session, tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id).as_dict() def _full_mail_settings_delta_response( session: Session, *, principal: ApiPrincipal, scope_type: str, scope_id: str | None, include_inactive: bool, campaign_id: str | None, ) -> MailSettingsDeltaResponse: profiles = list_mail_server_profiles( session, 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, session=session, principal=principal, target_scope_type=scope_type, target_scope_id=scope_id, include_inactive=include_inactive, administrative=True, ) for profile in profiles ], 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=principal.tenant_id), has_more=False, full=True, ) @router.get("/address-lookup", response_model=MailAddressLookupResponse) def lookup_mail_addresses( query: str = Query(min_length=1), limit: int = Query(default=25, ge=1, le=100), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(get_api_principal), ) -> MailAddressLookupResponse: _require_scope(principal, "mail:profile:use") capability = _registry_capability(CAPABILITY_ADDRESSES_LOOKUP) if capability is None or not hasattr(capability, "lookup"): return MailAddressLookupResponse(available=False, candidates=[]) candidates = getattr(capability, "lookup")(session, principal, query=query, limit=limit) return MailAddressLookupResponse( available=True, candidates=[MailAddressLookupCandidate.model_validate(_capability_payload(candidate)) for candidate in candidates], ) @router.get("/settings/delta", response_model=MailSettingsDeltaResponse) def mail_settings_delta( scope_type: str = Query(default="tenant"), scope_id: str | None = Query(default=None), include_inactive: bool = False, campaign_id: str | None = Query(default=None), since: str | None = None, limit: int = Query(default=100, ge=1, le=500), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): 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, principal=principal, scope_type=clean_scope_type, scope_id=scope_id, include_inactive=include_inactive, campaign_id=campaign_id, ) entries, has_more = _mail_settings_entries(session, tenant_id=principal.tenant_id, since=since, limit=limit) if entries is None: return _full_mail_settings_delta_response( session, 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 visible_profile_entries } visible_profiles = { profile.id: profile for profile in list_mail_server_profiles( session, 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 } policy_changed = any(entry.collection == MAIL_POLICIES_COLLECTION for entry in entries) changed_sections = [] if changed_profile_ids: changed_sections.append("profiles") if policy_changed: changed_sections.append("policy") deleted = [ _mail_deleted_item(entry) for entry in visible_profile_entries if entry.resource_id not in visible_profiles ] return MailSettingsDeltaResponse( profiles=[ _profile_response( profile, session=session, principal=principal, target_scope_type=clean_scope_type, target_scope_id=scope_id, include_inactive=include_inactive, administrative=True, ) for profile in visible_profiles.values() ], policy=_policy_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope_type, scope_id=scope_id, campaign_id=campaign_id) if policy_changed else None, changed_sections=changed_sections, deleted=deleted, watermark=_mail_settings_response_watermark(session, tenant_id=principal.tenant_id, entries=entries, has_more=has_more), has_more=has_more, full=False, ) @router.get("/profiles", response_model=MailServerProfileListResponse) def list_profiles( include_inactive: bool = False, campaign_id: str | None = Query(default=None), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_any_scope(principal, "mail:profile:read", "system:settings:read") try: profiles = list_mail_server_profiles( session, 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, session=session, principal=principal, target_scope_type="campaign" if campaign_id else None, target_scope_id=campaign_id, include_inactive=include_inactive, administrative=include_inactive, ) for profile in profiles ] ) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @router.post( "/profiles/{profile_id}/servers", response_model=MailServerEndpointResponse, status_code=status.HTTP_201_CREATED, ) def create_profile_server( profile_id: str, payload: MailServerEndpointCreateRequest, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): try: profile = _profile_for_mutation( session, principal=principal, profile_id=profile_id, ) server = create_mail_server_endpoint( session, profile=profile, protocol=payload.protocol, name=payload.name, config=payload.config, user_id=principal.user.id, inherit_to_lower_scopes=payload.inherit_to_lower_scopes, is_default=payload.is_default, is_active=payload.is_active, ) _record_profile_child_change( session, principal=principal, profile=profile, operation="created", resource_type=MAIL_SERVER_RESOURCE, resource_id=server.id, ) session.commit() session.refresh(server) return _server_response( session, principal=principal, profile=profile, server=server, ) except (MailProfileError, MailServerHierarchyError) as exc: session.rollback() raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc), ) from exc @router.patch( "/profiles/{profile_id}/servers/{server_id}", response_model=MailServerEndpointResponse, ) def update_profile_server( profile_id: str, server_id: str, payload: MailServerEndpointUpdateRequest, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): try: profile, server = _server_for_mutation( session, principal=principal, profile_id=profile_id, server_id=server_id, ) update_mail_server_endpoint( session, server=server, user_id=principal.user.id, name=payload.name, config=payload.config, inherit_to_lower_scopes=payload.inherit_to_lower_scopes, is_default=payload.is_default, is_active=payload.is_active, ) _record_profile_child_change( session, principal=principal, profile=profile, operation="updated", resource_type=MAIL_SERVER_RESOURCE, resource_id=server.id, ) session.commit() session.refresh(server) return _server_response( session, principal=principal, profile=profile, server=server, ) except MailServerHierarchyError as exc: session.rollback() raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc), ) from exc @router.delete( "/profiles/{profile_id}/servers/{server_id}", response_model=MailServerEndpointResponse, ) def deactivate_profile_server( profile_id: str, server_id: str, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): try: profile, server = _server_for_mutation( session, principal=principal, profile_id=profile_id, server_id=server_id, ) update_mail_server_endpoint( session, server=server, user_id=principal.user.id, is_active=False, ) _record_profile_child_change( session, principal=principal, profile=profile, operation="deactivated", resource_type=MAIL_SERVER_RESOURCE, resource_id=server.id, ) session.commit() session.refresh(server) return _server_response( session, principal=principal, profile=profile, server=server, ) except MailServerHierarchyError as exc: session.rollback() raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc), ) from exc def _credential_hierarchy_context(principal: ApiPrincipal, profile) -> MailHierarchyContext: can_manage_all = has_scope(principal, "mail:secret:manage") or ( profile.scope_type == "system" and has_scope(principal, "system:settings:write") ) return hierarchy_context_for_profile( profile, user_id=principal.user.id, group_ids=principal.group_ids, administrative=can_manage_all, ) @router.get( "/profiles/{profile_id}/servers/{server_id}/available-credentials", response_model=MailCredentialListResponse, ) def list_profile_server_available_credentials( profile_id: str, server_id: str, include_inactive: bool = False, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): profile = _get_profile_for_principal( session, principal=principal, profile_id=profile_id, administrative_visibility=True, ) _require_profile_credentials_scope( principal, profile.scope_type or "tenant", profile.scope_id, ) context = _credential_hierarchy_context(principal, profile) try: server = get_mail_server_endpoint( session, profile=profile, server_id=server_id, context=context, require_active=False, ) credentials = list_available_mail_credentials( session, context=context, server_id=server.id, include_inactive=include_inactive, ) return MailCredentialListResponse( credentials=[ MailCredentialEnvelopeResponse.model_validate( mail_credential_payload(credential) ) for credential in credentials ] ) except MailServerHierarchyError as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc), ) from exc @router.post( "/profiles/{profile_id}/servers/{server_id}/credentials", response_model=MailCredentialEnvelopeResponse, status_code=status.HTTP_201_CREATED, ) def create_profile_server_credential( profile_id: str, server_id: str, payload: MailCredentialCreateRequest, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): try: profile, server = _server_for_mutation( session, principal=principal, profile_id=profile_id, server_id=server_id, ) _require_profile_credentials_scope( principal, profile.scope_type or "tenant", profile.scope_id, ) credential = create_and_bind_mail_credential( session, profile=profile, server=server, name=payload.name, description=payload.description, credential_kind=payload.credential_kind, username=payload.username, password=payload.password, public_data=payload.public_data, secret_data=payload.secret_data, user_id=principal.user.id, inherit_to_lower_scopes=payload.inherit_to_lower_scopes, allowed_modules=payload.allowed_modules, allowed_server_refs=payload.allowed_server_refs, is_default=payload.is_default, ) binding, _ = get_bound_mail_credential( session, server=server, credential_id=credential.id, context=_credential_hierarchy_context(principal, profile), require_active=False, ) _record_profile_child_change( session, principal=principal, profile=profile, operation="created", resource_type=MAIL_CREDENTIAL_RESOURCE, resource_id=credential.id, ) session.commit() session.refresh(credential) return MailCredentialEnvelopeResponse.model_validate( mail_credential_payload( credential, server_id=server.id, binding_id=binding.id, is_default=binding.is_default, ) ) except (MailProfileError, MailServerHierarchyError) as exc: session.rollback() raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc), ) from exc @router.post( "/profiles/{profile_id}/campaign-credentials", response_model=MailCredentialEnvelopeResponse, status_code=status.HTTP_201_CREATED, ) def create_campaign_profile_credential( profile_id: str, payload: MailCampaignCredentialCreateRequest, campaign_id: str = Query(...), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_any_scope( principal, "campaigns:campaign:update", "mail:secret:manage", ) _require_campaign_context_visibility( session, principal=principal, campaign_id=campaign_id, ) try: profile = next( ( item for item in list_mail_server_profiles( session, tenant_id=principal.tenant_id, campaign_id=campaign_id, **_profile_actor_kwargs( principal, administrative_visibility=False, ), ) if item.id == profile_id ), None, ) if profile is None: raise MailServerHierarchyError("Mail-server profile not found") context = hierarchy_context_for_profile( profile, user_id=principal.user.id, group_ids=principal.group_ids, target_scope_type="campaign", target_scope_id=campaign_id, ) server_ids = list(dict.fromkeys(item.strip() for item in payload.server_ids if item.strip())) servers = [ get_mail_server_endpoint( session, profile=profile, server_id=server_id, context=context, require_active=True, for_update=True, ) for server_id in server_ids ] if not servers: raise MailServerHierarchyError("Select at least one mail server") allowed_server_refs = [ ref for ref in (mail_server_ref(server.id) for server in servers) if ref is not None ] credential = create_and_bind_mail_credential( session, profile=profile, server=servers[0], name=payload.name, username=payload.username, password=payload.password.get_secret_value(), user_id=principal.user.id, inherit_to_lower_scopes=False, allowed_modules=("mail",), allowed_server_refs=allowed_server_refs, credential_scope_type="campaign", credential_scope_id=campaign_id, ) for server in servers[1:]: bind_mail_credential( session, server=server, credential=credential, user_id=principal.user.id, ) first_binding, _ = get_bound_mail_credential( session, server=servers[0], credential_id=credential.id, context=context, ) _record_profile_child_change( session, principal=principal, profile=profile, operation="created", resource_type=MAIL_CREDENTIAL_RESOURCE, resource_id=credential.id, ) session.commit() session.refresh(credential) return MailCredentialEnvelopeResponse.model_validate( mail_credential_payload( credential, server_id=servers[0].id, binding_id=first_binding.id, is_default=False, ) ) except (MailProfileError, MailServerHierarchyError) as exc: session.rollback() raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc), ) from exc @router.post( "/profiles/{profile_id}/servers/{server_id}/credential-bindings", response_model=MailCredentialEnvelopeResponse, status_code=status.HTTP_201_CREATED, ) def bind_profile_server_credential( profile_id: str, server_id: str, payload: MailCredentialBindRequest, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): try: profile, server = _server_for_mutation( session, principal=principal, profile_id=profile_id, server_id=server_id, ) _require_profile_credentials_scope( principal, profile.scope_type or "tenant", profile.scope_id, ) context = _credential_hierarchy_context(principal, profile) credential = get_available_mail_credential( session, credential_id=payload.credential_id, context=context, server_id=server.id, ) binding = bind_mail_credential( session, server=server, credential=credential, user_id=principal.user.id, is_default=payload.is_default, ) _record_profile_child_change( session, principal=principal, profile=profile, operation="bound", resource_type=MAIL_CREDENTIAL_RESOURCE, resource_id=credential.id, ) session.commit() session.refresh(credential) return MailCredentialEnvelopeResponse.model_validate( mail_credential_payload( credential, server_id=server.id, binding_id=binding.id, is_default=binding.is_default, ) ) except (MailProfileError, MailServerHierarchyError) as exc: session.rollback() raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc), ) from exc @router.patch( "/profiles/{profile_id}/servers/{server_id}/credentials/{credential_id}", response_model=MailCredentialEnvelopeResponse, ) def update_profile_server_credential( profile_id: str, server_id: str, credential_id: str, payload: MailCredentialUpdateRequest, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): try: profile, server = _server_for_mutation( session, principal=principal, profile_id=profile_id, server_id=server_id, ) _require_profile_credentials_scope( principal, profile.scope_type or "tenant", profile.scope_id, ) context = _credential_hierarchy_context(principal, profile) binding, credential = get_bound_mail_credential( session, server=server, credential_id=credential_id, context=context, require_active=False, for_update=True, ) update_bound_mail_credential( session, server=server, credential=credential, context=context, user_id=principal.user.id, name=payload.name, description=payload.description, description_supplied="description" in payload.model_fields_set, username=payload.username, username_supplied="username" in payload.model_fields_set, password=payload.password, password_supplied="password" in payload.model_fields_set, public_data=payload.public_data, secret_data=payload.secret_data, allowed_modules=payload.allowed_modules, allowed_server_refs=payload.allowed_server_refs, inherit_to_lower_scopes=payload.inherit_to_lower_scopes, is_active=payload.is_active, is_default=payload.is_default, ) binding, credential = get_bound_mail_credential( session, server=server, credential_id=credential_id, context=context, require_active=False, ) _record_profile_child_change( session, principal=principal, profile=profile, operation="updated", resource_type=MAIL_CREDENTIAL_RESOURCE, resource_id=credential.id, ) session.commit() session.refresh(credential) return MailCredentialEnvelopeResponse.model_validate( mail_credential_payload( credential, server_id=server.id, binding_id=binding.id, is_default=binding.is_default, ) ) except (MailProfileError, MailServerHierarchyError) as exc: session.rollback() raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc), ) from exc @router.delete( "/profiles/{profile_id}/servers/{server_id}/credentials/{credential_id}", status_code=status.HTTP_204_NO_CONTENT, ) def unlink_profile_server_credential( profile_id: str, server_id: str, credential_id: str, retire_if_unused: bool = Query(default=False), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): try: profile, server = _server_for_mutation( session, principal=principal, profile_id=profile_id, server_id=server_id, ) _require_profile_credentials_scope( principal, profile.scope_type or "tenant", profile.scope_id, ) context = _credential_hierarchy_context(principal, profile) _, credential = get_bound_mail_credential( session, server=server, credential_id=credential_id, context=context, require_active=False, ) unlink_mail_credential( session, server=server, credential=credential, context=context, user_id=principal.user.id, retire_if_unused=retire_if_unused, ) _record_profile_child_change( session, principal=principal, profile=profile, operation="unlinked", resource_type=MAIL_CREDENTIAL_RESOURCE, resource_id=credential.id, ) session.commit() return None except (MailProfileError, MailServerHierarchyError) as exc: session.rollback() raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc), ) from exc @router.post("/profiles", response_model=MailServerProfileResponse, status_code=status.HTTP_201_CREATED) def create_profile( payload: MailServerProfileCreateRequest, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): authorization_scope_id = ( payload.scope_id or principal.user.id if payload.scope_type == "user" else payload.scope_id ) _require_profile_write_scope(principal, payload.scope_type, authorization_scope_id) _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, authorization_scope_id, ) try: smtp_config = payload.smtp_config() imap_config = payload.imap_config() profile = create_mail_server_profile( session, tenant_id=principal.tenant_id, user_id=principal.user.id, name=payload.name, slug=payload.slug, description=payload.description, smtp=smtp_config, imap=imap_config, is_active=payload.is_active, inherit_to_lower_scopes=payload.inherit_to_lower_scopes, scope_type=payload.scope_type, scope_id=payload.scope_id, ) initialize_profile_hierarchy( session, profile=profile, smtp=smtp_config, imap=imap_config, user_id=principal.user.id, ) _record_mail_change( session, collection=MAIL_PROFILES_COLLECTION, resource_type=MAIL_PROFILE_RESOURCE, resource_id=profile.id, operation="created", principal=principal, tenant_id=profile.tenant_id, payload={"scope_type": profile.scope_type, "scope_id": profile.scope_id}, ) session.commit() session.refresh(profile) return _profile_response( profile, session=session, principal=principal, target_scope_type=profile.scope_type, target_scope_id=profile.scope_id, include_inactive=True, administrative=True, ) except (MailProfileError, MailServerHierarchyError) as exc: session.rollback() raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc @router.get("/profiles/{profile_id}", response_model=MailServerProfileResponse) def get_profile( profile_id: str, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_any_scope(principal, "mail:profile:read", "system:settings:read") try: profile = _get_profile_for_principal( session, principal=principal, profile_id=profile_id, administrative_visibility=True, ) return _profile_response( profile, session=session, principal=principal, target_scope_type=profile.scope_type, target_scope_id=profile.scope_id, include_inactive=True, administrative=True, ) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @router.patch("/profiles/{profile_id}", response_model=MailServerProfileResponse) def update_profile( profile_id: str, payload: MailServerProfileUpdateRequest, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): try: profile = _profile_for_mutation( session, principal=principal, profile_id=profile_id, ) _require_campaign_profile_mutation_access( session, principal=principal, scope_type=profile.scope_type or "tenant", scope_id=profile.scope_id, ) smtp_config = payload.smtp_config() if payload.smtp is None and "smtp" in payload.credentials.model_fields_set: smtp_data = dict(profile.smtp_config or {}) smtp_data.update(payload.credentials.smtp.model_dump(mode="json", exclude_unset=True)) smtp_config = SmtpConfig.model_validate(smtp_data) imap_config = payload.imap_config() if payload.imap is None and "imap" in payload.credentials.model_fields_set and profile.imap_config: imap_data = dict(profile.imap_config or {}) imap_data.update(payload.credentials.imap.model_dump(mode="json", exclude_unset=True)) imap_config = ImapConfig.model_validate(imap_data) if _profile_update_requires_credentials_scope( profile, payload=payload, smtp=smtp_config, imap=imap_config, ): _require_profile_credentials_scope( principal, profile.scope_type or "tenant", profile.scope_id, ) update_mail_server_profile( session, profile, tenant_id=principal.tenant_id, user_id=principal.user.id, api_key_id=principal.api_key_id, name=payload.name, slug=payload.slug, description=payload.description if "description" in payload.model_fields_set else None, is_active=payload.is_active, inherit_to_lower_scopes=payload.inherit_to_lower_scopes, smtp=smtp_config, imap=imap_config, clear_imap=payload.clear_imap, ) smtp_server = None if smtp_config is not None: smtp_server = sync_default_profile_server( session, profile=profile, protocol="smtp", config=smtp_config.model_dump(mode="json"), user_id=principal.user.id, ) imap_server = None if imap_config is not None or payload.clear_imap: imap_server = sync_default_profile_server( session, profile=profile, protocol="imap", config=( imap_config.model_dump(mode="json") if imap_config is not None and not payload.clear_imap else None ), user_id=principal.user.id, ) if smtp_server is not None and "smtp" in payload.credentials.model_fields_set: smtp_credentials = payload.credentials.smtp sync_default_profile_credential( session, profile=profile, server=smtp_server, username=smtp_credentials.username, username_supplied="username" in smtp_credentials.model_fields_set, password=smtp_credentials.password, password_supplied="password" in smtp_credentials.model_fields_set, user_id=principal.user.id, ) if imap_server is not None and "imap" in payload.credentials.model_fields_set: imap_credentials = payload.credentials.imap sync_default_profile_credential( session, profile=profile, server=imap_server, username=imap_credentials.username, username_supplied="username" in imap_credentials.model_fields_set, password=imap_credentials.password, password_supplied="password" in imap_credentials.model_fields_set, user_id=principal.user.id, ) _record_mail_change( session, collection=MAIL_PROFILES_COLLECTION, resource_type=MAIL_PROFILE_RESOURCE, resource_id=profile.id, operation="updated", principal=principal, tenant_id=profile.tenant_id, payload={"scope_type": profile.scope_type, "scope_id": profile.scope_id}, ) session.commit() session.refresh(profile) return _profile_response( profile, session=session, principal=principal, target_scope_type=profile.scope_type, target_scope_id=profile.scope_id, include_inactive=True, administrative=True, ) except (MailProfileError, MailServerHierarchyError) as exc: session.rollback() raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc except Exception: session.rollback() raise @router.delete("/profiles/{profile_id}", response_model=MailServerProfileResponse) def deactivate_profile( profile_id: str, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): try: profile = _profile_for_mutation( session, principal=principal, profile_id=profile_id, ) _require_campaign_profile_mutation_access( session, principal=principal, scope_type=profile.scope_type or "tenant", scope_id=profile.scope_id, ) if _profile_has_stored_credentials(profile): _require_profile_credentials_scope( principal, profile.scope_type or "tenant", profile.scope_id, ) was_active = bool(profile.is_active) profile.is_active = False deleted_protocols = delete_mail_profile_credentials( session, profile=profile, deletion_reason="profile_deactivated", 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( session, collection=MAIL_PROFILES_COLLECTION, resource_type=MAIL_PROFILE_RESOURCE, resource_id=profile.id, operation="deleted", principal=principal, tenant_id=profile.tenant_id, payload={ "scope_type": profile.scope_type, "scope_id": profile.scope_id, "credentials_deleted": bool(deleted_protocols), "deleted_credential_protocols": list(deleted_protocols), }, ) session.commit() session.refresh(profile) return _profile_response( profile, session=session, principal=principal, target_scope_type=profile.scope_type, target_scope_id=profile.scope_id, include_inactive=True, administrative=True, ) except MailProfileError as exc: session.rollback() raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except Exception: session.rollback() raise @router.get("/policies/{scope_type}", response_model=MailProfilePolicyResponse) def read_mail_profile_policy( scope_type: str, scope_id: str | None = Query(default=None), campaign_id: str | None = Query(default=None), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): 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: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @router.put("/policies/{scope_type}", response_model=MailProfilePolicyResponse) def write_mail_profile_policy( scope_type: str, payload: MailProfilePolicyUpdateRequest, scope_id: str | None = Query(default=None), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): 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, tenant_id=principal.tenant_id, scope_type=scope_type, scope_id=scope_id, policy=payload.policy.model_dump(mode="json"), ) _record_mail_change( session, collection=MAIL_POLICIES_COLLECTION, resource_type=MAIL_POLICY_RESOURCE, resource_id=f"{scope_type}:{scope_id or ''}", operation="updated", principal=principal, tenant_id=None if scope_type == "system" else principal.tenant_id, payload={"scope_type": scope_type, "scope_id": scope_id}, ) session.commit() return _policy_response(session, tenant_id=principal.tenant_id, scope_type=scope_type, scope_id=scope_id) except MailProfileError as exc: session.rollback() raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc @router.post("/profiles/{profile_id}/test-smtp", response_model=MailConnectionTestResponse) def test_profile_smtp( profile_id: str, server_id: str | None = Query(default=None), credential_id: str | None = Query(default=None), campaign_id: str | None = Query(default=None), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_scope(principal, "mail:profile:test") _require_scope(principal, "mail:profile:use") try: profile = _get_profile_for_principal( session, principal=principal, profile_id=profile_id, require_active=True, ) if callable(getattr(session, "execute", None)): smtp = resolve_mail_transport( session, profile=profile, protocol="smtp", context=_transport_context_for_principal( session, principal=principal, campaign_id=campaign_id, ), server_id=server_id, credential_id=credential_id, ).config else: 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}) except (MailProfileError, MailServerHierarchyError) as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except Exception as exc: return MailConnectionTestResponse(ok=False, protocol="smtp", message=_safe_error_message(exc), details={"error_type": exc.__class__.__name__}) @router.post("/profiles/{profile_id}/test-imap", response_model=MailConnectionTestResponse) def test_profile_imap( profile_id: str, server_id: str | None = Query(default=None), credential_id: str | None = Query(default=None), campaign_id: str | None = Query(default=None), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_scope(principal, "mail:profile:test") _require_scope(principal, "mail:profile:use") try: profile = _get_profile_for_principal( session, principal=principal, profile_id=profile_id, require_active=True, ) if callable(getattr(session, "execute", None)): imap = resolve_mail_transport( session, profile=profile, protocol="imap", context=_transport_context_for_principal( session, principal=principal, campaign_id=campaign_id, ), server_id=server_id, credential_id=credential_id, ).config else: imap = imap_config_from_profile(profile) if imap is None: raise MailProfileError("Mail-server profile has no IMAP configuration") result = test_imap_login(imap_config=imap) return MailConnectionTestResponse(ok=True, protocol="imap", host=result.host, port=result.port, security=result.security, message="IMAP connection successful.", details={"authenticated": result.authenticated}) except (MailProfileError, MailServerHierarchyError) as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except Exception as exc: return MailConnectionTestResponse(ok=False, protocol="imap", message=_safe_error_message(exc), details={"error_type": exc.__class__.__name__}) @router.post("/profiles/{profile_id}/list-imap-folders", response_model=MailImapFolderListResponse) def list_profile_imap_folders( profile_id: str, server_id: str | None = Query(default=None), credential_id: str | None = Query(default=None), campaign_id: str | None = Query(default=None), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_scope(principal, "mail:profile:test") _require_scope(principal, "mail:profile:use") try: profile = _get_profile_for_principal( session, principal=principal, profile_id=profile_id, require_active=True, ) if callable(getattr(session, "execute", None)): imap = resolve_mail_transport( session, profile=profile, protocol="imap", context=_transport_context_for_principal( session, principal=principal, campaign_id=campaign_id, ), server_id=server_id, credential_id=credential_id, ).config else: imap = imap_config_from_profile(profile) if imap is None: raise MailProfileError("Mail-server profile has no IMAP configuration") result = list_imap_folders(imap_config=imap) return _mailbox_folder_response(result) except (MailProfileError, MailServerHierarchyError) as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except Exception as exc: return MailImapFolderListResponse(ok=False, message=_safe_error_message(exc), folders=[], detected_sent_folder=None, details={"error_type": exc.__class__.__name__}) @router.get("/profiles/{profile_id}/mailbox/folders", response_model=MailImapFolderListResponse) def list_profile_mailbox_folders( profile_id: str, include_status: bool = Query(default=False), refresh: bool = False, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_mailbox_read_scope(principal) try: 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 and not cached.stale: result = SimpleNamespace( host=imap.host, port=imap.port, security=imap.security.value, folders=cached.folders, detected_sent_folder=None, ) 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() return _mailbox_folder_response(result) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except ImapConfigurationError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc except ImapAppendError as exc: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc @router.get("/profiles/{profile_id}/mailbox/bootstrap", response_model=MailMailboxBootstrapResponse) def bootstrap_profile_mailbox( profile_id: str, 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), refresh: bool = False, include_status: bool = Query(default=False), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ) -> MailMailboxBootstrapResponse: _require_mailbox_read_scope(principal) try: 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 and not cached_folders.stale: selected_folder = _cached_folder_selection(cached_folders.folders, folder) cached_messages = cached_mailbox_message_page( session, tenant_id=principal.tenant_id, profile_id=profile_id, folder=selected_folder, limit=limit, offset=offset, ) if cached_messages is not None and not cached_messages.stale: folder_result = SimpleNamespace( host=imap.host, port=imap.port, security=imap.security.value, folders=cached_folders.folders, detected_sent_folder=None, ) next_cursor, cursor_stable = _next_mailbox_cursor( tenant_id=principal.tenant_id, profile_id=profile_id, folder=cached_messages.folder, limit=cached_messages.limit, offset=cached_messages.offset, total_count=cached_messages.total_count, uidvalidity=cached_messages.uidvalidity, messages=cached_messages.messages, ) return MailMailboxBootstrapResponse( profile_id=profile_id, folder=cached_messages.folder, folders=_mailbox_folder_response( folder_result, from_cache=True, refreshing=False, indexed_at=cached_folders.indexed_at, ), messages=_mailbox_messages_response( profile_id=profile_id, folder=cached_messages.folder, host=imap.host, port=imap.port, security=imap.security.value, total_count=cached_messages.total_count, offset=cached_messages.offset, limit=cached_messages.limit, cursor=None, next_cursor=next_cursor, cursor_stable=cursor_stable, full=not cursor_stable, messages=cached_messages.messages, from_cache=True, refreshing=False, indexed_at=cached_messages.indexed_at, ), ) result = load_imap_mailbox_bootstrap( imap_config=imap, folder=folder, limit=limit, offset=offset, include_folder_status=include_status, ) cache_mailbox_folders(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result.folders) cache_mailbox_messages(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result.messages) session.commit() next_cursor, cursor_stable = _next_mailbox_cursor( tenant_id=principal.tenant_id, profile_id=profile_id, folder=result.messages.folder, limit=result.messages.limit, offset=result.messages.offset, total_count=result.messages.total_count, uidvalidity=result.messages.uidvalidity, messages=result.messages.messages, ) return MailMailboxBootstrapResponse( profile_id=profile_id, folder=result.messages.folder, folders=_mailbox_folder_response(result.folders), messages=_mailbox_messages_response( profile_id=profile_id, folder=result.messages.folder, host=result.messages.host, port=result.messages.port, security=result.messages.security, total_count=result.messages.total_count, offset=result.messages.offset, limit=result.messages.limit, cursor=None, next_cursor=next_cursor, cursor_stable=cursor_stable, full=not cursor_stable, messages=result.messages.messages, ), ) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except ImapConfigurationError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc except ImapAppendError as exc: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc @router.get("/profiles/{profile_id}/mailbox/messages", response_model=MailMailboxMessageListResponse) def list_profile_mailbox_messages( profile_id: str, 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), cursor: str | None = None, refresh: bool = False, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_mailbox_read_scope(principal) try: 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) if not refresh: cached = cached_mailbox_message_page( session, tenant_id=principal.tenant_id, profile_id=profile_id, folder=folder, limit=effective_limit, offset=effective_offset, ) 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 and not cached.stale: next_cursor, cursor_stable = _next_mailbox_cursor( tenant_id=principal.tenant_id, profile_id=profile_id, folder=cached.folder, limit=cached.limit, offset=cached.offset, total_count=cached.total_count, uidvalidity=cached.uidvalidity, messages=cached.messages, ) return _mailbox_messages_response( profile_id=profile_id, folder=cached.folder, host=imap.host, port=imap.port, security=imap.security.value, total_count=cached.total_count, offset=cached.offset, limit=cached.limit, cursor=cursor, next_cursor=next_cursor, cursor_stable=cursor_stable, full=not cursor_stable, messages=cached.messages, from_cache=True, refreshing=False, indexed_at=cached.indexed_at, ) result = list_imap_messages( imap_config=imap, folder=folder, limit=effective_limit, offset=effective_offset, after_uid=after_uid, expected_uidvalidity=expected_uidvalidity, ) cache_mailbox_messages(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result) session.commit() full = bool(cursor_values is not None and result.cursor_reset) next_cursor, cursor_stable = _next_mailbox_cursor( tenant_id=principal.tenant_id, profile_id=profile_id, folder=result.folder, limit=result.limit, offset=result.offset, total_count=result.total_count, uidvalidity=result.uidvalidity, messages=result.messages, ) return _mailbox_messages_response( profile_id=profile_id, folder=result.folder, host=result.host, port=result.port, security=result.security, total_count=result.total_count, offset=result.offset, limit=result.limit, cursor=cursor, next_cursor=next_cursor, cursor_stable=cursor_stable, full=full or not cursor_stable, messages=result.messages, ) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except ImapConfigurationError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc except ImapAppendError as exc: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc @router.get("/profiles/{profile_id}/mailbox/messages/{message_uid}", response_model=MailMailboxMessageResponse) def get_profile_mailbox_message( profile_id: str, message_uid: str, folder: str = Query(default="INBOX", min_length=1, max_length=255), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_mailbox_read_scope(principal) try: 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, folder=result.folder, host=result.host, port=result.port, security=result.security, message=_mailbox_detail_response(result.message), ) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except ImapConfigurationError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc except ImapAppendError as exc: status_code = status.HTTP_404_NOT_FOUND if "not found" in str(exc).casefold() else status.HTTP_502_BAD_GATEWAY raise HTTPException(status_code=status_code, detail=str(exc)) from exc def _safe_error_message(exc: Exception) -> str: text = str(exc).strip() return text or exc.__class__.__name__ @router.post("/test-smtp", response_model=MailConnectionTestResponse) def test_smtp_settings( payload: MailSmtpTestRequest, principal: ApiPrincipal = Depends(get_api_principal), ): """Test SMTP connectivity/login without sending any message.""" if not has_scope(principal, "mail:profile:test"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: mail_servers:test") if not has_scope(principal, "mail:secret:manage"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Testing raw credentials requires mail_servers:manage_credentials") try: result = test_smtp_login(smtp_config=payload) return MailConnectionTestResponse( ok=True, protocol="smtp", host=result.host, port=result.port, security=result.security, message="SMTP connection successful.", details={"authenticated": result.authenticated}, ) except Exception as exc: return MailConnectionTestResponse( ok=False, protocol="smtp", host=payload.host, port=payload.port, security=payload.security.value, message=_safe_error_message(exc), details={"error_type": exc.__class__.__name__}, ) @router.post("/test-imap", response_model=MailConnectionTestResponse) def test_imap_settings( payload: MailImapTestRequest, principal: ApiPrincipal = Depends(get_api_principal), ): """Test IMAP connectivity/login without selecting or appending messages.""" if not has_scope(principal, "mail:profile:test"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: mail_servers:test") if not has_scope(principal, "mail:secret:manage"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Testing raw credentials requires mail_servers:manage_credentials") try: result = test_imap_login(imap_config=payload) return MailConnectionTestResponse( ok=True, protocol="imap", host=result.host, port=result.port, security=result.security, message="IMAP connection successful.", details={"authenticated": result.authenticated}, ) except Exception as exc: return MailConnectionTestResponse( ok=False, protocol="imap", host=payload.host, port=payload.port, security=payload.security.value, message=_safe_error_message(exc), details={"error_type": exc.__class__.__name__}, ) @router.post("/list-imap-folders", response_model=MailImapFolderListResponse) def list_imap_folder_settings( payload: MailImapTestRequest, principal: ApiPrincipal = Depends(get_api_principal), ): """List visible IMAP folders and return the best Sent-folder guess.""" if not has_scope(principal, "mail:profile:test"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: mail_servers:test") if not has_scope(principal, "mail:secret:manage"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Testing raw credentials requires mail_servers:manage_credentials") try: result = list_imap_folders(imap_config=payload) folders = [MailImapFolderResponse(name=item.name, flags=item.flags, message_count=item.message_count, unseen_count=item.unseen_count) for item in result.folders] return MailImapFolderListResponse( ok=True, host=result.host, port=result.port, security=result.security, message=f"Found {len(folders)} IMAP folder(s).", folders=folders, detected_sent_folder=result.detected_sent_folder, ) except Exception as exc: return MailImapFolderListResponse( ok=False, host=payload.host, port=payload.port, security=payload.security.value, message=_safe_error_message(exc), folders=[], detected_sent_folder=None, details={"error_type": exc.__class__.__name__}, )