From 19e8cd350a47b89a8c40145a9820acbc3d914807 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 21 Aug 2026 13:22:04 +0200 Subject: [PATCH] feat(mandates): add governed DSAR coverage --- .../backend/dsar_provider.py | 208 ++++++++++++++++++ src/govoplan_mandates/backend/manifest.py | 74 ++++++- tests/test_dsar_provider.py | 118 ++++++++++ 3 files changed, 394 insertions(+), 6 deletions(-) create mode 100644 src/govoplan_mandates/backend/dsar_provider.py create mode 100644 tests/test_dsar_provider.py diff --git a/src/govoplan_mandates/backend/dsar_provider.py b/src/govoplan_mandates/backend/dsar_provider.py new file mode 100644 index 0000000..ec506a2 --- /dev/null +++ b/src/govoplan_mandates/backend/dsar_provider.py @@ -0,0 +1,208 @@ +from __future__ import annotations + +from collections.abc import Sequence +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_mandates.backend.db.models import MandateRevision + + +MANDATES_DSAR_CAPABILITY = dsar_capability_name("mandates") +_MAX_RECORDS = 5_000 +_CONFLICT = object() + + +class MandatesDsarProvider: + provider_id = "mandates" + module_id = "mandates" + + def search_subject( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + ) -> Sequence[DsarRecordRef]: + db = _session(session) + selectors = _selectors(subject) + if selectors is None: + return () + account_id, mandate_id, revision_id = selectors + query = db.query(MandateRevision).filter( + MandateRevision.tenant_id == tenant_id, + MandateRevision.created_by == account_id, + ) + if mandate_id: + query = query.filter(MandateRevision.mandate_id == mandate_id) + if revision_id: + query = query.filter(MandateRevision.id == revision_id) + rows = ( + query.order_by(MandateRevision.recorded_at, MandateRevision.id) + .limit(_MAX_RECORDS + 1) + .all() + ) + if len(rows) > _MAX_RECORDS: + raise ValueError("Mandates DSAR result limit exceeded; narrow selectors.") + return tuple(_record(row) for row in rows) + + def plan_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + records: Sequence[DsarRecordRef], + ) -> Sequence[DsarErasureActionRef]: + del tenant_id + _session(session) + if _selectors(subject) is None: + raise ValueError("Mandates DSAR subject selectors conflict.") + actions = [] + for record in records: + _validate_record(record) + actions.append( + DsarErasureActionRef( + action_id=f"mandates:retain:{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 "Mandate attribution remains institutional evidence." + ), + 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 _selectors(subject) is None: + raise ValueError("Mandates DSAR subject selectors conflict.") + results = [] + for action in actions: + _validate_action(action) + if action.executable or action.kind != "retain": + raise ValueError("Mandates DSAR publishes retain actions only.") + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary="Mandate attribution remains institutional evidence.", + evidence={"request_id": request_id}, + ) + ) + return tuple(results) + + +def _selectors(subject: DsarSubjectRef) -> tuple[str, str | None, str | None] | None: + references = subject.external_references + account = _coalesce( + subject.account_id, + references.get("mandates.account"), + references.get("access.account"), + ) + mandate_id = _coalesce( + references.get("mandates.mandate"), references.get("mandates.mandate_id") + ) + revision_id = _coalesce( + references.get("mandates.revision"), + references.get("mandates.revision_id"), + ) + if account is _CONFLICT or mandate_id is _CONFLICT or revision_id is _CONFLICT: + return None + if not isinstance(account, str) or not account: + return None + return ( + account, + mandate_id if isinstance(mandate_id, str) else None, + revision_id if isinstance(revision_id, str) else None, + ) + + +def _record(row: MandateRevision) -> DsarRecordRef: + return DsarRecordRef( + provider_id="mandates", + module_id="mandates", + resource_type="mandate_actor_attribution", + resource_id=row.id, + category="mandate_governance_attribution", + title="Mandate-revision actor attribution", + data={ + "mandate_id": row.mandate_id, + "revision_id": row.id, + "revision": row.revision, + "status": row.status, + "valid_from": _iso(row.valid_from), + "valid_to": _iso(row.valid_to), + "recorded_at": _iso(row.recorded_at), + "superseded_at": _iso(row.superseded_at), + "activity": "recorded_mandate_revision", + }, + observed_at=_aware(row.recorded_at), + immutable_evidence=True, + retention_reason=( + "Mandate-revision attribution is retained with institutional authority history." + ), + ) + + +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 _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("Mandates DSAR requires a SQLAlchemy Session.") + return value + + +def _validate_record(record: DsarRecordRef) -> None: + if record.provider_id != "mandates" or record.module_id != "mandates": + raise ValueError("Mandates DSAR cannot plan a foreign provider record.") + if record.resource_type != "mandate_actor_attribution" or not record.resource_id: + raise ValueError("Mandates DSAR record identity is invalid.") + + +def _validate_action(action: DsarErasureActionRef) -> None: + if action.provider_id != "mandates" or action.module_id != "mandates": + raise ValueError("Mandates DSAR cannot execute a foreign provider action.") + if not action.action_id.startswith("mandates:retain:"): + raise ValueError("Mandates DSAR action identity is invalid.") + + +__all__ = ["MANDATES_DSAR_CAPABILITY", "MandatesDsarProvider"] diff --git a/src/govoplan_mandates/backend/manifest.py b/src/govoplan_mandates/backend/manifest.py index 2408302..3de99a4 100644 --- a/src/govoplan_mandates/backend/manifest.py +++ b/src/govoplan_mandates/backend/manifest.py @@ -21,6 +21,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_mandates.backend.db import models as mandate_models +from govoplan_mandates.backend.dsar_provider import ( + MANDATES_DSAR_CAPABILITY, + MandatesDsarProvider, +) from govoplan_mandates.backend.service import SqlMandateResolver @@ -56,6 +60,10 @@ def _resolver(_context: ModuleContext) -> SqlMandateResolver: return SqlMandateResolver() +def _dsar_provider(_context: ModuleContext) -> MandatesDsarProvider: + return MandatesDsarProvider() + + manifest = ModuleManifest( id=MODULE_ID, name=MODULE_NAME, @@ -64,11 +72,22 @@ manifest = ModuleManifest( provides_interfaces=( ModuleInterfaceProvider(name="mandates.definition", version="0.1.0"), ModuleInterfaceProvider(name="mandates.resolution", version="0.1.0"), + ModuleInterfaceProvider(name=MANDATES_DSAR_CAPABILITY, version="0.1.0"), ), permissions=( - _permission(READ_SCOPE, "View mandates", "View mandate definitions, history, and resolution evidence."), - _permission(WRITE_SCOPE, "Manage mandates", "Create and revise mandate definitions."), - _permission(ADMIN_SCOPE, "Administer mandates", "Administer mandate lifecycle and recovery."), + _permission( + READ_SCOPE, + "View mandates", + "View mandate definitions, history, and resolution evidence.", + ), + _permission( + WRITE_SCOPE, "Manage mandates", "Create and revise mandate definitions." + ), + _permission( + ADMIN_SCOPE, + "Administer mandates", + "Administer mandate lifecycle and recovery.", + ), ), role_templates=( RoleTemplate( @@ -85,13 +104,23 @@ manifest = ModuleManifest( ), ), route_factory=_router, - capability_factories={CAPABILITY_MANDATE_RESOLVER: _resolver}, + capability_factories={ + CAPABILITY_MANDATE_RESOLVER: _resolver, + MANDATES_DSAR_CAPABILITY: _dsar_provider, + }, capability_documentation={ CAPABILITY_MANDATE_RESOLVER: CapabilityDocumentation( label="Mandate resolver", summary="Resolves effective institutional competence deterministically and fail-closed.", contract_version="0.1.0", - ) + ), + MANDATES_DSAR_CAPABILITY: CapabilityDocumentation( + label="Mandates data-subject request provider", + summary=( + "Exports minimized mandate-author attribution without institutional payloads." + ), + contract_version="0.1.0", + ), }, migration_spec=MigrationSpec( module_id=MODULE_ID, @@ -111,6 +140,34 @@ manifest = ModuleManifest( ), ), documentation=( + DocumentationTopic( + id="mandates.data-subject-requests", + title="Mandate data-subject requests", + summary=( + "Export mandate-author activity without inferring people from authority payloads." + ), + body=( + "Mandates correlates only an exact tenant account identifier and can " + "narrow an already verified search to one mandate or revision. It " + "returns minimized lifecycle attribution with revision, status, validity, " + "and recording timestamps. The arbitrary mandate definition payload, " + "legal bases, organization and function references, competence criteria, " + "change reasons, and conflict references are not searched for identity or " + "included in the export. Mandate history establishes institutional " + "authority and remains immutable evidence rather than being automatically erased." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("user", "operator", "module_admin", "auditor"), + related_modules=("core", "organizations", "idm", "audit"), + metadata={ + "help_contexts": ["privacy.data-subject-requests"], + "consequence_classes": { + "export_mandate_attribution": "Returns minimized immutable revision activity.", + "exclude_mandate_payload": "Does not infer or export subjects from authority definitions.", + }, + }, + ), DocumentationTopic( id="mandates.definition-and-resolution", title="Institutional mandates", @@ -141,7 +198,12 @@ manifest = ModuleManifest( known_limits=("No dedicated WebUI is included; administration is API-first.",), supported_authority_modes=("native_authoritative",), owned_concepts=("mandate", "jurisdiction authority", "competence history"), - non_owned_concepts=("organization structure", "function incumbency", "application permission", "formal decision"), + non_owned_concepts=( + "organization structure", + "function incumbency", + "application permission", + "formal decision", + ), reference_packages=("product.service-to-decision",), migration_docs=("docs/MANDATES_DOMAIN.md",), recovery_docs=("docs/MANDATES_DOMAIN.md",), diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..ffec16a --- /dev/null +++ b/tests/test_dsar_provider.py @@ -0,0 +1,118 @@ +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_mandates.backend.db.models import MandateRevision +from govoplan_mandates.backend.dsar_provider import ( + MANDATES_DSAR_CAPABILITY, + MandatesDsarProvider, +) +from govoplan_mandates.backend.manifest import manifest + + +NOW = datetime(2026, 8, 22, 13, 0, tzinfo=UTC) + + +class MandatesDsarProviderTests(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 = MandatesDsarProvider() + self.session.add_all( + ( + MandateRevision( + id="revision-1", + tenant_id="tenant-1", + mandate_id="mandate-1", + revision="2", + status="active", + valid_from=NOW, + recorded_at=NOW, + payload={ + "person_ref": "payload-person-do-not-correlate", + "legal_basis": "legal-basis-do-not-export", + "change_reason": "change-reason-do-not-export", + }, + created_by="account-1", + ), + MandateRevision( + id="revision-other", + tenant_id="tenant-2", + mandate_id="mandate-other", + revision="1", + status="active", + recorded_at=NOW, + payload={}, + created_by="account-1", + ), + ) + ) + self.session.commit() + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def test_search_is_minimized_tenant_safe_and_narrowable(self) -> None: + self.assertIsInstance(self.provider, DsarProvider) + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef(account_id="account-1"), + ) + self.assertEqual(["revision-1"], [record.resource_id for record in records]) + exported = json.dumps([record.to_dict() for record in records]) + self.assertNotIn("payload-person-do-not-correlate", exported) + self.assertNotIn("legal-basis-do-not-export", exported) + self.assertNotIn("change-reason-do-not-export", exported) + narrowed = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + account_id="account-1", + external_references={"mandates.mandate": "mandate-1"}, + ), + ) + self.assertEqual(1, len(narrowed)) + + def test_account_is_required_and_history_is_retained(self) -> None: + self.assertEqual( + (), + self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + external_references={"mandates.mandate": "mandate-1"} + ), + ), + ) + subject = DsarSubjectRef(account_id="account-1") + records = self.provider.search_subject( + self.session, tenant_id="tenant-1", subject=subject + ) + actions = self.provider.plan_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + records=records, + ) + self.assertTrue(all(action.kind == "retain" for action in actions)) + + def test_manifest_registers_provider_and_documentation(self) -> None: + self.assertIn(MANDATES_DSAR_CAPABILITY, manifest.capability_factories) + self.assertIn( + "mandates.data-subject-requests", + {topic.id for topic in manifest.documentation}, + ) + + +if __name__ == "__main__": + unittest.main()