GovOPlaN/govoplan-postbox
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: protect postbox message content

Browse Source
This commit is contained in:
zemion
2026-08-02 03:40:57 +02:00
parent 7d310d5c33
commit 8f8d259a76
12 changed files with 611 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_manifest.py
+9
View File
@@ -10,6 +10,7 @@ from govoplan_core.core.postbox import (
CAPABILITY_POSTBOX_MESSAGES,
CAPABILITY_POSTBOX_ROUTING,
)
from govoplan_core.core.encryption import CAPABILITY_ENCRYPTION_CONTENT_CIPHER
from govoplan_postbox.backend.manifest import get_manifest
@@ -39,6 +40,14 @@ class PostboxManifestTests(unittest.TestCase):
"idm.function_assignments",
manifest.required_capabilities,
)
self.assertIn("encryption", manifest.optional_dependencies)
self.assertTrue(
any(
requirement.name == CAPABILITY_ENCRYPTION_CONTENT_CIPHER
and requirement.optional
for requirement in manifest.requires_interfaces
)
)
if __name__ == "__main__":
tests/test_migration.py
+21
View File
@@ -26,6 +26,10 @@ class PostboxMigrationTests(unittest.TestCase):
"govoplan_postbox.backend.migrations.versions."
"a6d9e1f4c8b3_v013_external_recipient_tokens"
)
protection_migration = importlib.import_module(
"govoplan_postbox.backend.migrations.versions."
"d8e3f6a9b2c5_postbox_content_protection"
)
engine = create_engine("sqlite:///:memory:")
try:
with engine.begin() as connection:
@@ -34,15 +38,18 @@ class PostboxMigrationTests(unittest.TestCase):
route_original = route_migration.op
occ_original = occ_migration.op
envelope_original = envelope_migration.op
protection_original = protection_migration.op
migration.op = operations
route_migration.op = operations
occ_migration.op = operations
envelope_migration.op = operations
protection_migration.op = operations
try:
migration.upgrade()
route_migration.upgrade()
occ_migration.upgrade()
envelope_migration.upgrade()
protection_migration.upgrade()
tables = set(inspect(connection).get_table_names())
self.assertIn("postboxes", tables)
self.assertIn("postbox_messages", tables)
@@ -63,8 +70,20 @@ class PostboxMigrationTests(unittest.TestCase):
"key_epoch",
"expires_at",
"withdrawn_at",
"body_ciphertext",
"encryption_envelope_id",
"encryption_resource_id",
}.issubset(message_columns)
)
self.assertIn(
"encryption_vault_id",
{
column["name"]
for column in inspect(connection).get_columns(
"postbox_template_revisions"
)
},
)
self.assertIn("authoring_key", message_columns)
for table_name in (
"postbox_templates",
@@ -91,6 +110,7 @@ class PostboxMigrationTests(unittest.TestCase):
route_columns
)
)
protection_migration.downgrade()
envelope_migration.downgrade()
occ_migration.downgrade()
route_migration.downgrade()
@@ -107,6 +127,7 @@ class PostboxMigrationTests(unittest.TestCase):
route_migration.op = route_original
occ_migration.op = occ_original
envelope_migration.op = envelope_original
protection_migration.op = protection_original
finally:
engine.dispose()
tests/test_service.py
+82
View File
@@ -2,6 +2,8 @@ from __future__ import annotations
import unittest
from datetime import timedelta
from types import SimpleNamespace
from unittest.mock import patch
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
@@ -51,6 +53,7 @@ from govoplan_postbox.backend.db.models import (
PostboxTemplate,
PostboxTemplateRevision,
)
from govoplan_postbox.backend.content_protection import PostboxContentProtectionError
from govoplan_postbox.backend.service import PostboxError, PostboxService
@@ -1236,6 +1239,85 @@ class PostboxServiceTests(unittest.TestCase):
),
)
def test_server_envelope_body_is_not_persisted_in_plaintext_and_fails_closed(
self,
) -> None:
self.idm.assignments.append(self.assignment)
protected = SimpleNamespace(
ciphertext=b"encrypted-message-body",
envelope=SimpleNamespace(
envelope_id="envelope-1",
ciphertext_ref="postbox-db://message/body",
algorithm_suite="AES-256-GCM",
wrapped_key_refs=("wrapped-key-1",),
),
)
with Session(self.engine) as session:
postbox = self.service.create_exact_postbox(
session,
tenant_id="tenant-1",
name="Protected intake",
organization_unit_id="unit-1",
function_id="function-1",
address_key=None,
description=None,
classification="internal",
actor_id="admin-1",
encryption_profile="server_envelope_v1",
encryption_vault_id="vault-1",
)
with patch(
"govoplan_postbox.backend.content_protection.protect_message_body",
return_value=protected,
):
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-protected",
idempotency_key="protected-message",
subject="Protected notice",
body_text="clear message body",
),
)
stored = session.get(PostboxMessage, delivered.message_id)
assert stored is not None
self.assertIsNone(stored.body_text)
self.assertEqual(b"encrypted-message-body", stored.body_ciphertext)
self.assertEqual("envelope-1", stored.encryption_envelope_id)
with patch(
"govoplan_postbox.backend.content_protection.unprotect_message_body",
return_value="clear message body",
):
opened = self.service.get_message(
session,
tenant_id="tenant-1",
message_id=delivered.message_id,
actor=self.actor,
)
assert opened is not None
self.assertEqual("clear message body", opened.body_text)
with patch(
"govoplan_postbox.backend.content_protection.unprotect_message_body",
side_effect=PostboxContentProtectionError("provider unavailable"),
):
with self.assertRaisesRegex(
PostboxError,
"provider unavailable",
):
self.service.get_message(
session,
tenant_id="tenant-1",
message_id=delivered.message_id,
actor=self.actor,
)
def test_hierarchy_linked_copy_snapshots_path_and_independent_state(
self,
) -> None: