Add ciphertext-safe Postbox search source

2026-08-04 03:03:27 +02:00
parent a97eb3bcef
commit 889b5e61f0
4 changed files with 490 additions and 0 deletions
@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from types import SimpleNamespace
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+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_postbox.backend.db.models import (
+    Postbox,
+    PostboxAddress,
+    PostboxMessage,
+)
+from govoplan_postbox.backend.search_source import (
+    PROVIDER_ID,
+    RESOURCE_TYPE,
+    PostboxSearchSource,
+)
+
+
+class _AccessService:
+    def __init__(self, allowed: bool = True) -> None:
+        self.allowed = allowed
+
+    def can_read_message(self, session, **kwargs):
+        del session, kwargs
+        return self.allowed
+
+
+class PostboxSearchSourceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite://")
+        Base.metadata.create_all(
+            self.engine,
+            tables=(
+                PostboxAddress.__table__,
+                Postbox.__table__,
+                PostboxMessage.__table__,
+            ),
+        )
+        self.session = Session(self.engine)
+        self.session.add_all(
+            (
+                PostboxAddress(
+                    id="address-1",
+                    tenant_id="tenant-1",
+                    address_key="permits",
+                    address="permits@example.test",
+                ),
+                Postbox(
+                    id="postbox-1",
+                    tenant_id="tenant-1",
+                    address_id="address-1",
+                    name="Permit office",
+                ),
+                PostboxMessage(
+                    id="message-1",
+                    tenant_id="tenant-1",
+                    postbox_id="postbox-1",
+                    subject="Permit decision",
+                    body_text=None,
+                    body_ciphertext=b"ciphertext",
+                    encryption_profile="server_envelope_v1",
+                    delivered_at=datetime(2026, 8, 5, tzinfo=timezone.utc),
+                ),
+            )
+        )
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_encrypted_content_is_not_indexed_and_access_is_rechecked(self) -> None:
+        source = PostboxSearchSource(_AccessService())  # type: ignore[arg-type]
+        page = source.backfill(
+            self.session,
+            request=SearchBackfillRequest(
+                tenant_id="tenant-1",
+                provider_id=PROVIDER_ID,
+                resource_type=RESOURCE_TYPE,
+                rebuild_id="rebuild-1",
+            ),
+        )
+        self.assertEqual(1, len(page.documents))
+        self.assertIsNone(page.documents[0].body)
+        self.assertNotIn("ciphertext", str(page.documents[0].metadata).casefold())
+        reference = SearchResourceReference(
+            tenant_id="tenant-1",
+            module_id="postbox",
+            resource_type=RESOURCE_TYPE,
+            resource_id="message-1",
+        )
+        request = SearchAuthorizationRequest(reference=reference, source_revision="1")
+        self.assertTrue(
+            source.authorize(
+                self.session,
+                _principal(),
+                requests=(request,),
+            )[reference.key]
+        )
+        denied = PostboxSearchSource(_AccessService(False))  # type: ignore[arg-type]
+        self.assertFalse(
+            denied.authorize(
+                self.session,
+                _principal(),
+                requests=(request,),
+            )[reference.key]
+        )
+
+
+def _principal() -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="user-1",
+            identity_id="identity-1",
+            tenant_id="tenant-1",
+            scopes=frozenset({"postbox:postbox:read"}),
+        ),
+        account=SimpleNamespace(id="account-1"),
+        user=SimpleNamespace(id="user-1"),
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()