From 72d6f8432c02acac31be4fefd0564f52f8466c90 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 4 Aug 2026 03:03:26 +0200 Subject: [PATCH] Index authorized mailbox cache changes in Search --- src/govoplan_mail/backend/mailbox_index.py | 89 ++++++- src/govoplan_mail/backend/manifest.py | 33 ++- src/govoplan_mail/backend/search_source.py | 292 +++++++++++++++++++++ tests/test_mailbox_index.py | 3 + tests/test_search_source.py | 144 ++++++++++ 5 files changed, 559 insertions(+), 2 deletions(-) create mode 100644 src/govoplan_mail/backend/search_source.py create mode 100644 tests/test_search_source.py diff --git a/src/govoplan_mail/backend/mailbox_index.py b/src/govoplan_mail/backend/mailbox_index.py index 58fcb72..123260b 100644 --- a/src/govoplan_mail/backend/mailbox_index.py +++ b/src/govoplan_mail/backend/mailbox_index.py @@ -7,11 +7,17 @@ from threading import Lock from sqlalchemy import func from sqlalchemy.orm import Session +from govoplan_core.core.change_sequence import record_change from govoplan_core.security.time import ensure_aware_utc, utc_now -from govoplan_mail.backend.db.models import MailMailboxFolderIndex, MailMailboxMessageIndex +from govoplan_mail.backend.db.models import ( + MailMailboxFolderIndex, + MailMailboxMessageIndex, + new_uuid, +) from govoplan_mail.backend.sending.imap import ImapFolderListResult, ImapMailboxInfo, ImapMailboxMessageListResult, ImapMailboxMessageSummary MAILBOX_INDEX_TTL_SECONDS = 30 +MAILBOX_MESSAGES_COLLECTION = "mail.mailbox_messages" _refresh_lock = Lock() _refreshing_keys: set[tuple[str, str, str]] = set() @@ -44,6 +50,13 @@ def clear_mailbox_index(session: Session, *, profile_id: str) -> tuple[int, int] transport change. Message rows are removed before their folder metadata. """ + messages = ( + session.query(MailMailboxMessageIndex) + .filter(MailMailboxMessageIndex.profile_id == profile_id) + .all() + ) + for message in messages: + _record_message_change(session, message, operation="deleted") deleted_messages = ( session.query(MailMailboxMessageIndex) .filter(MailMailboxMessageIndex.profile_id == profile_id) @@ -89,6 +102,17 @@ def cache_mailbox_folders( expected_names = {folder.name for folder in result.folders} removed_names = set(existing) - expected_names if removed_names: + removed_messages = ( + session.query(MailMailboxMessageIndex) + .filter( + MailMailboxMessageIndex.tenant_id == tenant_id, + MailMailboxMessageIndex.profile_id == profile_id, + MailMailboxMessageIndex.folder.in_(removed_names), + ) + .all() + ) + for message in removed_messages: + _record_message_change(session, message, operation="deleted") ( session.query(MailMailboxMessageIndex) .filter( @@ -139,6 +163,17 @@ def cache_mailbox_messages( uids = [message.uid for message in result.messages] if result.total_count <= 0: + removed_messages = ( + session.query(MailMailboxMessageIndex) + .filter( + MailMailboxMessageIndex.tenant_id == tenant_id, + MailMailboxMessageIndex.profile_id == profile_id, + MailMailboxMessageIndex.folder == result.folder, + ) + .all() + ) + for message in removed_messages: + _record_message_change(session, message, operation="deleted") ( session.query(MailMailboxMessageIndex) .filter( @@ -161,6 +196,7 @@ def cache_mailbox_messages( if uids: stale_query = stale_query.filter(MailMailboxMessageIndex.uid.notin_(uids)) for row in stale_query.all(): + _record_message_change(session, row, operation="deleted") session.delete(row) existing = {} if uids: @@ -177,13 +213,16 @@ def cache_mailbox_messages( } for index, message in enumerate(result.messages): row = existing.get(message.uid) + created = row is None if row is None: row = MailMailboxMessageIndex( + id=new_uuid(), tenant_id=tenant_id, profile_id=profile_id, folder=result.folder, uid=message.uid, ) + previous = None if created else _message_state(row) row.uid_int = _uid_int(message.uid) row.sort_position = result.offset + index row.subject = message.subject @@ -198,6 +237,12 @@ def cache_mailbox_messages( row.attachment_count = message.attachment_count row.indexed_at = indexed_at session.add(row) + if created or previous != _message_state(row): + _record_message_change( + session, + row, + operation="created" if created else "updated", + ) def cached_mailbox_folders( @@ -308,6 +353,48 @@ def _message_from_index(row: MailMailboxMessageIndex) -> ImapMailboxMessageSumma ) +def _message_state(row: MailMailboxMessageIndex) -> tuple[object, ...]: + return ( + row.folder, + row.uid, + row.sort_position, + row.subject, + row.from_header, + row.to_header, + row.cc_header, + row.date, + row.message_id, + tuple(row.flags or ()), + row.size_bytes, + row.body_preview, + row.attachment_count, + ) + + +def _record_message_change( + session: Session, + row: MailMailboxMessageIndex, + *, + operation: str, +) -> None: + record_change( + session, + module_id="mail", + collection=MAILBOX_MESSAGES_COLLECTION, + resource_type="mailbox_message", + resource_id=row.id, + operation=operation, + tenant_id=row.tenant_id, + actor_type="system", + payload={ + "profile_id": row.profile_id, + "folder": row.folder, + "uid": row.uid, + "message_id": row.message_id, + }, + ) + + def _uid_int(uid: str) -> int: try: return int(str(uid)) diff --git a/src/govoplan_mail/backend/manifest.py b/src/govoplan_mail/backend/manifest.py index 5c1960a..2a3c0a5 100644 --- a/src/govoplan_mail/backend/manifest.py +++ b/src/govoplan_mail/backend/manifest.py @@ -35,6 +35,7 @@ from govoplan_core.core.provider_governance import ( ProviderObjectDeclaration, declared_module_architecture, ) +from govoplan_core.core.search import SearchSourceProviderRegistration from govoplan_core.core.views import ViewSurface from govoplan_core.db.base import Base from govoplan_mail.backend.documentation import ( @@ -48,6 +49,7 @@ from govoplan_mail.backend.provider_state import ( smtp_provider_states, ) from govoplan_mail.backend.db import models as mail_models # noqa: F401 - populate Mail ORM metadata +from govoplan_mail.backend.search_source import create_mail_search_source _mail_table_retirement_provider = drop_table_retirement_provider( @@ -299,7 +301,7 @@ manifest = ModuleManifest( name="Mail", version="0.1.10", required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR), - optional_dependencies=("campaigns", "addresses", "calendar"), + optional_dependencies=("campaigns", "addresses", "calendar", "search"), provides_interfaces=( ModuleInterfaceProvider(name="mail.campaign_delivery", version="0.2.0"), ModuleInterfaceProvider(name="mail.delivery_commands", version="0.1.0"), @@ -332,10 +334,22 @@ manifest = ModuleManifest( version_max_exclusive="0.2.0", optional=True, ), + ModuleInterfaceRequirement( + name="search.source", + version_min="1.0.0", + version_max_exclusive="2.0.0", + optional=True, + ), ), permissions=PERMISSIONS, route_factory=_mail_router, role_templates=ROLE_TEMPLATES, + search_sources=( + SearchSourceProviderRegistration( + id="mail.mailbox_messages", + factory=create_mail_search_source, + ), + ), nav_items=(NavItem(path="/mail", label="Mail", icon="mail", required_any=("mail:mailbox:read", "mail:bounce:read", "mail:bounce:manage"), order=50),), frontend=FrontendModule( module_id="mail", @@ -404,6 +418,23 @@ manifest = ModuleManifest( ).SqlMailBounceProcessingProvider(), }, documentation=( + DocumentationTopic( + id="mail.search.mailbox-messages", + title="Search authorized mailbox messages", + summary="Expose bounded cached mailbox message metadata and previews to permission-aware platform Search.", + body=( + "When Search is installed, Mail contributes its read-only mailbox cache, never credentials or " + "unbounded raw messages. Results require both mailbox-read and profile-use authority and recheck " + "the current profile scope, policy, activity, tenant, user, group, or Campaign visibility before " + "returning a result. Mailbox refreshes publish durable index changes; a rebuild remains available " + "for reconciliation." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("mail_user", "mail_admin", "administrator"), + related_modules=("search",), + order=38, + ), DocumentationTopic( id="mail.profiles-and-policy", title="Mail profiles and policy hierarchy", diff --git a/src/govoplan_mail/backend/search_source.py b/src/govoplan_mail/backend/search_source.py new file mode 100644 index 0000000..7f0bcbb --- /dev/null +++ b/src/govoplan_mail/backend/search_source.py @@ -0,0 +1,292 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from urllib.parse import quote + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.events import PlatformEvent +from govoplan_core.core.modules import ModuleContext +from govoplan_core.core.search import ( + SearchAuthorizationRequest, + SearchBackfillPage, + SearchBackfillRequest, + SearchDocument, + SearchIndexChange, + SearchResourceReference, + SearchResourceType, +) +from govoplan_mail.backend.db.models import ( + MailMailboxMessageIndex, + MailServerProfile, +) +from govoplan_mail.backend.mail_profiles import mail_profile_visible_to_actor + + +PROVIDER_ID = "mail.mailbox_messages" +RESOURCE_TYPE = "mailbox_message" +READ_SCOPE = "mail:mailbox:read" +USE_SCOPE = "mail:profile:use" + + +class MailSearchSource: + def resource_types(self) -> Sequence[SearchResourceType]: + return ( + SearchResourceType( + provider_id=PROVIDER_ID, + module_id="mail", + resource_type=RESOURCE_TYPE, + label="Mailbox messages", + requires_authorization_recheck=True, + ), + ) + + def backfill( + self, + session: object, + *, + request: SearchBackfillRequest, + ) -> SearchBackfillPage: + _assert_source(request.provider_id, request.resource_type) + db = _session(session) + statement = ( + select(MailMailboxMessageIndex, MailServerProfile) + .join( + MailServerProfile, + MailServerProfile.id == MailMailboxMessageIndex.profile_id, + ) + .where(MailMailboxMessageIndex.tenant_id == request.tenant_id) + ) + if request.cursor: + statement = statement.where( + MailMailboxMessageIndex.id > request.cursor + ) + rows = list( + db.execute( + statement.order_by(MailMailboxMessageIndex.id).limit( + request.limit + 1 + ) + ).all() + ) + has_more = len(rows) > request.limit + selected = rows[: request.limit] + high_watermark = db.scalar( + select(func.max(MailMailboxMessageIndex.indexed_at)).where( + MailMailboxMessageIndex.tenant_id == request.tenant_id + ) + ) + return SearchBackfillPage( + documents=tuple( + _document(message, profile=profile) + for message, profile in selected + ), + next_cursor=( + selected[-1][0].id if has_more and selected else None + ), + complete=not has_more, + high_watermark=( + high_watermark.isoformat() + if high_watermark is not None + else None + ), + ) + + def authorize( + self, + session: object, + principal: object, + *, + requests: Sequence[SearchAuthorizationRequest], + ) -> Mapping[str, bool]: + decisions = {item.reference.key: False for item in requests} + if not isinstance(principal, ApiPrincipal) or not ( + principal.has(READ_SCOPE) and principal.has(USE_SCOPE) + ): + return decisions + valid = tuple( + item + for item in requests + if item.reference.tenant_id == principal.tenant_id + and item.reference.module_id == "mail" + and item.reference.resource_type == RESOURCE_TYPE + ) + if not valid: + return decisions + db = _session(session) + ids = {item.reference.resource_id for item in valid} + messages = { + message.id: (message, profile) + for message, profile in db.execute( + select(MailMailboxMessageIndex, MailServerProfile) + .join( + MailServerProfile, + MailServerProfile.id + == MailMailboxMessageIndex.profile_id, + ) + .where( + MailMailboxMessageIndex.id.in_(ids), + MailMailboxMessageIndex.tenant_id == principal.tenant_id, + ) + ).all() + } + user_id = str( + getattr(principal.user, "id", "") or principal.membership_id or "" + ) + for item in valid: + match = messages.get(item.reference.resource_id) + if match is None: + continue + _message, profile = match + try: + allowed = mail_profile_visible_to_actor( + db, + profile=profile, + tenant_id=principal.tenant_id, + user_id=user_id, + group_ids=principal.group_ids, + tenant_admin=principal.has("tenant:*"), + require_active=True, + ) + except (RuntimeError, ValueError): + allowed = False + decisions[item.reference.key] = allowed + return decisions + + def index_changes_for_event( + self, + session: object, + *, + event: PlatformEvent, + delivery_key: str, + ) -> Sequence[SearchIndexChange]: + if ( + event.module_id != "mail" + or event.tenant is None + or event.resource is None + or event.resource.type != RESOURCE_TYPE + or event.resource.id is None + ): + return () + db = _session(session) + row = db.get(MailMailboxMessageIndex, event.resource.id) + profile = ( + db.get(MailServerProfile, row.profile_id) + if row is not None and row.tenant_id == event.tenant.id + else None + ) + deleted = ( + row is None + or row.tenant_id != event.tenant.id + or profile is None + ) + cursor = event.event_id + document = ( + None + if deleted + else _document(row, profile=profile, change_cursor=cursor) + ) + reference = SearchResourceReference( + tenant_id=event.tenant.id, + module_id="mail", + resource_type=RESOURCE_TYPE, + resource_id=event.resource.id, + ) + return ( + SearchIndexChange( + change_id=f"{delivery_key}:{PROVIDER_ID}", + provider_id=PROVIDER_ID, + kind="delete" if deleted else "upsert", + reference=reference, + source_revision=( + document.source_revision if document is not None else cursor + ), + cursor=cursor, + document=document, + occurred_at=event.occurred_at, + ), + ) + + +def create_mail_search_source(_context: ModuleContext) -> MailSearchSource: + return MailSearchSource() + + +def _document( + message: MailMailboxMessageIndex, + *, + profile: MailServerProfile, + change_cursor: str | None = None, +) -> SearchDocument: + tokens = [f"scope:{READ_SCOPE}", f"scope:{USE_SCOPE}"] + scope_type = str(profile.scope_type or "tenant") + if scope_type == "user" and profile.scope_id: + tokens.append(f"membership:{profile.scope_id}") + elif scope_type == "group" and profile.scope_id: + tokens.append(f"group:{profile.scope_id}") + title = (message.subject or "(No subject)")[:500] + summary = " | ".join( + value for value in (message.from_header, message.date) if value + )[:4000] + body = " ".join( + value + for value in ( + message.from_header, + message.to_header, + message.cc_header, + message.body_preview, + ) + if value + )[:200_000] + updated_at = message.updated_at or message.indexed_at + return SearchDocument( + tenant_id=message.tenant_id, + module_id="mail", + provider_id=PROVIDER_ID, + resource_type=RESOURCE_TYPE, + resource_id=message.id, + title=title, + url=( + "/mail?" + f"profileId={quote(message.profile_id, safe='')}" + f"&folder={quote(message.folder, safe='')}" + f"&uid={quote(message.uid, safe='')}" + ), + summary=summary or None, + body=body or None, + keywords=(message.folder[:200], profile.name[:200]), + visibility="restricted", + acl_tokens=tuple(dict.fromkeys(tokens)), + metadata={ + "profile_id": message.profile_id, + "profile_name": profile.name, + "folder": message.folder, + "uid": message.uid, + "date": message.date, + "attachment_count": message.attachment_count, + }, + source_revision=updated_at.isoformat(), + change_cursor=change_cursor, + source_updated_at=updated_at, + requires_authorization_recheck=True, + ) + + +def _assert_source(provider_id: str, resource_type: str) -> None: + if provider_id != PROVIDER_ID or resource_type != RESOURCE_TYPE: + raise ValueError("Unsupported Mail search source.") + + +def _session(value: object) -> Session: + if not isinstance(value, Session): + raise TypeError("Mail search requires a SQLAlchemy session.") + return value + + +__all__ = [ + "MailSearchSource", + "PROVIDER_ID", + "RESOURCE_TYPE", + "create_mail_search_source", +] diff --git a/tests/test_mailbox_index.py b/tests/test_mailbox_index.py index 650298c..ef5f783 100644 --- a/tests/test_mailbox_index.py +++ b/tests/test_mailbox_index.py @@ -14,6 +14,9 @@ class _DeleteQuery: def filter(self, *_criteria): return self + def all(self): + return [] + def delete(self, *, synchronize_session: bool) -> int: if synchronize_session is not False: raise AssertionError("bulk invalidation must not synchronize loaded cache rows") diff --git a/tests/test_search_source.py b/tests/test_search_source.py new file mode 100644 index 0000000..dbe152c --- /dev/null +++ b/tests/test_search_source.py @@ -0,0 +1,144 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from govoplan_access.backend.db.models import Account, User +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.search import ( + SearchAuthorizationRequest, + SearchBackfillRequest, + SearchResourceReference, +) +from govoplan_core.db.base import Base +from govoplan_mail.backend.db.models import ( + MailMailboxMessageIndex, + MailServerProfile, +) +from govoplan_mail.backend.search_source import ( + MailSearchSource, + PROVIDER_ID, + RESOURCE_TYPE, +) + + +class MailSearchSourceTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite://") + Base.metadata.create_all( + self.engine, + tables=( + Account.__table__, + User.__table__, + MailServerProfile.__table__, + MailMailboxMessageIndex.__table__, + ), + ) + self.session = Session(self.engine) + self.session.add_all( + ( + Account( + id="account-1", + email="one@example.test", + normalized_email="one@example.test", + ), + User( + id="user-1", + tenant_id="tenant-1", + account_id="account-1", + email="one@example.test", + ), + MailServerProfile( + id="profile-1", + tenant_id="tenant-1", + scope_type="user", + scope_id="user-1", + name="Personal mailbox", + slug="personal", + smtp_config={}, + imap_config={"host": "imap.example.test"}, + ), + MailMailboxMessageIndex( + id="message-1", + tenant_id="tenant-1", + profile_id="profile-1", + folder="INBOX", + uid="42", + subject="Permit status", + from_header="service@example.test", + body_preview="Your permit is ready", + indexed_at=datetime(2026, 8, 5, tzinfo=timezone.utc), + ), + ) + ) + self.session.commit() + self.source = MailSearchSource() + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def test_mailbox_backfill_and_profile_recheck_are_bounded(self) -> None: + page = self.source.backfill( + self.session, + request=SearchBackfillRequest( + tenant_id="tenant-1", + provider_id=PROVIDER_ID, + resource_type=RESOURCE_TYPE, + rebuild_id="rebuild-1", + ), + ) + self.assertEqual(("message-1",), tuple(doc.resource_id for doc in page.documents)) + self.assertNotIn("password", str(page.documents[0].metadata).casefold()) + reference = SearchResourceReference( + tenant_id="tenant-1", + module_id="mail", + resource_type=RESOURCE_TYPE, + resource_id="message-1", + ) + request = SearchAuthorizationRequest(reference=reference, source_revision="1") + with patch( + "govoplan_mail.backend.search_source.mail_profile_visible_to_actor", + return_value=True, + ): + allowed = self.source.authorize( + self.session, + _principal(), + requests=(request,), + ) + self.assertTrue(allowed[reference.key]) + self.assertFalse( + self.source.authorize( + self.session, + _principal(scopes={"mail:mailbox:read"}), + requests=(request,), + )[reference.key] + ) + + +def _principal( + *, + scopes: set[str] | None = None, +) -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id="account-1", + membership_id="user-1", + tenant_id="tenant-1", + scopes=frozenset( + scopes or {"mail:mailbox:read", "mail:profile:use"} + ), + ), + account=SimpleNamespace(id="account-1"), + user=SimpleNamespace(id="user-1"), + ) + + +if __name__ == "__main__": + unittest.main()