Add encrypted envelope recipient state

tests/test_migration.py

@@ -22,6 +22,10 @@ class PostboxMigrationTests(unittest.TestCase):
             "govoplan_postbox.backend.migrations.versions."
             "f5c8d0e3b7a2_v012_authoring_and_occ"
         )
+        envelope_migration = importlib.import_module(
+            "govoplan_postbox.backend.migrations.versions."
+            "a6d9e1f4c8b3_v013_external_recipient_tokens"
+        )
         engine = create_engine("sqlite:///:memory:")
         try:
             with engine.begin() as connection:
@@ -29,13 +33,16 @@ class PostboxMigrationTests(unittest.TestCase):
                 original = migration.op
                 route_original = route_migration.op
                 occ_original = occ_migration.op
+                envelope_original = envelope_migration.op
                 migration.op = operations
                 route_migration.op = operations
                 occ_migration.op = operations
+                envelope_migration.op = operations
                 try:
                     migration.upgrade()
                     route_migration.upgrade()
                     occ_migration.upgrade()
+                    envelope_migration.upgrade()
                     tables = set(inspect(connection).get_table_names())
                     self.assertIn("postboxes", tables)
                     self.assertIn("postbox_messages", tables)
@@ -52,6 +59,7 @@ class PostboxMigrationTests(unittest.TestCase):
                             "ciphertext_ref",
                             "signed_manifest_ref",
                             "wrapped_keys",
+                            "external_recipient_tokens",
                             "key_epoch",
                             "expires_at",
                             "withdrawn_at",
@@ -83,6 +91,7 @@ class PostboxMigrationTests(unittest.TestCase):
                             route_columns
                         )
                     )
+                    envelope_migration.downgrade()
                     occ_migration.downgrade()
                     route_migration.downgrade()
                     migration.downgrade()
@@ -97,6 +106,7 @@ class PostboxMigrationTests(unittest.TestCase):
                     migration.op = original
                     route_migration.op = route_original
                     occ_migration.op = occ_original
+                    envelope_migration.op = envelope_original
         finally:
             engine.dispose()
 

tests/test_router.py

@@ -241,6 +241,15 @@ class PostboxRouterTests(unittest.TestCase):
         self.patch.stop()
         self.engine.dispose()
 
+    def test_assignment_context_override_must_belong_to_principal(self) -> None:
+        response = self.client.get(
+            "/api/v1/postbox/directory",
+            params={"assignment_context_id": "assignment-not-granted"},
+        )
+
+        self.assertEqual(response.status_code, 403)
+        self.assertIn("not active for this principal", response.text)
+
     def test_directory_delivery_message_and_receipt_round_trip(self) -> None:
         directory = self.client.get("/api/v1/postbox/directory")
         self.assertEqual(200, directory.status_code, directory.text)

tests/test_service.py

@@ -29,7 +29,9 @@ from govoplan_core.core.organizations import (
 from govoplan_core.core.postbox import (
     PostboxActorRef,
     PostboxDeliveryRequest,
+    PostboxExternalRecipientTokenRef,
     PostboxTargetRef,
+    PostboxWrappedKeyRef,
 )
 from govoplan_core.db.base import Base
 from govoplan_core.security.time import utc_now
@@ -1173,6 +1175,67 @@ class PostboxServiceTests(unittest.TestCase):
             self.assertFalse(summaries[expired.delivery_id].currently_readable)
             self.assertEqual(1, summaries[expired.delivery_id].expired_message_count)
 
+    def test_encrypted_envelope_and_external_grant_state_cross_capability(self) -> None:
+        self.idm.assignments.append(self.assignment)
+        expires_at = utc_now() + timedelta(days=1)
+        with Session(self.engine) as session:
+            postbox = self._create_exact(session)
+            delivered = self.service.deliver(
+                session,
+                PostboxDeliveryRequest(
+                    tenant_id="tenant-1",
+                    target=PostboxTargetRef(postbox_id=postbox.id),
+                    producer_module="campaigns",
+                    producer_resource_type="campaign_recipient",
+                    producer_resource_id="recipient-encrypted",
+                    idempotency_key="encrypted-message",
+                    subject="Encrypted decision",
+                    ciphertext_ref="files:ciphertext-1",
+                    signed_manifest_ref="files:manifest-1",
+                    wrapped_keys=(
+                        PostboxWrappedKeyRef(
+                            recipient_type="function_postbox",
+                            recipient_id=postbox.id,
+                            key_epoch=postbox.key_epoch,
+                            wrapped_key_ref="trust:wrapped-key-1",
+                            algorithm="HPKE-v1",
+                        ),
+                    ),
+                    external_recipient_tokens=(
+                        PostboxExternalRecipientTokenRef(
+                            token_id="grant-1",
+                            state="available",
+                            expires_at=expires_at,
+                            one_time=True,
+                            assurance_profile="email-otp",
+                        ),
+                    ),
+                ),
+            )
+
+            message = self.service.get_message(
+                session,
+                tenant_id="tenant-1",
+                message_id=delivered.message_id,
+                actor=self.actor,
+            )
+
+            assert message is not None
+            self.assertEqual("files:ciphertext-1", message.ciphertext_ref)
+            self.assertEqual(
+                "trust:wrapped-key-1",
+                message.wrapped_keys[0].wrapped_key_ref,
+            )
+            self.assertEqual("grant-1", message.external_recipient_tokens[0].token_id)
+            self.assertEqual("available", message.external_recipient_tokens[0].state)
+            assert message.external_recipient_tokens[0].expires_at is not None
+            self.assertEqual(
+                expires_at.replace(microsecond=0),
+                message.external_recipient_tokens[0].expires_at.replace(
+                    microsecond=0
+                ),
+            )
+
     def test_hierarchy_linked_copy_snapshots_path_and_independent_state(
         self,
     ) -> None: