From b89d9fb434ba29b9ec0aa08d9a5914aef1a226a8 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 21 Aug 2026 13:33:30 +0200 Subject: [PATCH] feat(dist-lists): add governed DSAR coverage --- .../backend/dsar_provider.py | 718 ++++++++++++++++++ src/govoplan_dist_lists/backend/manifest.py | 121 ++- tests/test_dsar_provider.py | 282 +++++++ 3 files changed, 1112 insertions(+), 9 deletions(-) create mode 100644 src/govoplan_dist_lists/backend/dsar_provider.py create mode 100644 tests/test_dsar_provider.py diff --git a/src/govoplan_dist_lists/backend/dsar_provider.py b/src/govoplan_dist_lists/backend/dsar_provider.py new file mode 100644 index 0000000..bb399a0 --- /dev/null +++ b/src/govoplan_dist_lists/backend/dsar_provider.py @@ -0,0 +1,718 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timezone +from email.utils import parseaddr + +from sqlalchemy import or_ +from sqlalchemy.orm import Session + +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarExecutionResultRef, + DsarRecordRef, + DsarSubjectRef, + dsar_capability_name, +) +from govoplan_dist_lists.backend.db.models import ( + DistributionList, + DistributionListEntry, + DistributionListRevision, + DistributionListSnapshot, +) + + +DIST_LISTS_DSAR_CAPABILITY = dsar_capability_name("dist_lists") +_MAX_RECORDS = 5_000 +_MAX_SNAPSHOTS = 5_000 +_MAX_SNAPSHOT_RECIPIENTS = 20_000 +_MAX_CHANNELS = 20 +_CONFLICT = object() + + +@dataclass(frozen=True, slots=True) +class _Selectors: + account_id: str | None + identity_id: str | None + email: str | None + contact_id: str | None + list_id: str | None + snapshot_id: str | None + entry_id: str | None + + @property + def has_subject_identity(self) -> bool: + return bool( + self.account_id or self.identity_id or self.email or self.contact_id + ) + + +class DistributionListsDsarProvider: + provider_id = "dist_lists" + module_id = "dist_lists" + + def search_subject( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + ) -> Sequence[DsarRecordRef]: + db = _session(session) + selectors = _selectors(subject) + if selectors is None or not selectors.has_subject_identity: + return () + + records: list[DsarRecordRef] = [] + seen: set[tuple[str, str]] = set() + + def append(record: DsarRecordRef) -> None: + key = (record.resource_type, record.resource_id) + if key in seen: + return + if len(records) >= _MAX_RECORDS: + raise ValueError( + "Distribution Lists DSAR result limit exceeded; narrow selectors." + ) + seen.add(key) + records.append(record) + + self._append_subject_scopes( + db, append=append, tenant_id=tenant_id, selectors=selectors + ) + self._append_definition_entries( + db, append=append, tenant_id=tenant_id, selectors=selectors + ) + self._append_snapshot_recipients( + db, append=append, tenant_id=tenant_id, selectors=selectors + ) + self._append_actor_attribution( + db, append=append, tenant_id=tenant_id, selectors=selectors + ) + return tuple( + sorted(records, key=lambda item: (item.resource_type, item.resource_id)) + ) + + def plan_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + records: Sequence[DsarRecordRef], + ) -> Sequence[DsarErasureActionRef]: + del tenant_id + _session(session) + selectors = _selectors(subject) + if selectors is None or not selectors.has_subject_identity: + raise ValueError("Distribution Lists DSAR subject selectors are invalid.") + actions = [] + for record in records: + _validate_record(record) + review = record.resource_type in { + "distribution_list_subject_scope", + "distribution_list_definition_entry", + } + actions.append( + DsarErasureActionRef( + action_id=( + f"dist_lists:{'manual_review' if review else 'retain'}:" + f"{record.resource_type}:{record.resource_id}" + ), + provider_id=self.provider_id, + module_id=self.module_id, + kind="manual_review" if review else "retain", + resource_type=record.resource_type, + resource_id=record.resource_id, + title=("Review " if review else "Retain ") + record.title, + rationale=( + "Create a new list revision or retire the subject-owned list only " + "after reviewing downstream consumers; prior revisions remain evidence." + if review + else record.retention_reason + or "Distribution-list execution evidence remains retained." + ), + executable=False, + ) + ) + return tuple(actions) + + def execute_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + actions: Sequence[DsarErasureActionRef], + request_id: str, + ) -> Sequence[DsarExecutionResultRef]: + del tenant_id + _session(session) + selectors = _selectors(subject) + if selectors is None or not selectors.has_subject_identity: + raise ValueError("Distribution Lists DSAR subject selectors are invalid.") + results = [] + for action in actions: + _validate_action(action) + if action.executable or action.kind not in {"manual_review", "retain"}: + raise ValueError( + "Distribution Lists DSAR publishes non-executable actions only." + ) + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary=( + "Distribution-list data remains unchanged pending a governed " + "revision and downstream-consumer review." + if action.kind == "manual_review" + else "Immutable distribution-list evidence remains retained." + ), + evidence={"request_id": request_id}, + ) + ) + return tuple(results) + + def _append_subject_scopes( + self, + db: Session, + *, + append, + tenant_id: str, + selectors: _Selectors, + ) -> None: + if not selectors.account_id or selectors.snapshot_id or selectors.entry_id: + return + query = db.query(DistributionList).filter( + DistributionList.tenant_id == tenant_id, + DistributionList.scope_type == "user", + DistributionList.scope_id == selectors.account_id, + ) + query = _list_filter(query, DistributionList, selectors) + for row in _limited(query, DistributionList.updated_at, DistributionList.id): + append( + _record( + "distribution_list_subject_scope", + row.id, + "subject_owned_distribution_list", + "Subject-scoped distribution list", + { + "distribution_list_id": row.id, + "scope_type": row.scope_type, + "scope_id": row.scope_id, + "status": row.status, + "current_revision": row.current_revision, + "deleted_at": _iso(row.deleted_at), + "created_at": _iso(row.created_at), + "updated_at": _iso(row.updated_at), + }, + observed_at=row.updated_at, + retention_reason=( + "List retirement requires downstream-consumer and retained-snapshot review." + ), + ) + ) + + def _append_definition_entries( + self, + db: Session, + *, + append, + tenant_id: str, + selectors: _Selectors, + ) -> None: + if selectors.snapshot_id: + return + query = db.query(DistributionListEntry).filter( + DistributionListEntry.tenant_id == tenant_id + ) + if selectors.entry_id: + query = query.filter(DistributionListEntry.id == selectors.entry_id) + if selectors.list_id: + query = query.join(DistributionListRevision).filter( + DistributionListRevision.distribution_list_id == selectors.list_id + ) + for row in _limited( + query, DistributionListEntry.created_at, DistributionListEntry.id + ): + if not _entry_matches(row, selectors): + continue + append( + _record( + "distribution_list_definition_entry", + row.id, + "distribution_list_recipient_definition", + "Distribution-list recipient definition", + { + "entry_id": row.id, + "revision_id": row.revision_id, + "kind": row.kind, + "mode": row.mode, + "source_provider": row.source_provider, + "source_resource_type": row.source_resource_type, + "source_resource_id": _bounded(row.source_resource_id, 500), + "purpose": _bounded(row.purpose, 120), + "requested_channels": [ + str(value)[:40] for value in row.requested_channels[:10] + ], + "effective_from": _iso(row.effective_from), + "effective_until": _iso(row.effective_until), + "created_at": _iso(row.created_at), + }, + observed_at=row.created_at, + immutable=True, + retention_reason=( + "Saved distribution-list revisions remain evidence for existing snapshots." + ), + ) + ) + + def _append_snapshot_recipients( + self, + db: Session, + *, + append, + tenant_id: str, + selectors: _Selectors, + ) -> None: + if selectors.entry_id: + return + query = db.query(DistributionListSnapshot).filter( + DistributionListSnapshot.tenant_id == tenant_id + ) + query = _list_filter(query, DistributionListSnapshot, selectors) + if selectors.snapshot_id: + query = query.filter(DistributionListSnapshot.id == selectors.snapshot_id) + snapshots = ( + query.order_by( + DistributionListSnapshot.created_at, DistributionListSnapshot.id + ) + .limit(_MAX_SNAPSHOTS + 1) + .all() + ) + if len(snapshots) > _MAX_SNAPSHOTS: + raise ValueError( + "Distribution Lists DSAR snapshot scan limit exceeded; narrow selectors." + ) + for snapshot in snapshots: + total = len(snapshot.recipients) + len(snapshot.excluded) + if total > _MAX_SNAPSHOT_RECIPIENTS: + raise ValueError( + "Distribution Lists DSAR snapshot recipient limit exceeded." + ) + for placement, rows in ( + ("included", snapshot.recipients), + ("excluded", snapshot.excluded), + ): + for index, recipient in enumerate(rows): + if not isinstance(recipient, dict) or not _recipient_matches( + recipient, selectors + ): + continue + append( + _snapshot_recipient_record( + snapshot, + recipient=recipient, + placement=placement, + index=index, + ) + ) + + def _append_actor_attribution( + self, + db: Session, + *, + append, + tenant_id: str, + selectors: _Selectors, + ) -> None: + if not selectors.account_id: + return + if selectors.entry_id: + return + account_id = selectors.account_id + if not selectors.snapshot_id: + lists = db.query(DistributionList).filter( + DistributionList.tenant_id == tenant_id, + or_( + DistributionList.created_by_account_id == account_id, + DistributionList.updated_by_account_id == account_id, + ), + ) + lists = _list_filter(lists, DistributionList, selectors) + for row in _limited( + lists, DistributionList.updated_at, DistributionList.id + ): + fields = [] + if row.created_by_account_id == account_id: + fields.append("created_by_account_id") + if row.updated_by_account_id == account_id: + fields.append("updated_by_account_id") + append( + _attribution_record( + "distribution_list_actor_attribution", + row.id, + "Distribution-list actor attribution", + { + "distribution_list_id": row.id, + "activities": fields, + "status": row.status, + "current_revision": row.current_revision, + "created_at": _iso(row.created_at), + "updated_at": _iso(row.updated_at), + }, + row.updated_at, + ) + ) + + revisions = db.query(DistributionListRevision).filter( + DistributionListRevision.tenant_id == tenant_id, + DistributionListRevision.created_by_account_id == account_id, + ) + if selectors.list_id: + revisions = revisions.filter( + DistributionListRevision.distribution_list_id == selectors.list_id + ) + for row in _limited( + revisions, + DistributionListRevision.created_at, + DistributionListRevision.id, + ): + append( + _attribution_record( + "distribution_list_revision_actor_attribution", + row.id, + "Distribution-list revision actor attribution", + { + "distribution_list_id": row.distribution_list_id, + "revision_id": row.id, + "revision": row.revision, + "definition_kind": row.definition_kind, + "activity": "created_revision", + "created_at": _iso(row.created_at), + }, + row.created_at, + ) + ) + + snapshots = db.query(DistributionListSnapshot).filter( + DistributionListSnapshot.tenant_id == tenant_id, + DistributionListSnapshot.created_by_account_id == account_id, + ) + snapshots = _list_filter(snapshots, DistributionListSnapshot, selectors) + if selectors.snapshot_id: + snapshots = snapshots.filter( + DistributionListSnapshot.id == selectors.snapshot_id + ) + for row in _limited( + snapshots, DistributionListSnapshot.created_at, DistributionListSnapshot.id + ): + append( + _attribution_record( + "distribution_list_snapshot_actor_attribution", + row.id, + "Distribution-list snapshot actor attribution", + { + "distribution_list_id": row.distribution_list_id, + "snapshot_id": row.id, + "revision_number": row.revision_number, + "effective_at": _iso(row.effective_at), + "activity": "froze_expansion_snapshot", + "created_at": _iso(row.created_at), + }, + row.created_at, + ) + ) + + +def _selectors(subject: DsarSubjectRef) -> _Selectors | None: + references = subject.external_references + values = { + "account_id": _coalesce( + subject.account_id, + references.get("dist_lists.account"), + references.get("access.account"), + ), + "identity_id": _coalesce( + subject.identity_id, + references.get("dist_lists.identity"), + references.get("identity.id"), + ), + "email": _coalesce_email(subject.email, references.get("dist_lists.email")), + "contact_id": _coalesce( + references.get("dist_lists.contact"), + references.get("addresses.contact"), + ), + "list_id": _coalesce( + references.get("dist_lists.list"), + references.get("dist_lists.list_id"), + ), + "snapshot_id": _coalesce( + references.get("dist_lists.snapshot"), + references.get("dist_lists.snapshot_id"), + ), + "entry_id": _coalesce( + references.get("dist_lists.entry"), + references.get("dist_lists.entry_id"), + ), + } + if any(value is _CONFLICT for value in values.values()): + return None + return _Selectors(**{key: _optional(value) for key, value in values.items()}) + + +def _entry_matches(row: DistributionListEntry, selectors: _Selectors) -> bool: + resource_id = row.source_resource_id.strip() + if row.kind == "raw_email" and selectors.email: + return _normalized_email(resource_id) == selectors.email + if row.kind == "internal_mail" and selectors.account_id: + return resource_id == selectors.account_id + if row.kind == "idm_identity" and selectors.identity_id: + return resource_id == selectors.identity_id + if row.kind == "address_contact" and selectors.contact_id: + return resource_id == selectors.contact_id + return False + + +def _recipient_matches(recipient: Mapping[str, object], selectors: _Selectors) -> bool: + if ( + selectors.account_id + and _text(recipient.get("account_id")) == selectors.account_id + ): + return True + if ( + selectors.identity_id + and _text(recipient.get("identity_id")) == selectors.identity_id + ): + return True + if ( + selectors.contact_id + and _text(recipient.get("contact_id")) == selectors.contact_id + ): + return True + if not selectors.email: + return False + if _normalized_recipient_key(recipient.get("recipient_key")) == selectors.email: + return True + channels = recipient.get("channels") + if not isinstance(channels, list) or len(channels) > _MAX_CHANNELS: + return False + return any( + isinstance(channel, dict) + and ( + _normalized_email(channel.get("target")) == selectors.email + or _normalized_recipient_key(channel.get("target_key")) == selectors.email + ) + for channel in channels + ) + + +def _snapshot_recipient_record( + snapshot: DistributionListSnapshot, + *, + recipient: Mapping[str, object], + placement: str, + index: int, +) -> DsarRecordRef: + channels = recipient.get("channels") + safe_channels = [] + if isinstance(channels, list): + if len(channels) > _MAX_CHANNELS: + raise ValueError( + "Distribution Lists DSAR recipient channel limit exceeded." + ) + for channel in channels: + if not isinstance(channel, dict): + continue + safe_channels.append( + { + "channel": _bounded(channel.get("channel"), 40), + "target": _bounded(channel.get("target"), 1_000), + "target_key": _bounded(channel.get("target_key"), 1_000), + "status": _bounded(channel.get("status"), 40), + "contact_point_id": _bounded(channel.get("contact_point_id"), 120), + "locale": _bounded(channel.get("locale"), 40), + "reason_code": _bounded(channel.get("reason_code"), 120), + } + ) + return _record( + "distribution_list_snapshot_recipient", + f"{snapshot.id}:{placement}:{index}", + "frozen_recipient_projection", + "Frozen distribution-list recipient", + { + "snapshot_id": snapshot.id, + "distribution_list_id": snapshot.distribution_list_id, + "revision_number": snapshot.revision_number, + "effective_at": _iso(snapshot.effective_at), + "placement": placement, + "recipient_key": _bounded(recipient.get("recipient_key"), 1_000), + "display_name": _bounded(recipient.get("display_name"), 500), + "status": _bounded(recipient.get("status"), 40), + "account_id": _bounded(recipient.get("account_id"), 120), + "identity_id": _bounded(recipient.get("identity_id"), 120), + "contact_id": _bounded(recipient.get("contact_id"), 120), + "channels": safe_channels, + }, + observed_at=snapshot.effective_at, + immutable=True, + retention_reason=( + "Frozen recipient decisions are immutable downstream execution evidence." + ), + ) + + +def _attribution_record( + resource_type: str, + resource_id: str, + title: str, + data: dict[str, object], + observed_at: datetime | None, +) -> DsarRecordRef: + return _record( + resource_type, + resource_id, + "distribution_list_governance_attribution", + title, + data, + observed_at=observed_at, + immutable=True, + retention_reason=( + "Distribution-list lifecycle attribution is retained for accountability." + ), + ) + + +def _record( + resource_type: str, + resource_id: str, + category: str, + title: str, + data: dict[str, object], + *, + observed_at: datetime | None, + immutable: bool = False, + retention_reason: str | None = None, +) -> DsarRecordRef: + return DsarRecordRef( + provider_id="dist_lists", + module_id="dist_lists", + resource_type=resource_type, + resource_id=resource_id, + category=category, + title=title, + data=data, + observed_at=_aware(observed_at), + immutable_evidence=immutable, + retention_reason=retention_reason, + source_path="/distribution-lists", + ) + + +def _list_filter(query, model, selectors: _Selectors): + if selectors.list_id: + column = model.id if model is DistributionList else model.distribution_list_id + query = query.filter(column == selectors.list_id) + return query + + +def _limited(query, first, second): + rows = query.order_by(first, second).limit(_MAX_RECORDS + 1).all() + if len(rows) > _MAX_RECORDS: + raise ValueError( + "Distribution Lists DSAR query limit exceeded; narrow selectors." + ) + return rows + + +def _coalesce(*values: object) -> str | None | object: + normalized = {_text(value) for value in values if _text(value)} + if len(normalized) > 1: + return _CONFLICT + return next(iter(normalized), None) + + +def _coalesce_email(*values: object) -> str | None | object: + normalized = { + _normalized_email(value) for value in values if _normalized_email(value) + } + if len(normalized) > 1: + return _CONFLICT + return next(iter(normalized), None) + + +def _normalized_email(value: object) -> str | None: + text = _text(value) + if not text: + return None + _, address = parseaddr(text) + normalized = (address or text).strip().casefold() + return normalized if "@" in normalized and len(normalized) <= 320 else None + + +def _normalized_recipient_key(value: object) -> str | None: + text = _text(value) + if text and text.casefold().startswith("email:"): + return _normalized_email(text[6:]) + return None + + +def _text(value: object) -> str | None: + if value is None: + return None + normalized = str(value).strip() + return normalized or None + + +def _bounded(value: object, limit: int) -> str | None: + text = _text(value) + return text[:limit] if text else None + + +def _optional(value: object) -> str | None: + return value if isinstance(value, str) and value else None + + +def _iso(value: datetime | None) -> str | None: + aware = _aware(value) + return aware.isoformat() if aware else None + + +def _aware(value: datetime | None) -> datetime | None: + if value is None or value.tzinfo is not None: + return value + return value.replace(tzinfo=timezone.utc) + + +def _session(value: object) -> Session: + if not isinstance(value, Session): + raise TypeError("Distribution Lists DSAR requires a SQLAlchemy Session.") + return value + + +_RESOURCE_TYPES = { + "distribution_list_subject_scope", + "distribution_list_definition_entry", + "distribution_list_snapshot_recipient", + "distribution_list_actor_attribution", + "distribution_list_revision_actor_attribution", + "distribution_list_snapshot_actor_attribution", +} + + +def _validate_record(record: DsarRecordRef) -> None: + if record.provider_id != "dist_lists" or record.module_id != "dist_lists": + raise ValueError("Distribution Lists DSAR cannot plan a foreign record.") + if record.resource_type not in _RESOURCE_TYPES or not record.resource_id: + raise ValueError("Distribution Lists DSAR record identity is invalid.") + + +def _validate_action(action: DsarErasureActionRef) -> None: + if action.provider_id != "dist_lists" or action.module_id != "dist_lists": + raise ValueError("Distribution Lists DSAR cannot execute a foreign action.") + if not action.action_id.startswith("dist_lists:"): + raise ValueError("Distribution Lists DSAR action identity is invalid.") + + +__all__ = ["DIST_LISTS_DSAR_CAPABILITY", "DistributionListsDsarProvider"] diff --git a/src/govoplan_dist_lists/backend/manifest.py b/src/govoplan_dist_lists/backend/manifest.py index b8b4729..8fc9506 100644 --- a/src/govoplan_dist_lists/backend/manifest.py +++ b/src/govoplan_dist_lists/backend/manifest.py @@ -30,6 +30,7 @@ from govoplan_core.core.module_guards import ( persistent_table_uninstall_guard, ) from govoplan_core.core.modules import ( + CapabilityDocumentation, DocumentationTopic, FrontendModule, FrontendRoute, @@ -48,6 +49,10 @@ from govoplan_core.core.organizations import CAPABILITY_ORGANIZATION_DIRECTORY from govoplan_core.core.views import ViewSurface from govoplan_core.db.base import Base from govoplan_dist_lists.backend.db import models as dist_list_models +from govoplan_dist_lists.backend.dsar_provider import ( + DIST_LISTS_DSAR_CAPABILITY, + DistributionListsDsarProvider, +) MODULE_ID = "dist_lists" MODULE_NAME = "Distribution Lists" @@ -73,9 +78,21 @@ def _permission(scope: str, label: str, description: str) -> PermissionDefinitio PERMISSIONS = ( - _permission(READ_SCOPE, "View distribution lists", "Read operational distribution lists and expansion previews."), - _permission(WRITE_SCOPE, "Manage distribution lists", "Create and edit operational distribution lists."), - _permission(ADMIN_SCOPE, "Administer distribution lists", "Configure distribution-list policies and provider integrations."), + _permission( + READ_SCOPE, + "View distribution lists", + "Read operational distribution lists and expansion previews.", + ), + _permission( + WRITE_SCOPE, + "Manage distribution lists", + "Create and edit operational distribution lists.", + ), + _permission( + ADMIN_SCOPE, + "Administer distribution lists", + "Configure distribution-list policies and provider integrations.", + ), ) ROLE_TEMPLATES = ( @@ -88,6 +105,41 @@ ROLE_TEMPLATES = ( ) DOCUMENTATION = ( + DocumentationTopic( + id=f"{MODULE_ID}.data-subject-requests", + title="Distribution-list data-subject requests", + summary="Export exact recipient projections without exposing the rest of an audience.", + body=( + "Distribution Lists matches exact account, identity, contact, and normalized " + "email selectors against saved recipient entries and frozen expansion snapshots. " + "A list, snapshot, or entry reference may only narrow a verified subject match; " + "it never establishes subject identity by itself. Each frozen snapshot result " + "contains only the matching recipient and minimized channel facts, never other " + "recipients, request parameters, diagnostics, provider evidence, provenance, or " + "expansion hashes. Exact account attribution for list creation, revisions, and " + "snapshot creation is exported separately. Frozen snapshots, saved revisions, " + "and lifecycle attribution remain evidence. Removing a current recipient entry " + "or subject-scoped list requires a new revision and downstream-consumer review." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("user", "operator", "module_admin", "auditor"), + related_modules=("core", "addresses", "identity", "campaigns", "audit"), + order=90, + metadata={ + "seed": True, + "help_contexts": [ + "dist_lists.page", + "dist_lists.preview", + "privacy.data-subject-requests", + ], + "consequence_classes": { + "export_subject_recipient": "Returns only the exactly matched recipient projection.", + "review_recipient_removal": "Requires a new revision and consumer-impact review.", + "retain_frozen_snapshot": "Preserves immutable expansion and execution evidence.", + }, + }, + ), DocumentationTopic( id=f"{MODULE_ID}.boundary", title="Distribution list boundary", @@ -100,7 +152,17 @@ DOCUMENTATION = ( layer="available", documentation_types=("admin", "user"), audience=("operator", "module_admin", "product_owner"), - related_modules=("addresses", "campaigns", "mail", "postbox", "notifications", "scheduling", "poll", "workflow_engine", "tasks"), + related_modules=( + "addresses", + "campaigns", + "mail", + "postbox", + "notifications", + "scheduling", + "poll", + "workflow_engine", + "tasks", + ), metadata={ "seed": True, "help_contexts": [ @@ -163,7 +225,15 @@ DOCUMENTATION = ( layer="available", documentation_types=("admin", "user"), audience=("operator", "module_admin", "product_owner"), - related_modules=("addresses", "campaigns", "dataflow", "idm", "policy", "reporting", "workflow_engine"), + related_modules=( + "addresses", + "campaigns", + "dataflow", + "idm", + "policy", + "reporting", + "workflow_engine", + ), metadata={ "seed": True, "help_contexts": [ @@ -199,6 +269,10 @@ def _capability(context: ModuleContext): return capability(context) +def _dsar_provider(_context: ModuleContext) -> DistributionListsDsarProvider: + return DistributionListsDsarProvider() + + def _tenant_summary(session, tenant_id: str) -> dict[str, int]: return { "distribution_lists": ( @@ -269,6 +343,10 @@ manifest = ModuleManifest( name=CAPABILITY_DISTRIBUTION_LIST_WRITER, version=MODULE_VERSION, ), + ModuleInterfaceProvider( + name=DIST_LISTS_DSAR_CAPABILITY, + version="0.1.0", + ), ), requires_interfaces=( ModuleInterfaceRequirement( @@ -328,7 +406,10 @@ manifest = ModuleManifest( label="i18n:govoplan-core.product_area.communication", icon="mail", description="i18n:govoplan-core.product_area.communication_description", - surface_ids=("dist_lists.nav.distribution.lists", "dist_lists.route.distribution.lists"), + surface_ids=( + "dist_lists.nav.distribution.lists", + "dist_lists.route.distribution.lists", + ), order=40, ), ), @@ -368,6 +449,17 @@ manifest = ModuleManifest( CAPABILITY_DISTRIBUTION_LIST_SOURCE: _capability, CAPABILITY_DISTRIBUTION_LIST_EXPAND: _capability, CAPABILITY_DISTRIBUTION_LIST_WRITER: _capability, + DIST_LISTS_DSAR_CAPABILITY: _dsar_provider, + }, + capability_documentation={ + DIST_LISTS_DSAR_CAPABILITY: CapabilityDocumentation( + label="Distribution Lists data-subject request provider", + summary=( + "Exports exact recipient projections and minimized lifecycle attribution " + "without disclosing other audience members." + ), + contract_version="0.1.0", + ), }, tenant_summary_providers=(_tenant_summary,), migration_spec=MigrationSpec( @@ -403,10 +495,21 @@ manifest = ModuleManifest( maturity="vertical_slice", documentation_ref="docs/DISTRIBUTION_LISTS_ARCHITECTURE.md", test_ref="tests/test_service.py", - known_limits=("Portal self-service and every AdreMa migration path are not yet implemented.",), + known_limits=( + "Portal self-service and every AdreMa migration path are not yet implemented.", + ), supported_authority_modes=("native_authoritative", "external_mirror"), - owned_concepts=("distribution list", "distribution-list revision", "frozen expansion"), - non_owned_concepts=("contact point", "campaign recipient snapshot", "identity", "dataflow output"), + owned_concepts=( + "distribution list", + "distribution-list revision", + "frozen expansion", + ), + non_owned_concepts=( + "contact point", + "campaign recipient snapshot", + "identity", + "dataflow output", + ), recovery_docs=("docs/DISTRIBUTION_LISTS_ARCHITECTURE.md",), security_docs=("docs/DISTRIBUTION_LISTS_ARCHITECTURE.md",), operations_docs=("README.md",), diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..31299a8 --- /dev/null +++ b/tests/test_dsar_provider.py @@ -0,0 +1,282 @@ +from __future__ import annotations + +import json +import unittest +from datetime import UTC, datetime + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef +from govoplan_core.db.base import Base +from govoplan_dist_lists.backend.db.models import ( + DistributionList, + DistributionListEntry, + DistributionListRevision, + DistributionListSnapshot, +) +from govoplan_dist_lists.backend.dsar_provider import ( + DIST_LISTS_DSAR_CAPABILITY, + DistributionListsDsarProvider, +) +from govoplan_dist_lists.backend.manifest import manifest + + +NOW = datetime(2026, 8, 22, 16, 0, tzinfo=UTC) + + +def _recipient( + *, + key: str, + email: str, + account_id: str | None = None, + identity_id: str | None = None, +) -> dict[str, object]: + return { + "recipient_key": key, + "display_name": f"Recipient {email}", + "status": "usable", + "channels": [ + { + "channel": "email", + "target": email, + "target_key": f"email:{email.casefold()}", + "status": "usable", + "contact_point_id": f"contact-point-{key}", + "decision_provenance": {"secret": "do-not-export-provenance"}, + } + ], + "account_id": account_id, + "identity_id": identity_id, + "contact_id": None, + "source_entry_ids": ["entry-1"], + "attributes": {"secret": "do-not-export-attributes"}, + "provenance": {"secret": "do-not-export-recipient-provenance"}, + } + + +class DistributionListsDsarProviderTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all(self.engine) + self.session = Session(self.engine) + self.provider = DistributionListsDsarProvider() + self._seed() + self.session.commit() + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def _seed(self) -> None: + distribution_list = DistributionList( + id="list-1", + tenant_id="tenant-1", + scope_type="user", + scope_id="account-1", + name="Sensitive list name do not export", + description="Sensitive description do not export", + status="active", + current_revision_id="revision-1", + current_revision=1, + resource_revision=1, + created_by_account_id="account-1", + updated_by_account_id="account-1", + metadata_={"secret": "list-metadata-do-not-export"}, + created_at=NOW, + updated_at=NOW, + ) + revision = DistributionListRevision( + id="revision-1", + tenant_id="tenant-1", + distribution_list_id="list-1", + revision=1, + definition_kind="static", + definition_hash="definition-hash-do-not-export", + parameter_schema=[{"secret": "parameter-do-not-export"}], + constraints={"secret": "constraint-do-not-export"}, + source_fingerprints=[{"secret": "fingerprint-do-not-export"}], + created_by_account_id="account-1", + created_at=NOW, + updated_at=NOW, + ) + entry = DistributionListEntry( + id="entry-1", + tenant_id="tenant-1", + revision_id="revision-1", + entry_key="subject-email", + kind="raw_email", + mode="include", + source_provider="local", + source_resource_type="email", + source_resource_id="Subject@Example.test", + source_revision="source-revision-do-not-export", + source_fingerprint="source-fingerprint-do-not-export", + source_label="source-label-do-not-export", + source_metadata={"secret": "source-metadata-do-not-export"}, + label="entry-label-do-not-export", + purpose="service_notice", + requested_channels=["email"], + order_index=0, + configuration={"secret": "configuration-do-not-export"}, + created_at=NOW, + updated_at=NOW, + ) + snapshot = DistributionListSnapshot( + id="snapshot-1", + tenant_id="tenant-1", + distribution_list_id="list-1", + revision_id="revision-1", + revision_number=1, + idempotency_key="idempotency-key-do-not-export", + request_={"secret": "request-do-not-export"}, + expansion_hash="expansion-hash-do-not-export", + effective_at=NOW, + recipient_count=2, + excluded_count=0, + recipients=[ + _recipient( + key="identity-1", + email="subject@example.test", + account_id="account-1", + identity_id="identity-1", + ), + _recipient( + key="identity-other", + email="other-person-do-not-export@example.test", + account_id="account-other-do-not-export", + identity_id="identity-other-do-not-export", + ), + ], + excluded=[], + diagnostics=[{"secret": "diagnostics-do-not-export"}], + provider_evidence=[{"secret": "provider-evidence-do-not-export"}], + stale=False, + truncated=False, + created_by_account_id="account-1", + provenance={"secret": "snapshot-provenance-do-not-export"}, + created_at=NOW, + updated_at=NOW, + ) + other_tenant = DistributionList( + id="list-other-tenant", + tenant_id="tenant-2", + scope_type="user", + scope_id="account-1", + name="Other tenant do not export", + status="active", + current_revision_id="revision-other", + current_revision=1, + resource_revision=1, + ) + self.session.add_all( + (distribution_list, revision, entry, snapshot, other_tenant) + ) + + def test_search_exports_only_exact_subject_projection(self) -> None: + self.assertIsInstance(self.provider, DsarProvider) + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + account_id="account-1", + identity_id="identity-1", + email=" SUBJECT@example.test ", + ), + ) + self.assertEqual( + { + "distribution_list_subject_scope", + "distribution_list_definition_entry", + "distribution_list_snapshot_recipient", + "distribution_list_actor_attribution", + "distribution_list_revision_actor_attribution", + "distribution_list_snapshot_actor_attribution", + }, + {record.resource_type for record in records}, + ) + exported = json.dumps([record.to_dict() for record in records]) + self.assertIn("subject@example.test", exported.casefold()) + for excluded in ( + "other-person-do-not-export@example.test", + "account-other-do-not-export", + "identity-other-do-not-export", + "list-metadata-do-not-export", + "definition-hash-do-not-export", + "parameter-do-not-export", + "constraint-do-not-export", + "fingerprint-do-not-export", + "source-revision-do-not-export", + "source-fingerprint-do-not-export", + "source-label-do-not-export", + "source-metadata-do-not-export", + "configuration-do-not-export", + "idempotency-key-do-not-export", + "request-do-not-export", + "expansion-hash-do-not-export", + "diagnostics-do-not-export", + "provider-evidence-do-not-export", + "snapshot-provenance-do-not-export", + "do-not-export-provenance", + "do-not-export-attributes", + ): + self.assertNotIn(excluded, exported) + + def test_resource_references_only_narrow_verified_subjects(self) -> None: + self.assertEqual( + (), + self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + external_references={"dist_lists.snapshot": "snapshot-1"} + ), + ), + ) + narrowed = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + email="subject@example.test", + external_references={"dist_lists.snapshot": "snapshot-1"}, + ), + ) + self.assertEqual( + ["distribution_list_snapshot_recipient"], + [record.resource_type for record in narrowed], + ) + conflict = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + email="subject@example.test", + external_references={"dist_lists.email": "other@example.test"}, + ), + ) + self.assertEqual((), conflict) + + def test_definition_change_requires_review_and_snapshot_is_retained(self) -> None: + subject = DsarSubjectRef(email="subject@example.test") + records = self.provider.search_subject( + self.session, tenant_id="tenant-1", subject=subject + ) + actions = self.provider.plan_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + records=records, + ) + by_type = {action.resource_type: action.kind for action in actions} + self.assertEqual("manual_review", by_type["distribution_list_definition_entry"]) + self.assertEqual("retain", by_type["distribution_list_snapshot_recipient"]) + + def test_manifest_registers_provider_and_documentation(self) -> None: + self.assertIn(DIST_LISTS_DSAR_CAPABILITY, manifest.capability_factories) + self.assertIn( + "dist_lists.data-subject-requests", + {topic.id for topic in manifest.documentation}, + ) + + +if __name__ == "__main__": + unittest.main()