feat(dist-lists): add governed DSAR coverage

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"]

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(
-    _permission(WRITE_SCOPE, "Manage distribution lists", "Create and edit operational distribution lists."),
+        READ_SCOPE,
-    _permission(ADMIN_SCOPE, "Administer distribution lists", "Configure distribution-list policies and provider integrations."),
+        "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"),
+        owned_concepts=(
-        non_owned_concepts=("contact point", "campaign recipient snapshot", "identity", "dataflow output"),
+            "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",),

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()