From e6457b3f6bd5c1e4fcda2327f5ecc26803fa33fd Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 7 Aug 2026 14:53:07 +0200 Subject: [PATCH] feat: add governed DSAR workflow --- .../b47e6f809a13_data_subject_requests.py | 23 + .../b47e6f809a13_data_subject_requests.py | 77 ++ docs/DATA_SUBJECT_REQUESTS.md | 91 +++ docs/DOCUMENTATION_MAP.md | 1 + docs/GOVERNANCE_MODEL.md | 3 +- src/govoplan_core/celery_app.py | 66 ++ src/govoplan_core/core/campaigns.py | 16 + src/govoplan_core/core/dsar.py | 197 +++++ src/govoplan_core/core/files.py | 57 ++ src/govoplan_core/core/mail.py | 52 ++ src/govoplan_core/core/postbox.py | 38 + src/govoplan_core/core/templates.py | 38 + src/govoplan_core/privacy/dsar_workflow.py | 753 ++++++++++++++++++ src/govoplan_core/server/app.py | 2 + src/govoplan_core/server/dsar.py | 406 ++++++++++ tests/test_dsar_api.py | 162 ++++ tests/test_dsar_workflow.py | 236 ++++++ tests/test_template_contract.py | 21 + tests/test_wheel_runtime.py | 2 +- 19 files changed, 2239 insertions(+), 2 deletions(-) create mode 100644 alembic/dev_versions/b47e6f809a13_data_subject_requests.py create mode 100644 alembic/versions/b47e6f809a13_data_subject_requests.py create mode 100644 docs/DATA_SUBJECT_REQUESTS.md create mode 100644 src/govoplan_core/core/dsar.py create mode 100644 src/govoplan_core/privacy/dsar_workflow.py create mode 100644 src/govoplan_core/server/dsar.py create mode 100644 tests/test_dsar_api.py create mode 100644 tests/test_dsar_workflow.py diff --git a/alembic/dev_versions/b47e6f809a13_data_subject_requests.py b/alembic/dev_versions/b47e6f809a13_data_subject_requests.py new file mode 100644 index 0000000..2bd4dcd --- /dev/null +++ b/alembic/dev_versions/b47e6f809a13_data_subject_requests.py @@ -0,0 +1,23 @@ +from __future__ import annotations + +from importlib.util import module_from_spec, spec_from_file_location +from pathlib import Path + + +_path = ( + Path(__file__).resolve().parents[1] + / "versions" + / "b47e6f809a13_data_subject_requests.py" +) +_spec = spec_from_file_location("govoplan_data_subject_requests_migration", _path) +if _spec is None or _spec.loader is None: + raise RuntimeError(f"Unable to load migration implementation from {_path}") +_module = module_from_spec(_spec) +_spec.loader.exec_module(_module) + +revision = _module.revision +down_revision = _module.down_revision +branch_labels = _module.branch_labels +depends_on = _module.depends_on +upgrade = _module.upgrade +downgrade = _module.downgrade diff --git a/alembic/versions/b47e6f809a13_data_subject_requests.py b/alembic/versions/b47e6f809a13_data_subject_requests.py new file mode 100644 index 0000000..e68b3eb --- /dev/null +++ b/alembic/versions/b47e6f809a13_data_subject_requests.py @@ -0,0 +1,77 @@ +"""add governed data-subject request workflow + +Revision ID: b47e6f809a13 +Revises: a36d8e4f9b12 +Create Date: 2026-08-07 00:00:00.000000 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "b47e6f809a13" +down_revision = "a36d8e4f9b12" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + inspector = sa.inspect(op.get_bind()) + if "core_data_subject_requests" in inspector.get_table_names(): + return + op.create_table( + "core_data_subject_requests", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("reference", sa.String(length=120), nullable=False), + sa.Column("request_kind", sa.String(length=30), nullable=False), + sa.Column("status", sa.String(length=30), nullable=False), + sa.Column("subject", sa.JSON(), nullable=False), + sa.Column("purpose", sa.String(length=1000), nullable=False), + sa.Column("legal_basis", sa.String(length=1000), nullable=True), + sa.Column("due_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("requested_by_account_id", sa.String(length=36), nullable=False), + sa.Column("search_result", sa.JSON(), nullable=False), + sa.Column("erasure_plan", sa.JSON(), nullable=False), + sa.Column("execution_result", sa.JSON(), nullable=False), + sa.Column("coverage", sa.JSON(), nullable=False), + sa.Column("evidence_sha256", sa.String(length=64), nullable=True), + sa.Column("resource_revision", sa.Integer(), nullable=False), + sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("notes", sa.Text(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id", name=op.f("pk_core_data_subject_requests")), + ) + op.create_index( + op.f("ix_core_data_subject_requests_tenant_id"), + "core_data_subject_requests", + ["tenant_id"], + unique=False, + ) + op.create_index( + op.f("ix_core_data_subject_requests_status"), + "core_data_subject_requests", + ["status"], + unique=False, + ) + op.create_index( + op.f("ix_core_data_subject_requests_due_at"), + "core_data_subject_requests", + ["due_at"], + unique=False, + ) + op.create_index( + "ix_core_data_subject_requests_tenant_status", + "core_data_subject_requests", + ["tenant_id", "status"], + unique=False, + ) + + +def downgrade() -> None: + inspector = sa.inspect(op.get_bind()) + if "core_data_subject_requests" in inspector.get_table_names(): + op.drop_table("core_data_subject_requests") diff --git a/docs/DATA_SUBJECT_REQUESTS.md b/docs/DATA_SUBJECT_REQUESTS.md new file mode 100644 index 0000000..7968dfe --- /dev/null +++ b/docs/DATA_SUBJECT_REQUESTS.md @@ -0,0 +1,91 @@ +# Data-Subject Request Contract + +This document defines the provider-neutral workflow for access and erasure +requests. It is an operational control and evidence mechanism. It does not +replace legal review, identity verification, retention policy, or the +institution's statutory response process. + +## Ownership + +Core owns the request aggregate, lifecycle API, optimistic concurrency, +provider discovery, export manifest, execution orchestration, and audit event +names. Modules that store subject-related data own their search, explanation, +retention, and mutation behavior through a `privacy.dsar.` capability. +Core never scans module tables or guesses how a foreign resource may be +erased. + +Access owns the first provider. It finds tenant memberships plus safe account, +identity, assignment, API-key, and session metadata. It does not export secret +hashes, session tokens, IP addresses, or browser fingerprints. Tenant-local +membership data can be anonymized and authentication material can be revoked. +Global accounts and identities require manual system-level review because they +may serve more than one tenant. + +## Lifecycle + +1. A privacy officer records a verified selector, purpose, legal basis, due + date, and internal reference. +2. Search invokes every available tenant capability independently. A provider + failure is isolated and recorded; it cannot turn an incomplete search into + a successful one. +3. The JSON export contains the request, records, provider runs, coverage, + retention reasons, execution evidence, and a SHA-256 manifest digest. +4. An erasure request produces stable provider-owned actions. Immutable + evidence generates an explicit non-executable `retain` decision. +5. Execution accepts only selected executable actions from the current plan. + It requires `If-Match`, the current resource revision, the dedicated erase + permission, and the exact `ERASE ` confirmation phrase. +6. Provider execution is idempotent. Completed or unchanged effects remain + durable in the request's execution evidence. + +The API is rooted at +`/api/v1/admin/privacy/data-subject-requests`. Access exposes the independent +permissions `access:privacy:read`, `access:privacy:manage`, +`access:privacy:export`, and `access:privacy:erase`; the built-in privacy +officer role contains all four. + +## Provider Rules + +A provider must: + +- enforce tenant ownership for every record and action; +- return stable, unique resource and action identities; +- avoid credentials, hashes, tokens, unnecessary telemetry, and unrelated + third-party data; +- distinguish mutable personal data from immutable institutional evidence; +- state a retention reason for immutable evidence; +- propose manual review instead of an automatic action when authority is + ambiguous or a resource spans tenants; +- return exactly one execution result per requested action; +- make execution idempotent and avoid committing the caller's transaction; +- keep all actual mutations inside the owning module. + +Each active module without a DSAR provider is listed in coverage. This is a +deliberate fail-visible state, not proof that the module stores personal data. +An institution may call an export complete only after it has reviewed both the +provider runs and that coverage list. + +## Retention And Evidence + +Erasure and retention are separate decisions. Stable object IDs, authorization +history, function incumbency, formal decisions, delivery evidence, and audit +records may remain necessary for accountability. Providers expose those items +with a concrete reason and Core prevents them from being selected as executable +actions. Policy may further restrict an action, but it must never silently +loosen a provider's retention decision. + +All lifecycle mutations and exports produce tenant audit events. The request +stores an evidence digest after every revision. This digest detects accidental +or unauthorized mutation of the aggregate; it is not a digital signature or a +substitute for signed recovery evidence. + +## Current Limits + +- Access is the first native provider. Other enabled modules appear in the + coverage list until they add a provider or an explicit no-subject-data + declaration is standardized. +- Verification of the requester's identity and statutory deadline escalation + remain institutional workflows outside this API. +- Global account or identity erasure is deliberately manual. +- Exports are JSON. A human-readable signed response package remains a later + Reporting/Templates integration. diff --git a/docs/DOCUMENTATION_MAP.md b/docs/DOCUMENTATION_MAP.md index 62d3247..710f0ad 100644 --- a/docs/DOCUMENTATION_MAP.md +++ b/docs/DOCUMENTATION_MAP.md @@ -20,6 +20,7 @@ operator, and roadmap pages. | Provider-neutral record filing | `RECORDS_FILING_CONTRACT.md` | Exact source-revision identity, current source authorization, idempotent filing, capability discovery, and ownership boundary. | | Temporal data read context | `TEMPORAL_DATA_CONTEXT.md` | Valid-time and recorded-time titlebar selection, HTTP/cache contract, security boundary, and module-adoption rule. | | Cross-module information governance adoption | `INFORMATION_GOVERNANCE_ADOPTION.md` | Manifest evidence and enforcement rules for temporal browsing, purpose-aware access, retention, and institutional context. | +| Data-subject access and erasure requests | `DATA_SUBJECT_REQUESTS.md` | Provider-owned search and mutation, explicit coverage, governed export, retained evidence, permissions, and idempotent execution. | | Context-sensitive F1 help | `CONTEXTUAL_HELP_CONTRACT.md` | Focus, route, module-manifest documentation contexts, Docs projection, and hosted fallback. | | German localization and help quality gate | `LOCALIZATION_AND_HELP_QUALITY.md` | German reference locale, new-installation default, catalog completeness, automatic page associations, and explicit-help review priorities. | | Postbox E2EE target architecture | `POSTBOX_E2EE_ARCHITECTURE.md` | Strategic encrypted postbox/mailbox model, key ownership, role mailbox semantics, and retraction limits. | diff --git a/docs/GOVERNANCE_MODEL.md b/docs/GOVERNANCE_MODEL.md index 775cc00..e842171 100644 --- a/docs/GOVERNANCE_MODEL.md +++ b/docs/GOVERNANCE_MODEL.md @@ -221,7 +221,8 @@ Admin lists use bounded container grids: - recipient import with column mapping; - session/device revocation UI; - backup/restore, monitoring, and update procedures; -- DSAR workflows and evidence bundle verifier; +- additional module providers and signed human-readable response packages for + the implemented DSAR workflow described in `DATA_SUBJECT_REQUESTS.md`; - campaign ownership transfer workflow; - policy impact analysis before delete/disable/unshare/change; - LDAP/OIDC/SAML provisioning; diff --git a/src/govoplan_core/celery_app.py b/src/govoplan_core/celery_app.py index 4f7d6d3..0966e75 100644 --- a/src/govoplan_core/celery_app.py +++ b/src/govoplan_core/celery_app.py @@ -18,7 +18,9 @@ from celery.signals import ( from govoplan_core.core.campaigns import ( CAPABILITY_CAMPAIGNS_DELIVERY_TASKS, + CAPABILITY_CAMPAIGNS_SCHEDULES, CampaignDeliveryTaskProvider, + CampaignScheduleProvider, ) from govoplan_core.core.calendar import ( CAPABILITY_CALENDAR_OUTBOX, @@ -100,6 +102,7 @@ celery.conf.update( task_routes={ "govoplan.campaigns.send_email": {"queue": "send_email"}, "govoplan.campaigns.append_sent": {"queue": "append_sent"}, + "govoplan.campaigns.dispatch_schedules": {"queue": "default"}, "govoplan.notifications.deliver": {"queue": "notifications"}, "govoplan.notifications.deliver_pending": {"queue": "notifications"}, "govoplan.mail.dispatch_outbox": {"queue": "mail"}, @@ -132,6 +135,11 @@ celery.conf.update( "schedule": 60.0, "args": (None, 100), }, + "campaign-schedules-every-minute": { + "task": "govoplan.campaigns.dispatch_schedules", + "schedule": 60.0, + "args": (None, 50), + }, "mail-outbox-every-five-seconds": { "task": "govoplan.mail.dispatch_outbox", "schedule": 5.0, @@ -555,6 +563,18 @@ def _campaign_delivery_tasks( return capability +def _campaign_schedules( + registry: PlatformRegistry | None = None, +) -> CampaignScheduleProvider | None: + registry = registry or _platform_registry() + if not registry.has_capability(CAPABILITY_CAMPAIGNS_SCHEDULES): + return None + capability = registry.require_capability(CAPABILITY_CAMPAIGNS_SCHEDULES) + if not isinstance(capability, CampaignScheduleProvider): + raise RuntimeError("Campaign schedule capability is invalid") + return capability + + def _notification_dispatch( registry: PlatformRegistry | None = None, ) -> NotificationDispatchProvider: @@ -699,6 +719,52 @@ def _idm_assignment_lifecycle( return capability +@celery.task( + name="govoplan.campaigns.dispatch_schedules", + bind=True, + max_retries=0, +) +def dispatch_campaign_schedules( + self, + tenant_id: str | None = None, + limit: int = 50, +): + """Prepare due campaign drafts; delivery always remains a separate action.""" + + from govoplan_core.db.session import get_database + + with get_database().SessionLocal() as session: + registry = _platform_registry() + defaults = { + "selected": 0, + "prepared": 0, + "failed": 0, + "completed": 0, + "coalesced": 0, + "campaign_ids": [], + "operator_actions": [], + } + if not registry.has_capability(CAPABILITY_CAMPAIGNS_SCHEDULES): + return defaults + result = _run_tenant_worker_batches( + registry, + session, + capability_name=CAPABILITY_CAMPAIGNS_SCHEDULES, + tenant_id=tenant_id, + operation=lambda effective_tenant_id: _campaign_schedules( + registry + ).dispatch_due( # type: ignore[union-attr] + session, + tenant_id=effective_tenant_id, + limit=limit, + ), + defaults=defaults, + work_state="new", + ) + session.commit() + return result + + @celery.task(name="govoplan.campaigns.send_email", bind=True, max_retries=0) def send_email(self, job_id: str): """Send one explicitly queued campaign job. diff --git a/src/govoplan_core/core/campaigns.py b/src/govoplan_core/core/campaigns.py index 1aed444..f35bfe7 100644 --- a/src/govoplan_core/core/campaigns.py +++ b/src/govoplan_core/core/campaigns.py @@ -10,6 +10,7 @@ CAPABILITY_CAMPAIGNS_MAIL_POLICY_CONTEXT = "campaigns.mailPolicyContext" CAPABILITY_CAMPAIGNS_ACCESS = "campaigns.access" CAPABILITY_CAMPAIGNS_POLICY_CONTEXT = "campaigns.policyContext" CAPABILITY_CAMPAIGNS_DELIVERY_TASKS = "campaigns.deliveryTasks" +CAPABILITY_CAMPAIGNS_SCHEDULES = "campaigns.schedules" CAPABILITY_CAMPAIGNS_RETENTION = "campaigns.retention" @@ -105,6 +106,21 @@ class CampaignDeliveryTaskProvider(Protocol): ... +@runtime_checkable +class CampaignScheduleProvider(Protocol): + """Durable boundary for preparing due recurring Campaign drafts.""" + + def dispatch_due( + self, + session: object, + *, + tenant_id: str | None = None, + now: datetime | None = None, + limit: int = 50, + ) -> Mapping[str, object]: + ... + + @runtime_checkable class CampaignRetentionProvider(Protocol): def apply_retention( diff --git a/src/govoplan_core/core/dsar.py b/src/govoplan_core/core/dsar.py new file mode 100644 index 0000000..d60a942 --- /dev/null +++ b/src/govoplan_core/core/dsar.py @@ -0,0 +1,197 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from datetime import datetime +from typing import Literal, Protocol, runtime_checkable + + +DSAR_CAPABILITY_PREFIX = "privacy.dsar." + +DsarRequestKind = Literal["access", "erasure", "access_and_erasure"] +DsarActionKind = Literal[ + "delete", + "anonymize", + "revoke", + "detach", + "retain", + "manual_review", +] + + +@dataclass(frozen=True, slots=True) +class DsarSubjectRef: + account_id: str | None = None + identity_id: str | None = None + membership_id: str | None = None + email: str | None = None + external_references: Mapping[str, str] = field(default_factory=dict) + + def has_selector(self) -> bool: + return bool( + self.account_id + or self.identity_id + or self.membership_id + or self.email + or self.external_references + ) + + def to_dict(self) -> dict[str, object]: + return { + "account_id": self.account_id, + "identity_id": self.identity_id, + "membership_id": self.membership_id, + "email": self.email, + "external_references": dict(self.external_references), + } + + +@dataclass(frozen=True, slots=True) +class DsarRecordRef: + provider_id: str + module_id: str + resource_type: str + resource_id: str + category: str + title: str + data: Mapping[str, object] = field(default_factory=dict) + observed_at: datetime | None = None + immutable_evidence: bool = False + retention_reason: str | None = None + source_path: str | None = None + + def to_dict(self) -> dict[str, object]: + return { + "provider_id": self.provider_id, + "module_id": self.module_id, + "resource_type": self.resource_type, + "resource_id": self.resource_id, + "category": self.category, + "title": self.title, + "data": dict(self.data), + "observed_at": self.observed_at.isoformat() if self.observed_at else None, + "immutable_evidence": self.immutable_evidence, + "retention_reason": self.retention_reason, + "source_path": self.source_path, + } + + +@dataclass(frozen=True, slots=True) +class DsarErasureActionRef: + action_id: str + provider_id: str + module_id: str + kind: DsarActionKind + resource_type: str + resource_id: str + title: str + rationale: str + executable: bool + irreversible: bool = False + metadata: Mapping[str, object] = field(default_factory=dict) + + def to_dict(self) -> dict[str, object]: + return { + "action_id": self.action_id, + "provider_id": self.provider_id, + "module_id": self.module_id, + "kind": self.kind, + "resource_type": self.resource_type, + "resource_id": self.resource_id, + "title": self.title, + "rationale": self.rationale, + "executable": self.executable, + "irreversible": self.irreversible, + "metadata": dict(self.metadata), + } + + +@dataclass(frozen=True, slots=True) +class DsarExecutionResultRef: + action_id: str + status: Literal["executed", "unchanged", "failed", "blocked"] + summary: str + evidence: Mapping[str, object] = field(default_factory=dict) + + def to_dict(self) -> dict[str, object]: + return { + "action_id": self.action_id, + "status": self.status, + "summary": self.summary, + "evidence": dict(self.evidence), + } + + +@runtime_checkable +class DsarProvider(Protocol): + provider_id: str + module_id: str + + def search_subject( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + ) -> Sequence[DsarRecordRef]: ... + + def plan_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + records: Sequence[DsarRecordRef], + ) -> Sequence[DsarErasureActionRef]: ... + + def execute_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + actions: Sequence[DsarErasureActionRef], + request_id: str, + ) -> Sequence[DsarExecutionResultRef]: ... + + +def dsar_capability_name(module_id: str) -> str: + normalized = module_id.strip().casefold() + if not normalized or not normalized.replace("_", "").isalnum(): + raise ValueError("DSAR module id must be an identifier.") + return f"{DSAR_CAPABILITY_PREFIX}{normalized}" + + +def dsar_provider_names(registry: object | None) -> tuple[str, ...]: + if registry is None or not hasattr(registry, "capability_names"): + return () + return tuple( + name + for name in registry.capability_names() + if name.startswith(DSAR_CAPABILITY_PREFIX) + ) + + +def dsar_provider( + registry: object, + capability_name: str, +) -> DsarProvider: + provider = registry.require_capability(capability_name) + if not isinstance(provider, DsarProvider): + raise TypeError(f"{capability_name} does not implement DsarProvider") + return provider + + +__all__ = [ + "DSAR_CAPABILITY_PREFIX", + "DsarActionKind", + "DsarErasureActionRef", + "DsarExecutionResultRef", + "DsarProvider", + "DsarRecordRef", + "DsarRequestKind", + "DsarSubjectRef", + "dsar_capability_name", + "dsar_provider", + "dsar_provider_names", +] diff --git a/src/govoplan_core/core/files.py b/src/govoplan_core/core/files.py index 397ba6f..1dae299 100644 --- a/src/govoplan_core/core/files.py +++ b/src/govoplan_core/core/files.py @@ -9,6 +9,7 @@ from govoplan_core.core.access import ResourceAccessExplanationProvider CAPABILITY_FILES_ACCESS = "files.access" CAPABILITY_FILES_ARTIFACT_STORE = "files.artifact_store" +CAPABILITY_FILES_POSTBOX_REFERENCES = "files.postbox_references" @dataclass(frozen=True, slots=True) @@ -34,6 +35,30 @@ class ManagedArtifactRef: provenance: Mapping[str, object] = field(default_factory=dict) +@dataclass(frozen=True, slots=True) +class PostboxFileReferenceRequest: + reference_type: str + reference_id: str + postbox_id: str + message_id: str + + +@dataclass(frozen=True, slots=True) +class PostboxFileReferenceRef: + reference_type: str + reference_id: str + available: bool + reason_code: str + file_asset_id: str | None = None + file_version_id: str | None = None + filename: str | None = None + content_type: str | None = None + size_bytes: int | None = None + sha256: str | None = None + download_path: str | None = None + provenance: Mapping[str, object] = field(default_factory=dict) + + @runtime_checkable class FileAccessProvider(ResourceAccessExplanationProvider, Protocol): """Resource-level access explanation provider for Files-owned resources.""" @@ -50,3 +75,35 @@ class ManagedArtifactStore(Protocol): *, request: ManagedArtifactWriteRequest, ) -> ManagedArtifactRef: ... + + +@runtime_checkable +class PostboxFileReferenceProvider(Protocol): + """Resolve Files-owned references after Postbox and Files authorization.""" + + def resolve_postbox_references( + self, + session: object, + principal: object, + *, + tenant_id: str, + requests: tuple[PostboxFileReferenceRequest, ...], + ) -> tuple[PostboxFileReferenceRef, ...]: ... + + +def postbox_file_reference_provider( + registry: object | None, +) -> PostboxFileReferenceProvider | None: + if ( + registry is None + or not hasattr(registry, "has_capability") + or not registry.has_capability(CAPABILITY_FILES_POSTBOX_REFERENCES) + ): + return None + provider = registry.require_capability(CAPABILITY_FILES_POSTBOX_REFERENCES) + if not isinstance(provider, PostboxFileReferenceProvider): + raise TypeError( + "files.postbox_references provider does not implement " + "PostboxFileReferenceProvider" + ) + return provider diff --git a/src/govoplan_core/core/mail.py b/src/govoplan_core/core/mail.py index 0571f90..80d2960 100644 --- a/src/govoplan_core/core/mail.py +++ b/src/govoplan_core/core/mail.py @@ -9,6 +9,7 @@ from typing import Protocol, runtime_checkable CAPABILITY_MAIL_DELIVERY_OUTBOX = "mail.delivery_outbox" CAPABILITY_MAIL_NOTIFICATION_DELIVERY = "mail.notificationDelivery" CAPABILITY_MAIL_BOUNCE_PROCESSING = "mail.bounce_processing" +CAPABILITY_MAIL_POSTBOX_BRIDGE = "mail.postbox_bridge" @dataclass(frozen=True, slots=True) @@ -81,6 +82,28 @@ class MailBounceObservationRef: evidence: Mapping[str, object] = field(default_factory=dict) +@dataclass(frozen=True, slots=True) +class MailPostboxBridgeRequest: + tenant_id: str + target: object + profile_id: str + folder: str + uid: str + uidvalidity: str + raw_message: bytes + classification: str = "internal" + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class MailPostboxBridgeResult: + postbox_id: str + message_id: str + delivery_id: str + duplicate: bool + source_digest: str + + @runtime_checkable class MailBounceProcessingProvider(Protocol): """Mail-owned DSN ingestion and durable correlation boundary.""" @@ -116,6 +139,17 @@ class MailBounceProcessingProvider(Protocol): ... +@runtime_checkable +class MailPostboxBridgeProvider(Protocol): + """Translate one immutable Mail observation into Postbox delivery.""" + + def bridge_message( + self, + session: object, + request: MailPostboxBridgeRequest, + ) -> MailPostboxBridgeResult: ... + + def notification_mail_delivery_provider( registry: object | None, ) -> NotificationMailDeliveryProvider | None: @@ -150,3 +184,21 @@ def mail_bounce_processing_provider( "MailBounceProcessingProvider" ) return provider + + +def mail_postbox_bridge_provider( + registry: object | None, +) -> MailPostboxBridgeProvider | None: + if ( + registry is None + or not hasattr(registry, "has_capability") + or not registry.has_capability(CAPABILITY_MAIL_POSTBOX_BRIDGE) + ): + return None + provider = registry.require_capability(CAPABILITY_MAIL_POSTBOX_BRIDGE) + if not isinstance(provider, MailPostboxBridgeProvider): + raise TypeError( + "mail.postbox_bridge provider does not implement " + "MailPostboxBridgeProvider" + ) + return provider diff --git a/src/govoplan_core/core/postbox.py b/src/govoplan_core/core/postbox.py index 39e46b3..aa18004 100644 --- a/src/govoplan_core/core/postbox.py +++ b/src/govoplan_core/core/postbox.py @@ -15,6 +15,7 @@ CAPABILITY_POSTBOX_MESSAGES = "postbox.messages" CAPABILITY_POSTBOX_DELIVERY = "postbox.delivery" CAPABILITY_POSTBOX_EVIDENCE = "postbox.evidence" CAPABILITY_POSTBOX_ROUTING = "postbox.routing" +CAPABILITY_POSTBOX_PORTAL = "postbox.portal_projection" PostboxAction = Literal[ "discover", @@ -323,6 +324,14 @@ class PostboxDeliveryResult: evidence: Mapping[str, object] = field(default_factory=dict) +@dataclass(frozen=True, slots=True) +class PostboxPortalEntryRef: + postbox: PostboxDirectoryEntryRef + unread_count: int = 0 + latest_message_at: datetime | None = None + route_path: str = "/postbox" + + class PostboxDeliveryRejected(RuntimeError): """A delivery was rejected before the provider accepted any effect.""" @@ -495,6 +504,20 @@ class PostboxRoutingProvider(Protocol): ... +@runtime_checkable +class PostboxPortalProjectionProvider(Protocol): + """Project portal-enabled Postboxes without transferring access ownership.""" + + def list_portal_entries( + self, + session: object, + principal: object, + *, + tenant_id: str, + limit: int = 100, + ) -> Sequence[PostboxPortalEntryRef]: ... + + def _postbox_provider( registry: object | None, *, @@ -573,3 +596,18 @@ def postbox_routing_provider( provider_type=PostboxRoutingProvider, ) return provider if isinstance(provider, PostboxRoutingProvider) else None + + +def postbox_portal_projection_provider( + registry: object | None, +) -> PostboxPortalProjectionProvider | None: + provider = _postbox_provider( + registry, + capability_name=CAPABILITY_POSTBOX_PORTAL, + provider_type=PostboxPortalProjectionProvider, + ) + return ( + provider + if isinstance(provider, PostboxPortalProjectionProvider) + else None + ) diff --git a/src/govoplan_core/core/templates.py b/src/govoplan_core/core/templates.py index 2236309..ad4dd1e 100644 --- a/src/govoplan_core/core/templates.py +++ b/src/govoplan_core/core/templates.py @@ -8,6 +8,7 @@ from typing import Literal, Protocol, runtime_checkable CAPABILITY_TEMPLATE_CATALOG = "templates.catalog" CAPABILITY_TEMPLATE_RENDERER = "templates.renderer" +CAPABILITY_TEMPLATE_CONTENT_LIBRARY = "templates.content_library" TemplateType = Literal[ "label", @@ -17,6 +18,7 @@ TemplateType = Literal[ "form_letter", "list_layout", "email", + "content_fragment", "generic", ] TemplateOutputFormat = Literal["html", "text"] @@ -79,6 +81,10 @@ class TemplateRevisionRef: locale: str required_fields: tuple[TemplateFieldRequirement, ...] output_profiles: tuple[TemplateOutputProfile, ...] + content_text: str | None = None + content_html: str | None = None + layout: Mapping[str, object] = field(default_factory=dict) + metadata: Mapping[str, object] = field(default_factory=dict) published_at: datetime | None = None provenance: Mapping[str, object] = field(default_factory=dict) @@ -166,6 +172,22 @@ class TemplateRenderResult: payload: bytes | None = None +@dataclass(frozen=True, slots=True) +class TemplateContentDraftRequest: + """Provider-neutral request for a reusable text/HTML content draft.""" + + name: str + template_type: TemplateType + usages: tuple[str, ...] + content_text: str | None = None + content_html: str | None = None + description: str | None = None + locale: str = "de" + scope_type: Literal["tenant", "group", "user"] = "tenant" + scope_id: str | None = None + metadata: Mapping[str, object] = field(default_factory=dict) + + @runtime_checkable class TemplateCatalogProvider(Protocol): def list_templates( @@ -213,13 +235,29 @@ class TemplateRendererProvider(Protocol): ) -> TemplateRenderResult: ... +@runtime_checkable +class TemplateContentLibraryProvider(Protocol): + """Create reusable content drafts while Templates retains ownership.""" + + def create_content_draft( + self, + session: object, + principal: object, + *, + request: TemplateContentDraftRequest, + ) -> TemplateRef: ... + + __all__ = [ "CAPABILITY_TEMPLATE_CATALOG", + "CAPABILITY_TEMPLATE_CONTENT_LIBRARY", "CAPABILITY_TEMPLATE_RENDERER", "TemplateArtifactRef", "TemplateCatalogProvider", "TemplateCompatibility", "TemplateCompatibilityError", + "TemplateContentDraftRequest", + "TemplateContentLibraryProvider", "TemplateContractError", "TemplateFieldRequirement", "TemplateNotFoundError", diff --git a/src/govoplan_core/privacy/dsar_workflow.py b/src/govoplan_core/privacy/dsar_workflow.py new file mode 100644 index 0000000..560e698 --- /dev/null +++ b/src/govoplan_core/privacy/dsar_workflow.py @@ -0,0 +1,753 @@ +from __future__ import annotations + +import hashlib +import json +import uuid +from collections import defaultdict +from collections.abc import Iterable, Mapping, Sequence +from datetime import datetime, timezone +from typing import Any + +from sqlalchemy import DateTime, Index, Integer, JSON, String, Text +from sqlalchemy.orm import Mapped, Session, mapped_column + +from govoplan_core.core.concurrency import RevisionConflictError, strong_resource_etag +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarExecutionResultRef, + DsarProvider, + DsarRecordRef, + DsarRequestKind, + DsarSubjectRef, + dsar_provider_names, +) +from govoplan_core.db.base import Base, TimestampMixin + + +MAX_RECORDS_PER_PROVIDER = 10_000 +MAX_PROVIDER_RESULT_BYTES = 10 * 1024 * 1024 + + +class DataSubjectRequest(Base, TimestampMixin): + __tablename__ = "core_data_subject_requests" + __table_args__ = ( + Index( + "ix_core_data_subject_requests_tenant_status", + "tenant_id", + "status", + ), + ) + + id: Mapped[str] = mapped_column( + String(36), primary_key=True, default=lambda: str(uuid.uuid4()) + ) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + reference: Mapped[str] = mapped_column(String(120), nullable=False) + request_kind: Mapped[str] = mapped_column(String(30), nullable=False) + status: Mapped[str] = mapped_column( + String(30), default="draft", nullable=False, index=True + ) + subject: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False) + purpose: Mapped[str] = mapped_column(String(1000), nullable=False) + legal_basis: Mapped[str | None] = mapped_column(String(1000), nullable=True) + due_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True, index=True + ) + requested_by_account_id: Mapped[str] = mapped_column(String(36), nullable=False) + search_result: Mapped[dict[str, Any]] = mapped_column( + JSON, default=dict, nullable=False + ) + erasure_plan: Mapped[dict[str, Any]] = mapped_column( + JSON, default=dict, nullable=False + ) + execution_result: Mapped[dict[str, Any]] = mapped_column( + JSON, default=dict, nullable=False + ) + coverage: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False) + evidence_sha256: Mapped[str | None] = mapped_column(String(64), nullable=True) + resource_revision: Mapped[int] = mapped_column(Integer, default=1, nullable=False) + completed_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + notes: Mapped[str | None] = mapped_column(Text, nullable=True) + + @property + def strong_etag(self) -> str: + return strong_resource_etag( + "data_subject_request", + self.id, + self.resource_revision, + ) + + +def create_data_subject_request( + session: Session, + *, + tenant_id: str, + reference: str, + request_kind: DsarRequestKind, + subject: DsarSubjectRef, + purpose: str, + legal_basis: str | None, + due_at: datetime | None, + requested_by_account_id: str, + notes: str | None = None, +) -> DataSubjectRequest: + if not subject.has_selector(): + raise ValueError("At least one data-subject selector is required.") + row = DataSubjectRequest( + tenant_id=tenant_id, + reference=reference.strip(), + request_kind=request_kind, + subject=subject.to_dict(), + purpose=purpose.strip(), + legal_basis=(legal_basis or "").strip() or None, + due_at=due_at, + requested_by_account_id=requested_by_account_id, + notes=(notes or "").strip() or None, + ) + session.add(row) + session.flush() + row.evidence_sha256 = _evidence_digest(row) + return row + + +def get_data_subject_request( + session: Session, + *, + tenant_id: str, + request_id: str, + for_update: bool = False, +) -> DataSubjectRequest: + query = session.query(DataSubjectRequest).filter( + DataSubjectRequest.id == request_id, + DataSubjectRequest.tenant_id == tenant_id, + ) + if for_update: + query = query.with_for_update() + row = query.one_or_none() + if row is None or row.tenant_id != tenant_id: + raise LookupError("Data-subject request not found.") + return row + + +def list_data_subject_requests( + session: Session, + *, + tenant_id: str, + limit: int = 200, +) -> tuple[DataSubjectRequest, ...]: + return tuple( + session.query(DataSubjectRequest) + .filter(DataSubjectRequest.tenant_id == tenant_id) + .order_by( + DataSubjectRequest.created_at.desc(), + DataSubjectRequest.id.desc(), + ) + .limit(max(1, min(limit, 500))) + .all() + ) + + +def search_data_subject_request( + session: Session, + *, + registry: object, + row: DataSubjectRequest, + expected_revision: int, +) -> DataSubjectRequest: + _assert_revision(row, expected_revision) + subject = _subject(row.subject) + records: list[dict[str, object]] = [] + provider_runs: list[dict[str, object]] = [] + providers, discovery = _providers( + registry, + session, + tenant_id=row.tenant_id, + ) + for capability_name, provider in providers: + started_at = _now() + try: + with session.begin_nested(): + provider_records = tuple( + provider.search_subject( + session, + tenant_id=row.tenant_id, + subject=subject, + ) + ) + _validate_records(provider, provider_records) + if len(provider_records) > MAX_RECORDS_PER_PROVIDER: + raise ValueError("DSAR provider result exceeds the record limit.") + encoded = _json_bytes([item.to_dict() for item in provider_records]) + if len(encoded) > MAX_PROVIDER_RESULT_BYTES: + raise ValueError("DSAR provider result exceeds the payload limit.") + records.extend(item.to_dict() for item in provider_records) + provider_runs.append( + { + "capability": capability_name, + "provider_id": provider.provider_id, + "module_id": provider.module_id, + "status": "complete", + "record_count": len(provider_records), + "started_at": started_at, + "completed_at": _now(), + } + ) + except Exception as exc: + provider_runs.append( + { + "capability": capability_name, + "provider_id": getattr(provider, "provider_id", capability_name), + "module_id": getattr(provider, "module_id", "unknown"), + "status": "failed", + "record_count": 0, + "error": _safe_error(exc), + "started_at": started_at, + "completed_at": _now(), + } + ) + failures = sum(1 for item in provider_runs if item["status"] == "failed") + row.search_result = { + "schema": "govoplan.dsars.search.v1", + "searched_at": _now(), + "records": records, + "provider_runs": provider_runs, + "record_count": len(records), + } + row.coverage = discovery + row.erasure_plan = {} + row.execution_result = {} + row.status = "search_partial" if failures else "searched" + row.completed_at = None + _advance(row) + return row + + +def plan_data_subject_erasure( + session: Session, + *, + registry: object, + row: DataSubjectRequest, + expected_revision: int, +) -> DataSubjectRequest: + _assert_revision(row, expected_revision) + if not row.search_result.get("searched_at"): + raise ValueError("Run the data-subject search before planning erasure.") + if row.request_kind == "access": + raise ValueError("This request does not include erasure.") + subject = _subject(row.subject) + records = tuple(_record(item) for item in row.search_result.get("records", [])) + by_provider: dict[str, list[DsarRecordRef]] = defaultdict(list) + for record in records: + by_provider[record.provider_id].append(record) + providers, discovery = _providers( + registry, + session, + tenant_id=row.tenant_id, + ) + actions: list[DsarErasureActionRef] = [] + provider_runs: list[dict[str, object]] = [] + for capability_name, provider in providers: + started_at = _now() + try: + with session.begin_nested(): + proposed = tuple( + provider.plan_erasure( + session, + tenant_id=row.tenant_id, + subject=subject, + records=tuple(by_provider.get(provider.provider_id, ())), + ) + ) + _validate_actions(provider, proposed) + actions.extend(proposed) + provider_runs.append( + { + "capability": capability_name, + "provider_id": provider.provider_id, + "module_id": provider.module_id, + "status": "complete", + "action_count": len(proposed), + "started_at": started_at, + "completed_at": _now(), + } + ) + except Exception as exc: + provider_runs.append( + { + "capability": capability_name, + "provider_id": getattr(provider, "provider_id", capability_name), + "module_id": getattr(provider, "module_id", "unknown"), + "status": "failed", + "action_count": 0, + "error": _safe_error(exc), + "started_at": started_at, + "completed_at": _now(), + } + ) + existing_resources = { + (item.provider_id, item.resource_type, item.resource_id) for item in actions + } + for record in records: + key = (record.provider_id, record.resource_type, record.resource_id) + if not record.immutable_evidence or key in existing_resources: + continue + digest = hashlib.sha256("\0".join(key).encode("utf-8")).hexdigest()[:24] + actions.append( + DsarErasureActionRef( + action_id=f"retain:{digest}", + provider_id=record.provider_id, + module_id=record.module_id, + kind="retain", + resource_type=record.resource_type, + resource_id=record.resource_id, + title=f"Retain {record.title}", + rationale=( + record.retention_reason + or "Immutable institutional evidence must be retained." + ), + executable=False, + ) + ) + duplicate_ids = _duplicates(item.action_id for item in actions) + if duplicate_ids: + raise ValueError( + f"DSAR providers returned duplicate action ids: {', '.join(duplicate_ids)}" + ) + failures = sum(1 for item in provider_runs if item["status"] == "failed") + row.erasure_plan = { + "schema": "govoplan.dsars.erasure-plan.v1", + "planned_at": _now(), + "actions": [item.to_dict() for item in actions], + "provider_runs": provider_runs, + "executable_count": sum(1 for item in actions if item.executable), + "retained_count": sum(1 for item in actions if item.kind == "retain"), + } + row.coverage = {**discovery, "plan_provider_runs": provider_runs} + row.execution_result = {} + row.status = "plan_partial" if failures else "plan_ready" + row.completed_at = None + _advance(row) + return row + + +def execute_data_subject_erasure( + session: Session, + *, + registry: object, + row: DataSubjectRequest, + expected_revision: int, + action_ids: Sequence[str], +) -> DataSubjectRequest: + _assert_revision(row, expected_revision) + raw_actions = row.erasure_plan.get("actions") + if not isinstance(raw_actions, list): + raise ValueError("Create an erasure plan before execution.") + actions_by_id = { + action.action_id: action for action in (_action(item) for item in raw_actions) + } + selected_ids = tuple(dict.fromkeys(str(item) for item in action_ids if str(item))) + if not selected_ids: + raise ValueError("Select at least one executable erasure action.") + missing = [item for item in selected_ids if item not in actions_by_id] + if missing: + raise ValueError("The erasure plan changed; refresh before executing it.") + selected = tuple(actions_by_id[item] for item in selected_ids) + blocked = [item.action_id for item in selected if not item.executable] + if blocked: + raise ValueError("Retained or review-only actions cannot be executed.") + + providers, _discovery = _providers( + registry, + session, + tenant_id=row.tenant_id, + ) + providers_by_id = {provider.provider_id: provider for _name, provider in providers} + grouped: dict[str, list[DsarErasureActionRef]] = defaultdict(list) + for action in selected: + grouped[action.provider_id].append(action) + previous = { + str(item.get("action_id")): item + for item in row.execution_result.get("results", []) + if isinstance(item, Mapping) + and item.get("status") in {"executed", "unchanged"} + } + results: list[dict[str, object]] = list(previous.values()) + for provider_id, provider_actions in grouped.items(): + provider = providers_by_id.get(provider_id) + if provider is None: + results.extend( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary="The owning DSAR provider is not currently available.", + ).to_dict() + for action in provider_actions + ) + continue + pending = tuple( + action for action in provider_actions if action.action_id not in previous + ) + if not pending: + continue + try: + with session.begin_nested(): + executed = tuple( + provider.execute_erasure( + session, + tenant_id=row.tenant_id, + subject=_subject(row.subject), + actions=pending, + request_id=row.id, + ) + ) + _validate_execution_results(pending, executed) + results.extend(item.to_dict() for item in executed) + except Exception as exc: + results.extend( + DsarExecutionResultRef( + action_id=action.action_id, + status="failed", + summary=_safe_error(exc), + ).to_dict() + for action in pending + ) + result_by_id = {str(item["action_id"]): item for item in results} + all_executable = { + action.action_id for action in actions_by_id.values() if action.executable + } + successful = { + action_id + for action_id, item in result_by_id.items() + if item.get("status") in {"executed", "unchanged"} + } + failed = { + action_id + for action_id, item in result_by_id.items() + if item.get("status") in {"failed", "blocked"} + } + row.execution_result = { + "schema": "govoplan.dsars.execution.v1", + "executed_at": _now(), + "results": list(result_by_id.values()), + "successful_count": len(successful), + "failed_count": len(failed), + } + if all_executable.issubset(successful): + row.status = "completed" + row.completed_at = datetime.now(timezone.utc) + elif failed: + row.status = "execution_partial" + row.completed_at = None + else: + row.status = "execution_pending" + row.completed_at = None + _advance(row) + return row + + +def data_subject_export(row: DataSubjectRequest) -> bytes: + payload: dict[str, object] = { + "schema": "govoplan.dsars.export.v1", + "generated_at": _now(), + "request": data_subject_request_dict(row, include_subject=True), + "search": row.search_result, + "erasure_plan": row.erasure_plan, + "execution": row.execution_result, + "coverage": row.coverage, + "limitations": [ + "Only providers listed as complete contributed data.", + "Retained immutable evidence is exported with its stated retention reason.", + "An export does not itself erase or alter source data.", + ], + } + digest = hashlib.sha256(_json_bytes(payload)).hexdigest() + payload["manifest_sha256"] = digest + return _json_bytes(payload, pretty=True) + + +def data_subject_request_dict( + row: DataSubjectRequest, + *, + include_subject: bool, +) -> dict[str, object]: + return { + "id": row.id, + "tenant_id": row.tenant_id, + "reference": row.reference, + "request_kind": row.request_kind, + "status": row.status, + "subject": dict(row.subject) if include_subject else {}, + "purpose": row.purpose, + "legal_basis": row.legal_basis, + "due_at": row.due_at.isoformat() if row.due_at else None, + "requested_by_account_id": row.requested_by_account_id, + "record_count": int(row.search_result.get("record_count", 0)), + "executable_action_count": int( + row.erasure_plan.get("executable_count", 0) + ), + "coverage": dict(row.coverage), + "evidence_sha256": row.evidence_sha256, + "resource_revision": row.resource_revision, + "etag": row.strong_etag, + "created_at": row.created_at.isoformat(), + "updated_at": row.updated_at.isoformat(), + "completed_at": row.completed_at.isoformat() if row.completed_at else None, + "notes": row.notes, + } + + +def _providers( + registry: object, + session: Session, + *, + tenant_id: str, +) -> tuple[tuple[tuple[str, DsarProvider], ...], dict[str, object]]: + values: list[tuple[str, DsarProvider]] = [] + failures: list[dict[str, str]] = [] + active_modules = _effective_module_ids(registry, session, tenant_id=tenant_id) + names = dsar_provider_names(registry) + active_names: list[str] = [] + inactive_names: list[str] = [] + for name in names: + owner = ( + registry.capability_owner(name) + if hasattr(registry, "capability_owner") + else None + ) + if owner and owner not in active_modules: + inactive_names.append(name) + continue + active_names.append(name) + try: + if hasattr(registry, "require_tenant_capability"): + candidate = registry.require_tenant_capability( + name, + session, + tenant_id=tenant_id, + work_state="interactive", + ) + else: + candidate = registry.require_capability(name) + if not isinstance(candidate, DsarProvider): + raise TypeError("Capability does not implement DsarProvider.") + values.append((name, candidate)) + except Exception as exc: + failures.append({"capability": name, "error": _safe_error(exc)}) + covered_modules = sorted({provider.module_id for _name, provider in values}) + return tuple(values), { + "provider_capabilities": active_names, + "inactive_provider_capabilities": inactive_names, + "covered_modules": covered_modules, + "modules_without_provider": sorted(set(active_modules) - set(covered_modules)), + "provider_discovery_failures": failures, + } + + +def _effective_module_ids( + registry: object, + session: Session, + *, + tenant_id: str, +) -> list[str]: + resolver_factory = getattr(registry, "tenant_entitlement_resolver", None) + if callable(resolver_factory): + state = resolver_factory().resolve(session, tenant_id) + return sorted(str(item) for item in state.effective_modules) + if hasattr(registry, "manifests"): + return sorted(str(manifest.id) for manifest in registry.manifests()) + return [] + + +def _subject(value: Mapping[str, object]) -> DsarSubjectRef: + raw_refs = value.get("external_references") + return DsarSubjectRef( + account_id=_optional_text(value.get("account_id")), + identity_id=_optional_text(value.get("identity_id")), + membership_id=_optional_text(value.get("membership_id")), + email=_optional_text(value.get("email")), + external_references={ + str(key): str(item) + for key, item in (raw_refs.items() if isinstance(raw_refs, Mapping) else ()) + }, + ) + + +def _record(value: object) -> DsarRecordRef: + if not isinstance(value, Mapping): + raise ValueError("Stored DSAR record is invalid.") + observed_at = _parse_datetime(value.get("observed_at")) + data = value.get("data") + return DsarRecordRef( + provider_id=str(value.get("provider_id") or ""), + module_id=str(value.get("module_id") or ""), + resource_type=str(value.get("resource_type") or ""), + resource_id=str(value.get("resource_id") or ""), + category=str(value.get("category") or ""), + title=str(value.get("title") or ""), + data=dict(data) if isinstance(data, Mapping) else {}, + observed_at=observed_at, + immutable_evidence=bool(value.get("immutable_evidence")), + retention_reason=_optional_text(value.get("retention_reason")), + source_path=_optional_text(value.get("source_path")), + ) + + +def _action(value: object) -> DsarErasureActionRef: + if not isinstance(value, Mapping): + raise ValueError("Stored DSAR action is invalid.") + metadata = value.get("metadata") + kind = str(value.get("kind") or "manual_review") + if kind not in { + "delete", + "anonymize", + "revoke", + "detach", + "retain", + "manual_review", + }: + raise ValueError("Stored DSAR action kind is invalid.") + return DsarErasureActionRef( + action_id=str(value.get("action_id") or ""), + provider_id=str(value.get("provider_id") or ""), + module_id=str(value.get("module_id") or ""), + kind=kind, # type: ignore[arg-type] + resource_type=str(value.get("resource_type") or ""), + resource_id=str(value.get("resource_id") or ""), + title=str(value.get("title") or ""), + rationale=str(value.get("rationale") or ""), + executable=bool(value.get("executable")), + irreversible=bool(value.get("irreversible")), + metadata=dict(metadata) if isinstance(metadata, Mapping) else {}, + ) + + +def _validate_records( + provider: DsarProvider, + records: Sequence[DsarRecordRef], +) -> None: + for record in records: + if record.provider_id != provider.provider_id or record.module_id != provider.module_id: + raise ValueError("DSAR record ownership does not match its provider.") + if not record.resource_type or not record.resource_id or not record.title: + raise ValueError("DSAR records require resource identity and title.") + + +def _validate_actions( + provider: DsarProvider, + actions: Sequence[DsarErasureActionRef], +) -> None: + for action in actions: + if action.provider_id != provider.provider_id or action.module_id != provider.module_id: + raise ValueError("DSAR action ownership does not match its provider.") + if not action.action_id or not action.resource_type or not action.resource_id: + raise ValueError("DSAR actions require stable identities.") + if action.kind in {"retain", "manual_review"} and action.executable: + raise ValueError("Retain and manual-review actions cannot be executable.") + + +def _validate_execution_results( + actions: Sequence[DsarErasureActionRef], + results: Sequence[DsarExecutionResultRef], +) -> None: + expected = {action.action_id for action in actions} + returned = {result.action_id for result in results} + if expected != returned or len(returned) != len(results): + raise ValueError("DSAR provider did not return one result per action.") + + +def _assert_revision(row: DataSubjectRequest, expected_revision: int) -> None: + if row.resource_revision != expected_revision: + raise RevisionConflictError( + resource_type="data_subject_request", + resource_id=row.id, + current_revision=row.resource_revision, + submitted_base_revision=expected_revision, + current_etag=row.strong_etag, + refresh_path=f"/api/v1/admin/privacy/data-subject-requests/{row.id}", + ) + + +def _advance(row: DataSubjectRequest) -> None: + row.resource_revision += 1 + row.evidence_sha256 = _evidence_digest(row) + + +def _evidence_digest(row: DataSubjectRequest) -> str: + return hashlib.sha256( + _json_bytes( + { + "id": row.id, + "tenant_id": row.tenant_id, + "reference": row.reference, + "request_kind": row.request_kind, + "status": row.status, + "subject": row.subject, + "search_result": row.search_result, + "erasure_plan": row.erasure_plan, + "execution_result": row.execution_result, + "coverage": row.coverage, + "resource_revision": row.resource_revision, + } + ) + ).hexdigest() + + +def _duplicates(values: Iterable[str]) -> tuple[str, ...]: + seen: set[str] = set() + duplicate: set[str] = set() + for value in values: + if value in seen: + duplicate.add(value) + seen.add(value) + return tuple(sorted(duplicate)) + + +def _json_bytes(value: object, *, pretty: bool = False) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + indent=2 if pretty else None, + separators=None if pretty else (",", ":"), + default=str, + ).encode("utf-8") + + +def _safe_error(exc: Exception) -> str: + text = " ".join(str(exc).split()) + return (text or exc.__class__.__name__)[:1000] + + +def _optional_text(value: object | None) -> str | None: + text = str(value or "").strip() + return text or None + + +def _parse_datetime(value: object | None) -> datetime | None: + if isinstance(value, datetime): + return value + if isinstance(value, str) and value: + try: + return datetime.fromisoformat(value) + except ValueError: + return None + return None + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +__all__ = [ + "DataSubjectRequest", + "create_data_subject_request", + "data_subject_export", + "data_subject_request_dict", + "execute_data_subject_erasure", + "get_data_subject_request", + "list_data_subject_requests", + "plan_data_subject_erasure", + "search_data_subject_request", +] diff --git a/src/govoplan_core/server/app.py b/src/govoplan_core/server/app.py index 22dd7c2..2ce8469 100644 --- a/src/govoplan_core/server/app.py +++ b/src/govoplan_core/server/app.py @@ -10,6 +10,7 @@ from govoplan_core.server.fastapi import create_govoplan_app from govoplan_core.server.platform import create_platform_router from govoplan_core.server.bootstrap import create_bootstrap_router from govoplan_core.server.credentials import router as credential_router +from govoplan_core.server.dsar import router as dsar_router from govoplan_core.server.ownership import router as ownership_router from govoplan_core.server.registry import available_module_manifests, build_platform_registry from govoplan_core.server.route_validation import validate_no_route_collisions @@ -72,6 +73,7 @@ def _server_api_router(server_config: GovoplanServerConfig, registry) -> APIRout api_router.include_router(create_platform_router(settings=server_config.settings)) api_router.include_router(create_bootstrap_router(server_config.settings)) api_router.include_router(credential_router) + api_router.include_router(dsar_router) api_router.include_router(ownership_router) for router in server_config.post_module_routers: api_router.include_router(router) diff --git a/src/govoplan_core/server/dsar.py b/src/govoplan_core/server/dsar.py new file mode 100644 index 0000000..0c6af6c --- /dev/null +++ b/src/govoplan_core/server/dsar.py @@ -0,0 +1,406 @@ +from __future__ import annotations + +import re +from datetime import datetime +from typing import Any, Literal + +from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, Response, status +from pydantic import BaseModel, Field, field_validator +from sqlalchemy.orm import Session + +from govoplan_core.audit.logging import audit_event +from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope +from govoplan_core.core.concurrency import ( + ConcurrencyError, + MissingPreconditionError, + RevisionConflictError, + assert_revision_precondition, +) +from govoplan_core.core.dsar import DsarSubjectRef +from govoplan_core.db.session import get_session +from govoplan_core.privacy.dsar_workflow import ( + DataSubjectRequest, + create_data_subject_request, + data_subject_export, + data_subject_request_dict, + execute_data_subject_erasure, + get_data_subject_request, + list_data_subject_requests, + plan_data_subject_erasure, + search_data_subject_request, +) + + +READ_SCOPE = "access:privacy:read" +MANAGE_SCOPE = "access:privacy:manage" +EXPORT_SCOPE = "access:privacy:export" +ERASE_SCOPE = "access:privacy:erase" + + +class DataSubjectSelectorRequest(BaseModel): + account_id: str | None = Field(default=None, max_length=36) + identity_id: str | None = Field(default=None, max_length=36) + membership_id: str | None = Field(default=None, max_length=36) + email: str | None = Field(default=None, max_length=320) + external_references: dict[str, str] = Field(default_factory=dict) + + @field_validator("external_references") + @classmethod + def validate_external_references( + cls, + value: dict[str, str], + ) -> dict[str, str]: + if len(value) > 50: + raise ValueError("At most 50 external subject references are allowed.") + normalized: dict[str, str] = {} + for raw_key, raw_value in value.items(): + key = str(raw_key).strip() + item = str(raw_value).strip() + if not key or not item: + continue + if len(key) > 120 or len(item) > 500: + raise ValueError( + "External subject-reference namespaces are limited to 120 " + "characters and values to 500 characters." + ) + normalized[key] = item + return normalized + + +class DataSubjectRequestCreate(BaseModel): + reference: str = Field(min_length=1, max_length=120) + request_kind: Literal["access", "erasure", "access_and_erasure"] + subject: DataSubjectSelectorRequest + purpose: str = Field(min_length=1, max_length=1000) + legal_basis: str | None = Field(default=None, max_length=1000) + due_at: datetime | None = None + notes: str | None = Field(default=None, max_length=10_000) + + +class RevisionMutationRequest(BaseModel): + base_revision: int = Field(ge=1) + + +class DataSubjectExecutionRequest(RevisionMutationRequest): + action_ids: list[str] = Field(min_length=1, max_length=10_000) + confirmation: str = Field(min_length=1, max_length=100) + + +class DataSubjectRequestResponse(BaseModel): + request: dict[str, Any] + search: dict[str, Any] = Field(default_factory=dict) + erasure_plan: dict[str, Any] = Field(default_factory=dict) + execution: dict[str, Any] = Field(default_factory=dict) + + +class DataSubjectRequestListResponse(BaseModel): + items: list[dict[str, Any]] = Field(default_factory=list) + + +router = APIRouter( + prefix="/admin/privacy/data-subject-requests", + tags=["data-subject-requests"], +) + + +@router.get("", response_model=DataSubjectRequestListResponse) +def list_requests( + limit: int = Query(default=200, ge=1, le=500), + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +) -> DataSubjectRequestListResponse: + _require(principal, READ_SCOPE) + rows = list_data_subject_requests( + session, + tenant_id=principal.tenant_id, + limit=limit, + ) + return DataSubjectRequestListResponse( + items=[ + data_subject_request_dict(row, include_subject=True) + for row in rows + ] + ) + + +@router.post( + "", + response_model=DataSubjectRequestResponse, + status_code=status.HTTP_201_CREATED, +) +def create_request( + payload: DataSubjectRequestCreate, + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +) -> DataSubjectRequestResponse: + _require(principal, MANAGE_SCOPE) + try: + row = create_data_subject_request( + session, + tenant_id=principal.tenant_id, + reference=payload.reference, + request_kind=payload.request_kind, + subject=_subject(payload.subject), + purpose=payload.purpose, + legal_basis=payload.legal_basis, + due_at=payload.due_at, + requested_by_account_id=principal.account_id, + notes=payload.notes, + ) + _audit(session, principal, row, "privacy.dsar.created") + session.commit() + session.refresh(row) + return _detail(row) + except ValueError as exc: + session.rollback() + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +@router.get("/{request_id}", response_model=DataSubjectRequestResponse) +def get_request( + request_id: str, + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +) -> DataSubjectRequestResponse: + _require(principal, READ_SCOPE) + return _detail(_row(session, principal, request_id)) + + +@router.post("/{request_id}/search", response_model=DataSubjectRequestResponse) +def search_request( + request_id: str, + payload: RevisionMutationRequest, + request: Request, + if_match: str | None = Header(default=None, alias="If-Match"), + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +) -> DataSubjectRequestResponse: + _require(principal, MANAGE_SCOPE) + return _mutate( + session, + principal, + request_id, + payload.base_revision, + if_match, + lambda row: search_data_subject_request( + session, + registry=_registry(request), + row=row, + expected_revision=payload.base_revision, + ), + "privacy.dsar.searched", + ) + + +@router.post("/{request_id}/erasure-plan", response_model=DataSubjectRequestResponse) +def plan_erasure( + request_id: str, + payload: RevisionMutationRequest, + request: Request, + if_match: str | None = Header(default=None, alias="If-Match"), + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +) -> DataSubjectRequestResponse: + _require(principal, MANAGE_SCOPE) + return _mutate( + session, + principal, + request_id, + payload.base_revision, + if_match, + lambda row: plan_data_subject_erasure( + session, + registry=_registry(request), + row=row, + expected_revision=payload.base_revision, + ), + "privacy.dsar.erasure_planned", + ) + + +@router.post("/{request_id}/execute", response_model=DataSubjectRequestResponse) +def execute_erasure( + request_id: str, + payload: DataSubjectExecutionRequest, + request: Request, + if_match: str | None = Header(default=None, alias="If-Match"), + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +) -> DataSubjectRequestResponse: + _require(principal, ERASE_SCOPE) + if payload.confirmation != f"ERASE {request_id}": + raise HTTPException( + status_code=400, + detail=f'Type "ERASE {request_id}" to confirm the selected actions.', + ) + return _mutate( + session, + principal, + request_id, + payload.base_revision, + if_match, + lambda row: execute_data_subject_erasure( + session, + registry=_registry(request), + row=row, + expected_revision=payload.base_revision, + action_ids=payload.action_ids, + ), + "privacy.dsar.erasure_executed", + details={"action_ids": payload.action_ids}, + ) + + +@router.get("/{request_id}/export") +def export_request( + request_id: str, + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +) -> Response: + _require(principal, EXPORT_SCOPE) + row = _row(session, principal, request_id) + content = data_subject_export(row) + _audit( + session, + principal, + row, + "privacy.dsar.exported", + details={"export_bytes": len(content)}, + ) + session.commit() + safe_reference = re.sub(r"[^A-Za-z0-9._-]+", "-", row.reference).strip("-") + filename = f"dsar-{safe_reference or row.id}.json" + return Response( + content=content, + media_type="application/json", + headers={"Content-Disposition": f'attachment; filename="{filename}"'}, + ) + + +def _mutate( + session: Session, + principal: ApiPrincipal, + request_id: str, + base_revision: int, + if_match: str | None, + operation: Any, + audit_action: str, + *, + details: dict[str, Any] | None = None, +) -> DataSubjectRequestResponse: + try: + assert_revision_precondition( + if_match, + resource_type="data_subject_request", + resource_id=request_id, + submitted_base_revision=base_revision, + ) + row = get_data_subject_request( + session, + tenant_id=principal.tenant_id, + request_id=request_id, + for_update=True, + ) + operation(row) + _audit(session, principal, row, audit_action, details=details) + session.commit() + session.refresh(row) + return _detail(row) + except LookupError as exc: + session.rollback() + raise HTTPException(status_code=404, detail=str(exc)) from exc + except MissingPreconditionError as exc: + session.rollback() + raise HTTPException(status_code=428, detail=exc.as_dict()) from exc + except RevisionConflictError as exc: + session.rollback() + raise HTTPException(status_code=409, detail=exc.as_dict()) from exc + except ConcurrencyError as exc: + session.rollback() + raise HTTPException(status_code=412, detail=str(exc)) from exc + except ValueError as exc: + session.rollback() + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +def _row( + session: Session, + principal: ApiPrincipal, + request_id: str, +) -> DataSubjectRequest: + try: + return get_data_subject_request( + session, + tenant_id=principal.tenant_id, + request_id=request_id, + ) + except LookupError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + + +def _detail(row: DataSubjectRequest) -> DataSubjectRequestResponse: + return DataSubjectRequestResponse( + request=data_subject_request_dict(row, include_subject=True), + search=dict(row.search_result), + erasure_plan=dict(row.erasure_plan), + execution=dict(row.execution_result), + ) + + +def _subject(payload: DataSubjectSelectorRequest) -> DsarSubjectRef: + return DsarSubjectRef( + account_id=_text(payload.account_id), + identity_id=_text(payload.identity_id), + membership_id=_text(payload.membership_id), + email=_text(payload.email), + external_references={ + str(key).strip(): str(value).strip() + for key, value in payload.external_references.items() + if str(key).strip() and str(value).strip() + }, + ) + + +def _registry(request: Request) -> object: + registry = getattr(request.app.state, "govoplan_registry", None) + if registry is None: + raise HTTPException(status_code=503, detail="Module registry is unavailable.") + return registry + + +def _require(principal: ApiPrincipal, scope: str) -> None: + if not has_scope(principal, scope): + raise HTTPException(status_code=403, detail=f"Missing scope: {scope}") + + +def _audit( + session: Session, + principal: ApiPrincipal, + row: DataSubjectRequest, + action: str, + *, + details: dict[str, Any] | None = None, +) -> None: + audit_event( + session, + tenant_id=principal.tenant_id, + action=action, + user_id=principal.membership_id, + api_key_id=principal.api_key_id, + object_type="data_subject_request", + object_id=row.id, + details={ + "reference": row.reference, + "status": row.status, + "resource_revision": row.resource_revision, + **(details or {}), + }, + ) + + +def _text(value: str | None) -> str | None: + normalized = (value or "").strip() + return normalized or None + + +__all__ = ["router"] diff --git a/tests/test_dsar_api.py b/tests/test_dsar_api.py new file mode 100644 index 0000000..ecd7e25 --- /dev/null +++ b/tests/test_dsar_api.py @@ -0,0 +1,162 @@ +from __future__ import annotations + +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +from fastapi import FastAPI +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool + +from govoplan_core.auth import ApiPrincipal, get_api_principal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.dsar import DsarRecordRef +from govoplan_core.db.base import Base +from govoplan_core.db.session import get_session +from govoplan_core.privacy.dsar_workflow import DataSubjectRequest +from govoplan_core.server.dsar import router + + +class _Provider: + provider_id = "example" + module_id = "example" + + def search_subject(self, session, *, tenant_id, subject): + del session, tenant_id, subject + return ( + DsarRecordRef( + provider_id="example", + module_id="example", + resource_type="profile", + resource_id="profile-1", + category="profile", + title="Example profile", + ), + ) + + def plan_erasure(self, session, *, tenant_id, subject, records): + del session, tenant_id, subject, records + return () + + def execute_erasure(self, session, *, tenant_id, subject, actions, request_id): + del session, tenant_id, subject, actions, request_id + return () + + +class _Registry: + provider = _Provider() + + def capability_names(self): + return ("privacy.dsar.example",) + + def require_tenant_capability(self, name, session, **kwargs): + del name, session, kwargs + return self.provider + + def manifests(self): + return (SimpleNamespace(id="example"),) + + +class DsarApiTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine( + "sqlite+pysqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + DataSubjectRequest.__table__.create(self.engine) + self.session = sessionmaker(bind=self.engine, expire_on_commit=False)() + self.principal = _principal( + "access:privacy:read", + "access:privacy:manage", + "access:privacy:export", + "access:privacy:erase", + ) + app = FastAPI() + app.state.govoplan_registry = _Registry() + app.include_router(router, prefix="/api/v1") + app.dependency_overrides[get_session] = lambda: self.session + app.dependency_overrides[get_api_principal] = lambda: self.principal + self.audit_patch = patch("govoplan_core.server.dsar.audit_event") + self.audit_patch.start() + self.client = TestClient(app) + + def tearDown(self) -> None: + self.client.close() + self.audit_patch.stop() + self.session.close() + Base.metadata.drop_all(self.engine, tables=[DataSubjectRequest.__table__]) + self.engine.dispose() + + def test_mutation_requires_strong_revision_and_export_is_available(self) -> None: + created = self.client.post( + "/api/v1/admin/privacy/data-subject-requests", + json={ + "reference": "DSAR-1", + "request_kind": "access", + "subject": {"email": "ada@example.test"}, + "purpose": "Verified access request", + }, + ) + self.assertEqual(201, created.status_code, created.text) + item = created.json()["request"] + + missing = self.client.post( + f"/api/v1/admin/privacy/data-subject-requests/{item['id']}/search", + json={"base_revision": item["resource_revision"]}, + ) + self.assertEqual(428, missing.status_code, missing.text) + + searched = self.client.post( + f"/api/v1/admin/privacy/data-subject-requests/{item['id']}/search", + headers={"If-Match": item["etag"]}, + json={"base_revision": item["resource_revision"]}, + ) + self.assertEqual(200, searched.status_code, searched.text) + self.assertEqual(1, searched.json()["search"]["record_count"]) + + stale = self.client.post( + f"/api/v1/admin/privacy/data-subject-requests/{item['id']}/search", + headers={"If-Match": item["etag"]}, + json={"base_revision": item["resource_revision"]}, + ) + self.assertEqual(409, stale.status_code, stale.text) + + exported = self.client.get( + f"/api/v1/admin/privacy/data-subject-requests/{item['id']}/export" + ) + self.assertEqual(200, exported.status_code, exported.text) + self.assertIn("govoplan.dsars.export.v1", exported.text) + + def test_privacy_scopes_are_independent(self) -> None: + self.principal = _principal("access:privacy:read") + denied = self.client.post( + "/api/v1/admin/privacy/data-subject-requests", + json={ + "reference": "DSAR-2", + "request_kind": "access", + "subject": {"email": "ada@example.test"}, + "purpose": "Verified access request", + }, + ) + self.assertEqual(403, denied.status_code, denied.text) + self.assertEqual(200, self.client.get("/api/v1/admin/privacy/data-subject-requests").status_code) + + +def _principal(*scopes: str) -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id="account-operator", + membership_id="membership-operator", + tenant_id="tenant-1", + scopes=frozenset(scopes), + ), + account=SimpleNamespace(id="account-operator"), + user=SimpleNamespace(id="membership-operator"), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_dsar_workflow.py b/tests/test_dsar_workflow.py new file mode 100644 index 0000000..93bc56d --- /dev/null +++ b/tests/test_dsar_workflow.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +import json +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session, sessionmaker +from sqlalchemy.pool import StaticPool + +from govoplan_core.core.concurrency import RevisionConflictError +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarExecutionResultRef, + DsarRecordRef, + DsarSubjectRef, +) +from govoplan_core.db.base import Base +from govoplan_core.privacy.dsar_workflow import ( + DataSubjectRequest, + create_data_subject_request, + data_subject_export, + execute_data_subject_erasure, + plan_data_subject_erasure, + search_data_subject_request, +) + + +class _Provider: + provider_id = "example" + module_id = "example" + + def __init__(self) -> None: + self.executions = 0 + + def search_subject(self, session, *, tenant_id, subject): + del session, tenant_id, subject + return ( + DsarRecordRef( + provider_id=self.provider_id, + module_id=self.module_id, + resource_type="profile", + resource_id="profile-1", + category="personal", + title="Example profile", + data={"name": "Ada"}, + ), + DsarRecordRef( + provider_id=self.provider_id, + module_id=self.module_id, + resource_type="audit_evidence", + resource_id="event-1", + category="evidence", + title="Decision event", + immutable_evidence=True, + retention_reason="Required decision evidence.", + ), + ) + + def plan_erasure(self, session, *, tenant_id, subject, records): + del session, tenant_id, subject, records + return ( + DsarErasureActionRef( + action_id="example:anonymize:profile-1", + provider_id=self.provider_id, + module_id=self.module_id, + kind="anonymize", + resource_type="profile", + resource_id="profile-1", + title="Anonymize profile", + rationale="Remove profile data.", + executable=True, + irreversible=True, + ), + ) + + def execute_erasure(self, session, *, tenant_id, subject, actions, request_id): + del session, tenant_id, subject, request_id + self.executions += 1 + return tuple( + DsarExecutionResultRef( + action_id=action.action_id, + status="executed", + summary="Profile anonymized.", + ) + for action in actions + ) + + +class _Registry: + def __init__(self, provider: _Provider) -> None: + self.provider = provider + + def capability_names(self): + return ("privacy.dsar.example", "privacy.dsar.inactive") + + def capability_owner(self, name): + return { + "privacy.dsar.example": "example", + "privacy.dsar.inactive": "inactive", + }[name] + + def tenant_entitlement_resolver(self): + class _Resolver: + @staticmethod + def resolve(session, tenant_id): + del session, tenant_id + return type( + "State", + (), + {"effective_modules": ("example", "without-provider")}, + )() + + return _Resolver() + + def require_tenant_capability(self, name, session, **kwargs): + del name, session, kwargs + return self.provider + + def manifests(self): + return ( + type("Manifest", (), {"id": "example"})(), + type("Manifest", (), {"id": "without-provider"})(), + ) + + +class DsarWorkflowTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine( + "sqlite+pysqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + DataSubjectRequest.__table__.create(self.engine) + self.session: Session = sessionmaker( + bind=self.engine, + expire_on_commit=False, + )() + self.provider = _Provider() + self.registry = _Registry(self.provider) + + def tearDown(self) -> None: + self.session.close() + Base.metadata.drop_all(self.engine, tables=[DataSubjectRequest.__table__]) + self.engine.dispose() + + def test_search_plan_export_and_idempotent_execution(self) -> None: + row = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-2026-001", + request_kind="access_and_erasure", + subject=DsarSubjectRef(email="ada@example.test"), + purpose="Respond to a verified data-subject request.", + legal_basis="Article 15 and 17 GDPR", + due_at=None, + requested_by_account_id="account-operator", + ) + self.session.commit() + + search_data_subject_request( + self.session, + registry=self.registry, + row=row, + expected_revision=1, + ) + self.assertEqual("searched", row.status) + self.assertEqual(2, row.search_result["record_count"]) + self.assertEqual(["without-provider"], row.coverage["modules_without_provider"]) + self.assertEqual( + ["privacy.dsar.inactive"], + row.coverage["inactive_provider_capabilities"], + ) + + plan_data_subject_erasure( + self.session, + registry=self.registry, + row=row, + expected_revision=2, + ) + actions = row.erasure_plan["actions"] + self.assertEqual(2, len(actions)) + self.assertEqual(1, row.erasure_plan["retained_count"]) + executable_id = next(item["action_id"] for item in actions if item["executable"]) + + execute_data_subject_erasure( + self.session, + registry=self.registry, + row=row, + expected_revision=3, + action_ids=[executable_id], + ) + self.assertEqual("completed", row.status) + self.assertEqual(1, self.provider.executions) + + export = json.loads(data_subject_export(row)) + self.assertEqual("govoplan.dsars.export.v1", export["schema"]) + self.assertEqual("DSAR-2026-001", export["request"]["reference"]) + self.assertEqual(64, len(export["manifest_sha256"])) + + with self.assertRaises(RevisionConflictError): + search_data_subject_request( + self.session, + registry=self.registry, + row=row, + expected_revision=1, + ) + + def test_access_only_request_cannot_create_erasure_plan(self) -> None: + row = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-2026-002", + request_kind="access", + subject=DsarSubjectRef(account_id="account-1"), + purpose="Provide access information.", + legal_basis=None, + due_at=None, + requested_by_account_id="account-operator", + ) + search_data_subject_request( + self.session, + registry=self.registry, + row=row, + expected_revision=1, + ) + with self.assertRaisesRegex(ValueError, "does not include erasure"): + plan_data_subject_erasure( + self.session, + registry=self.registry, + row=row, + expected_revision=2, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_template_contract.py b/tests/test_template_contract.py index e03cdf5..4303117 100644 --- a/tests/test_template_contract.py +++ b/tests/test_template_contract.py @@ -10,8 +10,11 @@ from govoplan_core.core.files import ( ) from govoplan_core.core.templates import ( CAPABILITY_TEMPLATE_CATALOG, + CAPABILITY_TEMPLATE_CONTENT_LIBRARY, CAPABILITY_TEMPLATE_RENDERER, TemplateCatalogProvider, + TemplateContentDraftRequest, + TemplateContentLibraryProvider, TemplateRenderRequest, TemplateRendererProvider, ) @@ -37,6 +40,12 @@ class _Renderer: return None +class _ContentLibrary: + def create_content_draft(self, session, principal, *, request): + del session, principal, request + return None + + class _Store: def store_artifact(self, session, principal, *, request): del session, principal @@ -54,10 +63,14 @@ class _Store: class TemplateContractTests(unittest.TestCase): def test_capability_names_and_runtime_protocols_are_stable(self) -> None: self.assertEqual("templates.catalog", CAPABILITY_TEMPLATE_CATALOG) + self.assertEqual( + "templates.content_library", CAPABILITY_TEMPLATE_CONTENT_LIBRARY + ) self.assertEqual("templates.renderer", CAPABILITY_TEMPLATE_RENDERER) self.assertEqual("files.artifact_store", CAPABILITY_FILES_ARTIFACT_STORE) self.assertIsInstance(_Catalog(), TemplateCatalogProvider) self.assertIsInstance(_Renderer(), TemplateRendererProvider) + self.assertIsInstance(_ContentLibrary(), TemplateContentLibraryProvider) self.assertIsInstance(_Store(), ManagedArtifactStore) def test_requests_do_not_expose_consumer_or_files_models(self) -> None: @@ -71,6 +84,14 @@ class TemplateContractTests(unittest.TestCase): self.assertEqual("preview", render.mode) self.assertEqual("Generated", artifact.folder) + content = TemplateContentDraftRequest( + name="Closing paragraph", + template_type="content_fragment", + usages=("campaign.content",), + content_text="Kind regards", + ) + self.assertEqual("de", content.locale) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_wheel_runtime.py b/tests/test_wheel_runtime.py index 1107c4e..ea9f055 100644 --- a/tests/test_wheel_runtime.py +++ b/tests/test_wheel_runtime.py @@ -69,7 +69,7 @@ class WheelRuntimeTests(unittest.TestCase): self.assertNotEqual(repository_root, runtime_root) self.assertTrue((runtime_root / "alembic.ini").is_file()) self.assertTrue((runtime_root / "alembic" / "env.py").is_file()) - self.assertEqual(["e14b8c2d6f90"], result["heads"]) + self.assertEqual(["b47e6f809a13"], result["heads"]) self.assertIn("core_scopes", result["tables"]) self.assertIn("core_system_settings", result["tables"])