diff --git a/README.md b/README.md index 6242a02..e2ecdb6 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,15 @@ Notifications owns email delivery, and Portal owns the applicant-facing page. Its runtime module ID is `forms_runtime`; the repository and Python distribution retain the hyphenated `govoplan-forms-runtime` name. +The module also contributes `privacy.dsar.forms_runtime`. Exact-tenant actor, +email-status, confirmation, acknowledgement, and explicit runtime selectors +produce bounded submission and lifecycle exports. Assisted operators receive +only attribution unless they are also the identified applicant. Credential-like +form keys, access hashes, replay keys, opaque details, validation internals, and +evidence identifiers are excluded. Immutable submission/handoff evidence is +retained; current drafts and active status grants require manual review through +their normal lifecycle, with no automatic DSAR mutation. + The module persists tenant-bound immutable revisions and events, exposes bounded owner/manager APIs and WebUI routes, and provides both `forms_runtime.registry` and `forms_runtime.service_launcher`. diff --git a/docs/FORMS_RUNTIME_DOMAIN_BOUNDARY.md b/docs/FORMS_RUNTIME_DOMAIN_BOUNDARY.md index 36fa175..2ad04a7 100644 --- a/docs/FORMS_RUNTIME_DOMAIN_BOUNDARY.md +++ b/docs/FORMS_RUNTIME_DOMAIN_BOUNDARY.md @@ -210,6 +210,27 @@ Destructive retirement is blocked while state exists and requires a verified database snapshot plus an export or retention decision for referenced evidence. No local generated files are required, so API and worker nodes remain stateless. +## Data-subject requests + +Forms Runtime publishes `privacy.dsar.forms_runtime`. It correlates exact-tenant +canonical actors, normalized email addresses against grant-specific hashes, +applicant confirmations and acknowledgements, and explicit instance or +lifecycle references. An assisted session prevents its operator from being +mistaken for the applicant when the session identifies a different actor; the +operator still receives a minimized accountability-attribution record. + +Corroborated subject instances export bounded form values and typed definition, +status, intake, acknowledgement, status-access, token-lifecycle, and handoff +facts. Credential-like value keys are redacted. Raw token/email hashes, +idempotency and provider keys, request/payload hashes, opaque metadata/details, +validation internals, evidence identifiers, errors, and unrelated submissions +are excluded. Immutable revisions, events, handoffs, confirmations, +acknowledgements, intake evidence, and minimized token lifecycle are retained. +Current drafts and active status grants receive non-executable manual-review +actions and may only be changed through authorized runtime lifecycles. Forms +owns definitions, Portal owns presentation, and Cases/Workflow Engine own +handoff targets. + Public and assisted intake, Files-backed attachment evidence, and authenticated acknowledgements implement the approved first profiles. Conditional multi-page definitions are resolved from Forms, and native Case/Workflow handoffs execute diff --git a/src/govoplan_forms_runtime/backend/dsar_provider.py b/src/govoplan_forms_runtime/backend/dsar_provider.py new file mode 100644 index 0000000..c34517b --- /dev/null +++ b/src/govoplan_forms_runtime/backend/dsar_provider.py @@ -0,0 +1,1158 @@ +from __future__ import annotations + +import hashlib +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_forms_runtime.backend.db.models import ( + FormAcknowledgement, + FormAssistedConfirmation, + FormHandoffEffect, + FormInstanceEvent, + FormInstanceIdentity, + FormInstanceRevision, + FormIntakeProfile, + FormIntakeSession, + FormStatusAccessGrant, + FormStatusAccessPolicy, + FormStatusAccessToken, +) + + +FORMS_RUNTIME_DSAR_CAPABILITY = dsar_capability_name("forms_runtime") +_MAX_RECORDS = 5_000 +_MAX_VALUE_DEPTH = 8 +_SENSITIVE_VALUE_MARKERS = ( + "password", + "passwd", + "token", + "secret", + "credential", + "private_key", + "api_key", +) + + +@dataclass(frozen=True, slots=True) +class _SubjectSelectors: + account_id: str | None + identity_id: str | None + membership_id: str | None + email: str | None + references: Mapping[str, str] + + @property + def actor_ids(self) -> tuple[str, ...]: + return tuple( + value + for value in (self.account_id, self.identity_id, self.membership_id) + if value + ) + + @property + def has_canonical_selector(self) -> bool: + return bool(self.actor_ids or self.email) + + @property + def has_direct_selector(self) -> bool: + return bool(self.references) + + +class FormsRuntimeDsarProvider: + provider_id = "forms_runtime" + module_id = "forms_runtime" + + def search_subject( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + ) -> Sequence[DsarRecordRef]: + db = _session(session) + selectors = _subject_selectors(subject) + if selectors is None or not ( + selectors.has_canonical_selector or selectors.has_direct_selector + ): + return () + + direct = _direct_resources(db, tenant_id=tenant_id, selectors=selectors) + if direct is None: + return () + direct_instance_ids = {instance_id for _kind, _id, instance_id in direct} + if len(direct_instance_ids) > 1: + return () + + identities = _identities( + db, + tenant_id=tenant_id, + selectors=selectors, + direct_instance_ids=direct_instance_ids, + ) + sessions = _intake_sessions( + db, + tenant_id=tenant_id, + selectors=selectors, + instance_ids={row.instance_id for row in identities} | direct_instance_ids, + ) + grants = _status_grants( + db, + tenant_id=tenant_id, + selectors=selectors, + direct_instance_ids=direct_instance_ids, + ) + confirmations = _confirmations( + db, + tenant_id=tenant_id, + selectors=selectors, + direct_instance_ids=direct_instance_ids, + ) + acknowledgements = _acknowledgements( + db, + tenant_id=tenant_id, + selectors=selectors, + direct_instance_ids=direct_instance_ids, + ) + + owner_instance_ids = _owner_instance_ids( + selectors, + identities=identities, + sessions=sessions, + grants=grants, + confirmations=confirmations, + acknowledgements=acknowledgements, + ) + if ( + selectors.has_canonical_selector + and direct_instance_ids - owner_instance_ids + ): + return () + subject_instance_ids = owner_instance_ids | direct_instance_ids + + identities = _rows_for_instances( + db, + FormInstanceIdentity, + tenant_id=tenant_id, + instance_ids=subject_instance_ids, + ) + revisions = _rows_for_instances( + db, + FormInstanceRevision, + tenant_id=tenant_id, + instance_ids=subject_instance_ids, + ) + events = _rows_for_instances( + db, + FormInstanceEvent, + tenant_id=tenant_id, + instance_ids=subject_instance_ids, + ) + handoffs = _rows_for_instances( + db, + FormHandoffEffect, + tenant_id=tenant_id, + instance_ids=subject_instance_ids, + ) + sessions = _rows_for_instances( + db, + FormIntakeSession, + tenant_id=tenant_id, + instance_ids=subject_instance_ids, + ) + grants = _rows_for_instances( + db, + FormStatusAccessGrant, + tenant_id=tenant_id, + instance_ids=subject_instance_ids, + ) + confirmations = _rows_for_instances( + db, + FormAssistedConfirmation, + tenant_id=tenant_id, + instance_ids=subject_instance_ids, + ) + acknowledgements = _rows_for_instances( + db, + FormAcknowledgement, + tenant_id=tenant_id, + instance_ids=subject_instance_ids, + ) + tokens = _status_tokens( + db, + tenant_id=tenant_id, + grant_ids={row.id for row in grants}, + ) + + records: list[DsarRecordRef] = [] + seen: set[tuple[str, str]] = set() + + def append(record: DsarRecordRef) -> None: + key = (record.resource_type, record.resource_id) + if key in seen: + return + if len(records) >= _MAX_RECORDS: + raise ValueError( + "Forms Runtime DSAR result limit exceeded; narrow the selectors." + ) + seen.add(key) + records.append(record) + + for identity in identities: + append(_identity_record(identity)) + for revision in revisions: + append(_revision_record(revision, selectors=selectors)) + if revision.superseded_at is None and revision.status in { + "draft", + "started", + }: + append(_current_draft_record(revision)) + for event in events: + append(_event_record(event, selectors=selectors)) + for handoff in handoffs: + append(_handoff_record(handoff)) + for intake_session in sessions: + append(_intake_session_record(intake_session, selectors=selectors)) + for grant in grants: + append(_status_grant_record(grant, selectors=selectors)) + for token in tokens: + append(_status_token_record(token)) + for confirmation in confirmations: + append(_confirmation_record(confirmation, selectors=selectors)) + for acknowledgement in acknowledgements: + append(_acknowledgement_record(acknowledgement, selectors=selectors)) + + _append_runtime_operator_attribution( + db, + tenant_id=tenant_id, + selectors=selectors, + append=append, + ) + _append_operator_configuration_attribution( + db, + tenant_id=tenant_id, + selectors=selectors, + append=append, + ) + return tuple(records) + + 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("Forms Runtime DSAR subject selectors conflict.") + actions: list[DsarErasureActionRef] = [] + for record in records: + _validate_record(record) + if record.immutable_evidence: + kind = "retain" + title = f"Retain {record.title}" + rationale = record.retention_reason or ( + "Form submission evidence must be retained." + ) + else: + kind = "manual_review" + title = f"Review {record.title}" + rationale = ( + "An authorized operator must expire, revoke, or supersede the " + "mutable runtime fact after reviewing submission, access, handoff, " + "and third-party consequences." + ) + actions.append( + DsarErasureActionRef( + action_id=( + f"forms_runtime:{kind}:{record.resource_type}:" + f"{record.resource_id}" + ), + provider_id=self.provider_id, + module_id=self.module_id, + kind=kind, + resource_type=record.resource_type, + resource_id=record.resource_id, + title=title, + rationale=rationale, + 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("Forms Runtime DSAR subject selectors conflict.") + results: list[DsarExecutionResultRef] = [] + for action in actions: + _validate_action(action) + if action.executable: + raise ValueError( + "Forms Runtime DSAR does not publish executable erasure actions." + ) + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary=( + "Use the governed form, intake, status-access, or handoff " + "lifecycle after evidence and third-party review." + ), + evidence={"request_id": request_id}, + ) + ) + return tuple(results) + + +def _direct_resources( + session: Session, + *, + tenant_id: str, + selectors: _SubjectSelectors, +) -> list[tuple[str, str, str]] | None: + definitions = ( + ("instance", FormInstanceIdentity, "instance_id"), + ("revision", FormInstanceRevision, "instance_id"), + ("event", FormInstanceEvent, "instance_id"), + ("handoff", FormHandoffEffect, "instance_id"), + ("intake_session", FormIntakeSession, "instance_id"), + ("status_grant", FormStatusAccessGrant, "instance_id"), + ("confirmation", FormAssistedConfirmation, "instance_id"), + ("acknowledgement", FormAcknowledgement, "instance_id"), + ) + result: list[tuple[str, str, str]] = [] + for kind, model, instance_field in definitions: + reference = selectors.references.get(kind) + if not reference: + continue + if kind == "instance": + condition = FormInstanceIdentity.instance_id == reference + else: + condition = model.id == reference + row = ( + session.query(model) + .filter(model.tenant_id == tenant_id, condition) + .one_or_none() + ) + if row is None: + return None + result.append((kind, reference, str(getattr(row, instance_field)))) + return result + + +def _identities( + session: Session, + *, + tenant_id: str, + selectors: _SubjectSelectors, + direct_instance_ids: set[str], +) -> list[FormInstanceIdentity]: + conditions = [] + if selectors.actor_ids: + conditions.append(FormInstanceIdentity.created_by.in_(selectors.actor_ids)) + if direct_instance_ids: + conditions.append(FormInstanceIdentity.instance_id.in_(direct_instance_ids)) + return _query(session, FormInstanceIdentity, tenant_id, conditions) + + +def _intake_sessions( + session: Session, + *, + tenant_id: str, + selectors: _SubjectSelectors, + instance_ids: set[str], +) -> list[FormIntakeSession]: + conditions = [] + if selectors.actor_ids: + conditions.extend( + ( + FormIntakeSession.actor_id.in_(selectors.actor_ids), + FormIntakeSession.created_by.in_(selectors.actor_ids), + ) + ) + if instance_ids: + conditions.append(FormIntakeSession.instance_id.in_(instance_ids)) + if reference := selectors.references.get("intake_session"): + conditions.append(FormIntakeSession.id == reference) + return _query(session, FormIntakeSession, tenant_id, conditions) + + +def _status_grants( + session: Session, + *, + tenant_id: str, + selectors: _SubjectSelectors, + direct_instance_ids: set[str], +) -> list[FormStatusAccessGrant]: + conditions = [] + if selectors.actor_ids: + conditions.append( + FormStatusAccessGrant.applicant_actor_id.in_(selectors.actor_ids) + ) + if direct_instance_ids: + conditions.append(FormStatusAccessGrant.instance_id.in_(direct_instance_ids)) + if reference := selectors.references.get("status_grant"): + conditions.append(FormStatusAccessGrant.id == reference) + rows = _query(session, FormStatusAccessGrant, tenant_id, conditions) + if selectors.email: + candidates = _query( + session, + FormStatusAccessGrant, + tenant_id, + (FormStatusAccessGrant.recipient_email_sha256.is_not(None),), + ) + by_id = {row.id: row for row in (*rows, *candidates)} + rows = list(by_id.values()) + return rows + + +def _confirmations( + session: Session, + *, + tenant_id: str, + selectors: _SubjectSelectors, + direct_instance_ids: set[str], +) -> list[FormAssistedConfirmation]: + conditions = [] + if selectors.actor_ids: + conditions.extend( + ( + FormAssistedConfirmation.confirmed_by_ref.in_(selectors.actor_ids), + FormAssistedConfirmation.operator_actor_id.in_(selectors.actor_ids), + ) + ) + if direct_instance_ids: + conditions.append(FormAssistedConfirmation.instance_id.in_(direct_instance_ids)) + if reference := selectors.references.get("confirmation"): + conditions.append(FormAssistedConfirmation.id == reference) + return _query(session, FormAssistedConfirmation, tenant_id, conditions) + + +def _acknowledgements( + session: Session, + *, + tenant_id: str, + selectors: _SubjectSelectors, + direct_instance_ids: set[str], +) -> list[FormAcknowledgement]: + conditions = [] + if selectors.actor_ids: + conditions.append(FormAcknowledgement.actor_id.in_(selectors.actor_ids)) + if direct_instance_ids: + conditions.append(FormAcknowledgement.instance_id.in_(direct_instance_ids)) + if reference := selectors.references.get("acknowledgement"): + conditions.append(FormAcknowledgement.id == reference) + return _query(session, FormAcknowledgement, tenant_id, conditions) + + +def _owner_instance_ids( + selectors: _SubjectSelectors, + *, + identities: Sequence[FormInstanceIdentity], + sessions: Sequence[FormIntakeSession], + grants: Sequence[FormStatusAccessGrant], + confirmations: Sequence[FormAssistedConfirmation], + acknowledgements: Sequence[FormAcknowledgement], +) -> set[str]: + assisted_operator_instances = { + row.instance_id + for row in sessions + if row.instance_id + and row.mode == "assisted" + and row.created_by in selectors.actor_ids + and row.actor_id not in selectors.actor_ids + } + result = { + row.instance_id + for row in identities + if row.created_by in selectors.actor_ids + and row.instance_id not in assisted_operator_instances + } + result.update( + row.instance_id + for row in sessions + if row.instance_id and row.actor_id in selectors.actor_ids + ) + result.update( + row.instance_id + for row in grants + if row.applicant_actor_id in selectors.actor_ids + or _grant_email_matches(row, selectors.email) + ) + result.update( + row.instance_id + for row in confirmations + if row.confirmed_by_ref in selectors.actor_ids + ) + result.update( + row.instance_id + for row in acknowledgements + if row.actor_id in selectors.actor_ids + ) + return result + + +def _rows_for_instances( + session: Session, + model: type, + *, + tenant_id: str, + instance_ids: set[str], +) -> list[object]: + if not instance_ids: + return [] + return _query(session, model, tenant_id, (model.instance_id.in_(instance_ids),)) + + +def _status_tokens( + session: Session, + *, + tenant_id: str, + grant_ids: set[str], +) -> list[FormStatusAccessToken]: + if not grant_ids: + return [] + return _query( + session, + FormStatusAccessToken, + tenant_id, + (FormStatusAccessToken.grant_id.in_(grant_ids),), + ) + + +def _identity_record(row: FormInstanceIdentity) -> DsarRecordRef: + return _record( + "forms_runtime_instance_identity", + row.id, + "form_instance_identity", + "Form instance identity", + { + "instance_id": row.instance_id, + "definition_id": row.definition_id, + "definition_revision": row.definition_revision, + "created_at": _iso(row.created_at), + }, + observed_at=row.created_at, + immutable=True, + retention_reason=( + "The stable form-instance identity is retained so submission, handoff, " + "receipt, and downstream evidence remain reconstructable." + ), + ) + + +def _revision_record( + row: FormInstanceRevision, + *, + selectors: _SubjectSelectors, +) -> DsarRecordRef: + snapshot = row.snapshot if isinstance(row.snapshot, Mapping) else {} + values = snapshot.get("values") + return _record( + "forms_runtime_instance_revision", + row.id, + "form_submission_history", + "Form instance revision", + { + "instance_id": row.instance_id, + "revision": row.revision, + "previous_revision_id": row.previous_revision_id, + "status": row.status, + "definition_ref": _safe_reference(snapshot.get("definition_ref")), + "service_ref": _safe_reference(snapshot.get("service_ref")), + "values": _safe_form_value(values, depth=0), + "attachment_count": _sequence_length(snapshot.get("attachment_refs")), + "signature_count": _sequence_length(snapshot.get("signature_refs")), + "handoff_count": _sequence_length(snapshot.get("handoff_refs")), + "receipt_present": bool(snapshot.get("receipt_id")), + "recorded_at": _iso(row.recorded_at), + "superseded_at": _iso(row.superseded_at), + "changed_by": ( + row.changed_by if row.changed_by in selectors.actor_ids else None + ), + }, + observed_at=row.recorded_at, + immutable=True, + retention_reason=( + "Form revisions are immutable submission and validation evidence; " + "corrections append a new governed revision." + ), + ) + + +def _current_draft_record(row: FormInstanceRevision) -> DsarRecordRef: + return _record( + "forms_runtime_current_draft", + row.instance_id, + "current_form_draft", + "Current form draft", + { + "revision_record_id": row.id, + "revision": row.revision, + "status": row.status, + "recorded_at": _iso(row.recorded_at), + }, + observed_at=row.recorded_at, + ) + + +def _event_record( + row: FormInstanceEvent, + *, + selectors: _SubjectSelectors, +) -> DsarRecordRef: + return _record( + "forms_runtime_instance_event", + row.id, + "form_lifecycle_evidence", + "Form instance lifecycle event", + { + "instance_id": row.instance_id, + "instance_revision": row.instance_revision, + "event_type": _bounded_text(row.event_type, 120), + "status": row.status, + "occurred_at": _iso(row.occurred_at), + "actor_id": row.actor_id if row.actor_id in selectors.actor_ids else None, + }, + observed_at=row.occurred_at, + immutable=True, + retention_reason="Form lifecycle events are immutable submission evidence.", + ) + + +def _handoff_record(row: FormHandoffEffect) -> DsarRecordRef: + return _record( + "forms_runtime_handoff_effect", + row.id, + "form_handoff_evidence", + "Form handoff effect", + { + "instance_id": row.instance_id, + "instance_revision": row.instance_revision, + "binding_kind": row.binding_kind, + "binding_reference": _bounded_text(row.binding_reference, 500), + "provider_capability": _bounded_text(row.provider_capability, 120), + "state": row.state, + "attempt_count": row.attempt_count, + "requested_at": _iso(row.requested_at), + "resolved_at": _iso(row.resolved_at), + "target_ref": _safe_reference(row.target_ref), + "evidence_count": len(row.evidence or ()), + }, + observed_at=row.resolved_at or row.requested_at, + immutable=True, + retention_reason=( + "Handoff intent, attempts, and outcomes are retained as replay and " + "downstream-effect evidence." + ), + ) + + +def _intake_session_record( + row: FormIntakeSession, + *, + selectors: _SubjectSelectors, +) -> DsarRecordRef: + return _record( + "forms_runtime_intake_session", + row.id, + "form_intake_evidence", + "Form intake session", + { + "instance_id": row.instance_id, + "mode": row.mode, + "status": row.status, + "actor_id": row.actor_id if row.actor_id in selectors.actor_ids else None, + "expires_at": _iso(row.expires_at), + "started_at": _iso(row.started_at), + "submitted_at": _iso(row.submitted_at), + "revoked_at": _iso(row.revoked_at), + }, + observed_at=row.updated_at, + immutable=True, + retention_reason=( + "Intake-session state is retained to explain public or assisted access " + "and the resulting submission. Token hashes and opaque details are excluded." + ), + ) + + +def _status_grant_record( + row: FormStatusAccessGrant, + *, + selectors: _SubjectSelectors, +) -> DsarRecordRef: + immutable = row.revoked_at is not None + return _record( + "forms_runtime_status_access_grant", + row.id, + "applicant_status_access", + "Applicant status access grant", + { + "instance_id": row.instance_id, + "mode": row.mode, + "applicant_actor_id": ( + row.applicant_actor_id + if row.applicant_actor_id in selectors.actor_ids + else None + ), + "email_selector_matched": _grant_email_matches(row, selectors.email), + "token_ttl_seconds": row.token_ttl_seconds, + "request_limit_per_hour": row.request_limit_per_hour, + "issued_at": _iso(row.issued_at), + "last_accessed_at": _iso(row.last_accessed_at), + "revoked_at": _iso(row.revoked_at), + }, + observed_at=row.updated_at, + immutable=immutable, + retention_reason=( + "Revoked status-access grants are retained as access evidence." + if immutable + else None + ), + ) + + +def _status_token_record(row: FormStatusAccessToken) -> DsarRecordRef: + return _record( + "forms_runtime_status_token_lifecycle", + row.id, + "status_access_evidence", + "Applicant status token lifecycle", + { + "issued_at": _iso(row.issued_at), + "expires_at": _iso(row.expires_at), + "revoked_at": _iso(row.revoked_at), + "last_used_at": _iso(row.last_used_at), + }, + observed_at=row.updated_at, + immutable=True, + retention_reason=( + "The minimized token lifecycle is retained as access evidence; token " + "identifiers, digests, notification references, and details are excluded." + ), + ) + + +def _confirmation_record( + row: FormAssistedConfirmation, + *, + selectors: _SubjectSelectors, +) -> DsarRecordRef: + return _record( + "forms_runtime_assisted_confirmation", + row.id, + "assisted_intake_evidence", + "Assisted-intake read-back confirmation", + { + "instance_id": row.instance_id, + "instance_revision": row.instance_revision, + "outcome": row.outcome, + "method": row.method, + "confirmed_by_ref": ( + row.confirmed_by_ref + if row.confirmed_by_ref in selectors.actor_ids + else None + ), + "operator_actor_id": ( + row.operator_actor_id + if row.operator_actor_id in selectors.actor_ids + else None + ), + "confirmed_at": _iso(row.confirmed_at), + "has_correction_note": bool(row.correction_note), + }, + observed_at=row.confirmed_at, + immutable=True, + retention_reason=( + "Read-back confirmation is immutable assisted-intake consent and " + "accountability evidence. Free text and hashes are excluded." + ), + ) + + +def _acknowledgement_record( + row: FormAcknowledgement, + *, + selectors: _SubjectSelectors, +) -> DsarRecordRef: + return _record( + "forms_runtime_acknowledgement", + row.id, + "applicant_acknowledgement_evidence", + "Form acknowledgement", + { + "instance_id": row.instance_id, + "instance_revision": row.instance_revision, + "statement_id": _bounded_text(row.statement_id, 255), + "statement_version": _bounded_text(row.statement_version, 255), + "actor_id": row.actor_id if row.actor_id in selectors.actor_ids else None, + "accepted_at": _iso(row.accepted_at), + }, + observed_at=row.accepted_at, + immutable=True, + retention_reason=( + "Applicant acknowledgement is immutable consent and submission evidence." + ), + ) + + +def _append_runtime_operator_attribution( + session: Session, + *, + tenant_id: str, + selectors: _SubjectSelectors, + append: object, +) -> None: + if not selectors.actor_ids: + return + definitions = ( + ( + FormInstanceIdentity, + FormInstanceIdentity.created_by, + "created_form_instance", + "instance_id", + None, + "created_at", + ), + ( + FormInstanceRevision, + FormInstanceRevision.changed_by, + "changed_form_instance_revision", + "instance_id", + "revision", + "recorded_at", + ), + ( + FormInstanceEvent, + FormInstanceEvent.actor_id, + "performed_form_lifecycle_event", + "instance_id", + "instance_revision", + "occurred_at", + ), + ( + FormIntakeSession, + FormIntakeSession.created_by, + "created_form_intake_session", + "instance_id", + None, + "created_at", + ), + ( + FormAssistedConfirmation, + FormAssistedConfirmation.operator_actor_id, + "recorded_assisted_confirmation", + "instance_id", + "instance_revision", + "confirmed_at", + ), + ) + for ( + model, + actor_field, + activity, + instance_field, + revision_field, + time_field, + ) in definitions: + rows = _query( + session, + model, + tenant_id, + (actor_field.in_(selectors.actor_ids),), + ) + for row in rows: + append( # type: ignore[operator] + _record( + "forms_runtime_operator_attribution", + f"{activity}:{row.id}", + "operator_accountability_evidence", + "Forms Runtime operator attribution", + { + "activity": activity, + "instance_id": getattr(row, instance_field), + "instance_revision": ( + getattr(row, revision_field) if revision_field else None + ), + "observed_at": _iso(getattr(row, time_field)), + }, + observed_at=getattr(row, time_field), + immutable=True, + retention_reason=( + "Operator attribution is retained as accountability evidence; " + "form values, intake details, payloads, hashes, and errors are excluded." + ), + ) + ) + + +def _append_operator_configuration_attribution( + session: Session, + *, + tenant_id: str, + selectors: _SubjectSelectors, + append: object, +) -> None: + if not selectors.actor_ids: + return + definitions = ( + (FormIntakeProfile, "intake_profile_configuration"), + (FormStatusAccessPolicy, "status_access_policy_configuration"), + ) + for model, activity in definitions: + rows = _query( + session, + model, + tenant_id, + ( + model.created_by.in_(selectors.actor_ids), + model.updated_by.in_(selectors.actor_ids), + ), + ) + for row in rows: + append( # type: ignore[operator] + _record( + "forms_runtime_operator_attribution", + f"{activity}:{row.id}", + "operator_accountability_evidence", + "Forms Runtime configuration attribution", + { + "activity": activity, + "definition_id": row.definition_id, + "definition_revision": row.definition_revision, + "revision": row.revision, + "created_at": _iso(row.created_at), + "updated_at": _iso(row.updated_at), + }, + observed_at=row.updated_at, + immutable=True, + retention_reason=( + "Configuration attribution is retained as accountability " + "evidence; opaque policy/profile details are excluded." + ), + ) + ) + + +def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None: + groups = { + "account_id": ( + subject.account_id, + subject.external_references.get("forms_runtime.account"), + subject.external_references.get("access.account"), + ), + "identity_id": ( + subject.identity_id, + subject.external_references.get("forms_runtime.identity"), + subject.external_references.get("identity.id"), + ), + "membership_id": ( + subject.membership_id, + subject.external_references.get("forms_runtime.membership"), + subject.external_references.get("tenancy.membership"), + ), + } + normalized: dict[str, str | None] = {} + for key, values in groups.items(): + distinct = {value for item in values if (value := _normalized_id(item))} + if len(distinct) > 1: + return None + normalized[key] = next(iter(distinct), None) + aliases = { + "forms_runtime.instance": "instance", + "forms_runtime.revision": "revision", + "forms_runtime.event": "event", + "forms_runtime.handoff": "handoff", + "forms_runtime.intake_session": "intake_session", + "forms_runtime.status_grant": "status_grant", + "forms_runtime.confirmation": "confirmation", + "forms_runtime.acknowledgement": "acknowledgement", + } + references = { + target: value + for source, target in aliases.items() + if (value := _normalized_id(subject.external_references.get(source))) + } + return _SubjectSelectors( + email=_normalized_email(subject.email), + references=references, + **normalized, + ) + + +def _query( + session: Session, + model: type, + tenant_id: str, + conditions: Sequence[object], +) -> list[object]: + if not conditions: + return [] + rows = ( + session.query(model) + .filter(model.tenant_id == tenant_id, or_(*conditions)) + .order_by(model.id.asc()) + .limit(_MAX_RECORDS + 1) + .all() + ) + if len(rows) > _MAX_RECORDS: + raise ValueError( + "Forms Runtime DSAR match limit exceeded; narrow the selectors." + ) + return rows + + +def _grant_email_matches( + grant: FormStatusAccessGrant, + email: str | None, +) -> bool: + if not email or not grant.recipient_email_sha256: + return False + digest = hashlib.sha256(f"{grant.grant_id}\0{email}".encode()).hexdigest() + return digest == grant.recipient_email_sha256 + + +def _safe_form_value(value: object, *, depth: int) -> object: + if depth >= _MAX_VALUE_DEPTH: + return "[nested value omitted]" + if isinstance(value, Mapping): + result: dict[str, object] = {} + for key, item in list(value.items())[:500]: + clean_key = str(key)[:255] + if any( + marker in clean_key.casefold() for marker in _SENSITIVE_VALUE_MARKERS + ): + result[clean_key] = "[credential-like value omitted]" + else: + result[clean_key] = _safe_form_value(item, depth=depth + 1) + return result + if isinstance(value, (list, tuple)): + return [_safe_form_value(item, depth=depth + 1) for item in value[:500]] + if isinstance(value, str): + return value[:20_000] + if value is None or isinstance(value, (bool, int, float)): + return value + return str(value)[:2_000] + + +def _safe_reference(value: object) -> dict[str, object] | None: + if not isinstance(value, Mapping): + return None + return { + "kind": _bounded_text(_optional_text(value.get("kind")), 80), + "owner_module": _bounded_text(_optional_text(value.get("owner_module")), 100), + "object_id": _bounded_text(_optional_text(value.get("object_id")), 500), + "version": _bounded_text(_optional_text(value.get("version")), 255), + } + + +def _sequence_length(value: object) -> int: + return len(value) if isinstance(value, (list, tuple)) else 0 + + +def _validate_record(record: DsarRecordRef) -> None: + if record.provider_id != "forms_runtime" or record.module_id != "forms_runtime": + raise ValueError("Forms Runtime DSAR received a foreign provider record.") + + +def _validate_action(action: DsarErasureActionRef) -> None: + if action.provider_id != "forms_runtime" or action.module_id != "forms_runtime": + raise ValueError("Forms Runtime DSAR received a foreign provider action.") + + +def _record( + resource_type: str, + resource_id: str, + category: str, + title: str, + data: Mapping[str, object], + *, + observed_at: datetime | None, + immutable: bool = False, + retention_reason: str | None = None, +) -> DsarRecordRef: + return DsarRecordRef( + provider_id="forms_runtime", + module_id="forms_runtime", + resource_type=resource_type, + resource_id=resource_id, + category=category, + title=title, + data=data, + observed_at=observed_at, + immutable_evidence=immutable, + retention_reason=retention_reason, + source_path="/forms-runtime", + ) + + +def _session(value: object) -> Session: + if not isinstance(value, Session): + raise TypeError("Forms Runtime DSAR provider requires a SQLAlchemy session.") + return value + + +def _normalized_email(value: object) -> str | None: + candidate = str(value or "").strip().casefold() + if ( + not candidate + or len(candidate) > 320 + or candidate.count("@") != 1 + or any(character.isspace() for character in candidate) + ): + return None + local, domain = candidate.rsplit("@", 1) + if not local or "." not in domain or domain.startswith(".") or domain.endswith("."): + return None + return candidate + + +def _normalized_id(value: object) -> str | None: + if value is None: + return None + normalized = str(value).strip() + return normalized or None + + +def _optional_text(value: object) -> str | None: + normalized = str(value or "").strip() + return normalized or None + + +def _bounded_text(value: str | None, limit: int) -> str | None: + return value[:limit] if value else None + + +def _iso(value: datetime | None) -> str | None: + if value is None: + return None + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + return value.isoformat() + + +__all__ = ["FORMS_RUNTIME_DSAR_CAPABILITY", "FormsRuntimeDsarProvider"] diff --git a/src/govoplan_forms_runtime/backend/manifest.py b/src/govoplan_forms_runtime/backend/manifest.py index 8825cfb..11879a9 100644 --- a/src/govoplan_forms_runtime/backend/manifest.py +++ b/src/govoplan_forms_runtime/backend/manifest.py @@ -41,6 +41,10 @@ from govoplan_core.core.search import SearchSourceProviderRegistration from govoplan_core.core.views import ViewSurface from govoplan_core.db.base import Base from govoplan_forms_runtime.backend.db import models as runtime_models +from govoplan_forms_runtime.backend.dsar_provider import ( + FORMS_RUNTIME_DSAR_CAPABILITY, + FormsRuntimeDsarProvider, +) from govoplan_forms_runtime.backend.record_source import ( CAPABILITY_RECORD_SOURCE_FORMS_RUNTIME, create_forms_runtime_record_source, @@ -210,6 +214,11 @@ def _service_launcher(context: ModuleContext) -> FormsServiceLauncher: return FormsServiceLauncher(context.registry) +def _dsar_provider(context: ModuleContext) -> FormsRuntimeDsarProvider: + del context + return FormsRuntimeDsarProvider() + + def _tenant_summary(session, tenant_id: str) -> dict[str, int]: current = session.query(runtime_models.FormInstanceRevision).filter( runtime_models.FormInstanceRevision.tenant_id == tenant_id, @@ -393,6 +402,10 @@ manifest = ModuleManifest( name=CAPABILITY_RECORD_SOURCE_FORMS_RUNTIME, version="1.0.0", ), + ModuleInterfaceProvider( + name=FORMS_RUNTIME_DSAR_CAPABILITY, + version="0.1.0", + ), ), requires_interfaces=( ModuleInterfaceRequirement( @@ -430,6 +443,7 @@ manifest = ModuleManifest( CAPABILITY_FORMS_RUNTIME_SERVICE_LAUNCHER: _service_launcher, CAPABILITY_RECORD_SOURCE_FORMS_RUNTIME: create_forms_runtime_record_source, CAPABILITY_APPLICATION_STATUS_PROJECTION: _status_projection, + FORMS_RUNTIME_DSAR_CAPABILITY: _dsar_provider, }, capability_documentation={ CAPABILITY_FORMS_RUNTIME_REGISTRY: CapabilityDocumentation( @@ -452,6 +466,11 @@ manifest = ModuleManifest( summary="Resolves tenant-bound, deliberately limited application status and configured access challenges.", contract_version="1.0.0", ), + FORMS_RUNTIME_DSAR_CAPABILITY: CapabilityDocumentation( + label="Forms Runtime data-subject request provider", + summary="Finds minimized tenant-scoped submission, intake, status-access, and handoff facts without exposing credentials.", + contract_version="0.1.0", + ), }, migration_spec=MigrationSpec( module_id=MODULE_ID, @@ -499,6 +518,26 @@ manifest = ModuleManifest( ), tenant_summary_providers=(_tenant_summary,), documentation=( + DocumentationTopic( + id="forms_runtime.data-subject-requests", + title="Form-submission data-subject requests", + summary="Export subject-owned form values and governed lifecycle evidence without exposing access credentials or unrelated assisted-intake data.", + body=( + "The Forms Runtime DSAR provider matches exact-tenant canonical actors, normalized email status grants, applicant confirmations and acknowledgements, and explicit runtime references. Assisted operators are recorded as minimized attribution but are not treated as the applicant when the assisted session identifies another actor. Corroborated subject instances export bounded form values and typed definition, status, intake, acknowledgement, status-access, and handoff facts. Credential-like form keys are redacted; token and email hashes, idempotency keys, request and payload hashes, opaque metadata/details, validation internals, evidence identifiers, provider replay keys, errors, and unrelated submissions are excluded. " + "Immutable revisions, events, handoffs, confirmations, acknowledgements, intake evidence, and minimized token lifecycle receive retention actions. Current drafts and active status grants require authorized manual review; Forms Runtime publishes no automatic DSAR mutation. Forms owns definitions, Portal presents public/status routes, and Cases or Workflow Engine own downstream handoff targets." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("user", "operator", "module_admin", "auditor"), + related_modules=("forms", "portal", "cases", "workflow_engine"), + links=( + DocumentationLink( + label="Forms Runtime security and recovery", + href="govoplan-forms-runtime/docs/FORMS_RUNTIME_DOMAIN_BOUNDARY.md", + kind="repository", + ), + ), + ), DocumentationTopic( id="forms_runtime.submissions", title="Complete and manage Forms", diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..64be977 --- /dev/null +++ b/tests/test_dsar_provider.py @@ -0,0 +1,701 @@ +from __future__ import annotations + +import hashlib +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 ( + DsarErasureActionRef, + DsarProvider, + DsarRecordRef, + 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_forms_runtime.backend.db.models import ( + FormAcknowledgement, + FormAssistedConfirmation, + FormHandoffEffect, + FormInstanceEvent, + FormInstanceIdentity, + FormInstanceRevision, + FormIntakeProfile, + FormIntakeSession, + FormStatusAccessGrant, + FormStatusAccessPolicy, + FormStatusAccessToken, +) +from govoplan_forms_runtime.backend.dsar_provider import ( + FORMS_RUNTIME_DSAR_CAPABILITY, + FormsRuntimeDsarProvider, +) +from govoplan_forms_runtime.backend.manifest import manifest + + +NOW = datetime(2026, 8, 21, 14, 0, tzinfo=UTC) + + +class _Registry: + def __init__( + self, + provider: FormsRuntimeDsarProvider, + *, + active: bool = True, + ) -> None: + self.provider = provider + self.active = active + + def capability_names(self): + return (FORMS_RUNTIME_DSAR_CAPABILITY,) + + def capability_owner(self, name): + self._assert_capability(name) + return "forms_runtime" + + def tenant_entitlement_resolver(self): + active = self.active + + class _Resolver: + @staticmethod + def resolve(session, tenant_id): + del session, tenant_id + return type( + "State", + (), + {"effective_modules": ("forms_runtime",) if active else ()}, + )() + + return _Resolver() + + def require_tenant_capability(self, name, session, **kwargs): + del session, kwargs + self._assert_capability(name) + return self.provider + + def manifests(self): + return (type("Manifest", (), {"id": "forms_runtime"})(),) + + @staticmethod + def _assert_capability(name: str) -> None: + if name != FORMS_RUNTIME_DSAR_CAPABILITY: + raise KeyError(name) + + +def _snapshot(instance_id: str, *, name: str) -> dict[str, object]: + return { + "instance_id": instance_id, + "definition_ref": { + "kind": "form", + "owner_module": "forms", + "object_id": "permit-form", + "version": "3", + }, + "service_ref": { + "kind": "service", + "owner_module": "services", + "object_id": "permit-service", + "version": "2", + }, + "values": { + "name": name, + "email": "subject@example.org", + "password_token": "credential-value-do-not-export", + "nested": {"address": "Example Street 1"}, + }, + "validation_results": [{"message": "private-validation-do-not-export"}], + "attachment_refs": [{"evidence_id": "private-attachment-do-not-export"}], + "signature_refs": [{"evidence_id": "private-signature-do-not-export"}], + "handoff_refs": [{"object_id": "case-1"}], + "receipt_id": "receipt-1", + "change_reason": "private-change-reason-do-not-export", + "metadata": {"secret": "private-metadata-do-not-export"}, + } + + +class FormsRuntimeDsarProviderTests(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 = FormsRuntimeDsarProvider() + self.assertIsInstance(self.provider, DsarProvider) + self._seed_subject_instance() + self._seed_assisted_instance() + self._seed_unrelated_instances() + self._seed_operator_configuration() + self.session.commit() + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def _seed_subject_instance(self) -> None: + identity = FormInstanceIdentity( + id="identity-row-1", + tenant_id="tenant-1", + instance_id="instance-1", + definition_id="permit-form", + definition_revision="3", + created_by="applicant-1", + ) + self.session.add(identity) + self.session.flush() + first = FormInstanceRevision( + id="revision-row-1", + tenant_id="tenant-1", + instance_id="instance-1", + identity_id=identity.id, + revision=1, + status="draft", + recorded_at=NOW, + superseded_at=NOW + timedelta(minutes=1), + snapshot=_snapshot("instance-1", name="Ada Example draft"), + changed_by="applicant-1", + ) + second = FormInstanceRevision( + id="revision-row-2", + tenant_id="tenant-1", + instance_id="instance-1", + identity_id=identity.id, + revision=2, + previous_revision_id=first.id, + status="submitted", + recorded_at=NOW + timedelta(minutes=1), + snapshot=_snapshot("instance-1", name="Ada Example"), + changed_by="applicant-1", + ) + events = ( + FormInstanceEvent( + id="event-row-1", + tenant_id="tenant-1", + instance_id="instance-1", + instance_revision=1, + event_id="event-public-1", + event_type="forms_runtime.instance.started", + status="draft", + occurred_at=NOW, + actor_id="applicant-1", + idempotency_key="private-event-key-do-not-export", + request_sha256="a" * 64, + payload={"secret": "private-event-payload-do-not-export"}, + ), + FormInstanceEvent( + id="event-row-2", + tenant_id="tenant-1", + instance_id="instance-1", + instance_revision=2, + event_id="event-public-2", + event_type="forms_runtime.instance.submitted", + status="submitted", + occurred_at=NOW + timedelta(minutes=1), + actor_id="applicant-1", + idempotency_key="private-submit-key-do-not-export", + request_sha256="b" * 64, + payload={"secret": "private-submit-payload-do-not-export"}, + ), + ) + handoff = FormHandoffEffect( + id="handoff-row-1", + tenant_id="tenant-1", + instance_id="instance-1", + effect_id="handoff-public-1", + instance_revision=2, + idempotency_key="private-handoff-key-do-not-export", + provider_key="private-provider-key-do-not-export", + request_sha256="c" * 64, + binding_kind="case", + binding_reference="permit-case", + provider_capability="cases.service_launcher", + state="succeeded", + attempt_count=1, + requested_at=NOW + timedelta(minutes=1), + resolved_at=NOW + timedelta(minutes=2), + target_ref={ + "kind": "case", + "owner_module": "cases", + "object_id": "case-1", + "version": "1", + }, + evidence=[{"secret": "private-handoff-evidence-do-not-export"}], + last_error="private-error-do-not-export", + details={"secret": "private-handoff-details-do-not-export"}, + ) + intake = FormIntakeSession( + id="intake-row-1", + tenant_id="tenant-1", + session_id="intake-public-1", + profile_id="profile-row-1", + token_sha256="d" * 64, + mode="authenticated", + status="submitted", + instance_id="instance-1", + actor_id="applicant-1", + idempotency_key="private-intake-key-do-not-export", + request_sha256="e" * 64, + expires_at=NOW + timedelta(hours=1), + started_at=NOW, + submitted_at=NOW + timedelta(minutes=1), + created_by="applicant-1", + details={"secret": "private-intake-details-do-not-export"}, + ) + grant_id = "status-grant-public-1" + grant = FormStatusAccessGrant( + id="grant-row-1", + tenant_id="tenant-1", + grant_id=grant_id, + policy_id="policy-row-1", + instance_id="instance-1", + tracking_id="private-tracking-id-do-not-export", + mode="email_link", + applicant_actor_id="applicant-1", + recipient_email_sha256=hashlib.sha256( + f"{grant_id}\0subject@example.org".encode() + ).hexdigest(), + token_ttl_seconds=900, + request_limit_per_hour=3, + issued_at=NOW + timedelta(minutes=1), + details={"secret": "private-grant-details-do-not-export"}, + ) + token = FormStatusAccessToken( + id="token-row-1", + tenant_id="tenant-1", + token_id="private-token-id-do-not-export", + grant_id=grant.id, + token_sha256="f" * 64, + issued_at=NOW + timedelta(minutes=1), + expires_at=NOW + timedelta(minutes=16), + notification_id="private-notification-id-do-not-export", + details={"secret": "private-token-details-do-not-export"}, + ) + confirmation = FormAssistedConfirmation( + id="confirmation-row-1", + tenant_id="tenant-1", + confirmation_id="confirmation-public-1", + intake_session_id=intake.id, + instance_id="instance-1", + instance_revision=2, + outcome="confirmed", + method="read_back", + confirmed_by_ref="applicant-1", + operator_actor_id="operator-other", + confirmed_at=NOW + timedelta(minutes=1), + payload_sha256="1" * 64, + idempotency_key="private-confirmation-key-do-not-export", + request_sha256="2" * 64, + correction_note="private-correction-note-do-not-export", + details={"secret": "private-confirmation-details-do-not-export"}, + ) + acknowledgement = FormAcknowledgement( + id="ack-row-1", + tenant_id="tenant-1", + acknowledgement_id="ack-public-1", + instance_id="instance-1", + instance_revision=2, + statement_id="truthful-submission", + statement_version="1", + actor_id="applicant-1", + accepted_at=NOW + timedelta(minutes=1), + payload_sha256="3" * 64, + idempotency_key="private-ack-key-do-not-export", + request_sha256="4" * 64, + details={"secret": "private-ack-details-do-not-export"}, + ) + self.session.add_all( + ( + first, + second, + *events, + handoff, + intake, + grant, + token, + confirmation, + acknowledgement, + ) + ) + + def _seed_assisted_instance(self) -> None: + identity = FormInstanceIdentity( + id="identity-assisted", + tenant_id="tenant-1", + instance_id="instance-assisted", + definition_id="permit-form", + definition_revision="3", + created_by="operator-1", + ) + self.session.add(identity) + self.session.flush() + self.session.add_all( + ( + FormInstanceRevision( + id="revision-assisted", + tenant_id="tenant-1", + instance_id="instance-assisted", + identity_id=identity.id, + revision=1, + status="draft", + recorded_at=NOW, + snapshot=_snapshot( + "instance-assisted", + name="Assisted Applicant Private Name", + ), + changed_by="operator-1", + ), + FormInstanceEvent( + id="event-assisted", + tenant_id="tenant-1", + instance_id="instance-assisted", + instance_revision=1, + event_id="event-assisted-public", + event_type="forms_runtime.instance.started", + status="draft", + occurred_at=NOW, + actor_id="operator-1", + idempotency_key="assisted-private-key", + request_sha256="5" * 64, + payload={"secret": "assisted-private-payload"}, + ), + FormIntakeSession( + id="intake-assisted", + tenant_id="tenant-1", + session_id="intake-assisted-public", + profile_id="profile-row-1", + token_sha256="6" * 64, + mode="assisted", + status="started", + instance_id="instance-assisted", + actor_id="applicant-assisted", + idempotency_key="assisted-intake-private-key", + request_sha256="7" * 64, + expires_at=NOW + timedelta(hours=1), + started_at=NOW, + created_by="operator-1", + details={"secret": "assisted-details-private"}, + ), + ) + ) + + def _seed_unrelated_instances(self) -> None: + for tenant_id, suffix, actor in ( + ("tenant-1", "unrelated", "unrelated-actor"), + ("tenant-2", "other-tenant", "applicant-1"), + ): + identity = FormInstanceIdentity( + id=f"identity-{suffix}", + tenant_id=tenant_id, + instance_id=f"instance-{suffix}", + definition_id="permit-form", + definition_revision="3", + created_by=actor, + ) + self.session.add(identity) + self.session.flush() + self.session.add( + FormInstanceRevision( + id=f"revision-{suffix}", + tenant_id=tenant_id, + instance_id=identity.instance_id, + identity_id=identity.id, + revision=1, + status="submitted", + recorded_at=NOW, + snapshot=_snapshot( + identity.instance_id, + name=f"private-{suffix}-name-do-not-export", + ), + changed_by=actor, + ) + ) + + def _seed_operator_configuration(self) -> None: + self.session.add_all( + ( + FormIntakeProfile( + id="profile-row-1", + tenant_id="tenant-1", + profile_id="profile-public-1", + public_id="public-profile-id", + definition_id="permit-form", + definition_revision="3", + mode="assisted", + enabled=True, + revision=1, + custodian_ref="unit-1", + draft_ttl_seconds=3600, + invitation_ttl_seconds=3600, + rate_limit_per_minute=10, + created_by="operator-1", + updated_by="operator-1", + details={"secret": "private-profile-details-do-not-export"}, + ), + FormStatusAccessPolicy( + id="policy-row-1", + tenant_id="tenant-1", + policy_id="policy-public-1", + definition_id="permit-form", + definition_revision="3", + mode="email_link", + enabled=True, + revision=1, + email_field_key="email", + token_ttl_seconds=900, + request_limit_per_hour=3, + created_by="operator-1", + updated_by="operator-1", + details={"secret": "private-policy-details-do-not-export"}, + ), + ) + ) + + def test_actor_search_exports_subject_instance_and_minimizes_internals( + self, + ) -> None: + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef(account_id="applicant-1"), + ) + types = {item.resource_type for item in records} + self.assertIn("forms_runtime_instance_revision", types) + self.assertIn("forms_runtime_handoff_effect", types) + self.assertIn("forms_runtime_status_access_grant", types) + self.assertIn("forms_runtime_status_token_lifecycle", types) + self.assertIn("forms_runtime_assisted_confirmation", types) + self.assertIn("forms_runtime_acknowledgement", types) + + exported = json.dumps([item.to_dict() for item in records], sort_keys=True) + self.assertIn("Ada Example", exported) + self.assertIn("Example Street 1", exported) + self.assertNotIn("credential-value-do-not-export", exported) + self.assertNotIn("private-attachment-do-not-export", exported) + self.assertNotIn("private-signature-do-not-export", exported) + self.assertNotIn("private-validation-do-not-export", exported) + self.assertNotIn("private-event-payload-do-not-export", exported) + self.assertNotIn("private-token-id-do-not-export", exported) + self.assertNotIn("private-tracking-id-do-not-export", exported) + self.assertNotIn("private-notification-id-do-not-export", exported) + self.assertNotIn("private-provider-key-do-not-export", exported) + self.assertNotIn("private-error-do-not-export", exported) + self.assertNotIn("private-correction-note-do-not-export", exported) + self.assertNotIn("private-unrelated-name-do-not-export", exported) + self.assertNotIn("private-other-tenant-name-do-not-export", exported) + + def test_email_selector_matches_grant_specific_digest(self) -> None: + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef(email=" Subject@Example.org "), + ) + self.assertTrue( + any( + item.resource_type == "forms_runtime_status_access_grant" + and item.data["email_selector_matched"] is True + for item in records + ) + ) + self.assertTrue( + any( + item.resource_type == "forms_runtime_instance_revision" + for item in records + ) + ) + + def test_assisted_operator_is_attributed_but_not_treated_as_applicant(self) -> None: + operator_records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef(account_id="operator-1"), + ) + exported = json.dumps( + [item.to_dict() for item in operator_records], sort_keys=True + ) + self.assertIn("forms_runtime_operator_attribution", exported) + self.assertIn("created_form_instance", exported) + self.assertIn("intake_profile_configuration", exported) + self.assertNotIn("Assisted Applicant Private Name", exported) + self.assertNotIn("assisted-private-payload", exported) + self.assertNotIn("private-profile-details-do-not-export", exported) + self.assertNotIn("private-policy-details-do-not-export", exported) + + applicant_records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef(account_id="applicant-assisted"), + ) + self.assertIn( + "Assisted Applicant Private Name", + json.dumps([item.to_dict() for item in applicant_records]), + ) + + def test_direct_and_canonical_conflicts_fail_closed(self) -> None: + direct = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + external_references={"forms_runtime.instance": "instance-1"} + ), + ) + self.assertTrue( + any( + item.resource_type == "forms_runtime_instance_revision" + for item in direct + ) + ) + conflict = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + account_id="applicant-1", + external_references={"forms_runtime.instance": "instance-unrelated"}, + ), + ) + self.assertEqual((), conflict) + alias_conflict = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + account_id="applicant-1", + external_references={"forms_runtime.account": "other-account"}, + ), + ) + self.assertEqual((), alias_conflict) + + def test_planning_retains_evidence_and_reviews_active_status_access(self) -> None: + subject = DsarSubjectRef(account_id="applicant-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.assertGreater(sum(item.kind == "retain" for item in actions), 5) + self.assertEqual(1, sum(item.kind == "manual_review" for item in actions)) + self.assertTrue(all(not item.executable for item in actions)) + results = self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=actions, + request_id="dsar-forms-1", + ) + self.assertTrue(all(item.status == "blocked" for item in results)) + + def test_foreign_records_and_actions_are_rejected(self) -> None: + subject = DsarSubjectRef(account_id="applicant-1") + foreign_record = DsarRecordRef( + provider_id="foreign", + module_id="foreign", + resource_type="foreign", + resource_id="foreign-1", + category="foreign", + title="Foreign record", + ) + with self.assertRaisesRegex(ValueError, "foreign provider record"): + self.provider.plan_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + records=(foreign_record,), + ) + foreign_action = DsarErasureActionRef( + action_id="foreign:delete:1", + provider_id="foreign", + module_id="foreign", + kind="delete", + resource_type="foreign", + resource_id="foreign-1", + title="Delete foreign", + rationale="No", + executable=True, + ) + with self.assertRaisesRegex(ValueError, "foreign provider action"): + self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=(foreign_action,), + request_id="dsar-forms-1", + ) + + def test_workflow_discovers_only_the_active_tenant_capability(self) -> None: + active = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-FORMS-1", + request_kind="access", + subject=DsarSubjectRef(account_id="applicant-1"), + purpose="Subject access request", + legal_basis=None, + due_at=None, + requested_by_account_id="privacy-operator", + ) + search_data_subject_request( + self.session, + registry=_Registry(self.provider), + row=active, + expected_revision=1, + ) + self.assertEqual("searched", active.status) + self.assertEqual( + [FORMS_RUNTIME_DSAR_CAPABILITY], + active.coverage["provider_capabilities"], + ) + self.assertEqual(["forms_runtime"], active.coverage["covered_modules"]) + + inactive = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-FORMS-2", + request_kind="access", + subject=DsarSubjectRef(account_id="applicant-1"), + purpose="Inactive module coverage", + legal_basis=None, + due_at=None, + requested_by_account_id="privacy-operator", + ) + search_data_subject_request( + self.session, + registry=_Registry(self.provider, active=False), + row=inactive, + expected_revision=1, + ) + self.assertEqual([], inactive.coverage["provider_capabilities"]) + self.assertEqual( + [FORMS_RUNTIME_DSAR_CAPABILITY], + inactive.coverage["inactive_provider_capabilities"], + ) + self.assertEqual(0, inactive.search_result["record_count"]) + + def test_manifest_registers_and_documents_the_capability(self) -> None: + self.assertIn(FORMS_RUNTIME_DSAR_CAPABILITY, manifest.capability_factories) + self.assertIn( + FORMS_RUNTIME_DSAR_CAPABILITY, + manifest.capability_documentation, + ) + self.assertIn( + FORMS_RUNTIME_DSAR_CAPABILITY, + {item.name for item in manifest.provides_interfaces}, + ) + self.assertTrue( + any( + topic.id == "forms_runtime.data-subject-requests" + and {"admin", "user"}.issubset(topic.documentation_types) + for topic in manifest.documentation + ) + ) + + +if __name__ == "__main__": + unittest.main()