From b5017acad35bccabb5e7cc710f7ec1e3d11229e0 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 21 Aug 2026 12:01:18 +0200 Subject: [PATCH] feat(identity): add governed DSAR coverage --- .../backend/dsar_provider.py | 247 ++++++++++++++++++ src/govoplan_identity/backend/manifest.py | 60 +++++ tests/test_dsar_provider.py | 234 +++++++++++++++++ 3 files changed, 541 insertions(+) create mode 100644 src/govoplan_identity/backend/dsar_provider.py create mode 100644 tests/test_dsar_provider.py diff --git a/src/govoplan_identity/backend/dsar_provider.py b/src/govoplan_identity/backend/dsar_provider.py new file mode 100644 index 0000000..ec31ceb --- /dev/null +++ b/src/govoplan_identity/backend/dsar_provider.py @@ -0,0 +1,247 @@ +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import datetime, timezone + +from sqlalchemy.orm import Session + +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarExecutionResultRef, + DsarRecordRef, + DsarSubjectRef, + dsar_capability_name, +) +from govoplan_identity.backend.db.models import Identity, IdentityAccountLink + + +IDENTITY_DSAR_CAPABILITY = dsar_capability_name("identity") +_CONFLICT = object() + + +@dataclass(frozen=True, slots=True) +class _SubjectSelectors: + identity_id: str + account_id: str | None + link_id: str | None + + +class IdentityDsarProvider: + provider_id = "identity" + module_id = "identity" + + def search_subject( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + ) -> Sequence[DsarRecordRef]: + del tenant_id + db = _session(session) + selectors = _subject_selectors(subject) + if selectors is None: + return () + query = ( + db.query(Identity, IdentityAccountLink) + .join( + IdentityAccountLink, + IdentityAccountLink.identity_id == Identity.id, + ) + .filter(Identity.id == selectors.identity_id) + ) + if selectors.account_id: + query = query.filter( + IdentityAccountLink.account_id == selectors.account_id + ) + if selectors.link_id: + query = query.filter(IdentityAccountLink.id == selectors.link_id) + matches = query.limit(2).all() + if len(matches) != 1: + return () + identity, link = matches[0] + return (_identity_record(identity, link),) + + def plan_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + records: Sequence[DsarRecordRef], + ) -> Sequence[DsarErasureActionRef]: + del tenant_id + _session(session) + selectors = _subject_selectors(subject) + if selectors is None: + raise ValueError("Identity DSAR subject selectors conflict or are incomplete.") + actions: list[DsarErasureActionRef] = [] + for record in records: + _validate_record(record) + if record.resource_id != selectors.identity_id: + raise ValueError("Identity DSAR record does not match the subject.") + actions.append( + DsarErasureActionRef( + action_id=f"identity:manual_review:canonical_identity:{record.resource_id}", + provider_id=self.provider_id, + module_id=self.module_id, + kind="manual_review", + resource_type=record.resource_type, + resource_id=record.resource_id, + title=f"Review {record.title}", + rationale=( + "Canonical identities and account links are system-scoped and may " + "support authentication or memberships in more than one tenant. " + "Identity, Access, and tenancy owners must review deactivation, " + "unlinking, or minimization together." + ), + 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 DSAR subject selectors conflict or are incomplete.") + results: list[DsarExecutionResultRef] = [] + for action in actions: + _validate_action(action) + if action.executable or action.kind != "manual_review": + raise ValueError("Identity DSAR publishes manual-review actions only.") + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary=( + "The system identity and account link remain unchanged pending " + "cross-tenant identity, authentication, and retention review." + ), + evidence={"request_id": request_id}, + ) + ) + return tuple(results) + + +def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None: + references = subject.external_references + identity_id = _coalesce( + subject.identity_id, + references.get("identity.id"), + references.get("identity.identity"), + ) + account_id = _coalesce( + subject.account_id, + references.get("identity.account"), + references.get("access.account"), + ) + link_id = _coalesce( + references.get("identity.link"), + references.get("identity.account_link"), + ) + if _CONFLICT in (identity_id, account_id, link_id): + return None + normalized_identity = _optional_string(identity_id) + normalized_account = _optional_string(account_id) + normalized_link = _optional_string(link_id) + if not normalized_identity or not (normalized_account or normalized_link): + return None + return _SubjectSelectors( + identity_id=normalized_identity, + account_id=normalized_account, + link_id=normalized_link, + ) + + +def _identity_record( + identity: Identity, + link: IdentityAccountLink, +) -> DsarRecordRef: + observed = max( + value for value in (identity.updated_at, link.updated_at) if value is not None + ) + return DsarRecordRef( + provider_id="identity", + module_id="identity", + resource_type="canonical_identity", + resource_id=identity.id, + category="system_identity_and_account_link", + title="Canonical identity and corroborated account link", + data={ + "identity_id": identity.id, + "display_name": (identity.display_name or "")[:255] or None, + "external_subject": (identity.external_subject or "")[:255] or None, + "source": identity.source, + "is_active": identity.is_active, + "created_at": _iso(identity.created_at), + "updated_at": _iso(identity.updated_at), + "matching_account_link": { + "id": link.id, + "account_id": link.account_id, + "is_primary": link.is_primary, + "source": link.source, + "created_at": _iso(link.created_at), + "updated_at": _iso(link.updated_at), + }, + }, + observed_at=_aware(observed), + retention_reason=( + "The canonical identity and account link are system-scoped and require " + "cross-tenant lifecycle review before alteration." + ), + ) + + +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 DSAR requires a SQLAlchemy Session.") + return value + + +def _validate_record(record: DsarRecordRef) -> None: + if record.provider_id != "identity" or record.module_id != "identity": + raise ValueError("Identity DSAR cannot plan a foreign provider record.") + if record.resource_type != "canonical_identity" or not record.resource_id: + raise ValueError("Identity DSAR record identity is invalid.") + + +def _validate_action(action: DsarErasureActionRef) -> None: + if action.provider_id != "identity" or action.module_id != "identity": + raise ValueError("Identity DSAR cannot execute a foreign provider action.") + if not action.action_id.startswith("identity:manual_review:"): + raise ValueError("Identity DSAR action identity is invalid.") + + +__all__ = ["IDENTITY_DSAR_CAPABILITY", "IdentityDsarProvider"] diff --git a/src/govoplan_identity/backend/manifest.py b/src/govoplan_identity/backend/manifest.py index 6ccd5bf..6e71edc 100644 --- a/src/govoplan_identity/backend/manifest.py +++ b/src/govoplan_identity/backend/manifest.py @@ -6,10 +6,12 @@ from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPA from govoplan_core.core.identity import CAPABILITY_IDENTITY_DIRECTORY, CAPABILITY_IDENTITY_SEARCH from govoplan_core.core.module_guards import persistent_table_uninstall_guard from govoplan_core.core.modules import ( + CapabilityDocumentation, DocumentationTopic, FrontendModule, MigrationSpec, ModuleContext, + ModuleInterfaceProvider, ModuleManifest, PermissionDefinition, RoleTemplate, @@ -18,6 +20,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.backend.db import models as identity_models # noqa: F401 - populate metadata +from govoplan_identity.backend.dsar_provider import ( + IDENTITY_DSAR_CAPABILITY, + IdentityDsarProvider, +) def _permission( @@ -90,6 +96,11 @@ def _identity_directory(context: ModuleContext) -> object: return SqlIdentityDirectory() +def _dsar_provider(context: ModuleContext) -> IdentityDsarProvider: + del context + return IdentityDsarProvider() + + manifest = ModuleManifest( id="identity", name="Identity", @@ -98,6 +109,9 @@ manifest = ModuleManifest( permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, route_factory=_route_factory, + provides_interfaces=( + ModuleInterfaceProvider(name=IDENTITY_DSAR_CAPABILITY, version="0.1.0"), + ), frontend=FrontendModule( module_id="identity", package_name="@govoplan/identity-webui", @@ -135,8 +149,54 @@ manifest = ModuleManifest( capability_factories={ CAPABILITY_IDENTITY_DIRECTORY: _identity_directory, CAPABILITY_IDENTITY_SEARCH: _identity_directory, + IDENTITY_DSAR_CAPABILITY: _dsar_provider, + }, + capability_documentation={ + IDENTITY_DSAR_CAPABILITY: CapabilityDocumentation( + label="Identity data-subject request provider", + summary=( + "Exports a corroborated system identity and matching account link " + "without automatically mutating cross-tenant identity state." + ), + contract_version="0.1.0", + ), }, documentation=( + DocumentationTopic( + id="identity.data-subject-requests", + title="Identity data-subject requests", + summary=( + "Export a canonical identity only after its exact identity and " + "account-link identifiers corroborate each other." + ), + body=( + "Identity records are system-scoped rather than tenant-owned. The " + "data-subject provider therefore requires an exact identity identifier " + "and either its exact linked account or account-link identifier before " + "returning display, external-subject, lifecycle, and matching-link data. " + "Other links and arbitrary identity settings are excluded. A tenant " + "request cannot automatically deactivate the identity or remove the " + "link because either action can affect authentication and memberships " + "outside that tenant. Erasure is recorded as a manual review requiring " + "Identity, Access, tenancy, and retention owners." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("user", "system_admin", "identity_admin", "auditor"), + related_modules=("core", "access", "tenancy"), + order=23, + metadata={ + "help_contexts": ["privacy.data-subject-requests"], + "consequence_classes": { + "corroborated_export": ( + "Discloses one matching identity/account-link pair only." + ), + "manual_erasure_review": ( + "Prevents a tenant request from changing system-wide identity state." + ), + }, + }, + ), DocumentationTopic( id="identity.model", title="Identity directory", diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..dcedb05 --- /dev/null +++ b/tests/test_dsar_provider.py @@ -0,0 +1,234 @@ +from __future__ import annotations + +import json +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from govoplan_core.core.dsar import DsarErasureActionRef, 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.backend.db.models import Identity, IdentityAccountLink +from govoplan_identity.backend.dsar_provider import ( + IDENTITY_DSAR_CAPABILITY, + IdentityDsarProvider, +) +from govoplan_identity.backend.manifest import manifest + + +class _Registry: + def __init__(self, provider: IdentityDsarProvider) -> None: + self.provider = provider + + def capability_names(self): + return (IDENTITY_DSAR_CAPABILITY,) + + def capability_owner(self, name): + if name != IDENTITY_DSAR_CAPABILITY: + raise KeyError(name) + return "identity" + + def tenant_entitlement_resolver(self): + class _Resolver: + @staticmethod + def resolve(session, tenant_id): + del session, tenant_id + return type("State", (), {"effective_modules": ("identity",)})() + + return _Resolver() + + def require_tenant_capability(self, name, session, **kwargs): + del session, kwargs + if name != IDENTITY_DSAR_CAPABILITY: + raise KeyError(name) + return self.provider + + def manifests(self): + return (type("Manifest", (), {"id": "identity"})(),) + + +class IdentityDsarProviderTests(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 = IdentityDsarProvider() + self.assertIsInstance(self.provider, DsarProvider) + identity = Identity( + id="identity-1", + display_name="Ada Example", + external_subject="external-ada", + source="directory", + is_active=True, + settings={"secret": "identity-setting-do-not-export"}, + ) + other = Identity( + id="identity-other", + display_name="Other Person", + external_subject="external-other", + source="local", + is_active=True, + settings={"private": "other-setting"}, + ) + self.session.add_all((identity, other)) + self.session.flush() + self.session.add_all( + ( + IdentityAccountLink( + id="link-1", + identity_id="identity-1", + account_id="account-1", + is_primary=True, + source="directory", + ), + IdentityAccountLink( + id="link-secondary", + identity_id="identity-1", + account_id="account-secondary", + is_primary=False, + source="local", + ), + IdentityAccountLink( + id="link-other", + identity_id="identity-other", + account_id="account-other", + is_primary=True, + source="local", + ), + ) + ) + self.session.commit() + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + @staticmethod + def _subject() -> DsarSubjectRef: + return DsarSubjectRef(identity_id="identity-1", account_id="account-1") + + def test_search_requires_and_exports_one_corroborated_link(self) -> None: + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=self._subject(), + ) + + self.assertEqual(1, len(records)) + exported = json.dumps(records[0].to_dict()) + self.assertIn("Ada Example", exported) + self.assertIn("external-ada", exported) + self.assertIn("link-1", exported) + self.assertNotIn("account-secondary", exported) + self.assertNotIn("Other Person", exported) + self.assertNotIn("identity-setting-do-not-export", exported) + + def test_incomplete_conflicting_and_mismatched_selectors_fail_closed(self) -> None: + subjects = ( + DsarSubjectRef(identity_id="identity-1"), + DsarSubjectRef(account_id="account-1"), + DsarSubjectRef(identity_id="identity-1", account_id="account-other"), + DsarSubjectRef( + identity_id="identity-1", + account_id="account-1", + external_references={"identity.account": "account-other"}, + ), + ) + for subject in subjects: + with self.subTest(subject=subject): + self.assertEqual( + (), + self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=subject, + ), + ) + + def test_exact_link_can_corroborate_identity(self) -> None: + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + identity_id="identity-1", + external_references={"identity.link": "link-1"}, + ), + ) + self.assertEqual(["identity-1"], [record.resource_id for record in records]) + + def test_erasure_requires_cross_tenant_manual_review(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"], [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-1", + ) + self.assertEqual(["blocked"], [result.status for result in results]) + self.assertEqual(3, self.session.query(IdentityAccountLink).count()) + + def test_foreign_actions_are_rejected(self) -> None: + with self.assertRaises(ValueError): + self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=self._subject(), + actions=( + DsarErasureActionRef( + action_id="other:manual_review:x", + provider_id="other", + module_id="other", + kind="manual_review", + resource_type="canonical_identity", + resource_id="identity-1", + title="Foreign", + rationale="Foreign", + executable=False, + ), + ), + request_id="dsar-1", + ) + + def test_manifest_and_core_workflow_discover_provider(self) -> None: + self.assertIn(IDENTITY_DSAR_CAPABILITY, manifest.capability_factories) + self.assertIn( + IDENTITY_DSAR_CAPABILITY, + {item.name for item in manifest.provides_interfaces}, + ) + row = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-IDENTITY-1", + request_kind="access", + subject=self._subject(), + purpose="Identity 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(1, row.search_result["record_count"]) + + +if __name__ == "__main__": + unittest.main()