From 8d1ca493bbff1802d6611f0decb5c62ae5c56200 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 21 Aug 2026 12:01:18 +0200 Subject: [PATCH] feat(identity-trust): add governed DSAR coverage --- .../backend/dsar_provider.py | 461 ++++++++++++++++++ .../backend/manifest.py | 62 +++ tests/test_dsar_provider.py | 297 +++++++++++ 3 files changed, 820 insertions(+) create mode 100644 src/govoplan_identity_trust/backend/dsar_provider.py create mode 100644 tests/test_dsar_provider.py diff --git a/src/govoplan_identity_trust/backend/dsar_provider.py b/src/govoplan_identity_trust/backend/dsar_provider.py new file mode 100644 index 0000000..16956f1 --- /dev/null +++ b/src/govoplan_identity_trust/backend/dsar_provider.py @@ -0,0 +1,461 @@ +from __future__ import annotations + +from collections.abc import Mapping, 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_identity_trust.backend.db.models import ( + AssuranceEvidence, + DevicePublicKey, + KeyAccessDecisionRecord, + TrustKeyEpoch, +) + + +IDENTITY_TRUST_DSAR_CAPABILITY = dsar_capability_name("identity_trust") +_MAX_RECORDS = 5_000 +_CONFLICT = object() + + +@dataclass(frozen=True, slots=True) +class _SubjectSelectors: + account_id: str | None + identity_id: str | None + device_id: str | None + key_id: str | None + + +class IdentityTrustDsarProvider: + provider_id = "identity_trust" + module_id = "identity_trust" + + 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 () + + devices = db.query(DevicePublicKey).filter( + DevicePublicKey.tenant_id == tenant_id + ) + if selectors.account_id: + devices = devices.filter( + DevicePublicKey.account_id == selectors.account_id + ) + if selectors.identity_id: + devices = devices.filter( + DevicePublicKey.identity_id == selectors.identity_id + ) + if selectors.device_id: + devices = devices.filter(DevicePublicKey.device_id == selectors.device_id) + if selectors.key_id: + devices = devices.filter(DevicePublicKey.key_id == selectors.key_id) + records: list[DsarRecordRef] = [ + _device_record(row) + for row in _limited( + devices, + DevicePublicKey.registered_at, + DevicePublicKey.id, + label="device key", + ) + ] + + if selectors.account_id: + assurances = db.query(AssuranceEvidence).filter( + AssuranceEvidence.tenant_id == tenant_id, + AssuranceEvidence.account_id == selectors.account_id, + ) + if selectors.key_id: + assurances = assurances.filter( + AssuranceEvidence.device_key_id == selectors.key_id + ) + records.extend( + _assurance_record(row) + for row in _limited( + assurances, + AssuranceEvidence.verified_at, + AssuranceEvidence.id, + label="assurance evidence", + ) + ) + decisions = db.query(KeyAccessDecisionRecord).filter( + KeyAccessDecisionRecord.tenant_id == tenant_id, + KeyAccessDecisionRecord.account_id == selectors.account_id, + ) + if selectors.key_id: + decisions = decisions.filter( + KeyAccessDecisionRecord.device_key_id == selectors.key_id + ) + records.extend( + _access_decision_record(row) + for row in _limited( + decisions, + KeyAccessDecisionRecord.created_at, + KeyAccessDecisionRecord.id, + label="key-access decision", + ) + ) + + subject_conditions = [] + if selectors.device_id: + subject_conditions.append( + (TrustKeyEpoch.subject_kind == "device") + & (TrustKeyEpoch.subject_id == selectors.device_id) + ) + elif not selectors.key_id: + if selectors.account_id: + subject_conditions.append( + (TrustKeyEpoch.subject_kind == "account") + & (TrustKeyEpoch.subject_id == selectors.account_id) + ) + if selectors.identity_id: + subject_conditions.append( + (TrustKeyEpoch.subject_kind == "identity") + & (TrustKeyEpoch.subject_id == selectors.identity_id) + ) + if subject_conditions: + epochs = db.query(TrustKeyEpoch).filter( + TrustKeyEpoch.tenant_id == tenant_id, + or_(*subject_conditions), + ) + records.extend( + _epoch_record(row) + for row in _limited( + epochs, + TrustKeyEpoch.effective_at, + TrustKeyEpoch.id, + label="key epoch", + ) + ) + + if len(records) > _MAX_RECORDS: + raise ValueError( + "Identity Trust DSAR result limit exceeded; narrow the 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("Identity Trust DSAR subject selectors conflict.") + actions: list[DsarErasureActionRef] = [] + for record in records: + _validate_record(record) + device = record.resource_type == "device_public_key" + actions.append( + DsarErasureActionRef( + action_id=( + f"identity_trust:{'manual_review' if device else 'retain'}:" + f"{record.resource_type}:{record.resource_id}" + ), + provider_id=self.provider_id, + module_id=self.module_id, + kind="manual_review" if device else "retain", + resource_type=record.resource_type, + resource_id=record.resource_id, + title=("Review " if device else "Retain ") + record.title, + rationale=( + "Revocation or removal must account for encrypted-resource " + "recovery, active sessions, and the current key epoch." + if device + else record.retention_reason + or "Trust and access 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("Identity Trust DSAR subject selectors conflict.") + results: list[DsarExecutionResultRef] = [] + for action in actions: + _validate_action(action) + if action.executable or action.kind not in {"manual_review", "retain"}: + raise ValueError( + "Identity Trust DSAR publishes non-executable actions only." + ) + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary=( + "The device key remains unchanged pending cryptographic " + "recovery and revocation review." + if action.kind == "manual_review" + else "Trust and key-access evidence remains immutable." + ), + evidence={"request_id": request_id}, + ) + ) + return tuple(results) + + +def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None: + references = subject.external_references + values = { + "account_id": _coalesce( + subject.account_id, + references.get("identity_trust.account"), + references.get("access.account"), + ), + "identity_id": _coalesce( + subject.identity_id, + references.get("identity_trust.identity"), + references.get("identity.id"), + ), + "device_id": _coalesce( + references.get("identity_trust.device"), + references.get("identity_trust.device_id"), + ), + "key_id": _coalesce( + references.get("identity_trust.key"), + references.get("identity_trust.key_id"), + ), + } + if any(value is _CONFLICT for value in values.values()): + return None + account_id = _optional_string(values["account_id"]) + identity_id = _optional_string(values["identity_id"]) + if not (account_id or identity_id): + return None + return _SubjectSelectors( + account_id=account_id, + identity_id=identity_id, + device_id=_optional_string(values["device_id"]), + key_id=_optional_string(values["key_id"]), + ) + + +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"Identity Trust DSAR {label} limit exceeded; narrow the selectors." + ) + return rows + + +def _device_record(row: DevicePublicKey) -> DsarRecordRef: + return DsarRecordRef( + provider_id="identity_trust", + module_id="identity_trust", + resource_type="device_public_key", + resource_id=row.id, + category="personal_device_trust", + title=f"Device public key {row.key_id[:255]}", + data={ + "identity_id": row.identity_id, + "account_id": row.account_id, + "device_id": row.device_id, + "key_id": row.key_id, + "algorithm": row.algorithm, + "public_jwk": _public_jwk(row.public_jwk), + "purpose": row.purpose, + "assurance_level": row.assurance_level, + "attestation_ref": (row.attestation_ref or "")[:1_000] or None, + "status": row.status, + "epoch": row.epoch, + "registered_at": _iso(row.registered_at), + "expires_at": _iso(row.expires_at), + "revoked_at": _iso(row.revoked_at), + "revocation_reason": (row.revocation_reason or "")[:4_000] or None, + }, + observed_at=_aware(row.updated_at), + retention_reason=( + "Device-key changes require recovery and active-resource review." + ), + ) + + +def _assurance_record(row: AssuranceEvidence) -> DsarRecordRef: + return DsarRecordRef( + provider_id="identity_trust", + module_id="identity_trust", + resource_type="assurance_evidence", + resource_id=row.id, + category="identity_assurance_evidence", + title="Identity assurance evidence", + data={ + "account_id": row.account_id, + "device_key_id": row.device_key_id, + "evidence_ref": row.evidence_ref[:1_000], + "assurance_level": row.assurance_level, + "provider_id": row.provider_id, + "verified_at": _iso(row.verified_at), + "expires_at": _iso(row.expires_at), + }, + observed_at=_aware(row.created_at), + immutable_evidence=True, + retention_reason=( + "Assurance verification is immutable security and accountability evidence." + ), + ) + + +def _epoch_record(row: TrustKeyEpoch) -> DsarRecordRef: + return DsarRecordRef( + provider_id="identity_trust", + module_id="identity_trust", + resource_type="key_epoch", + resource_id=row.id, + category="cryptographic_access_history", + title="Subject key epoch", + data={ + "subject_kind": row.subject_kind, + "subject_id": row.subject_id, + "epoch": row.epoch, + "previous_epoch": row.previous_epoch, + "state": row.state, + "history_policy": row.history_policy, + "reason": row.reason[:4_000], + "access_decision_ref": row.access_decision_ref[:1_000], + "effective_at": _iso(row.effective_at), + }, + observed_at=_aware(row.effective_at), + immutable_evidence=True, + retention_reason="Key-epoch history is immutable cryptographic-access evidence.", + ) + + +def _access_decision_record(row: KeyAccessDecisionRecord) -> DsarRecordRef: + return DsarRecordRef( + provider_id="identity_trust", + module_id="identity_trust", + resource_type="key_access_decision", + resource_id=row.id, + category="cryptographic_access_decision", + title="Key-access decision", + data={ + "decision_ref": row.decision_ref, + "account_id": row.account_id, + "device_key_id": row.device_key_id, + "subject_kind": row.subject_kind, + "subject_id": row.subject_id, + "key_epoch": row.key_epoch, + "access_decision_ref": row.access_decision_ref[:1_000], + "purpose": row.purpose[:255], + "allowed": row.allowed, + "reason": row.reason[:4_000], + "resource_ref": (row.resource_ref or "")[:1_000] or None, + "function_assignment_id": row.function_assignment_id, + "delegation_id": row.delegation_id, + "recorded_at": _iso(row.created_at), + }, + observed_at=_aware(row.created_at), + immutable_evidence=True, + retention_reason=( + "Key-access decisions are immutable authorization and security evidence." + ), + ) + + +def _public_jwk(value: object) -> dict[str, object]: + if not isinstance(value, Mapping): + raise ValueError("Identity Trust public JWK is invalid.") + projection: dict[str, object] = {} + for key in ("kty", "crv", "x", "y", "use", "alg"): + if key in value: + projection[key] = str(value[key])[:2_000] + key_ops = value.get("key_ops") + if key_ops is not None: + if not isinstance(key_ops, list) or len(key_ops) > 20: + raise ValueError("Identity Trust public JWK key operations are invalid.") + projection["key_ops"] = [str(item)[:120] for item in key_ops] + return projection + + +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("Identity Trust DSAR requires a SQLAlchemy Session.") + return value + + +_RESOURCE_TYPES = { + "device_public_key", + "assurance_evidence", + "key_epoch", + "key_access_decision", +} + + +def _validate_record(record: DsarRecordRef) -> None: + if ( + record.provider_id != "identity_trust" + or record.module_id != "identity_trust" + ): + raise ValueError("Identity Trust DSAR cannot plan a foreign provider record.") + if record.resource_type not in _RESOURCE_TYPES or not record.resource_id: + raise ValueError("Identity Trust DSAR record identity is invalid.") + + +def _validate_action(action: DsarErasureActionRef) -> None: + if ( + action.provider_id != "identity_trust" + or action.module_id != "identity_trust" + ): + raise ValueError("Identity Trust DSAR cannot execute a foreign action.") + if not action.action_id.startswith("identity_trust:"): + raise ValueError("Identity Trust DSAR action identity is invalid.") + + +__all__ = ["IDENTITY_TRUST_DSAR_CAPABILITY", "IdentityTrustDsarProvider"] diff --git a/src/govoplan_identity_trust/backend/manifest.py b/src/govoplan_identity_trust/backend/manifest.py index fae5af8..ace2af8 100644 --- a/src/govoplan_identity_trust/backend/manifest.py +++ b/src/govoplan_identity_trust/backend/manifest.py @@ -26,6 +26,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_identity_trust.backend.db import models +from govoplan_identity_trust.backend.dsar_provider import ( + IDENTITY_TRUST_DSAR_CAPABILITY, + IdentityTrustDsarProvider, +) from govoplan_identity_trust.backend.service import SqlIdentityTrustService @@ -64,6 +68,10 @@ def _service(_context: ModuleContext) -> SqlIdentityTrustService: return SqlIdentityTrustService() +def _dsar_provider(_context: ModuleContext) -> IdentityTrustDsarProvider: + return IdentityTrustDsarProvider() + + manifest = ModuleManifest( id=MODULE_ID, name=MODULE_NAME, @@ -72,6 +80,10 @@ manifest = ModuleManifest( provides_interfaces=( ModuleInterfaceProvider(name="identity_trust.directory", version="1.0.0"), ModuleInterfaceProvider(name="identity_trust.assurance", version="1.0.0"), + ModuleInterfaceProvider( + name=IDENTITY_TRUST_DSAR_CAPABILITY, + version="0.1.0", + ), ), permissions=( _permission( @@ -165,6 +177,7 @@ manifest = ModuleManifest( capability_factories={ CAPABILITY_IDENTITY_TRUST_DIRECTORY: _service, CAPABILITY_IDENTITY_TRUST_ASSURANCE: _service, + IDENTITY_TRUST_DSAR_CAPABILITY: _dsar_provider, }, capability_documentation={ CAPABILITY_IDENTITY_TRUST_DIRECTORY: CapabilityDocumentation( @@ -177,6 +190,14 @@ manifest = ModuleManifest( summary="Verifies bounded, recent assurance evidence for high-risk cryptographic operations.", contract_version="1.0.0", ), + IDENTITY_TRUST_DSAR_CAPABILITY: CapabilityDocumentation( + label="Identity Trust data-subject request provider", + summary=( + "Exports bounded device-key and assurance records while keeping " + "cryptographic history under governed retention and review." + ), + contract_version="0.1.0", + ), }, migration_spec=MigrationSpec( module_id=MODULE_ID, @@ -202,6 +223,47 @@ manifest = ModuleManifest( ), ), documentation=( + DocumentationTopic( + id="identity-trust.data-subject-requests", + title="Identity Trust data-subject requests", + summary=( + "Export tenant-scoped device trust, assurance, epoch, and key-access " + "evidence without private or operational key material." + ), + body=( + "Identity Trust correlates exact account and identity identifiers in " + "the active tenant and can narrow results to an exact device or key. " + "The access package includes bounded public-key registration fields, " + "assurance state, matching key epochs, and key-access decisions. It " + "never exports private JWK parameters, request digests, idempotency " + "keys, or arbitrary provenance payloads. Device-key revocation requires " + "manual review of recovery, active encrypted resources, and the current " + "epoch. Assurance, epoch, and access-decision records remain immutable " + "security evidence." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("user", "administrator", "security_officer", "auditor"), + related_modules=("core", "access", "encryption", "postbox"), + metadata={ + "help_contexts": [ + "identity_trust.settings.devices", + "identity_trust.admin.trust", + "privacy.data-subject-requests", + ], + "consequence_classes": { + "export_trust_state": ( + "Returns bounded public trust and assurance evidence only." + ), + "review_device_revocation": ( + "Requires recovery and encrypted-resource impact review." + ), + "retain_security_evidence": ( + "Preserves epoch, assurance, and access-decision history." + ), + }, + }, + ), DocumentationTopic( id="identity-trust.device-keys", title="Device keys and key epochs", diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..ef62c90 --- /dev/null +++ b/tests/test_dsar_provider.py @@ -0,0 +1,297 @@ +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_core.privacy.dsar_workflow import ( + create_data_subject_request, + search_data_subject_request, +) +from govoplan_identity_trust.backend.db.models import ( + AssuranceEvidence, + DevicePublicKey, + KeyAccessDecisionRecord, + TrustKeyEpoch, +) +from govoplan_identity_trust.backend.dsar_provider import ( + IDENTITY_TRUST_DSAR_CAPABILITY, + IdentityTrustDsarProvider, +) +from govoplan_identity_trust.backend.manifest import manifest + + +NOW = datetime(2026, 8, 21, 12, 0, tzinfo=UTC) + + +class _Registry: + def __init__(self, provider: IdentityTrustDsarProvider) -> None: + self.provider = provider + + def capability_names(self): + return (IDENTITY_TRUST_DSAR_CAPABILITY,) + + def capability_owner(self, name): + if name != IDENTITY_TRUST_DSAR_CAPABILITY: + raise KeyError(name) + return "identity_trust" + + def tenant_entitlement_resolver(self): + class _Resolver: + @staticmethod + def resolve(session, tenant_id): + del session, tenant_id + return type( + "State", (), {"effective_modules": ("identity_trust",)} + )() + + return _Resolver() + + def require_tenant_capability(self, name, session, **kwargs): + del session, kwargs + if name != IDENTITY_TRUST_DSAR_CAPABILITY: + raise KeyError(name) + return self.provider + + def manifests(self): + return (type("Manifest", (), {"id": "identity_trust"})(),) + + +class IdentityTrustDsarProviderTests(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 = IdentityTrustDsarProvider() + 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( + ( + DevicePublicKey( + id="device-key-row-1", + tenant_id="tenant-1", + identity_id="identity-1", + account_id="account-1", + device_id="device-1", + key_id="key-1", + algorithm="EdDSA", + public_jwk={ + "kty": "OKP", + "crv": "Ed25519", + "x": "public-coordinate", + "d": "private-jwk-do-not-export", + "unknown": "arbitrary-jwk-do-not-export", + }, + purpose="postbox", + assurance_level="substantial", + attestation_ref="evidence:attestation-1", + status="active", + epoch=1, + registration_digest="registration-digest-do-not-export", + idempotency_key="device-idempotency-do-not-export", + registered_at=NOW, + created_by="account-1", + updated_by="account-1", + ), + DevicePublicKey( + id="device-key-other-tenant", + tenant_id="tenant-2", + identity_id="identity-1", + account_id="account-1", + device_id="device-other-tenant", + key_id="key-other-tenant", + algorithm="EdDSA", + public_jwk={"kty": "OKP", "x": "other-tenant-public"}, + purpose="postbox", + assurance_level="substantial", + status="active", + epoch=1, + registration_digest="other-digest", + idempotency_key="other-key", + registered_at=NOW, + ), + ) + ) + self.session.add( + AssuranceEvidence( + id="assurance-1", + tenant_id="tenant-1", + account_id="account-1", + device_key_id="key-1", + evidence_ref="assurance:evidence-1", + assurance_level="substantial", + provider_id="bund-id", + verified_at=NOW, + expires_at=NOW, + provenance={"secret": "assurance-provenance-do-not-export"}, + recorded_by="security-officer", + ) + ) + self.session.add( + TrustKeyEpoch( + id="epoch-1", + tenant_id="tenant-1", + subject_kind="account", + subject_id="account-1", + epoch=2, + previous_epoch=1, + state="active", + history_policy="forward_only", + reason="Device rotation", + access_decision_ref="access:decision-1", + idempotency_key="epoch-idempotency-do-not-export", + request_digest="epoch-request-digest-do-not-export", + effective_at=NOW, + created_by="security-officer", + ) + ) + self.session.add( + KeyAccessDecisionRecord( + id="access-decision-1", + tenant_id="tenant-1", + decision_ref="trust-decision-1", + request_digest="access-request-digest-do-not-export", + account_id="account-1", + device_key_id="key-1", + subject_kind="postbox", + subject_id="postbox-1", + key_epoch=2, + access_decision_ref="access:authorization-1", + purpose="read-message", + allowed=True, + reason="Current device and epoch", + resource_ref="postbox:message-1", + provenance={"secret": "access-provenance-do-not-export"}, + ) + ) + + @staticmethod + def _subject() -> DsarSubjectRef: + return DsarSubjectRef(account_id="account-1", identity_id="identity-1") + + def test_search_exports_bounded_trust_records(self) -> None: + records = self.provider.search_subject( + self.session, tenant_id="tenant-1", subject=self._subject() + ) + self.assertEqual( + { + "device_public_key", + "assurance_evidence", + "key_epoch", + "key_access_decision", + }, + {record.resource_type for record in records}, + ) + exported = json.dumps([record.to_dict() for record in records]) + self.assertIn("public-coordinate", exported) + self.assertIn("assurance:evidence-1", exported) + self.assertIn("postbox:message-1", exported) + for excluded in ( + "private-jwk-do-not-export", + "arbitrary-jwk-do-not-export", + "registration-digest-do-not-export", + "device-idempotency-do-not-export", + "assurance-provenance-do-not-export", + "epoch-request-digest-do-not-export", + "access-request-digest-do-not-export", + "access-provenance-do-not-export", + "other-tenant-public", + ): + self.assertNotIn(excluded, exported) + + def test_key_narrowing_and_conflicts_fail_closed(self) -> None: + narrowed = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + account_id="account-1", + external_references={"identity_trust.key": "key-1"}, + ), + ) + conflict = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + account_id="account-1", + external_references={"identity_trust.account": "account-other"}, + ), + ) + reference_only = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + external_references={"identity_trust.key": "key-1"} + ), + ) + self.assertEqual( + {"device_public_key", "assurance_evidence", "key_access_decision"}, + {record.resource_type for record in narrowed}, + ) + self.assertEqual((), conflict) + self.assertEqual((), reference_only) + + def test_erasure_is_review_or_retain_only(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.assertEqual( + {"manual_review", "retain"}, {action.kind for action in actions} + ) + results = self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=self._subject(), + actions=actions, + request_id="dsar-trust-1", + ) + self.assertTrue(all(result.status == "blocked" for result in results)) + self.assertEqual( + "active", + self.session.get(DevicePublicKey, "device-key-row-1").status, + ) + + def test_manifest_and_core_workflow_discover_provider(self) -> None: + self.assertIn( + IDENTITY_TRUST_DSAR_CAPABILITY, manifest.capability_factories + ) + row = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-TRUST-1", + request_kind="access", + subject=self._subject(), + purpose="Trust 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(4, row.search_result["record_count"]) + + +if __name__ == "__main__": + unittest.main()