feat(postbox): add governed DSAR coverage

This commit is contained in:
2026-08-21 00:46:06 +02:00
parent 8a21876634
commit e5da713d5f
5 changed files with 1868 additions and 1 deletions
+605
View File
@@ -0,0 +1,605 @@
from __future__ import annotations
import unittest
from datetime import datetime, timedelta, timezone
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from govoplan_access.backend.db.models import Account, Group, User
from govoplan_core.core.change_sequence import ChangeSequenceEntry
from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef
from govoplan_core.db.base import Base
from govoplan_core.privacy.dsar_workflow import (
DataSubjectRequest,
create_data_subject_request,
plan_data_subject_erasure,
search_data_subject_request,
)
from govoplan_postbox.backend.db.models import (
Postbox,
PostboxAccessEvent,
PostboxAddress,
PostboxAttachmentReference,
PostboxDelivery,
PostboxGrouping,
PostboxGroupingSource,
PostboxMessage,
PostboxMessageReceipt,
PostboxParticipant,
PostboxProtectionTransition,
PostboxProtectionTransitionItem,
PostboxRoute,
PostboxTemplate,
PostboxTemplateRevision,
)
from govoplan_postbox.backend.dsar_provider import (
POSTBOX_DSAR_CAPABILITY,
PostboxDsarProvider,
)
from govoplan_postbox.backend.manifest import manifest
class _Registry:
def __init__(self, provider: object, *, active: bool = True):
self.provider = provider
self.active = active
def capability_names(self):
return (POSTBOX_DSAR_CAPABILITY,)
def capability_owner(self, name):
assert name == POSTBOX_DSAR_CAPABILITY
return "postbox"
def tenant_entitlement_resolver(self):
active = self.active
class Resolver:
@staticmethod
def resolve(session, tenant_id):
del session, tenant_id
return type(
"State",
(),
{"effective_modules": ("postbox",) if active else ()},
)()
return Resolver()
def require_tenant_capability(self, name, session, **kwargs):
del session, kwargs
assert name == POSTBOX_DSAR_CAPABILITY
return self.provider
class PostboxDsarProviderTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite:///:memory:", future=True)
Base.metadata.create_all(
self.engine,
tables=[
Account.__table__,
User.__table__,
Group.__table__,
ChangeSequenceEntry.__table__,
DataSubjectRequest.__table__,
PostboxTemplate.__table__,
PostboxTemplateRevision.__table__,
PostboxAddress.__table__,
Postbox.__table__,
PostboxMessage.__table__,
PostboxParticipant.__table__,
PostboxAttachmentReference.__table__,
PostboxDelivery.__table__,
PostboxRoute.__table__,
PostboxMessageReceipt.__table__,
PostboxGrouping.__table__,
PostboxGroupingSource.__table__,
PostboxAccessEvent.__table__,
PostboxProtectionTransition.__table__,
PostboxProtectionTransitionItem.__table__,
],
)
self.session = sessionmaker(bind=self.engine, future=True)()
now = datetime.now(timezone.utc)
account = Account(
id="account-subject",
email="subject@example.test",
normalized_email="subject@example.test",
display_name="Subject",
)
user = User(
id="membership-subject",
tenant_id="tenant-1",
account_id=account.id,
email="subject@example.test",
display_name="Subject",
)
template = PostboxTemplate(
id="template-subject",
tenant_id="tenant-1",
slug="subject-template",
name="Subject configured template",
status="published",
created_by=account.id,
updated_by=account.id,
)
revision = PostboxTemplateRevision(
id="template-revision-subject",
tenant_id="tenant-1",
template_id=template.id,
revision=1,
created_by=account.id,
published_at=now,
)
address = PostboxAddress(
id="address-1",
tenant_id="tenant-1",
address_key="office",
address="office.postbox",
status="active",
)
postbox = Postbox(
id="postbox-1",
tenant_id="tenant-1",
address_id=address.id,
name="Office Postbox",
classification="confidential",
)
message = PostboxMessage(
id="message-subject",
tenant_id="tenant-1",
postbox_id=postbox.id,
subject="Subject request context",
body_text="Bounded plaintext concerning the subject",
status="delivered",
classification="personal",
producer_module="postbox",
producer_resource_type="account_authored_message",
producer_resource_id=account.id,
authoring_key="authoring-secret-do-not-export",
delivered_at=now,
metadata_={"secret": "message-metadata-do-not-export"},
)
matching_participant = PostboxParticipant(
id="participant-subject",
tenant_id="tenant-1",
message_id=message.id,
kind="recipient",
reference_type="account",
reference_id=account.id,
label="Subject Person",
address="Subject@Example.Test",
position=1,
metadata_={"secret": "participant-metadata-do-not-export"},
)
unrelated_participant = PostboxParticipant(
id="participant-other",
tenant_id="tenant-1",
message_id=message.id,
kind="recipient",
reference_type="account",
reference_id="account-other",
label="Unrelated Person Do Not Export",
address="other@example.test",
position=2,
)
attachment = PostboxAttachmentReference(
id="attachment-subject",
tenant_id="tenant-1",
message_id=message.id,
reference_type="file_version",
reference_id="file-version-1",
name="subject-evidence.pdf",
digest="digest-do-not-export",
metadata_={"secret": "attachment-metadata-do-not-export"},
)
encrypted_message = PostboxMessage(
id="message-encrypted",
tenant_id="tenant-1",
postbox_id=postbox.id,
subject="Encrypted subject context",
body_ciphertext=b"ciphertext-do-not-export",
status="delivered",
classification="personal",
encryption_profile="server_envelope_v1",
encryption_envelope_id="envelope-do-not-export",
encryption_resource_id="resource-do-not-export",
wrapped_keys=[{"wrapped_key_ref": "wrapped-key-do-not-export"}],
external_recipient_tokens=[{"token_id": "external-token-do-not-export"}],
delivered_at=now,
)
encrypted_participant = PostboxParticipant(
id="participant-encrypted-subject",
tenant_id="tenant-1",
message_id=encrypted_message.id,
kind="recipient",
reference_type="identity",
reference_id="identity-subject",
position=1,
)
unrelated_message = PostboxMessage(
id="message-other",
tenant_id="tenant-1",
postbox_id=postbox.id,
subject="Unrelated message do not export",
body_text="Unrelated body do not export",
status="delivered",
delivered_at=now,
)
tenant_two_address = PostboxAddress(
id="address-tenant-2",
tenant_id="tenant-2",
address_key="other",
address="other.postbox",
)
tenant_two_postbox = Postbox(
id="postbox-tenant-2",
tenant_id="tenant-2",
address_id=tenant_two_address.id,
name="Tenant two Postbox",
)
tenant_two_message = PostboxMessage(
id="message-tenant-2",
tenant_id="tenant-2",
postbox_id=tenant_two_postbox.id,
subject="Tenant two message do not export",
body_text="Tenant two body do not export",
delivered_at=now,
)
tenant_two_participant = PostboxParticipant(
id="participant-tenant-2",
tenant_id="tenant-2",
message_id=tenant_two_message.id,
kind="recipient",
reference_type="account",
reference_id=account.id,
)
delivery = PostboxDelivery(
id="delivery-subject",
tenant_id="tenant-1",
postbox_id=postbox.id,
message_id=message.id,
producer_module="postbox",
producer_resource_type="account_authored_message",
producer_resource_id=account.id,
idempotency_key="delivery-idempotency-do-not-export",
status="accepted",
holder_count=1,
target_snapshot={"secret": "target-snapshot-do-not-export"},
accepted_at=now,
metadata_={"secret": "delivery-metadata-do-not-export"},
)
route = PostboxRoute(
id="route-subject",
tenant_id="tenant-1",
delivery_id=delivery.id,
source_postbox_id=postbox.id,
source_message_id=message.id,
target_postbox_id=postbox.id,
target_message_id=message.id,
route_kind="linked_copy",
status="completed",
depth=1,
processed_at=now,
policy_snapshot={"secret": "route-policy-do-not-export"},
)
receipt = PostboxMessageReceipt(
id="receipt-subject",
tenant_id="tenant-1",
message_id=message.id,
account_id=account.id,
identity_id="identity-subject",
assignment_id="assignment-subject",
read_at=now,
acknowledged_at=now,
metadata_={"secret": "receipt-metadata-do-not-export"},
)
grouping = PostboxGrouping(
id="grouping-subject",
tenant_id="tenant-1",
account_id=account.id,
name="My work",
is_default=True,
settings={"secret": "grouping-settings-do-not-export"},
)
grouping.sources.append(
PostboxGroupingSource(
id="grouping-source-subject",
tenant_id="tenant-1",
postbox_id=postbox.id,
position=0,
)
)
access_event = PostboxAccessEvent(
id="access-event-subject",
tenant_id="tenant-1",
postbox_id=postbox.id,
message_id=message.id,
account_id=account.id,
identity_id="identity-subject",
assignment_id="assignment-subject",
action="read_message",
outcome="allowed",
reason_code="assigned",
occurred_at=now,
details={"secret": "access-details-do-not-export"},
)
transition = PostboxProtectionTransition(
id="transition-subject",
tenant_id="tenant-1",
postbox_id=postbox.id,
idempotency_key="transition-idempotency-do-not-export",
source_profile="plaintext_v1",
target_profile="server_envelope_v1",
history_mode="migrate",
authority_mode="institutional",
required_quorum=1,
evidence_refs=["evidence-ref-do-not-export"],
reason="private transition reason do not export",
state="completed",
message_count=1,
completed_count=1,
requested_by=account.id,
activated_at=now,
completed_at=now,
configuration_snapshot={"secret": "transition-config-do-not-export"},
)
transition_item = PostboxProtectionTransitionItem(
id="transition-item-subject",
tenant_id="tenant-1",
transition_id=transition.id,
message_id=message.id,
source_profile="plaintext_v1",
target_profile="server_envelope_v1",
state="completed",
source_digest="source-digest-do-not-export",
target_digest="target-digest-do-not-export",
completed_by=account.id,
completed_at=now,
evidence={"secret": "transition-item-evidence-do-not-export"},
)
self.session.add_all(
[
account,
user,
template,
revision,
address,
postbox,
message,
matching_participant,
unrelated_participant,
attachment,
encrypted_message,
encrypted_participant,
unrelated_message,
tenant_two_address,
tenant_two_postbox,
tenant_two_message,
tenant_two_participant,
delivery,
route,
receipt,
grouping,
access_event,
transition,
transition_item,
]
)
self.session.commit()
self.provider = PostboxDsarProvider()
self.subject = DsarSubjectRef(
account_id=account.id,
identity_id="identity-subject",
membership_id=user.id,
email="subject@example.test",
external_references={"postbox.assignment": "assignment-subject"},
)
def tearDown(self) -> None:
self.session.close()
self.engine.dispose()
def test_manifest_and_minimized_tenant_scoped_search(self) -> None:
self.assertIn(
POSTBOX_DSAR_CAPABILITY,
{item.name for item in manifest.provides_interfaces},
)
self.assertIsInstance(
manifest.capability_factories[POSTBOX_DSAR_CAPABILITY](None),
DsarProvider,
)
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=self.subject,
)
self.assertTrue(
{
"postbox_message",
"postbox_participant",
"postbox_attachment_reference",
"postbox_delivery",
"postbox_route",
"postbox_message_receipt",
"postbox_grouping",
"postbox_access_event",
"postbox_template",
"postbox_template_revision",
"postbox_protection_transition",
"postbox_protection_transition_item",
}.issubset({record.resource_type for record in records})
)
encrypted = next(
record
for record in records
if record.resource_type == "postbox_message"
and record.resource_id == "message-encrypted"
)
self.assertEqual(
"institution_managed_envelope", encrypted.data["content_state"]
)
serialized = repr([record.to_dict() for record in records])
for hidden in (
"Unrelated Person Do Not Export",
"other@example.test",
"participant-other",
"Unrelated message do not export",
"Unrelated body do not export",
"message-tenant-2",
"Tenant two message do not export",
"ciphertext-do-not-export",
"envelope-do-not-export",
"resource-do-not-export",
"wrapped-key-do-not-export",
"external-token-do-not-export",
"authoring-secret-do-not-export",
"message-metadata-do-not-export",
"participant-metadata-do-not-export",
"digest-do-not-export",
"attachment-metadata-do-not-export",
"delivery-idempotency-do-not-export",
"target-snapshot-do-not-export",
"delivery-metadata-do-not-export",
"route-policy-do-not-export",
"receipt-metadata-do-not-export",
"grouping-settings-do-not-export",
"access-details-do-not-export",
"transition-idempotency-do-not-export",
"evidence-ref-do-not-export",
"private transition reason do not export",
"transition-config-do-not-export",
"source-digest-do-not-export",
"target-digest-do-not-export",
"transition-item-evidence-do-not-export",
):
self.assertNotIn(hidden, serialized)
def test_conflicting_selectors_fail_closed(self) -> None:
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
account_id="account-subject",
external_references={"postbox.account": "account-other"},
),
)
self.assertEqual((), records)
def test_grouping_erasure_is_revalidated_and_idempotent(self) -> None:
records = self.provider.search_subject(
self.session, tenant_id="tenant-1", subject=self.subject
)
actions = self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=self.subject,
records=records,
)
self.assertTrue(
{"retain", "manual_review", "delete"}.issubset(
{action.kind for action in actions}
)
)
delete = next(action for action in actions if action.kind == "delete")
grouping = self.session.get(PostboxGrouping, "grouping-subject")
assert grouping is not None
grouping.resource_revision += 1
self.session.commit()
stale = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=self.subject,
actions=(delete,),
request_id="request-1",
)
self.assertEqual("blocked", stale[0].status)
refreshed_records = self.provider.search_subject(
self.session, tenant_id="tenant-1", subject=self.subject
)
refreshed_actions = self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=self.subject,
records=refreshed_records,
)
refreshed_delete = next(
action for action in refreshed_actions if action.kind == "delete"
)
executed = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=self.subject,
actions=(refreshed_delete,),
request_id="request-1",
)
self.assertEqual("executed", executed[0].status)
self.assertIsNone(self.session.get(PostboxGrouping, "grouping-subject"))
self.assertIsNotNone(self.session.get(PostboxMessage, "message-subject"))
replay = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=self.subject,
actions=(refreshed_delete,),
request_id="request-1",
)
self.assertEqual("unchanged", replay[0].status)
def test_core_workflow_discovers_active_and_skips_disabled_provider(self) -> None:
request = create_data_subject_request(
self.session,
tenant_id="tenant-1",
reference="DSAR-POSTBOX-1",
request_kind="access_and_erasure",
subject=self.subject,
purpose="Authorized request",
legal_basis="GDPR",
due_at=datetime.now(timezone.utc) + timedelta(days=30),
requested_by_account_id="privacy-officer",
)
self.session.commit()
search_data_subject_request(
self.session,
registry=_Registry(self.provider),
row=request,
expected_revision=1,
)
self.assertEqual(["postbox"], request.coverage["covered_modules"])
plan_data_subject_erasure(
self.session,
registry=_Registry(self.provider),
row=request,
expected_revision=2,
)
self.assertTrue(
any(action["executable"] for action in request.erasure_plan["actions"])
)
disabled = create_data_subject_request(
self.session,
tenant_id="tenant-1",
reference="DSAR-POSTBOX-OFF",
request_kind="access",
subject=self.subject,
purpose="Coverage",
legal_basis=None,
due_at=None,
requested_by_account_id="privacy-officer",
)
search_data_subject_request(
self.session,
registry=_Registry(self.provider, active=False),
row=disabled,
expected_revision=1,
)
self.assertEqual(
[POSTBOX_DSAR_CAPABILITY],
disabled.coverage["inactive_provider_capabilities"],
)
if __name__ == "__main__":
unittest.main()
+5 -1
View File
@@ -12,6 +12,7 @@ from govoplan_core.core.postbox import (
CAPABILITY_POSTBOX_ROUTING,
)
from govoplan_core.core.encryption import CAPABILITY_ENCRYPTION_CONTENT_CIPHER
from govoplan_postbox.backend.dsar_provider import POSTBOX_DSAR_CAPABILITY
from govoplan_postbox.backend.manifest import get_manifest
@@ -33,11 +34,14 @@ class PostboxManifestTests(unittest.TestCase):
CAPABILITY_POSTBOX_EVIDENCE,
CAPABILITY_POSTBOX_ROUTING,
CAPABILITY_POSTBOX_PORTAL,
POSTBOX_DSAR_CAPABILITY,
},
set(manifest.capability_factories),
)
self.assertEqual("@govoplan/postbox-webui", manifest.frontend.package_name)
self.assertEqual(["/postbox"], [route.path for route in manifest.frontend.routes])
self.assertEqual(
["/postbox"], [route.path for route in manifest.frontend.routes]
)
self.assertIn(
"idm.function_assignments",
manifest.required_capabilities,