From 5bc651a8e7098d64eca57d1f198700b6d30b0150 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 21 Aug 2026 13:20:28 +0200 Subject: [PATCH] feat(forms): add governed DSAR coverage --- src/govoplan_forms/backend/dsar_provider.py | 209 ++++++++++++++++++++ src/govoplan_forms/backend/manifest.py | 52 ++++- tests/test_dsar_provider.py | 120 +++++++++++ 3 files changed, 379 insertions(+), 2 deletions(-) create mode 100644 src/govoplan_forms/backend/dsar_provider.py create mode 100644 tests/test_dsar_provider.py diff --git a/src/govoplan_forms/backend/dsar_provider.py b/src/govoplan_forms/backend/dsar_provider.py new file mode 100644 index 0000000..85c3889 --- /dev/null +++ b/src/govoplan_forms/backend/dsar_provider.py @@ -0,0 +1,209 @@ +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_forms.backend.db.models import FormDefinitionRevision + + +FORMS_DSAR_CAPABILITY = dsar_capability_name("forms") +_MAX_RECORDS = 5_000 +_CONFLICT = object() + + +class FormsDsarProvider: + provider_id = "forms" + module_id = "forms" + + 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, form_id, revision_id = selectors + query = db.query(FormDefinitionRevision).filter( + FormDefinitionRevision.tenant_id == tenant_id, + FormDefinitionRevision.changed_by == account_id, + ) + if form_id: + query = query.filter(FormDefinitionRevision.form_id == form_id) + if revision_id: + query = query.filter(FormDefinitionRevision.id == revision_id) + rows = ( + query.order_by( + FormDefinitionRevision.recorded_at, + FormDefinitionRevision.id, + ) + .limit(_MAX_RECORDS + 1) + .all() + ) + if len(rows) > _MAX_RECORDS: + raise ValueError("Forms 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("Forms DSAR subject selectors conflict.") + actions = [] + for record in records: + _validate_record(record) + actions.append( + DsarErasureActionRef( + action_id=f"forms: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 "Form-definition attribution remains governance 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("Forms DSAR subject selectors conflict.") + results = [] + for action in actions: + _validate_action(action) + if action.executable or action.kind != "retain": + raise ValueError("Forms DSAR publishes retain actions only.") + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary="Form-definition attribution remains governance 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("forms.account"), + references.get("access.account"), + ) + form_id = _coalesce(references.get("forms.form"), references.get("forms.form_id")) + revision_id = _coalesce( + references.get("forms.revision"), references.get("forms.revision_id") + ) + if account is _CONFLICT or form_id is _CONFLICT or revision_id is _CONFLICT: + return None + if not isinstance(account, str) or not account: + return None + return ( + account, + form_id if isinstance(form_id, str) else None, + revision_id if isinstance(revision_id, str) else None, + ) + + +def _record(row: FormDefinitionRevision) -> DsarRecordRef: + return DsarRecordRef( + provider_id="forms", + module_id="forms", + resource_type="form_definition_actor_attribution", + resource_id=row.id, + category="form_definition_governance_attribution", + title="Form-definition actor attribution", + data={ + "form_id": row.form_id, + "revision_id": row.id, + "revision": row.revision, + "publication_state": row.publication_state, + "recorded_at": _iso(row.recorded_at), + "superseded_at": _iso(row.superseded_at), + "activity": "recorded_form_definition_revision", + }, + observed_at=_aware(row.recorded_at), + immutable_evidence=True, + retention_reason=( + "Form-definition author attribution is retained with immutable schema 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("Forms DSAR requires a SQLAlchemy Session.") + return value + + +def _validate_record(record: DsarRecordRef) -> None: + if record.provider_id != "forms" or record.module_id != "forms": + raise ValueError("Forms DSAR cannot plan a foreign provider record.") + if ( + record.resource_type != "form_definition_actor_attribution" + or not record.resource_id + ): + raise ValueError("Forms DSAR record identity is invalid.") + + +def _validate_action(action: DsarErasureActionRef) -> None: + if action.provider_id != "forms" or action.module_id != "forms": + raise ValueError("Forms DSAR cannot execute a foreign provider action.") + if not action.action_id.startswith("forms:retain:"): + raise ValueError("Forms DSAR action identity is invalid.") + + +__all__ = ["FORMS_DSAR_CAPABILITY", "FormsDsarProvider"] diff --git a/src/govoplan_forms/backend/manifest.py b/src/govoplan_forms/backend/manifest.py index d437aeb..52588cd 100644 --- a/src/govoplan_forms/backend/manifest.py +++ b/src/govoplan_forms/backend/manifest.py @@ -30,6 +30,10 @@ from govoplan_core.core.provider_governance import declared_module_architecture from govoplan_core.core.views import ViewSurface from govoplan_core.db.base import Base from govoplan_forms.backend.db import models as form_models +from govoplan_forms.backend.dsar_provider import ( + FORMS_DSAR_CAPABILITY, + FormsDsarProvider, +) from govoplan_forms.backend.service import SqlFormDefinitionProvider @@ -72,6 +76,10 @@ def _definitions(_context: ModuleContext) -> SqlFormDefinitionProvider: return SqlFormDefinitionProvider() +def _dsar_provider(_context: ModuleContext) -> FormsDsarProvider: + return FormsDsarProvider() + + manifest = ModuleManifest( id=MODULE_ID, name=MODULE_NAME, @@ -84,6 +92,7 @@ manifest = ModuleManifest( ), provides_interfaces=( ModuleInterfaceProvider(name="forms.definitions", version="0.1.0"), + ModuleInterfaceProvider(name=FORMS_DSAR_CAPABILITY, version="0.1.0"), ), permissions=( _permission( @@ -174,13 +183,24 @@ manifest = ModuleManifest( ), ), ), - capability_factories={CAPABILITY_FORM_DEFINITIONS: _definitions}, + capability_factories={ + CAPABILITY_FORM_DEFINITIONS: _definitions, + FORMS_DSAR_CAPABILITY: _dsar_provider, + }, capability_documentation={ CAPABILITY_FORM_DEFINITIONS: CapabilityDocumentation( label="Immutable form definitions", summary="Resolves exact tenant-bound form schemas without exposing Forms tables.", contract_version="0.1.0", - ) + ), + FORMS_DSAR_CAPABILITY: CapabilityDocumentation( + label="Forms data-subject request provider", + summary=( + "Exports minimized form-definition author attribution without schema " + "or semantic content." + ), + contract_version="0.1.0", + ), }, migration_spec=MigrationSpec( module_id=MODULE_ID, @@ -200,6 +220,34 @@ manifest = ModuleManifest( ), ), documentation=( + DocumentationTopic( + id="forms.data-subject-requests", + title="Form-definition data-subject requests", + summary=( + "Export definition-author activity without treating schemas as submitted values." + ), + body=( + "Forms correlates only an exact tenant account identifier and can narrow " + "an already verified search to one form or definition revision. It " + "returns the immutable revision identifier, lifecycle state, and timing " + "of the subject's definition work. Titles, search text, schema payloads, " + "field semantics, policy references, and change-reason content are not " + "included. Forms stores no submitted values; Forms Runtime and the " + "owning service export those records separately. Definition attribution " + "is retained with immutable schema history." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("user", "operator", "module_admin", "auditor"), + related_modules=("core", "forms_runtime", "docs", "audit"), + metadata={ + "help_contexts": ["forms.catalogue", "privacy.data-subject-requests"], + "consequence_classes": { + "export_definition_attribution": "Returns minimized immutable revision activity.", + "exclude_form_semantics": "Does not return schema, field, or submitted-value content.", + }, + }, + ), DocumentationTopic( id="forms.definitions", title="Reusable form definitions", diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..4eb8c97 --- /dev/null +++ b/tests/test_dsar_provider.py @@ -0,0 +1,120 @@ +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_forms.backend.db.models import FormDefinitionRevision +from govoplan_forms.backend.dsar_provider import ( + FORMS_DSAR_CAPABILITY, + FormsDsarProvider, +) +from govoplan_forms.backend.manifest import manifest + + +NOW = datetime(2026, 8, 22, 12, 0, tzinfo=UTC) + + +class FormsDsarProviderTests(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 = FormsDsarProvider() + self.session.add_all( + ( + FormDefinitionRevision( + id="revision-1", + tenant_id="tenant-1", + form_id="form-1", + form_key="secret-form-key-do-not-export", + revision="2", + publication_state="published", + title="Sensitive semantic title do not export", + recorded_at=NOW, + search_text="search-content-do-not-export", + payload={"secret": "schema-payload-do-not-export"}, + changed_by="account-1", + ), + FormDefinitionRevision( + id="revision-other", + tenant_id="tenant-2", + form_id="form-other", + form_key="other", + revision="1", + publication_state="draft", + title="Other tenant", + recorded_at=NOW, + search_text="other", + payload={}, + changed_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) + subject = DsarSubjectRef(account_id="account-1") + records = self.provider.search_subject( + self.session, tenant_id="tenant-1", subject=subject + ) + self.assertEqual(["revision-1"], [record.resource_id for record in records]) + exported = json.dumps([record.to_dict() for record in records]) + for excluded in ( + "secret-form-key-do-not-export", + "Sensitive semantic title do not export", + "search-content-do-not-export", + "schema-payload-do-not-export", + ): + self.assertNotIn(excluded, exported) + narrowed = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + account_id="account-1", + external_references={"forms.form": "form-1"}, + ), + ) + self.assertEqual(1, len(narrowed)) + + def test_requires_account_and_retains_definition_history(self) -> None: + self.assertEqual( + (), + self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef(email="designer@example.test"), + ), + ) + 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(FORMS_DSAR_CAPABILITY, manifest.capability_factories) + self.assertIn( + "forms.data-subject-requests", + {topic.id for topic in manifest.documentation}, + ) + + +if __name__ == "__main__": + unittest.main()