diff --git a/src/govoplan_encryption/backend/dsar_provider.py b/src/govoplan_encryption/backend/dsar_provider.py new file mode 100644 index 0000000..aff984b --- /dev/null +++ b/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"] diff --git a/src/govoplan_encryption/backend/manifest.py b/src/govoplan_encryption/backend/manifest.py index dbb8cc0..661fbc2 100644 --- a/src/govoplan_encryption/backend/manifest.py +++ b/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", diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..c30f38f --- /dev/null +++ b/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() diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 323cc66..ff6ed7d 100644 --- a/tests/test_manifest.py +++ b/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}", },