feat: bridge observed messages to postboxes

src/govoplan_mail/backend/manifest.py

@@ -10,7 +10,9 @@ from govoplan_core.core.mail import (
     CAPABILITY_MAIL_BOUNCE_PROCESSING,
     CAPABILITY_MAIL_DELIVERY_OUTBOX,
     CAPABILITY_MAIL_NOTIFICATION_DELIVERY,
+    CAPABILITY_MAIL_POSTBOX_BRIDGE,
 )
+from govoplan_core.core.postbox import CAPABILITY_POSTBOX_DELIVERY
 from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard
 from govoplan_core.core.modules import (
     DocumentationCondition,
@@ -311,13 +313,14 @@ manifest = ModuleManifest(
     name="Mail",
     version="0.1.18",
     required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
-    optional_dependencies=("campaigns", "addresses", "calendar", "search"),
+    optional_dependencies=("campaigns", "addresses", "calendar", "postbox", "search"),
     provides_interfaces=(
         ModuleInterfaceProvider(name="mail.campaign_delivery", version="0.2.0"),
         ModuleInterfaceProvider(name="mail.delivery_commands", version="0.1.0"),
         ModuleInterfaceProvider(name="mail.delivery_outbox", version="0.1.0"),
         ModuleInterfaceProvider(name="mail.notification_delivery", version="0.1.0"),
         ModuleInterfaceProvider(name="mail.bounce_processing", version="0.1.0"),
+        ModuleInterfaceProvider(name=CAPABILITY_MAIL_POSTBOX_BRIDGE, version="1.0.0"),
     ),
     requires_interfaces=(
         ModuleInterfaceRequirement(
@@ -344,6 +347,12 @@ manifest = ModuleManifest(
             version_max_exclusive="0.2.0",
             optional=True,
         ),
+        ModuleInterfaceRequirement(
+            name=CAPABILITY_POSTBOX_DELIVERY,
+            version_min="0.1.0",
+            version_max_exclusive="0.2.0",
+            optional=True,
+        ),
         ModuleInterfaceRequirement(
             name="search.source",
             version_min="1.0.0",
@@ -454,8 +463,39 @@ manifest = ModuleManifest(
             "govoplan_mail.backend.bounce_processing",
             fromlist=["SqlMailBounceProcessingProvider"],
         ).SqlMailBounceProcessingProvider(),
+        CAPABILITY_MAIL_POSTBOX_BRIDGE: lambda context: __import__(
+            "govoplan_mail.backend.postbox_bridge",
+            fromlist=["create_postbox_bridge"],
+        ).create_postbox_bridge(context),
     },
     documentation=(
+        DocumentationTopic(
+            id="mail.postbox.bridge",
+            title="Bridge selected Mail observations into Postbox",
+            summary="Deliver one immutable IMAP observation to an explicit function Postbox without turning Postbox into a mailbox account.",
+            body=(
+                "Mail exposes an optional, idempotent bridge capability. A configured caller supplies an exact Mail profile, folder, UIDVALIDITY, UID, raw observation, "
+                "and typed Postbox target. Mail parses bounded headers, plaintext, participants, and attachment evidence; Postbox owns target resolution, "
+                "delivery, retention, receipts, and access. The bridge stores no credentials in Postbox and never treats account login as Postbox membership."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("mail_user", "mail_admin", "administrator"),
+            related_modules=("postbox", "idm", "organizations"),
+            translations={
+                "de": {
+                    "title": "Ausgewählte Mail-Beobachtungen an Postbox übergeben",
+                    "summary": "Eine unveränderliche IMAP-Beobachtung an ein ausdrücklich bestimmtes Funktionspostfach liefern, ohne Postbox zu einem Mailkonto zu machen.",
+                    "body": (
+                        "Mail stellt eine optionale, idempotente Bridge-Capability bereit. Ein konfigurierter Aufrufer übergibt ein exaktes Mailprofil, Ordner, UIDVALIDITY, UID, die rohe Beobachtung "
+                        "und ein typisiertes Postbox-Ziel. Mail liest begrenzte Kopfzeilen, Klartext, Beteiligte und Anlagennachweise; Postbox besitzt Zielauflösung, Zustellung, "
+                        "Aufbewahrung, Belege und Zugriff. Die Bridge speichert keine Zugangsdaten in Postbox und behandelt die Kontoanmeldung niemals als Postfachmitgliedschaft."
+                    ),
+                }
+            },
+            metadata={"kind": "guide", "help_contexts": ["mail.postbox-bridge"]},
+            order=36,
+        ),
         DocumentationTopic(
             id="mail.configuration-package.smtp-profile",
             title="Provision SMTP profiles from deployment receipts",

src/govoplan_mail/backend/postbox_bridge.py

@@ -0,0 +1,211 @@
+from __future__ import annotations
+
+import hashlib
+from email import policy
+from email.message import Message
+from email.parser import BytesParser
+from email.utils import getaddresses
+
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.mail import (
+    MailPostboxBridgeProvider,
+    MailPostboxBridgeRequest,
+    MailPostboxBridgeResult,
+)
+from govoplan_core.core.modules import ModuleContext
+from govoplan_core.core.postbox import (
+    PostboxAttachmentRef,
+    PostboxDeliveryRequest,
+    PostboxParticipantRef,
+    PostboxTargetRef,
+    postbox_delivery_provider,
+)
+from govoplan_mail.backend.db.models import MailServerProfile
+from govoplan_mail.backend.runtime import configure_runtime, get_registry
+
+
+MAX_BRIDGE_MESSAGE_BYTES = 50 * 1024 * 1024
+MAX_POSTBOX_BODY_CHARS = 500_000
+
+
+class MailPostboxBridgeError(RuntimeError):
+    pass
+
+
+class MailPostboxBridge(MailPostboxBridgeProvider):
+    """Translate immutable IMAP observations into native Postbox delivery."""
+
+    def bridge_message(
+        self,
+        session: object,
+        request: MailPostboxBridgeRequest,
+    ) -> MailPostboxBridgeResult:
+        if not isinstance(session, Session):
+            raise TypeError("Mail Postbox bridging requires a SQLAlchemy session.")
+        if not isinstance(request.target, PostboxTargetRef):
+            raise MailPostboxBridgeError("A typed Postbox target is required.")
+        if len(request.raw_message) > MAX_BRIDGE_MESSAGE_BYTES:
+            raise MailPostboxBridgeError("Mail message exceeds the Postbox bridge limit.")
+        profile = session.get(MailServerProfile, request.profile_id)
+        if profile is None or profile.tenant_id != request.tenant_id:
+            raise MailPostboxBridgeError("Mail profile not found.")
+        folder = request.folder.strip()
+        uid = request.uid.strip()
+        uidvalidity = request.uidvalidity.strip()
+        if not folder or not uid or not uidvalidity:
+            raise MailPostboxBridgeError(
+                "Mail folder, UIDVALIDITY, and immutable UID are required."
+            )
+        try:
+            message = BytesParser(policy=policy.default).parsebytes(request.raw_message)
+        except Exception as exc:
+            raise MailPostboxBridgeError("Mail message could not be parsed.") from exc
+
+        provider = postbox_delivery_provider(get_registry())
+        if provider is None:
+            raise MailPostboxBridgeError("Postbox delivery is not available.")
+        source_digest = hashlib.sha256(request.raw_message).hexdigest()
+        source_key = hashlib.sha256(
+            f"{request.tenant_id}\0{request.profile_id}\0{folder}\0{uidvalidity}\0{uid}".encode("utf-8")
+        ).hexdigest()
+        result = provider.deliver(
+            session,
+            PostboxDeliveryRequest(
+                tenant_id=request.tenant_id,
+                target=request.target,
+                producer_module="mail",
+                producer_resource_type="imap_message",
+                producer_resource_id=source_key,
+                idempotency_key=f"mail-postbox:{source_key}:{source_digest}",
+                subject=_header(message, "Subject") or "(No subject)",
+                body_text=_plain_text_body(message),
+                sender_label=_header(message, "From"),
+                classification=request.classification,
+                participants=_participants(message),
+                attachments=_attachments(
+                    message,
+                    profile_id=request.profile_id,
+                    folder=folder,
+                    uidvalidity=uidvalidity,
+                    uid=uid,
+                ),
+                metadata={
+                    **dict(request.metadata),
+                    "transport": "mail-imap",
+                    "mail_profile_id": request.profile_id,
+                    "mailbox_folder": folder,
+                    "mailbox_uidvalidity": uidvalidity,
+                    "mailbox_uid": uid,
+                    "rfc_message_id": _header(message, "Message-ID"),
+                    "raw_sha256": source_digest,
+                },
+            ),
+        )
+        return MailPostboxBridgeResult(
+            postbox_id=result.postbox_id,
+            message_id=result.message_id,
+            delivery_id=result.delivery_id,
+            duplicate=result.duplicate,
+            source_digest=source_digest,
+        )
+
+
+def _header(message: Message, name: str) -> str | None:
+    value = " ".join(str(message.get(name) or "").split())
+    return value[:1000] or None
+
+
+def _plain_text_body(message: Message) -> str | None:
+    candidates = message.walk() if message.is_multipart() else (message,)
+    for part in candidates:
+        if part.get_content_type() != "text/plain":
+            continue
+        if part.get_content_disposition() == "attachment":
+            continue
+        try:
+            value = part.get_content()
+        except Exception:
+            payload = part.get_payload(decode=True) or b""
+            value = payload.decode(part.get_content_charset() or "utf-8", errors="replace")
+        if isinstance(value, bytes):
+            value = value.decode(part.get_content_charset() or "utf-8", errors="replace")
+        normalized = str(value).strip()
+        if normalized:
+            return normalized[:MAX_POSTBOX_BODY_CHARS]
+    return None
+
+
+def _participants(message: Message) -> tuple[PostboxParticipantRef, ...]:
+    result: list[PostboxParticipantRef] = []
+    for kind, headers in (
+        ("sender", ("From",)),
+        ("to", ("To",)),
+        ("cc", ("Cc",)),
+        ("bcc", ("Bcc",)),
+    ):
+        for name, address in getaddresses(
+            [str(value) for header in headers for value in message.get_all(header, [])]
+        ):
+            clean_address = address.strip()[:500]
+            if not clean_address:
+                continue
+            result.append(
+                PostboxParticipantRef(
+                    kind=kind,
+                    reference_type="external_email",
+                    label=name.strip()[:500] or None,
+                    address=clean_address,
+                )
+            )
+    return tuple(result)
+
+
+def _attachments(
+    message: Message,
+    *,
+    profile_id: str,
+    folder: str,
+    uidvalidity: str,
+    uid: str,
+) -> tuple[PostboxAttachmentRef, ...]:
+    result: list[PostboxAttachmentRef] = []
+    for index, part in enumerate(message.walk()):
+        filename = part.get_filename()
+        if part.get_content_disposition() != "attachment" and not filename:
+            continue
+        payload = part.get_payload(decode=True) or b""
+        digest = hashlib.sha256(payload).hexdigest()
+        reference_id = hashlib.sha256(
+            f"{profile_id}\0{folder}\0{uidvalidity}\0{uid}\0{index}\0{digest}".encode("utf-8")
+        ).hexdigest()
+        result.append(
+            PostboxAttachmentRef(
+                reference_type="mail_attachment",
+                reference_id=reference_id,
+                name=str(filename or f"attachment-{index + 1}")[:1000],
+                media_type=part.get_content_type(),
+                size_bytes=len(payload),
+                digest=digest,
+                metadata={
+                    "mail_profile_id": profile_id,
+                    "mailbox_folder": folder,
+                    "mailbox_uidvalidity": uidvalidity,
+                    "mailbox_uid": uid,
+                    "mime_part_index": index,
+                },
+            )
+        )
+    return tuple(result)
+
+
+def create_postbox_bridge(context: ModuleContext) -> MailPostboxBridge:
+    configure_runtime(registry=context.registry, settings=context.settings)
+    return MailPostboxBridge()
+
+
+__all__ = [
+    "MailPostboxBridge",
+    "MailPostboxBridgeError",
+    "create_postbox_bridge",
+]

tests/test_postbox_bridge.py

@@ -0,0 +1,138 @@
+from __future__ import annotations
+
+import unittest
+from dataclasses import replace
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from sqlalchemy import Column, String, Table, create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from govoplan_core.core.mail import MailPostboxBridgeRequest
+from govoplan_core.core.postbox import PostboxTargetRef
+from govoplan_core.db.base import Base
+from govoplan_mail.backend.db.models import MailServerProfile
+from govoplan_mail.backend.postbox_bridge import MailPostboxBridge
+
+
+RAW_MESSAGE = b"""From: Ada Example <ada@example.test>
+To: Clerk <clerk@example.test>
+Cc: Archive <archive@example.test>
+Message-ID: <bridge-1@example.test>
+Subject: Submitted evidence
+MIME-Version: 1.0
+Content-Type: multipart/mixed; boundary=bridge
+
+--bridge
+Content-Type: text/plain; charset=utf-8
+
+Please process the attached evidence.
+--bridge
+Content-Type: application/pdf
+Content-Disposition: attachment; filename=evidence.pdf
+Content-Transfer-Encoding: base64
+
+UERG
+--bridge--
+"""
+
+
+class _DeliveryProvider:
+    def __init__(self) -> None:
+        self.requests = []
+
+    def deliver(self, session, request):
+        del session
+        self.requests.append(request)
+        return SimpleNamespace(
+            postbox_id="postbox-1",
+            message_id="message-1",
+            delivery_id="delivery-1",
+            duplicate=len(self.requests) > 1,
+        )
+
+
+class MailPostboxBridgeTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        access_users = Base.metadata.tables.get("access_users")
+        if access_users is None:
+            access_users = Table(
+                "access_users",
+                Base.metadata,
+                Column("id", String(36), primary_key=True),
+            )
+        Base.metadata.create_all(
+            self.engine,
+            tables=[access_users, MailServerProfile.__table__],
+        )
+        self.Session = sessionmaker(
+            bind=self.engine,
+            class_=Session,
+            expire_on_commit=False,
+        )
+        with self.Session() as session:
+            session.add(
+                MailServerProfile(
+                    id="profile-1",
+                    tenant_id="tenant-1",
+                    scope_type="tenant",
+                    scope_id="tenant-1",
+                    name="Inbound",
+                    slug="inbound",
+                    smtp_config={},
+                )
+            )
+            session.commit()
+        self.addCleanup(self.engine.dispose)
+
+    def test_bridge_parses_bounded_content_and_delegates_idempotently(self) -> None:
+        delivery = _DeliveryProvider()
+        request = MailPostboxBridgeRequest(
+            tenant_id="tenant-1",
+            target=PostboxTargetRef(postbox_id="postbox-1"),
+            profile_id="profile-1",
+            folder="INBOX",
+            uid="42",
+            uidvalidity="20260807",
+            raw_message=RAW_MESSAGE,
+        )
+        with self.Session() as session, patch(
+            "govoplan_mail.backend.postbox_bridge.postbox_delivery_provider",
+            return_value=delivery,
+        ):
+            first = MailPostboxBridge().bridge_message(session, request)
+            second = MailPostboxBridge().bridge_message(session, request)
+            MailPostboxBridge().bridge_message(
+                session,
+                replace(request, uidvalidity="20260808"),
+            )
+
+        self.assertFalse(first.duplicate)
+        self.assertTrue(second.duplicate)
+        self.assertEqual(first.source_digest, second.source_digest)
+        self.assertEqual(
+            delivery.requests[0].idempotency_key,
+            delivery.requests[1].idempotency_key,
+        )
+        self.assertNotEqual(
+            delivery.requests[0].idempotency_key,
+            delivery.requests[2].idempotency_key,
+        )
+        bridged = delivery.requests[0]
+        self.assertEqual("Submitted evidence", bridged.subject)
+        self.assertIn("Please process", bridged.body_text)
+        self.assertEqual(
+            ["sender", "to", "cc"],
+            [participant.kind for participant in bridged.participants],
+        )
+        self.assertEqual(1, len(bridged.attachments))
+        self.assertEqual("evidence.pdf", bridged.attachments[0].name)
+        self.assertEqual("mail_attachment", bridged.attachments[0].reference_type)
+        self.assertEqual("<bridge-1@example.test>", bridged.metadata["rfc_message_id"])
+        self.assertEqual("20260807", bridged.metadata["mailbox_uidvalidity"])
+        self.assertNotIn("raw_message", bridged.metadata)
+
+
+if __name__ == "__main__":
+    unittest.main()