From 975d90b012c22e25813b6b66034015a20be02fba Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 21 Aug 2026 04:08:21 +0200 Subject: [PATCH] feat(notifications): add governed DSAR coverage --- .../backend/dsar_provider.py | 655 ++++++++++++++++++ .../backend/manifest.py | 152 +++- tests/test_dsar_provider.py | 476 +++++++++++++ 3 files changed, 1270 insertions(+), 13 deletions(-) create mode 100644 src/govoplan_notifications/backend/dsar_provider.py create mode 100644 tests/test_dsar_provider.py diff --git a/src/govoplan_notifications/backend/dsar_provider.py b/src/govoplan_notifications/backend/dsar_provider.py new file mode 100644 index 0000000..4c63aa9 --- /dev/null +++ b/src/govoplan_notifications/backend/dsar_provider.py @@ -0,0 +1,655 @@ +from __future__ import annotations + +import math +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timezone + +from sqlalchemy import func, or_ +from sqlalchemy.orm import Session + +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarExecutionResultRef, + DsarRecordRef, + DsarSubjectRef, + dsar_capability_name, +) +from govoplan_notifications.backend.db.models import ( + NotificationMessage, + NotificationPreference, +) + + +NOTIFICATIONS_DSAR_CAPABILITY = dsar_capability_name("notifications") +_MAX_RECORDS = 1_000 +_MAX_ATTEMPTS_PER_MESSAGE = 100 +_MAX_BODY_CHARS = 100_000 +_MAX_JSON_NODES = 10_000 +_MAX_JSON_CHARS = 100_000 +_CONFLICT = object() +_SENSITIVE_KEY_PARTS = ( + "authorization", + "cookie", + "credential", + "password", + "secret", + "token", + "api_key", + "apikey", +) + + +@dataclass(frozen=True, slots=True) +class _SubjectSelectors: + actor_ids: tuple[str, ...] + email: str | None + notification_id: str | None + preference_id: str | None + + +@dataclass(slots=True) +class _JsonBudget: + nodes: int = 0 + characters: int = 0 + + +class NotificationsDsarProvider: + provider_id = "notifications" + module_id = "notifications" + + def search_subject( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + ) -> Sequence[DsarRecordRef]: + db = _session(session) + selectors = _subject_selectors(subject) + if selectors is None: + return () + + records: list[DsarRecordRef] = [] + if selectors.preference_id is None: + messages = db.query(NotificationMessage).filter( + NotificationMessage.tenant_id == tenant_id, + _recipient_filter(selectors), + ) + if selectors.notification_id: + messages = messages.filter( + NotificationMessage.id == selectors.notification_id + ) + rows = ( + messages.order_by( + NotificationMessage.created_at, + NotificationMessage.id, + ) + .limit(_MAX_RECORDS + 1) + .all() + ) + if len(rows) > _MAX_RECORDS: + raise ValueError( + "Notifications DSAR result limit exceeded; narrow the selectors." + ) + records.extend(_message_record(row) for row in rows) + + if selectors.notification_id is None and selectors.actor_ids: + preferences = db.query(NotificationPreference).filter( + NotificationPreference.tenant_id == tenant_id, + NotificationPreference.user_id.in_(selectors.actor_ids), + ) + if selectors.preference_id: + preferences = preferences.filter( + NotificationPreference.id == selectors.preference_id + ) + rows = ( + preferences.order_by(NotificationPreference.id) + .limit(_MAX_RECORDS + 1) + .all() + ) + if len(rows) > _MAX_RECORDS: + raise ValueError( + "Notification preference DSAR result limit exceeded; narrow the selectors." + ) + records.extend(_preference_record(row) for row in rows) + + if len(records) > _MAX_RECORDS: + raise ValueError( + "Notifications DSAR combined result limit exceeded; narrow the selectors." + ) + order = {"notification_preference": 10, "notification_message": 20} + return tuple( + sorted( + records, + key=lambda item: (order[item.resource_type], item.resource_id), + ) + ) + + def plan_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + records: Sequence[DsarRecordRef], + ) -> Sequence[DsarErasureActionRef]: + db = _session(session) + selectors = _subject_selectors(subject) + if selectors is None: + raise ValueError("Notifications DSAR subject selectors conflict.") + + actions: list[DsarErasureActionRef] = [] + for record in records: + _validate_record(record) + if record.resource_type == "notification_preference": + preference = ( + db.query(NotificationPreference) + .filter( + NotificationPreference.tenant_id == tenant_id, + NotificationPreference.id == record.resource_id, + ) + .one_or_none() + ) + if ( + preference is None + or preference.user_id not in selectors.actor_ids + or ( + selectors.preference_id + and preference.id != selectors.preference_id + ) + ): + actions.append( + _manual_action(record, "preference ownership mismatch") + ) + continue + actions.append(_delete_action(record, preference.updated_at)) + continue + + notification = ( + db.query(NotificationMessage) + .filter( + NotificationMessage.tenant_id == tenant_id, + NotificationMessage.id == record.resource_id, + _recipient_filter(selectors), + ) + .one_or_none() + ) + if notification is None or ( + selectors.notification_id + and notification.id != selectors.notification_id + ): + actions.append(_manual_action(record, "recipient or selector mismatch")) + elif notification.channel == "inbox": + actions.append(_delete_action(record, notification.updated_at)) + else: + actions.append( + _manual_action( + record, + "external-channel delivery and provider evidence require retention review", + ) + ) + return tuple(actions) + + def execute_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + actions: Sequence[DsarErasureActionRef], + request_id: str, + ) -> Sequence[DsarExecutionResultRef]: + db = _session(session) + selectors = _subject_selectors(subject) + if selectors is None: + raise ValueError("Notifications DSAR subject selectors conflict.") + + results: list[DsarExecutionResultRef] = [] + for action in actions: + _validate_action(action) + if not action.executable or action.kind != "delete": + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary=( + "External notification delivery evidence requires an " + "authorized retention review." + ), + ) + ) + continue + + if action.resource_type == "notification_preference": + row = ( + db.query(NotificationPreference) + .filter( + NotificationPreference.tenant_id == tenant_id, + NotificationPreference.id == action.resource_id, + ) + .one_or_none() + ) + if row is None: + results.append(_unchanged(action, request_id, "preference")) + continue + if ( + row.user_id not in selectors.actor_ids + or str(action.metadata.get("owner_id") or "") != row.user_id + or (selectors.preference_id and row.id != selectors.preference_id) + ): + results.append(_blocked(action, "preference ownership mismatch")) + continue + elif action.resource_type == "notification_message": + row = ( + db.query(NotificationMessage) + .filter( + NotificationMessage.tenant_id == tenant_id, + NotificationMessage.id == action.resource_id, + ) + .one_or_none() + ) + if row is None: + results.append(_unchanged(action, request_id, "notification")) + continue + if ( + row.channel != "inbox" + or not _recipient_matches(row, selectors) + or ( + selectors.notification_id + and row.id != selectors.notification_id + ) + ): + results.append(_blocked(action, "recipient or channel mismatch")) + continue + else: + raise ValueError("Unsupported executable Notifications DSAR action.") + + if action.metadata.get("updated_at") != _iso(row.updated_at): + results.append(_blocked(action, "the record changed after planning")) + continue + resource_id = str(row.id) + db.delete(row) + db.flush() + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="executed", + summary=( + "Deleted the personal Notifications record without " + "traversing or changing its source-module data." + ), + evidence={ + "request_id": request_id, + "resource_type": action.resource_type, + "resource_id": resource_id, + }, + ) + ) + return tuple(results) + + +def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None: + references = subject.external_references + values = { + "account_id": _coalesce( + subject.account_id, + references.get("notifications.account"), + references.get("access.account"), + ), + "membership_id": _coalesce( + subject.membership_id, + references.get("notifications.membership"), + references.get("tenancy.membership"), + ), + "identity_id": _coalesce( + subject.identity_id, + references.get("notifications.identity"), + references.get("identity.id"), + ), + "user_id": _coalesce( + references.get("notifications.user"), + references.get("idm.user"), + ), + "email": _coalesce( + _normalized_email(subject.email), + _normalized_email(references.get("notifications.email")), + ), + "notification_id": _coalesce( + references.get("notifications.notification"), + references.get("notifications.message"), + ), + "preference_id": _coalesce( + references.get("notifications.preference"), + references.get("notifications.preference_id"), + ), + } + if any(value is _CONFLICT for value in values.values()): + return None + actor_ids = tuple( + dict.fromkeys( + value + for key in ("account_id", "membership_id", "identity_id", "user_id") + if (value := _optional_string(values[key])) + ) + ) + email = _optional_string(values["email"]) + if not actor_ids and email is None: + return None + return _SubjectSelectors( + actor_ids=actor_ids, + email=email, + notification_id=_optional_string(values["notification_id"]), + preference_id=_optional_string(values["preference_id"]), + ) + + +def _coalesce(*values: str | None) -> str | None | object: + normalized = {str(value).strip() for value in values if str(value or "").strip()} + if len(normalized) > 1: + return _CONFLICT + return next(iter(normalized), None) + + +def _optional_string(value: object) -> str | None: + return value if isinstance(value, str) and value else None + + +def _normalized_email(value: str | None) -> str | None: + normalized = (value or "").strip().casefold() + return normalized or None + + +def _recipient_filter(selectors: _SubjectSelectors): + filters = [] + if selectors.actor_ids: + filters.append(NotificationMessage.recipient_id.in_(selectors.actor_ids)) + if selectors.email: + filters.extend( + ( + func.lower(NotificationMessage.recipient) == selectors.email, + func.lower(NotificationMessage.recipient_id) == selectors.email, + ) + ) + return or_(*filters) + + +def _recipient_matches( + notification: NotificationMessage, + selectors: _SubjectSelectors, +) -> bool: + if notification.recipient_id in selectors.actor_ids: + return True + return bool( + selectors.email + and selectors.email + in { + _normalized_email(notification.recipient), + _normalized_email(notification.recipient_id), + } + ) + + +def _message_record(notification: NotificationMessage) -> DsarRecordRef: + attempts = list(notification.attempts) + if len(attempts) > _MAX_ATTEMPTS_PER_MESSAGE: + raise ValueError( + "Notifications DSAR delivery-attempt limit exceeded; narrow the selectors." + ) + return DsarRecordRef( + provider_id="notifications", + module_id="notifications", + resource_type="notification_message", + resource_id=notification.id, + category="direct_notification", + title=f"Notification: {(notification.subject or notification.event_kind)[:200]}", + data={ + "source_module": notification.source_module, + "source_resource_type": notification.source_resource_type, + "source_resource_id": notification.source_resource_id, + "event_kind": notification.event_kind, + "channel": notification.channel, + "recipient": notification.recipient, + "recipient_type": notification.recipient_type, + "recipient_id": notification.recipient_id, + "recipient_label": notification.recipient_label, + "subject": notification.subject, + "body_text": _bounded_text(notification.body_text, "body text"), + "body_html": _bounded_text(notification.body_html, "body HTML"), + "action_url": notification.action_url, + "priority": notification.priority, + "status": notification.status, + "not_before_at": _iso(notification.not_before_at), + "queued_at": _iso(notification.queued_at), + "sent_at": _iso(notification.sent_at), + "failed_at": _iso(notification.failed_at), + "read_at": _iso(notification.read_at), + "acknowledged_at": _iso(notification.acknowledged_at), + "cancelled_at": _iso(notification.cancelled_at), + "attempt_count": notification.attempt_count, + "last_error": _bounded_text(notification.last_error, "last error"), + "external_message_id": notification.external_message_id, + "payload": _bounded_json(notification.payload, label="payload"), + "metadata": _bounded_json(notification.metadata_, label="metadata"), + "deleted_at": _iso(notification.deleted_at), + "attempts": [ + { + "id": attempt.id, + "attempt_no": attempt.attempt_no, + "channel": attempt.channel, + "provider": attempt.provider, + "status": attempt.status, + "started_at": _iso(attempt.started_at), + "finished_at": _iso(attempt.finished_at), + "external_message_id": attempt.external_message_id, + "error": _bounded_text(attempt.error, "attempt error"), + "details": _bounded_json( + attempt.details, + label="attempt details", + ), + } + for attempt in attempts + ], + }, + observed_at=_aware(notification.updated_at), + retention_reason=( + "External-channel messages require delivery and source-record retention review." + if notification.channel != "inbox" + else None + ), + ) + + +def _preference_record(preference: NotificationPreference) -> DsarRecordRef: + muted = preference.muted_source_modules + if not isinstance(muted, list) or len(muted) > 500: + raise ValueError("Notification muted-source preference exceeds the DSAR bound.") + return DsarRecordRef( + provider_id="notifications", + module_id="notifications", + resource_type="notification_preference", + resource_id=preference.id, + category="personal_communication_preference", + title="Personal notification preferences", + data={ + "user_id": preference.user_id, + "show_unread_badge": preference.show_unread_badge, + "email_enabled": preference.email_enabled, + "email_digest_enabled": preference.email_digest_enabled, + "muted_source_modules": [str(value)[:100] for value in muted], + "metadata": _bounded_json( + preference.metadata_, + label="preference metadata", + ), + }, + observed_at=_aware(preference.updated_at), + ) + + +def _bounded_text(value: str | None, label: str) -> str | None: + if value is None: + return None + if len(value) > _MAX_BODY_CHARS: + raise ValueError( + f"Notification {label} exceeds the DSAR export bound; narrow and review the record." + ) + return value + + +def _bounded_json(value: object, *, label: str) -> object: + budget = _JsonBudget() + + def project(item: object, depth: int) -> object: + budget.nodes += 1 + if budget.nodes > _MAX_JSON_NODES or depth > 12: + raise ValueError(f"Notification {label} exceeds the DSAR structure bound.") + if item is None or isinstance(item, (bool, int)): + return item + if isinstance(item, float): + if not math.isfinite(item): + raise ValueError(f"Notification {label} contains a non-finite number.") + return item + if isinstance(item, str): + budget.characters += len(item) + if budget.characters > _MAX_JSON_CHARS: + raise ValueError( + f"Notification {label} exceeds the DSAR character bound." + ) + return item + if isinstance(item, Mapping): + if len(item) > 1_000: + raise ValueError(f"Notification {label} contains too many fields.") + projected: dict[str, object] = {} + for raw_key, raw_value in item.items(): + key = str(raw_key) + if len(key) > 500: + raise ValueError(f"Notification {label} contains an oversized key.") + budget.characters += len(key) + if budget.characters > _MAX_JSON_CHARS: + raise ValueError( + f"Notification {label} exceeds the DSAR character bound." + ) + projected[key] = ( + "[redacted]" + if _is_sensitive_key(key) + else project(raw_value, depth + 1) + ) + return projected + if isinstance(item, (list, tuple)): + if len(item) > 1_000: + raise ValueError(f"Notification {label} contains too many items.") + return [project(entry, depth + 1) for entry in item] + raise ValueError(f"Notification {label} contains an unsupported value.") + + return project(value, 0) + + +def _is_sensitive_key(value: str) -> bool: + normalized = value.strip().casefold().replace("-", "_") + return any(part in normalized for part in _SENSITIVE_KEY_PARTS) + + +def _delete_action( + record: DsarRecordRef, + updated_at: datetime | None, +) -> DsarErasureActionRef: + metadata: dict[str, object] = {"updated_at": _iso(updated_at)} + if record.resource_type == "notification_preference": + metadata["owner_id"] = str(record.data.get("user_id") or "") + return DsarErasureActionRef( + action_id=f"notifications:delete:{record.resource_type}:{record.resource_id}", + provider_id="notifications", + module_id="notifications", + kind="delete", + resource_type=record.resource_type, + resource_id=record.resource_id, + title=f"Delete {record.title}", + rationale=( + "The record is a subject-owned preference or derived in-product " + "notification copy; source-module data remains unchanged." + ), + executable=True, + irreversible=True, + metadata=metadata, + ) + + +def _manual_action(record: DsarRecordRef, reason: str) -> DsarErasureActionRef: + return DsarErasureActionRef( + action_id=( + f"notifications:manual_review:{record.resource_type}:{record.resource_id}" + ), + provider_id="notifications", + module_id="notifications", + kind="manual_review", + resource_type=record.resource_type, + resource_id=record.resource_id, + title=f"Review {record.title}", + rationale=f"Automatic deletion stopped because {reason}.", + executable=False, + ) + + +def _unchanged( + action: DsarErasureActionRef, + request_id: str, + label: str, +) -> DsarExecutionResultRef: + return DsarExecutionResultRef( + action_id=action.action_id, + status="unchanged", + summary=f"The personal notification {label} was already absent.", + evidence={"request_id": request_id}, + ) + + +def _blocked( + action: DsarErasureActionRef, + reason: str, +) -> DsarExecutionResultRef: + return DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary=f"Notifications erasure stopped because of {reason}.", + ) + + +def _iso(value: datetime | None) -> str | None: + aware = _aware(value) + return aware.isoformat() if aware else None + + +def _aware(value: datetime | None) -> datetime | None: + if value is None or value.tzinfo is not None: + return value + return value.replace(tzinfo=timezone.utc) + + +def _session(value: object) -> Session: + if not isinstance(value, Session): + raise TypeError("Notifications DSAR requires a SQLAlchemy Session.") + return value + + +def _validate_record(record: DsarRecordRef) -> None: + if record.provider_id != "notifications" or record.module_id != "notifications": + raise ValueError("Notifications DSAR cannot plan a foreign provider record.") + if ( + record.resource_type + not in { + "notification_message", + "notification_preference", + } + or not record.resource_id + ): + raise ValueError("Notifications DSAR record identity is invalid.") + + +def _validate_action(action: DsarErasureActionRef) -> None: + if action.provider_id != "notifications" or action.module_id != "notifications": + raise ValueError("Notifications DSAR cannot execute a foreign provider action.") + if not action.action_id.startswith("notifications:"): + raise ValueError("Notifications DSAR action identity is invalid.") + + +__all__ = ["NOTIFICATIONS_DSAR_CAPABILITY", "NotificationsDsarProvider"] diff --git a/src/govoplan_notifications/backend/manifest.py b/src/govoplan_notifications/backend/manifest.py index f77e546..8f07525 100644 --- a/src/govoplan_notifications/backend/manifest.py +++ b/src/govoplan_notifications/backend/manifest.py @@ -2,9 +2,16 @@ from __future__ import annotations from pathlib import Path -from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER -from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard +from govoplan_core.core.access import ( + CAPABILITY_AUTH_PERMISSION_EVALUATOR, + CAPABILITY_AUTH_PRINCIPAL_RESOLVER, +) +from govoplan_core.core.module_guards import ( + drop_table_retirement_provider, + persistent_table_uninstall_guard, +) from govoplan_core.core.modules import ( + CapabilityDocumentation, DocumentationTopic, FrontendModule, FrontendRoute, @@ -21,6 +28,10 @@ from govoplan_core.core.notifications import CAPABILITY_NOTIFICATIONS_DISPATCH from govoplan_core.core.views import ViewSurface from govoplan_core.db.base import Base from govoplan_notifications.backend.db import models as notification_models # noqa: F401 - populate Notifications ORM metadata +from govoplan_notifications.backend.dsar_provider import ( + NOTIFICATIONS_DSAR_CAPABILITY, + NotificationsDsarProvider, +) MODULE_ID = "notifications" @@ -47,10 +58,26 @@ def _permission(scope: str, label: str, description: str) -> PermissionDefinitio PERMISSIONS = ( - _permission(READ_SCOPE, "View notifications", "Read the authenticated actor's notification inbox and delivery state."), - _permission(WRITE_SCOPE, "Manage notifications", "Create, update, cancel, read, and acknowledge notifications."), - _permission(DISPATCH_SCOPE, "Dispatch notifications", "Run notification delivery attempts and delivery workers."), - _permission(ADMIN_SCOPE, "Administer notifications", "Explicitly read and manage tenant-wide notification delivery state."), + _permission( + READ_SCOPE, + "View notifications", + "Read the authenticated actor's notification inbox and delivery state.", + ), + _permission( + WRITE_SCOPE, + "Manage notifications", + "Create, update, cancel, read, and acknowledge notifications.", + ), + _permission( + DISPATCH_SCOPE, + "Dispatch notifications", + "Run notification delivery attempts and delivery workers.", + ), + _permission( + ADMIN_SCOPE, + "Administer notifications", + "Explicitly read and manage tenant-wide notification delivery state.", + ), ) ROLE_TEMPLATES = ( @@ -73,8 +100,19 @@ def _tenant_summary(session, tenant_id: str) -> dict[str, int]: from govoplan_notifications.backend.db.models import NotificationMessage return { - "notifications": session.query(NotificationMessage).filter(NotificationMessage.tenant_id == tenant_id, NotificationMessage.deleted_at.is_(None)).count(), - "pending_notifications": session.query(NotificationMessage).filter(NotificationMessage.tenant_id == tenant_id, NotificationMessage.status.in_(("pending", "queued", "failed")), NotificationMessage.deleted_at.is_(None)).count(), + "notifications": session.query(NotificationMessage) + .filter( + NotificationMessage.tenant_id == tenant_id, + NotificationMessage.deleted_at.is_(None), + ) + .count(), + "pending_notifications": session.query(NotificationMessage) + .filter( + NotificationMessage.tenant_id == tenant_id, + NotificationMessage.status.in_(("pending", "queued", "failed")), + NotificationMessage.deleted_at.is_(None), + ) + .count(), } @@ -84,18 +122,89 @@ def _notifications_router(_context: ModuleContext): return router +def _dsar_provider(_context: ModuleContext) -> NotificationsDsarProvider: + return NotificationsDsarProvider() + + manifest = ModuleManifest( id=MODULE_ID, name=MODULE_NAME, version=MODULE_VERSION, - required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR), - optional_dependencies=("mail", "tasks", "portal", "workflow_engine", "calendar", "scheduling"), - provides_interfaces=(ModuleInterfaceProvider(name="notifications.dispatch", version="0.1.8"),), + required_capabilities=( + CAPABILITY_AUTH_PRINCIPAL_RESOLVER, + CAPABILITY_AUTH_PERMISSION_EVALUATOR, + ), + optional_dependencies=( + "mail", + "tasks", + "portal", + "workflow_engine", + "calendar", + "scheduling", + ), + provides_interfaces=( + ModuleInterfaceProvider(name="notifications.dispatch", version="0.1.8"), + ModuleInterfaceProvider( + name=NOTIFICATIONS_DSAR_CAPABILITY, + version="0.1.0", + ), + ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, route_factory=_notifications_router, tenant_summary_providers=(_tenant_summary,), documentation=( + DocumentationTopic( + id="notifications.data-subject-requests", + title="Notification data-subject requests", + summary=( + "Export subject-addressed notifications and preferences, delete " + "personal inbox copies, and review external delivery evidence." + ), + body=( + "Notifications correlates verified account, membership, identity, " + "user, and email selectors only inside the active tenant. The access " + "package contains bounded message content, recipient fields, lifecycle " + "timestamps, source references, payload and metadata, and sanitized " + "delivery attempts. Personal badge, email, digest, and source-muting " + "preferences are also included. Explicit notification or preference " + "references can narrow a request and must still match the subject. " + "In-product inbox messages are derived copies, so erasure can delete " + "them after recipient and change checks without following the source " + "reference into its owning module. Personal preferences can also be " + "deleted and then return to defaults. Mail and other external-channel " + "records require manual retention review because provider acceptance " + "and delivery evidence may remain outside Notifications. Repeated " + "execution is unchanged; unrelated tenants and recipients are never " + "included." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("user", "tenant_admin", "operator", "auditor"), + related_modules=("core", "mail", "tasks", "workflow_engine"), + metadata={ + "help_contexts": [ + "notifications.page.inbox", + "notifications.page.detail", + "notifications.page.delivery", + "notifications.settings.preferences", + "privacy.data-subject-requests", + ], + "consequence_classes": { + "delete_inbox_copy": ( + "Removes the derived in-product notification and attempts, " + "not the source-module record." + ), + "delete_preferences": ( + "Removes personal overrides; notification defaults apply again." + ), + "review_external_delivery": ( + "Requires an authorized retention review before changing " + "provider-delivery evidence." + ), + }, + }, + ), DocumentationTopic( id="notifications.center-and-preferences", title="Use the notification center", @@ -266,6 +375,17 @@ manifest = ModuleManifest( "govoplan_notifications.backend.capabilities", fromlist=["dispatch_capability"], ).dispatch_capability(context), + NOTIFICATIONS_DSAR_CAPABILITY: _dsar_provider, + }, + capability_documentation={ + NOTIFICATIONS_DSAR_CAPABILITY: CapabilityDocumentation( + label="Notifications data-subject request provider", + summary=( + "Exports subject-addressed notification data, deletes personal " + "inbox copies and preferences, and isolates external delivery evidence." + ), + contract_version="0.1.0", + ), }, architecture=declared_module_architecture( layer="communication_participation", @@ -273,8 +393,14 @@ manifest = ModuleManifest( maturity="vertical_slice", documentation_ref="docs/NOTIFICATION_INBOX_BOUNDARY.md", test_ref="tests/test_notifications.py", - known_limits=("Production email delivery depends on the optional Mail capability and does not provide an independent transport.",), - owned_concepts=("notification", "notification preference", "notification delivery attempt"), + known_limits=( + "Production email delivery depends on the optional Mail capability and does not provide an independent transport.", + ), + owned_concepts=( + "notification", + "notification preference", + "notification delivery attempt", + ), non_owned_concepts=("mail transport", "domain event", "portal message"), recovery_docs=("docs/EMAIL_DELIVERY.md",), operations_docs=("docs/EMAIL_DELIVERY.md",), diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..f5c48a4 --- /dev/null +++ b/tests/test_dsar_provider.py @@ -0,0 +1,476 @@ +from __future__ import annotations + +import json +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarProvider, + 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_notifications.backend.db.models import ( + NotificationDeliveryAttempt, + NotificationMessage, + NotificationPreference, +) +from govoplan_notifications.backend.dsar_provider import ( + NOTIFICATIONS_DSAR_CAPABILITY, + NotificationsDsarProvider, +) +from govoplan_notifications.backend.manifest import manifest + + +class _Registry: + def __init__( + self, + provider: NotificationsDsarProvider, + *, + active: bool = True, + ) -> None: + self.provider = provider + self.active = active + + def capability_names(self): + return (NOTIFICATIONS_DSAR_CAPABILITY,) + + def capability_owner(self, name): + self._assert_capability(name) + return "notifications" + + 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": ("notifications",) 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": "notifications"})(),) + + @staticmethod + def _assert_capability(name: str) -> None: + if name != NOTIFICATIONS_DSAR_CAPABILITY: + raise KeyError(name) + + +class NotificationsDsarProviderTests(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 = NotificationsDsarProvider() + self.assertIsInstance(self.provider, DsarProvider) + self._seed() + self.session.commit() + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def _message( + self, + notification_id: str, + *, + tenant_id: str = "tenant-1", + channel: str = "inbox", + recipient: str | None = None, + recipient_id: str | None = "membership-1", + subject: str, + ) -> NotificationMessage: + return NotificationMessage( + id=notification_id, + tenant_id=tenant_id, + source_module="tasks", + source_resource_type="task", + source_resource_id=f"task-{notification_id}", + event_kind="task_due", + channel=channel, + recipient=recipient, + recipient_type="email" if recipient else "membership", + recipient_id=recipient_id, + recipient_label="Resident Example", + subject=subject, + body_text=f"Private body for {notification_id}", + body_html=f"

Private HTML for {notification_id}

", + action_url="/tasks", + priority=2, + status="sent", + attempt_count=0, + payload={ + "case_reference": f"CASE-{notification_id}", + "access_token": "provider-secret-do-not-export", + }, + metadata_={"classification": "personal"}, + ) + + def _seed(self) -> None: + inbox = self._message( + "notification-inbox", + subject="Personal inbox notice", + ) + mail = self._message( + "notification-mail", + channel="mail", + recipient="Resident@Example.test", + recipient_id=None, + subject="Personal mail notice", + ) + mail.attempt_count = 1 + mail.attempts.append( + NotificationDeliveryAttempt( + id="attempt-mail", + tenant_id="tenant-1", + attempt_no=1, + channel="mail", + provider="mail.delivery_outbox", + status="accepted", + external_message_id="mail-command-1", + details={ + "provider_state": "accepted", + "authorization": "Bearer provider-secret-do-not-export", + }, + ) + ) + self.session.add_all( + ( + inbox, + mail, + self._message( + "notification-other-recipient", + recipient_id="membership-other", + subject="Other recipient private notice", + ), + self._message( + "notification-other-tenant", + tenant_id="tenant-2", + subject="Other tenant private notice", + ), + NotificationPreference( + id="preference-personal", + tenant_id="tenant-1", + user_id="membership-1", + show_unread_badge=False, + email_enabled=True, + email_digest_enabled=True, + muted_source_modules=["campaign", "calendar"], + metadata_={"digest_hour": 8}, + ), + NotificationPreference( + id="preference-other", + tenant_id="tenant-1", + user_id="membership-other", + show_unread_badge=True, + email_enabled=False, + email_digest_enabled=False, + muted_source_modules=[], + metadata_={}, + ), + ) + ) + + @staticmethod + def _subject() -> DsarSubjectRef: + return DsarSubjectRef( + account_id="account-1", + membership_id="membership-1", + identity_id="identity-1", + email="resident@example.test", + ) + + def test_search_is_tenant_recipient_scoped_and_exports_bounded_content( + self, + ) -> None: + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=self._subject(), + ) + + self.assertEqual( + [ + "preference-personal", + "notification-inbox", + "notification-mail", + ], + [record.resource_id for record in records], + ) + exported = json.dumps([record.to_dict() for record in records]) + self.assertIn("Private body for notification-inbox", exported) + self.assertIn("CASE-notification-mail", exported) + self.assertIn("mail-command-1", exported) + self.assertIn("digest_hour", exported) + self.assertIn("[redacted]", exported) + self.assertNotIn("provider-secret-do-not-export", exported) + self.assertNotIn("notification-other-recipient", exported) + self.assertNotIn("notification-other-tenant", exported) + + def test_resource_references_narrow_and_alias_conflicts_fail_closed(self) -> None: + notification = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + membership_id="membership-1", + external_references={ + "notifications.notification": "notification-inbox" + }, + ), + ) + preference = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + membership_id="membership-1", + external_references={"notifications.preference": "preference-personal"}, + ), + ) + conflict = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + account_id="account-1", + external_references={"notifications.account": "account-other"}, + ), + ) + reference_only = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + external_references={"notifications.notification": "notification-inbox"} + ), + ) + + self.assertEqual( + ["notification-inbox"], [item.resource_id for item in notification] + ) + self.assertEqual( + ["preference-personal"], [item.resource_id for item in preference] + ) + self.assertEqual((), conflict) + self.assertEqual((), reference_only) + + def test_erasure_deletes_preferences_and_inbox_but_reviews_mail(self) -> None: + subject = self._subject() + 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.assertEqual( + ["delete", "delete", "manual_review"], + [action.kind for action in actions], + ) + + first = self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=actions, + request_id="dsar-1", + ) + second = self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=actions, + request_id="dsar-1", + ) + + self.assertEqual( + ["executed", "executed", "blocked"], + [result.status for result in first], + ) + self.assertEqual( + ["unchanged", "unchanged", "blocked"], + [result.status for result in second], + ) + self.assertIsNone( + self.session.get(NotificationPreference, "preference-personal") + ) + self.assertIsNone(self.session.get(NotificationMessage, "notification-inbox")) + self.assertIsNotNone(self.session.get(NotificationMessage, "notification-mail")) + self.assertIsNotNone( + self.session.get(NotificationMessage, "notification-other-recipient") + ) + + def test_changed_and_foreign_resources_are_blocked(self) -> None: + subject = self._subject() + record = next( + item + for item in self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=subject, + ) + if item.resource_id == "notification-inbox" + ) + action = self.provider.plan_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + records=(record,), + )[0] + notification = self.session.get(NotificationMessage, "notification-inbox") + notification.subject = "Changed after planning" + self.session.flush() + changed = self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=(action,), + request_id="dsar-1", + ) + self.assertEqual("blocked", changed[0].status) + + with self.assertRaisesRegex(ValueError, "foreign provider record"): + self.provider.plan_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + records=( + DsarRecordRef( + provider_id="mail", + module_id="mail", + resource_type="notification_message", + resource_id="notification-inbox", + category="message", + title="Foreign message", + ), + ), + ) + with self.assertRaisesRegex(ValueError, "foreign provider action"): + self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=( + DsarErasureActionRef( + action_id="mail:delete:notification:notification-inbox", + provider_id="mail", + module_id="mail", + kind="delete", + resource_type="notification_message", + resource_id="notification-inbox", + title="Delete notification", + rationale="Foreign action", + executable=True, + ), + ), + request_id="dsar-1", + ) + + def test_oversized_content_fails_closed(self) -> None: + notification = self.session.get( + NotificationMessage, + "notification-inbox", + ) + notification.body_text = "x" * 100_001 + self.session.flush() + + with self.assertRaisesRegex(ValueError, "body text exceeds"): + self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + membership_id="membership-1", + external_references={ + "notifications.notification": "notification-inbox" + }, + ), + ) + + def test_core_workflow_and_manifest_register_provider(self) -> None: + row = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-NOTIFICATIONS-1", + request_kind="access", + subject=self._subject(), + purpose="Respond to a verified request.", + legal_basis="Article 15 GDPR", + due_at=None, + requested_by_account_id="privacy-officer", + ) + self.session.commit() + search_data_subject_request( + self.session, + registry=_Registry(self.provider), + row=row, + expected_revision=1, + ) + self.assertEqual( + [NOTIFICATIONS_DSAR_CAPABILITY], + row.coverage["provider_capabilities"], + ) + self.assertEqual(3, row.search_result["record_count"]) + + inactive = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-NOTIFICATIONS-2", + request_kind="access", + subject=self._subject(), + purpose="Respond to a verified request.", + legal_basis="Article 15 GDPR", + due_at=None, + requested_by_account_id="privacy-officer", + ) + self.session.commit() + search_data_subject_request( + self.session, + registry=_Registry(self.provider, active=False), + row=inactive, + expected_revision=1, + ) + self.assertEqual( + [NOTIFICATIONS_DSAR_CAPABILITY], + inactive.coverage["inactive_provider_capabilities"], + ) + + self.assertIn(NOTIFICATIONS_DSAR_CAPABILITY, manifest.capability_factories) + self.assertIn( + NOTIFICATIONS_DSAR_CAPABILITY, + manifest.capability_documentation, + ) + self.assertIn( + NOTIFICATIONS_DSAR_CAPABILITY, + {item.name for item in manifest.provides_interfaces}, + ) + self.assertTrue( + any( + topic.id == "notifications.data-subject-requests" + and {"admin", "user"}.issubset(topic.documentation_types) + for topic in manifest.documentation + ) + ) + + +if __name__ == "__main__": + unittest.main()