feat(encryption): add governed DSAR coverage

src/govoplan_encryption/backend/dsar_provider.py

@@ -0,0 +1,527 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from sqlalchemy import or_
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarRecordRef,
+    DsarSubjectRef,
+    dsar_capability_name,
+)
+from govoplan_encryption.backend.db.models import (
+    ContentProtectionRecord,
+    EncryptionKeyOperation,
+    EncryptionVault,
+    ProtectionMigration,
+    RecoveryApproval,
+    RecoveryCeremony,
+)
+
+
+ENCRYPTION_DSAR_CAPABILITY = dsar_capability_name("encryption")
+_MAX_RECORDS = 5_000
+_CONFLICT = object()
+
+
+@dataclass(frozen=True, slots=True)
+class _SubjectSelectors:
+    account_id: str
+    vault_id: str | None
+    key_operation_id: str | None
+    envelope_id: str | None
+    migration_id: str | None
+    recovery_id: str | None
+
+    @property
+    def narrowed(self) -> bool:
+        return any(
+            (
+                self.vault_id,
+                self.key_operation_id,
+                self.envelope_id,
+                self.migration_id,
+                self.recovery_id,
+            )
+        )
+
+
+class EncryptionDsarProvider:
+    provider_id = "encryption"
+    module_id = "encryption"
+
+    def search_subject(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+    ) -> Sequence[DsarRecordRef]:
+        db = _session(session)
+        selectors = _subject_selectors(subject)
+        if selectors is None:
+            return ()
+
+        records: list[DsarRecordRef] = []
+        if not selectors.narrowed or selectors.vault_id:
+            query = db.query(EncryptionVault).filter(
+                EncryptionVault.tenant_id == tenant_id,
+                or_(
+                    EncryptionVault.created_by == selectors.account_id,
+                    EncryptionVault.updated_by == selectors.account_id,
+                ),
+            )
+            if selectors.vault_id:
+                query = query.filter(EncryptionVault.vault_id == selectors.vault_id)
+            records.extend(
+                _vault_attribution(row, selectors.account_id)
+                for row in _limited(
+                    query,
+                    EncryptionVault.created_at,
+                    EncryptionVault.id,
+                    label="vault attribution",
+                )
+            )
+
+        if not selectors.narrowed or selectors.vault_id or selectors.key_operation_id:
+            query = db.query(EncryptionKeyOperation).filter(
+                EncryptionKeyOperation.tenant_id == tenant_id,
+                EncryptionKeyOperation.requested_by == selectors.account_id,
+            )
+            if selectors.vault_id:
+                query = query.filter(
+                    EncryptionKeyOperation.vault_id == selectors.vault_id
+                )
+            if selectors.key_operation_id:
+                query = query.filter(
+                    EncryptionKeyOperation.id == selectors.key_operation_id
+                )
+            records.extend(
+                _key_operation_attribution(row)
+                for row in _limited(
+                    query,
+                    EncryptionKeyOperation.created_at,
+                    EncryptionKeyOperation.id,
+                    label="key-operation attribution",
+                )
+            )
+
+        if not selectors.narrowed or selectors.vault_id or selectors.envelope_id:
+            query = db.query(ContentProtectionRecord).filter(
+                ContentProtectionRecord.tenant_id == tenant_id,
+                ContentProtectionRecord.registered_by == selectors.account_id,
+            )
+            if selectors.vault_id:
+                query = query.filter(
+                    ContentProtectionRecord.vault_id == selectors.vault_id
+                )
+            if selectors.envelope_id:
+                query = query.filter(
+                    ContentProtectionRecord.envelope_id == selectors.envelope_id
+                )
+            records.extend(
+                _protection_attribution(row)
+                for row in _limited(
+                    query,
+                    ContentProtectionRecord.created_at,
+                    ContentProtectionRecord.id,
+                    label="content-protection attribution",
+                )
+            )
+
+        if not selectors.narrowed or selectors.vault_id or selectors.migration_id:
+            query = db.query(ProtectionMigration).filter(
+                ProtectionMigration.tenant_id == tenant_id,
+                ProtectionMigration.requested_by == selectors.account_id,
+            )
+            if selectors.vault_id:
+                query = query.filter(
+                    ProtectionMigration.target_vault_id == selectors.vault_id
+                )
+            if selectors.migration_id:
+                query = query.filter(ProtectionMigration.id == selectors.migration_id)
+            records.extend(
+                _migration_attribution(row)
+                for row in _limited(
+                    query,
+                    ProtectionMigration.created_at,
+                    ProtectionMigration.id,
+                    label="migration attribution",
+                )
+            )
+
+        if not selectors.narrowed or selectors.vault_id or selectors.recovery_id:
+            ceremonies = db.query(RecoveryCeremony).filter(
+                RecoveryCeremony.tenant_id == tenant_id,
+                RecoveryCeremony.requester_account_id == selectors.account_id,
+            )
+            if selectors.vault_id:
+                ceremonies = ceremonies.filter(
+                    RecoveryCeremony.vault_id == selectors.vault_id
+                )
+            if selectors.recovery_id:
+                ceremonies = ceremonies.filter(
+                    RecoveryCeremony.id == selectors.recovery_id
+                )
+            records.extend(
+                _recovery_request_attribution(row)
+                for row in _limited(
+                    ceremonies,
+                    RecoveryCeremony.created_at,
+                    RecoveryCeremony.id,
+                    label="recovery-request attribution",
+                )
+            )
+
+            approvals = db.query(RecoveryApproval).filter(
+                RecoveryApproval.tenant_id == tenant_id,
+                RecoveryApproval.approver_account_id == selectors.account_id,
+            )
+            if selectors.recovery_id:
+                approvals = approvals.filter(
+                    RecoveryApproval.recovery_id == selectors.recovery_id
+                )
+            elif selectors.vault_id:
+                approvals = approvals.filter(
+                    RecoveryApproval.recovery_id.in_(
+                        db.query(RecoveryCeremony.id).filter(
+                            RecoveryCeremony.tenant_id == tenant_id,
+                            RecoveryCeremony.vault_id == selectors.vault_id,
+                        )
+                    )
+                )
+            records.extend(
+                _recovery_approval_attribution(row)
+                for row in _limited(
+                    approvals,
+                    RecoveryApproval.created_at,
+                    RecoveryApproval.id,
+                    label="recovery-approval attribution",
+                )
+            )
+
+        if len(records) > _MAX_RECORDS:
+            raise ValueError("Encryption DSAR result limit exceeded; narrow 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)
+        if _subject_selectors(subject) is None:
+            raise ValueError("Encryption DSAR subject selectors conflict.")
+        actions: list[DsarErasureActionRef] = []
+        for record in records:
+            _validate_record(record)
+            actions.append(
+                DsarErasureActionRef(
+                    action_id=(
+                        f"encryption:retain:{record.resource_type}:{record.resource_id}"
+                    ),
+                    provider_id=self.provider_id,
+                    module_id=self.module_id,
+                    kind="retain",
+                    resource_type=record.resource_type,
+                    resource_id=record.resource_id,
+                    title=f"Retain {record.title}",
+                    rationale=(
+                        record.retention_reason
+                        or "Cryptographic custody evidence remains immutable."
+                    ),
+                    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)
+        if _subject_selectors(subject) is None:
+            raise ValueError("Encryption DSAR subject selectors conflict.")
+        results: list[DsarExecutionResultRef] = []
+        for action in actions:
+            _validate_action(action)
+            if action.executable or action.kind != "retain":
+                raise ValueError("Encryption DSAR publishes retain actions only.")
+            results.append(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status="blocked",
+                    summary=(
+                        "Cryptographic lifecycle and custody attribution remains "
+                        "immutable security evidence."
+                    ),
+                    evidence={"request_id": request_id},
+                )
+            )
+        return tuple(results)
+
+
+def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
+    references = subject.external_references
+    account = _coalesce(
+        subject.account_id,
+        references.get("encryption.account"),
+        references.get("access.account"),
+    )
+    values = {
+        "vault_id": _coalesce(
+            references.get("encryption.vault"),
+            references.get("encryption.vault_id"),
+        ),
+        "key_operation_id": _coalesce(
+            references.get("encryption.key_operation"),
+            references.get("encryption.key_operation_id"),
+        ),
+        "envelope_id": _coalesce(
+            references.get("encryption.envelope"),
+            references.get("encryption.envelope_id"),
+        ),
+        "migration_id": _coalesce(
+            references.get("encryption.migration"),
+            references.get("encryption.migration_id"),
+        ),
+        "recovery_id": _coalesce(
+            references.get("encryption.recovery"),
+            references.get("encryption.recovery_id"),
+        ),
+    }
+    if account is _CONFLICT or any(value is _CONFLICT for value in values.values()):
+        return None
+    account_id = _optional_string(account)
+    if not account_id:
+        return None
+    return _SubjectSelectors(
+        account_id=account_id,
+        vault_id=_optional_string(values["vault_id"]),
+        key_operation_id=_optional_string(values["key_operation_id"]),
+        envelope_id=_optional_string(values["envelope_id"]),
+        migration_id=_optional_string(values["migration_id"]),
+        recovery_id=_optional_string(values["recovery_id"]),
+    )
+
+
+def _vault_attribution(row: EncryptionVault, account_id: str) -> DsarRecordRef:
+    activities = []
+    if row.created_by == account_id:
+        activities.append("created_vault")
+    if row.updated_by == account_id:
+        activities.append("updated_vault")
+    return _record(
+        resource_type="vault_actor_attribution",
+        resource_id=row.id,
+        title="Encryption vault actor attribution",
+        data={
+            "vault_id": row.vault_id,
+            "purpose": row.purpose[:255],
+            "profile_kind": row.profile_kind,
+            "scope_type": row.scope_type,
+            "state": row.state,
+            "revision": row.revision,
+            "current_key_version": row.current_key_version,
+            "recovery_quorum": row.recovery_quorum,
+            "activities": activities,
+            "created_at": _iso(row.created_at),
+            "updated_at": _iso(row.updated_at),
+        },
+        observed_at=row.updated_at,
+    )
+
+
+def _key_operation_attribution(row: EncryptionKeyOperation) -> DsarRecordRef:
+    return _record(
+        resource_type="key_operation_actor_attribution",
+        resource_id=row.id,
+        title="Encryption key-operation actor attribution",
+        data={
+            "key_operation_id": row.id,
+            "vault_id": row.vault_id,
+            "key_version": row.key_version,
+            "operation": row.operation,
+            "state": row.state,
+            "activity": "requested_key_operation",
+            "created_at": _iso(row.created_at),
+            "completed_at": _iso(row.completed_at),
+        },
+        observed_at=row.completed_at or row.created_at,
+    )
+
+
+def _protection_attribution(row: ContentProtectionRecord) -> DsarRecordRef:
+    return _record(
+        resource_type="content_protection_actor_attribution",
+        resource_id=row.id,
+        title="Content-protection registration attribution",
+        data={
+            "protection_record_id": row.id,
+            "owner_module": row.owner_module,
+            "resource_type": row.resource_type,
+            "profile_kind": row.profile_kind,
+            "state": row.state,
+            "activity": "registered_content_protection",
+            "created_at": _iso(row.created_at),
+            "updated_at": _iso(row.updated_at),
+        },
+        observed_at=row.updated_at,
+    )
+
+
+def _migration_attribution(row: ProtectionMigration) -> DsarRecordRef:
+    return _record(
+        resource_type="protection_migration_actor_attribution",
+        resource_id=row.id,
+        title="Protection-migration actor attribution",
+        data={
+            "migration_id": row.id,
+            "mode": row.mode,
+            "state": row.state,
+            "activity": "requested_protection_migration",
+            "created_at": _iso(row.created_at),
+            "completed_at": _iso(row.completed_at),
+        },
+        observed_at=row.completed_at or row.created_at,
+    )
+
+
+def _recovery_request_attribution(row: RecoveryCeremony) -> DsarRecordRef:
+    return _record(
+        resource_type="recovery_request_actor_attribution",
+        resource_id=row.id,
+        title="Encryption recovery-request attribution",
+        data={
+            "recovery_id": row.id,
+            "vault_id": row.vault_id,
+            "state": row.state,
+            "quorum": row.quorum,
+            "revision": row.revision,
+            "expires_at": _iso(row.expires_at),
+            "activity": "requested_recovery_ceremony",
+            "created_at": _iso(row.created_at),
+            "updated_at": _iso(row.updated_at),
+        },
+        observed_at=row.updated_at,
+    )
+
+
+def _recovery_approval_attribution(row: RecoveryApproval) -> DsarRecordRef:
+    return _record(
+        resource_type="recovery_approval_actor_attribution",
+        resource_id=row.id,
+        title="Encryption recovery-approval attribution",
+        data={
+            "recovery_approval_id": row.id,
+            "recovery_id": row.recovery_id,
+            "decision": row.decision,
+            "activity": "decided_recovery_approval",
+            "created_at": _iso(row.created_at),
+        },
+        observed_at=row.created_at,
+    )
+
+
+def _record(
+    *,
+    resource_type: str,
+    resource_id: str,
+    title: str,
+    data: dict[str, object],
+    observed_at: datetime | None,
+) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="encryption",
+        module_id="encryption",
+        resource_type=resource_type,
+        resource_id=resource_id,
+        category="cryptographic_custody_attribution",
+        title=title,
+        data=data,
+        observed_at=_aware(observed_at),
+        immutable_evidence=True,
+        retention_reason=(
+            "Cryptographic lifecycle and custody attribution is retained as immutable "
+            "security and accountability evidence."
+        ),
+    )
+
+
+def _limited(query, first, second, *, label: str):
+    rows = query.order_by(first, second).limit(_MAX_RECORDS + 1).all()
+    if len(rows) > _MAX_RECORDS:
+        raise ValueError(f"Encryption DSAR {label} limit exceeded; narrow selectors.")
+    return rows
+
+
+def _coalesce(*values: str | None) -> str | None | object:
+    normalized = {str(value).strip() for value in values if str(value or "").strip()}
+    if len(normalized) > 1:
+        return _CONFLICT
+    return next(iter(normalized), None)
+
+
+def _optional_string(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("Encryption DSAR requires a SQLAlchemy Session.")
+    return value
+
+
+_RESOURCE_TYPES = {
+    "vault_actor_attribution",
+    "key_operation_actor_attribution",
+    "content_protection_actor_attribution",
+    "protection_migration_actor_attribution",
+    "recovery_request_actor_attribution",
+    "recovery_approval_actor_attribution",
+}
+
+
+def _validate_record(record: DsarRecordRef) -> None:
+    if record.provider_id != "encryption" or record.module_id != "encryption":
+        raise ValueError("Encryption DSAR cannot plan a foreign provider record.")
+    if record.resource_type not in _RESOURCE_TYPES or not record.resource_id:
+        raise ValueError("Encryption DSAR record identity is invalid.")
+
+
+def _validate_action(action: DsarErasureActionRef) -> None:
+    if action.provider_id != "encryption" or action.module_id != "encryption":
+        raise ValueError("Encryption DSAR cannot execute a foreign provider action.")
+    if not action.action_id.startswith("encryption:retain:"):
+        raise ValueError("Encryption DSAR action identity is invalid.")
+
+
+__all__ = ["ENCRYPTION_DSAR_CAPABILITY", "EncryptionDsarProvider"]

src/govoplan_encryption/backend/manifest.py

@@ -32,6 +32,10 @@ from govoplan_core.core.modules import (
 from govoplan_core.core.provider_governance import declared_module_architecture
 from govoplan_core.db.base import Base
 from govoplan_encryption.backend.db import models
+from govoplan_encryption.backend.dsar_provider import (
+    ENCRYPTION_DSAR_CAPABILITY,
+    EncryptionDsarProvider,
+)
 from govoplan_encryption.backend.local_provider import (
     LOCAL_PROVIDER_ID,
     LocalAesGcmProvider,
@@ -88,6 +92,10 @@ def _local_provider(context: ModuleContext) -> LocalAesGcmProvider:
     return LocalAesGcmProvider(getattr(context.settings, "master_key_b64", None))
 
 
+def _dsar_provider(_context: ModuleContext) -> EncryptionDsarProvider:
+    return EncryptionDsarProvider()
+
+
 def _disable_guard(
     session: object | None,
     _module_id: str,
@@ -188,6 +196,7 @@ manifest = ModuleManifest(
             name=CAPABILITY_ENCRYPTION_DISABLE_PREFLIGHT,
             version="1.0.0",
         ),
+        ModuleInterfaceProvider(name=ENCRYPTION_DSAR_CAPABILITY, version="0.1.0"),
     ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
@@ -235,6 +244,7 @@ manifest = ModuleManifest(
         CAPABILITY_ENCRYPTION_CONTENT_CIPHER: _service,
         CAPABILITY_ENCRYPTION_RECOVERY: _service,
         CAPABILITY_ENCRYPTION_DISABLE_PREFLIGHT: _service,
+        ENCRYPTION_DSAR_CAPABILITY: _dsar_provider,
         f"{CAPABILITY_ENCRYPTION_KEY_MATERIAL_PROVIDER_PREFIX}{LOCAL_PROVIDER_ID}": _local_provider,
         f"{CAPABILITY_ENCRYPTION_CONTENT_CIPHER_PROVIDER_PREFIX}{LOCAL_PROVIDER_ID}": _local_provider,
     },
@@ -281,6 +291,14 @@ manifest = ModuleManifest(
             ),
             contract_version="1.0.0",
         ),
+        ENCRYPTION_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Encryption data-subject request provider",
+            summary=(
+                "Exports minimized cryptographic custody attribution without key "
+                "material, ciphertext references, or protected evidence."
+            ),
+            contract_version="0.1.0",
+        ),
     },
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,
@@ -322,6 +340,50 @@ manifest = ModuleManifest(
         ),
     ),
     documentation=(
+        DocumentationTopic(
+            id="encryption.data-subject-requests",
+            title="Encryption data-subject requests",
+            summary=(
+                "Export cryptographic custody participation without exposing protected "
+                "content or key material."
+            ),
+            body=(
+                "Encryption correlates only an exact tenant account identifier and can "
+                "narrow an already verified search to a vault, key operation, envelope, "
+                "migration, or recovery ceremony. The export reports minimized vault "
+                "administration, key-operation requests, protection registrations, "
+                "migrations, and recovery participation. It never returns provider or "
+                "public key references, wrapped keys, nonces, ciphertext locations, "
+                "resource identifiers, digests, request payloads, assurance and policy "
+                "references, recovery reasons, idempotency keys, or provenance. Feature "
+                "modules remain responsible for exporting the plaintext semantics of "
+                "their own protected resources. Cryptographic lifecycle and custody "
+                "attribution remains immutable security evidence and is retained rather "
+                "than automatically erased."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("user", "administrator", "security_officer", "auditor"),
+            related_modules=("core", "identity_trust", "audit", "policy"),
+            order=95,
+            metadata={
+                "help_contexts": [
+                    "encryption.admin.operations",
+                    "privacy.data-subject-requests",
+                ],
+                "consequence_classes": {
+                    "export_custody_attribution": (
+                        "Returns minimized lifecycle activity for the exact account."
+                    ),
+                    "exclude_cryptographic_secrets": (
+                        "Never returns key material, ciphertext references, nonces, or protected evidence."
+                    ),
+                    "retain_cryptographic_evidence": (
+                        "Preserves immutable custody and recovery accountability."
+                    ),
+                },
+            },
+        ),
         DocumentationTopic(
             id="encryption.boundary",
             title="Encryption and key-custody boundary",

tests/test_dsar_provider.py

@@ -0,0 +1,372 @@
+from __future__ import annotations
+
+import json
+import unittest
+from datetime import UTC, datetime, timedelta
+
+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_core.privacy.dsar_workflow import (
+    create_data_subject_request,
+    search_data_subject_request,
+)
+from govoplan_encryption.backend.db.models import (
+    ContentProtectionRecord,
+    EncryptionKeyOperation,
+    EncryptionVault,
+    ProtectionMigration,
+    RecoveryApproval,
+    RecoveryCeremony,
+)
+from govoplan_encryption.backend.dsar_provider import (
+    ENCRYPTION_DSAR_CAPABILITY,
+    EncryptionDsarProvider,
+)
+from govoplan_encryption.backend.manifest import manifest
+
+
+NOW = datetime(2026, 8, 22, 11, 0, tzinfo=UTC)
+
+
+class _Registry:
+    def __init__(self, provider: EncryptionDsarProvider) -> None:
+        self.provider = provider
+
+    def capability_names(self):
+        return (ENCRYPTION_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        if name != ENCRYPTION_DSAR_CAPABILITY:
+            raise KeyError(name)
+        return "encryption"
+
+    def tenant_entitlement_resolver(self):
+        class _Resolver:
+            @staticmethod
+            def resolve(session, tenant_id):
+                del session, tenant_id
+                return type("State", (), {"effective_modules": ("encryption",)})()
+
+        return _Resolver()
+
+    def require_tenant_capability(self, name, session, **kwargs):
+        del session, kwargs
+        if name != ENCRYPTION_DSAR_CAPABILITY:
+            raise KeyError(name)
+        return self.provider
+
+    def manifests(self):
+        return (type("Manifest", (), {"id": "encryption"})(),)
+
+
+class EncryptionDsarProviderTests(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 = EncryptionDsarProvider()
+        self.assertIsInstance(self.provider, DsarProvider)
+        self._seed()
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def _seed(self) -> None:
+        self.session.add_all(
+            (
+                EncryptionVault(
+                    id="vault-row-1",
+                    tenant_id="tenant-1",
+                    vault_id="vault-1",
+                    name="Resident postbox vault",
+                    provider_id="provider-secret-do-not-export",
+                    purpose="resident correspondence",
+                    profile_kind="institutional",
+                    scope_type="tenant",
+                    scope_id="scope-secret-do-not-export",
+                    policy_ref="policy-ref-do-not-export",
+                    recovery_quorum=2,
+                    state="active",
+                    revision=2,
+                    current_key_version=3,
+                    create_idempotency_key="vault-idempotency-do-not-export",
+                    create_request_digest="vault-digest-do-not-export",
+                    provenance={"secret": "vault-provenance-do-not-export"},
+                    created_by="account-1",
+                    updated_by="account-1",
+                    created_at=NOW,
+                    updated_at=NOW,
+                ),
+                EncryptionKeyOperation(
+                    id="key-operation-1",
+                    tenant_id="tenant-1",
+                    vault_id="vault-1",
+                    key_version=3,
+                    operation="rotate",
+                    provider_id="provider-secret-do-not-export",
+                    state="complete",
+                    idempotency_key="key-idempotency-do-not-export",
+                    request_digest="key-request-digest-do-not-export",
+                    request_payload={"secret": "request-payload-do-not-export"},
+                    error_code="provider-error-do-not-export",
+                    policy_decision_ref="policy-decision-do-not-export",
+                    assurance_evidence_ref="assurance-evidence-do-not-export",
+                    requested_by="account-1",
+                    completed_at=NOW,
+                    created_at=NOW,
+                    updated_at=NOW,
+                ),
+                ContentProtectionRecord(
+                    id="protection-row-1",
+                    envelope_id="envelope-1",
+                    tenant_id="tenant-1",
+                    owner_module="postbox",
+                    resource_type="message",
+                    resource_id="message-resource-do-not-export",
+                    profile_kind="institutional",
+                    profile_id="profile-secret-do-not-export",
+                    provider_id="provider-secret-do-not-export",
+                    vault_id="vault-1",
+                    key_version=3,
+                    algorithm_suite="AES-256-GCM",
+                    ciphertext_ref="ciphertext-ref-do-not-export",
+                    ciphertext_digest="ciphertext-digest-do-not-export",
+                    authenticated_context_digest="context-digest-do-not-export",
+                    wrapped_key_refs=["wrapped-key-ref-do-not-export"],
+                    state="active",
+                    source_envelope_id="source-envelope-do-not-export",
+                    migration_id="migration-secret-do-not-export",
+                    envelope_metadata={"secret": "envelope-metadata-do-not-export"},
+                    idempotency_key="envelope-idempotency-do-not-export",
+                    request_digest="envelope-request-digest-do-not-export",
+                    policy_decision_ref="envelope-policy-do-not-export",
+                    registered_by="account-1",
+                    created_at=NOW,
+                    updated_at=NOW,
+                ),
+                ProtectionMigration(
+                    id="migration-1",
+                    tenant_id="tenant-1",
+                    source_envelope_id="source-envelope-ref-do-not-export",
+                    target_envelope_id="target-envelope-ref-do-not-export",
+                    target_provider_id="target-provider-do-not-export",
+                    target_vault_id="vault-1",
+                    target_key_version=4,
+                    target_algorithm_suite="AES-256-GCM",
+                    mode="rewrap",
+                    state="complete",
+                    policy_decision_ref="migration-policy-do-not-export",
+                    assurance_evidence_ref="migration-assurance-do-not-export",
+                    idempotency_key="migration-idempotency-do-not-export",
+                    request_digest="migration-request-digest-do-not-export",
+                    evidence_refs=["migration-evidence-do-not-export"],
+                    error_code="migration-error-do-not-export",
+                    provenance={"secret": "migration-provenance-do-not-export"},
+                    requested_by="account-1",
+                    completed_at=NOW,
+                    created_at=NOW,
+                    updated_at=NOW,
+                ),
+                RecoveryCeremony(
+                    id="recovery-1",
+                    tenant_id="tenant-1",
+                    vault_id="vault-1",
+                    state="approved",
+                    requested_scope="scope-secret-do-not-export",
+                    reason="recovery-reason-do-not-export",
+                    quorum=2,
+                    revision=2,
+                    policy_decision_ref="recovery-policy-do-not-export",
+                    requester_assurance_ref="requester-assurance-do-not-export",
+                    requester_account_id="account-1",
+                    idempotency_key="recovery-idempotency-do-not-export",
+                    request_digest="recovery-request-digest-do-not-export",
+                    expires_at=NOW + timedelta(hours=1),
+                    execution_ref="execution-ref-do-not-export",
+                    provenance={"secret": "recovery-provenance-do-not-export"},
+                    created_at=NOW,
+                    updated_at=NOW,
+                ),
+                RecoveryApproval(
+                    id="recovery-approval-1",
+                    tenant_id="tenant-1",
+                    recovery_id="recovery-1",
+                    approver_account_id="account-1",
+                    decision="approve",
+                    reason="approval-reason-do-not-export",
+                    assurance_evidence_ref="approval-assurance-do-not-export",
+                    idempotency_key="approval-idempotency-do-not-export",
+                    request_digest="approval-request-digest-do-not-export",
+                    created_at=NOW,
+                    updated_at=NOW,
+                ),
+                EncryptionVault(
+                    id="vault-row-other",
+                    tenant_id="tenant-2",
+                    vault_id="vault-other",
+                    name="Other tenant vault",
+                    provider_id="local",
+                    purpose="other",
+                    profile_kind="institutional",
+                    scope_type="tenant",
+                    policy_ref="policy",
+                    recovery_quorum=2,
+                    state="active",
+                    revision=1,
+                    create_idempotency_key="other-key",
+                    create_request_digest="other-digest",
+                    provenance={},
+                    created_by="account-1",
+                    updated_by="account-1",
+                ),
+            )
+        )
+
+    @staticmethod
+    def _subject() -> DsarSubjectRef:
+        return DsarSubjectRef(account_id="account-1")
+
+    def test_search_exports_minimized_custody_attribution(self) -> None:
+        records = self.provider.search_subject(
+            self.session, tenant_id="tenant-1", subject=self._subject()
+        )
+        self.assertEqual(
+            {
+                "vault_actor_attribution",
+                "key_operation_actor_attribution",
+                "content_protection_actor_attribution",
+                "protection_migration_actor_attribution",
+                "recovery_request_actor_attribution",
+                "recovery_approval_actor_attribution",
+            },
+            {record.resource_type for record in records},
+        )
+        exported = json.dumps([record.to_dict() for record in records])
+        for excluded in (
+            "provider-secret-do-not-export",
+            "scope-secret-do-not-export",
+            "policy-ref-do-not-export",
+            "vault-idempotency-do-not-export",
+            "vault-digest-do-not-export",
+            "vault-provenance-do-not-export",
+            "key-idempotency-do-not-export",
+            "key-request-digest-do-not-export",
+            "request-payload-do-not-export",
+            "provider-error-do-not-export",
+            "policy-decision-do-not-export",
+            "assurance-evidence-do-not-export",
+            "message-resource-do-not-export",
+            "profile-secret-do-not-export",
+            "ciphertext-ref-do-not-export",
+            "ciphertext-digest-do-not-export",
+            "context-digest-do-not-export",
+            "wrapped-key-ref-do-not-export",
+            "source-envelope-do-not-export",
+            "envelope-metadata-do-not-export",
+            "target-provider-do-not-export",
+            "migration-policy-do-not-export",
+            "migration-assurance-do-not-export",
+            "migration-evidence-do-not-export",
+            "recovery-reason-do-not-export",
+            "requester-assurance-do-not-export",
+            "execution-ref-do-not-export",
+            "approval-reason-do-not-export",
+            "approval-assurance-do-not-export",
+        ):
+            self.assertNotIn(excluded, exported)
+
+    def test_requires_exact_account_and_enforces_tenant(self) -> None:
+        email_only = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(email="custodian@example.test"),
+        )
+        conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"encryption.account": "account-other"},
+            ),
+        )
+        records = self.provider.search_subject(
+            self.session, tenant_id="tenant-1", subject=self._subject()
+        )
+        self.assertEqual((), email_only)
+        self.assertEqual((), conflict)
+        self.assertNotIn("vault-row-other", {record.resource_id for record in records})
+
+    def test_recovery_narrowing_includes_request_and_approval_only(self) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"encryption.recovery": "recovery-1"},
+            ),
+        )
+        self.assertEqual(
+            {
+                "recovery_request_actor_attribution",
+                "recovery_approval_actor_attribution",
+            },
+            {record.resource_type for record in records},
+        )
+
+    def test_erasure_retains_cryptographic_custody_evidence(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(actions)
+        self.assertTrue(
+            all(action.kind == "retain" and not action.executable for action in actions)
+        )
+        results = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self._subject(),
+            actions=actions,
+            request_id="dsar-encryption-1",
+        )
+        self.assertTrue(all(result.status == "blocked" for result in results))
+
+    def test_manifest_and_core_workflow_discover_provider(self) -> None:
+        self.assertIn(ENCRYPTION_DSAR_CAPABILITY, manifest.capability_factories)
+        self.assertIn(
+            "encryption.data-subject-requests",
+            {topic.id for topic in manifest.documentation},
+        )
+        row = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-ENCRYPTION-1",
+            request_kind="access",
+            subject=self._subject(),
+            purpose="Encryption custody access request",
+            legal_basis=None,
+            due_at=None,
+            requested_by_account_id="operator-1",
+        )
+        search_data_subject_request(
+            self.session,
+            registry=_Registry(self.provider),
+            row=row,
+            expected_revision=row.resource_revision,
+        )
+        self.assertEqual("searched", row.status)
+        self.assertEqual(6, row.search_result["record_count"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_manifest.py

@@ -12,6 +12,7 @@ from govoplan_core.core.encryption import (
     CAPABILITY_ENCRYPTION_RECOVERY,
 )
 from govoplan_encryption.backend.manifest import get_manifest
+from govoplan_encryption.backend.dsar_provider import ENCRYPTION_DSAR_CAPABILITY
 from govoplan_encryption.backend.local_provider import LOCAL_PROVIDER_ID
 
 
@@ -28,6 +29,7 @@ class EncryptionManifestTests(unittest.TestCase):
                 CAPABILITY_ENCRYPTION_CONTENT_CIPHER,
                 CAPABILITY_ENCRYPTION_RECOVERY,
                 CAPABILITY_ENCRYPTION_DISABLE_PREFLIGHT,
+                ENCRYPTION_DSAR_CAPABILITY,
                 f"{CAPABILITY_ENCRYPTION_KEY_MATERIAL_PROVIDER_PREFIX}{LOCAL_PROVIDER_ID}",
                 f"{CAPABILITY_ENCRYPTION_CONTENT_CIPHER_PROVIDER_PREFIX}{LOCAL_PROVIDER_ID}",
             },