diff --git a/README.md b/README.md index 1eca8f7..bbe2983 100644 --- a/README.md +++ b/README.md @@ -92,6 +92,9 @@ services can cooperate without importing campaign internals: - `campaigns.policyContext` for retention/policy provenance - `campaigns.deliveryTasks` for queued send and append-to-Sent workers - `campaigns.retention` for campaign-owned retention cleanup +- `privacy.dsar.campaigns` for tenant-scoped recipient, version, delivery, + report-projection, and artifact-metadata discovery plus governed erasure + planning Keep these capability payloads narrow: stable ids, policy payloads, and task results only. diff --git a/docs/CAMPAIGN_HANDBOOK.md b/docs/CAMPAIGN_HANDBOOK.md index eefa5c9..131cd83 100644 --- a/docs/CAMPAIGN_HANDBOOK.md +++ b/docs/CAMPAIGN_HANDBOOK.md @@ -557,6 +557,30 @@ purpose, lawful basis, minimization, export control, and retention before the campaign starts; do not use Campaign as a substitute consent or address-master system. +The Core data-subject-request workflow discovers Campaign through the optional +`privacy.dsar.campaigns` capability. After the request's email, membership, and +namespaced Campaign references have been independently authorized and +corroborated, the provider searches only the effective tenant and isolates the +matching recipient entries and jobs. Its JSON result includes safe Campaign, +version, delivery-attempt, schedule, report-projection, share, import-mapping, +attachment, and generated-artifact metadata. Generated EML bytes and paths, +storage locators, delivery target snapshots, worker claims, idempotency +material, credentials, secret-like values, and unrelated recipients are never +embedded in that result. Authorized Campaign and Files review surfaces remain +the source for content that cannot safely be copied into the DSAR case. + +Built, locked, published, terminal, delivered, or corrected records are +retained with an explicit reason and continue through Campaign's configured +retention and redaction process. Draft recipient content and user-owned +attachment content require coordinated manual review because copies may span +version JSON, jobs, generated messages, and managed files. The provider can +idempotently revoke an active share aimed at the subject and delete the +subject's personal recipient-import mapping profile. It does not rewrite +delivery evidence, delete generated artifacts, or report derived Campaign +counts as a separate store. Re-running an approved action is safe: already +revoked or absent data is reported as unchanged, and tenant, subject, and row +ownership are revalidated immediately before mutation. + ### Audit and destructive actions Material authoring, validation, locking, review, queueing, send, retry, diff --git a/src/govoplan_campaign/backend/dsar_provider.py b/src/govoplan_campaign/backend/dsar_provider.py new file mode 100644 index 0000000..6b17d2b --- /dev/null +++ b/src/govoplan_campaign/backend/dsar_provider.py @@ -0,0 +1,1327 @@ +from __future__ import annotations + +from collections import Counter +from collections.abc import Mapping, Sequence +from datetime import datetime, timezone + +from sqlalchemy import Text, cast, func, or_ +from sqlalchemy.orm import Session + +from govoplan_campaign.backend.db.models import ( + AttachmentInstance, + Campaign, + CampaignIssue, + CampaignJob, + CampaignMessageAction, + CampaignMessageActionAttempt, + CampaignSchedule, + CampaignShare, + CampaignVersion, + ImapAppendAttempt, + PostboxDeliveryAttempt, + PrintOutputAttempt, + RecipientImportMappingProfile, + SendAttempt, +) +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarExecutionResultRef, + DsarRecordRef, + DsarSubjectRef, + dsar_capability_name, +) + + +CAMPAIGN_DSAR_CAPABILITY = dsar_capability_name("campaigns") +_MAX_RECORDS = 5_000 +_SECRET_PARTS = ( + "password", + "secret", + "token", + "credential", + "authorization", + "private_key", + "claim_token", + "idempotency_key", +) +_CONTENT_KEYS = { + "attachments", + "body", + "content", + "html", + "template", + "text", +} + + +class CampaignDsarProvider: + provider_id = "campaigns" + module_id = "campaigns" + + def search_subject( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + ) -> Sequence[DsarRecordRef]: + db = _session(session) + subject_user_id = _subject_user_id(subject) + subject_email = _subject_email(subject) + references = _campaign_references(subject) + if subject_user_id is None and subject_email is None and not references: + return () + + records: list[DsarRecordRef] = [] + + def append(record: DsarRecordRef) -> None: + if len(records) >= _MAX_RECORDS: + raise ValueError( + "Campaign DSAR match limit exceeded; narrow the subject selectors." + ) + records.append(record) + + jobs = _matching_jobs( + db, + tenant_id=tenant_id, + email=subject_email, + job_id=references.get("job"), + entry_id=references.get("entry"), + ) + job_ids = {row.id for row in jobs} + job_version_ids = {row.campaign_version_id for row in jobs} + campaign_ids = {row.campaign_id for row in jobs} + evidence_campaign_ids = set(campaign_ids) + + versions = _matching_versions( + db, + tenant_id=tenant_id, + subject_user_id=subject_user_id, + email=subject_email, + job_version_ids=job_version_ids, + version_id=references.get("version"), + ) + version_ids = {row.id for row in versions} + campaign_ids.update(row.campaign_id for row in versions) + if references.get("campaign"): + campaign_ids.add(references["campaign"]) + + campaigns = _matching_campaigns( + db, + tenant_id=tenant_id, + subject_user_id=subject_user_id, + campaign_ids=campaign_ids, + ) + campaign_ids = {row.id for row in campaigns} + + for campaign in campaigns: + immutable = campaign.status != "draft" + append( + _record( + "campaign", + campaign.id, + "campaign_workspace", + campaign.name, + { + "match_fields": _matching_fields( + campaign, + subject_user_id, + ("created_by_user_id", "owner_user_id"), + ), + "recipient_context": campaign.id + in {job.campaign_id for job in jobs}, + "external_id": campaign.external_id, + "name": campaign.name, + "description": campaign.description, + "status": campaign.status, + }, + observed_at=campaign.updated_at, + immutable=immutable, + retention_reason=( + "Non-draft Campaign state is institutional workflow and delivery evidence." + if immutable + else None + ), + source_path=f"/campaigns/{campaign.id}", + ) + ) + + jobs_by_version: Counter[str] = Counter(row.campaign_version_id for row in jobs) + for version in versions: + fragments = _matching_entry_fragments(version.raw_json, subject_email) + immutable = _version_is_evidence(version, jobs_by_version[version.id]) + match_fields = _matching_fields( + version, + subject_user_id, + ( + "locked_by_user_id", + "user_locked_by_user_id", + "archived_by_user_id", + ), + ) + if fragments: + match_fields.insert(0, "raw_json.recipient") + if version.id in job_version_ids: + match_fields.insert(0, "campaign_job.recipient") + append( + _record( + "campaign_version", + version.id, + "campaign_recipient_snapshot", + f"Campaign version {version.version_number}", + { + "match_fields": match_fields, + "campaign_id": version.campaign_id, + "version_number": version.version_number, + "workflow_state": version.workflow_state, + "published_at": _iso(version.published_at), + "locked_at": _iso(version.locked_at), + "archived_at": _iso(version.archived_at), + "matching_entries": fragments, + }, + observed_at=version.updated_at, + immutable=immutable, + retention_reason=( + "Built, locked, published, or terminal Campaign versions are retained as execution evidence and are redacted only by Campaign retention policy." + if immutable + else None + ), + source_path=f"/campaigns/{version.campaign_id}", + ) + ) + + immutable_jobs: dict[str, bool] = {} + for job in jobs: + immutable = _job_is_evidence(job) + immutable_jobs[job.id] = immutable + append( + _record( + "campaign_recipient_job", + job.id, + "campaign_recipient_delivery", + job.recipient_email or f"Recipient {job.entry_index}", + { + "campaign_id": job.campaign_id, + "campaign_version_id": job.campaign_version_id, + "entry_index": job.entry_index, + "entry_id": job.entry_id, + "recipient_email": job.recipient_email, + "subject": job.subject, + "build_status": job.build_status, + "validation_status": job.validation_status, + "queue_status": job.queue_status, + "send_status": job.send_status, + "postbox_status": job.postbox_status, + "print_status": job.print_status, + "imap_status": job.imap_status, + "attempt_count": job.attempt_count, + "message_sha256": job.eml_sha256, + "message_size_bytes": job.eml_size_bytes, + "queued_at": _iso(job.queued_at), + "sent_at": _iso(job.sent_at), + "matched_recipients": _matching_recipient_fragment( + job.resolved_recipients, + subject_email, + ), + }, + observed_at=job.updated_at, + immutable=immutable, + retention_reason=( + "Built message and delivery state is retained under Campaign delivery-evidence policy." + if immutable + else None + ), + source_path=f"/campaigns/{job.campaign_id}/report", + ) + ) + if ( + job.eml_sha256 + or job.eml_size_bytes + or job.eml_storage_key + or job.eml_local_path + ): + append( + _record( + "campaign_message_artifact", + job.id, + "generated_message_artifact", + f"Generated message for job {job.id}", + { + "campaign_id": job.campaign_id, + "campaign_version_id": job.campaign_version_id, + "job_id": job.id, + "sha256": job.eml_sha256, + "size_bytes": job.eml_size_bytes, + "artifact_present": bool( + job.eml_storage_key or job.eml_local_path + ), + }, + observed_at=job.updated_at, + immutable=True, + retention_reason=( + "Generated message bytes and their locator follow Campaign artifact retention and recovery controls." + ), + source_path=f"/campaigns/{job.campaign_id}/report", + ) + ) + + self._append_delivery_evidence( + db, + append=append, + tenant_id=tenant_id, + job_ids=job_ids, + immutable_jobs=immutable_jobs, + ) + self._append_report_projections(append=append, jobs=jobs) + self._append_member_resources( + db, + append=append, + tenant_id=tenant_id, + subject_user_id=subject_user_id, + campaign_ids=evidence_campaign_ids, + version_ids=version_ids, + ) + return tuple(records) + + def plan_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + records: Sequence[DsarRecordRef], + ) -> Sequence[DsarErasureActionRef]: + del session + subject_user_id = _subject_user_id(subject) + actions: list[DsarErasureActionRef] = [] + for record in records: + if ( + record.provider_id != self.provider_id + or record.module_id != self.module_id + ): + raise ValueError("Campaign DSAR received a foreign provider record.") + match_fields = {str(value) for value in record.data.get("match_fields", ())} + if record.immutable_evidence: + actions.append( + _action( + f"campaigns:retain:{record.resource_type}:{record.resource_id}", + "retain", + record, + f"Retain {record.title}", + record.retention_reason + or "Institutional Campaign evidence must be retained.", + executable=False, + ) + ) + else: + actions.append( + _action( + f"campaigns:review:{record.resource_type}:{record.resource_id}", + "manual_review", + record, + f"Review {record.title}", + ( + "Recipient data can span a Campaign version, generated message, " + "attachments, and delivery projections. Apply Campaign retention " + "or an approved coordinated correction instead of rewriting one row." + ), + executable=False, + ) + ) + if ( + record.resource_type == "campaign_share" + and "target_id" in match_fields + and record.data.get("revoked_at") is None + and subject_user_id is not None + ): + actions.append( + _action( + f"campaigns:revoke:campaign_share:{record.resource_id}", + "revoke", + record, + "Revoke active Campaign share", + "The active subject-targeted workspace share can be revoked without altering delivery evidence.", + executable=True, + metadata={ + "subject_user_id": subject_user_id, + "tenant_id": tenant_id, + }, + ) + ) + if ( + record.resource_type == "recipient_import_mapping_profile" + and "owner_user_id" in match_fields + and subject_user_id is not None + ): + actions.append( + _action( + f"campaigns:delete:recipient_import_mapping_profile:{record.resource_id}", + "delete", + record, + "Delete personal recipient import mapping", + "The user-owned import preference can be removed without changing Campaign or delivery evidence.", + executable=True, + irreversible=True, + metadata={ + "subject_user_id": subject_user_id, + "tenant_id": tenant_id, + }, + ) + ) + ids = [action.action_id for action in actions] + if len(ids) != len(set(ids)): + raise ValueError("Campaign DSAR produced duplicate action ids.") + 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) + subject_user_id = _subject_user_id(subject) + results: list[DsarExecutionResultRef] = [] + for action in actions: + if ( + subject_user_id is None + or action.provider_id != self.provider_id + or action.module_id != self.module_id + or action.metadata.get("subject_user_id") != subject_user_id + or action.metadata.get("tenant_id") != tenant_id + ): + results.append( + _blocked(action, "The Campaign DSAR action is stale or invalid.") + ) + continue + if action.action_id.startswith( + "campaigns:delete:recipient_import_mapping_profile:" + ): + results.append( + _delete_mapping_profile( + db, + tenant_id=tenant_id, + subject_user_id=subject_user_id, + action=action, + request_id=request_id, + ) + ) + elif action.action_id.startswith("campaigns:revoke:campaign_share:"): + results.append( + _revoke_share( + db, + tenant_id=tenant_id, + subject_user_id=subject_user_id, + action=action, + request_id=request_id, + ) + ) + else: + results.append( + _blocked(action, "Campaign does not execute this action kind.") + ) + db.flush() + return tuple(results) + + def _append_delivery_evidence( + self, + db: Session, + *, + append: object, + tenant_id: str, + job_ids: set[str], + immutable_jobs: Mapping[str, bool], + ) -> None: + if not job_ids: + return + issues = _bounded_rows( + db.query(CampaignIssue) + .filter( + CampaignIssue.tenant_id == tenant_id, + CampaignIssue.job_id.in_(job_ids), + ) + .order_by(CampaignIssue.id) + ) + for issue in issues: + append( # type: ignore[operator] + _record( + "campaign_delivery_issue", + issue.id, + "campaign_delivery_evidence", + issue.code, + { + "campaign_id": issue.campaign_id, + "campaign_version_id": issue.campaign_version_id, + "job_id": issue.job_id, + "severity": issue.severity, + "code": issue.code, + "behavior": issue.behavior, + }, + observed_at=issue.updated_at, + immutable=bool(immutable_jobs.get(str(issue.job_id))), + retention_reason=( + "Issues attached to an evidence-bearing delivery are retained with that delivery." + if immutable_jobs.get(str(issue.job_id)) + else None + ), + ) + ) + + evidence_specs = ( + ( + SendAttempt, + "campaign_send_attempt", + lambda row: { + "job_id": row.job_id, + "attempt_number": row.attempt_number, + "status": row.status, + "smtp_status_code": row.smtp_status_code, + "error_type": row.error_type, + "started_at": _iso(row.started_at), + "finished_at": _iso(row.finished_at), + }, + ), + ( + ImapAppendAttempt, + "campaign_imap_attempt", + lambda row: { + "job_id": row.job_id, + "attempt_number": row.attempt_number, + "status": row.status, + }, + ), + ( + PostboxDeliveryAttempt, + "campaign_postbox_attempt", + lambda row: { + "job_id": row.job_id, + "attempt_number": row.attempt_number, + "status": row.status, + "provider_delivery_id": row.provider_delivery_id, + "provider_message_id": row.provider_message_id, + "postbox_id": row.postbox_id, + "address": row.address, + "vacant": row.vacant, + "duplicate": row.duplicate, + "started_at": _iso(row.started_at), + "finished_at": _iso(row.finished_at), + }, + ), + ( + PrintOutputAttempt, + "campaign_print_attempt", + lambda row: { + "job_id": row.job_id, + "attempt_number": row.attempt_number, + "status": row.status, + "render_id": row.render_id, + "artifact_sha256": row.artifact_sha256, + "started_at": _iso(row.started_at), + "finished_at": _iso(row.finished_at), + }, + ), + ) + for model, resource_type, encoder in evidence_specs: + for row in _bounded_rows( + db.query(model).filter(model.job_id.in_(job_ids)).order_by(model.id) + ): + append( # type: ignore[operator] + _record( + resource_type, + row.id, + "campaign_delivery_evidence", + resource_type.replace("_", " "), + encoder(row), + observed_at=row.updated_at, + immutable=True, + retention_reason="Provider-attempt outcomes are immutable delivery evidence.", + ) + ) + + actions = _bounded_rows( + db.query(CampaignMessageAction) + .filter( + CampaignMessageAction.tenant_id == tenant_id, + CampaignMessageAction.job_id.in_(job_ids), + ) + .order_by(CampaignMessageAction.id) + ) + action_ids = {row.id for row in actions} + for row in actions: + append( # type: ignore[operator] + _record( + "campaign_message_action", + row.id, + "campaign_delivery_evidence", + f"Message action {row.kind}", + { + "job_id": row.job_id, + "kind": row.kind, + "status": row.status, + "message_sha256": row.message_sha256, + "recipient_count": row.recipient_count, + "accepted_count": row.accepted_count, + "refused_count": row.refused_count, + "completed_at": _iso(row.completed_at), + }, + observed_at=row.updated_at, + immutable=True, + retention_reason="Single-message actions are immutable delivery and correction evidence.", + ) + ) + if action_ids: + for attempt in _bounded_rows( + db.query(CampaignMessageActionAttempt) + .filter(CampaignMessageActionAttempt.action_id.in_(action_ids)) + .order_by(CampaignMessageActionAttempt.id) + ): + append( # type: ignore[operator] + _record( + "campaign_message_action_attempt", + attempt.id, + "campaign_delivery_evidence", + "Message action attempt", + { + "action_id": attempt.action_id, + "attempt_number": attempt.attempt_number, + "status": attempt.status, + "outcome_code": attempt.outcome_code, + "started_at": _iso(attempt.started_at), + "completed_at": _iso(attempt.completed_at), + }, + observed_at=attempt.updated_at, + immutable=True, + retention_reason="Action attempts are immutable correction evidence.", + ) + ) + + def _append_report_projections( + self, + *, + append: object, + jobs: Sequence[CampaignJob], + ) -> None: + by_campaign: dict[str, list[CampaignJob]] = {} + for job in jobs: + by_campaign.setdefault(job.campaign_id, []).append(job) + for campaign_id, rows in sorted(by_campaign.items()): + append( # type: ignore[operator] + _record( + "campaign_report_projection", + campaign_id, + "campaign_report", + "Recipient-specific Campaign report projection", + { + "campaign_id": campaign_id, + "matched_job_count": len(rows), + "send_status_counts": dict( + sorted(Counter(row.send_status for row in rows).items()) + ), + "postbox_status_counts": dict( + sorted(Counter(row.postbox_status for row in rows).items()) + ), + "print_status_counts": dict( + sorted(Counter(row.print_status for row in rows).items()) + ), + }, + immutable=True, + retention_reason=( + "Campaign reports are derived projections over retained recipient and delivery evidence." + ), + source_path=f"/campaigns/{campaign_id}/report", + ) + ) + + def _append_member_resources( + self, + db: Session, + *, + append: object, + tenant_id: str, + subject_user_id: str | None, + campaign_ids: set[str], + version_ids: set[str], + ) -> None: + if subject_user_id is not None: + shares = _bounded_rows( + db.query(CampaignShare) + .filter( + CampaignShare.tenant_id == tenant_id, + or_( + (CampaignShare.target_type == "user") + & (CampaignShare.target_id == subject_user_id), + CampaignShare.created_by_user_id == subject_user_id, + ), + ) + .order_by(CampaignShare.id) + ) + for share in shares: + match_fields = _matching_fields( + share, subject_user_id, ("created_by_user_id",) + ) + if share.target_type == "user" and share.target_id == subject_user_id: + match_fields.insert(0, "target_id") + append( # type: ignore[operator] + _record( + "campaign_share", + share.id, + "campaign_access_evidence", + f"Campaign share {share.id}", + { + "match_fields": match_fields, + "campaign_id": share.campaign_id, + "permission": share.permission, + "revoked_at": _iso(share.revoked_at), + }, + observed_at=share.updated_at, + immutable=True, + retention_reason="Campaign sharing history is institutional access evidence.", + source_path=f"/campaigns/{share.campaign_id}", + ) + ) + + for profile in _bounded_rows( + db.query(RecipientImportMappingProfile) + .filter( + RecipientImportMappingProfile.tenant_id == tenant_id, + RecipientImportMappingProfile.owner_user_id == subject_user_id, + ) + .order_by(RecipientImportMappingProfile.id) + ): + append( # type: ignore[operator] + _record( + "recipient_import_mapping_profile", + profile.id, + "personal_campaign_preference", + profile.name, + { + "match_fields": ["owner_user_id"], + "name": profile.name, + "column_count": profile.column_count, + "headers": _safe_value(profile.headers), + "mappings": _safe_value(profile.mappings), + "delimiter": profile.delimiter, + "header_rows": profile.header_rows, + }, + observed_at=profile.updated_at, + ) + ) + + schedule_conditions = [] + if subject_user_id is not None: + schedule_conditions.append( + CampaignSchedule.created_by_user_id == subject_user_id + ) + if version_ids: + schedule_conditions.append( + CampaignSchedule.source_version_id.in_(version_ids) + ) + if schedule_conditions: + for schedule in _bounded_rows( + db.query(CampaignSchedule) + .filter( + CampaignSchedule.tenant_id == tenant_id, + or_(*schedule_conditions), + ) + .order_by(CampaignSchedule.id) + ): + append( # type: ignore[operator] + _record( + "campaign_schedule", + schedule.id, + "campaign_schedule_evidence", + schedule.name, + { + "match_fields": _matching_fields( + schedule, subject_user_id, ("created_by_user_id",) + ) + + ( + ["source_version.recipient"] + if schedule.source_version_id in version_ids + else [] + ), + "campaign_id": schedule.campaign_id, + "source_version_id": schedule.source_version_id, + "delivery_mode": schedule.delivery_mode, + "recurrence_kind": schedule.recurrence_kind, + "active": schedule.active, + "starts_at": _iso(schedule.starts_at), + "next_fire_at": _iso(schedule.next_fire_at), + }, + observed_at=schedule.updated_at, + immutable=True, + retention_reason="Schedule configuration and occurrence intent are execution evidence.", + source_path=f"/campaigns/{schedule.campaign_id}", + ) + ) + + attachment_conditions = [] + if subject_user_id is not None: + attachment_conditions.append( + AttachmentInstance.owner_user_id == subject_user_id, + ) + if campaign_ids: + attachment_conditions.append( + AttachmentInstance.campaign_id.in_(campaign_ids) + ) + if not attachment_conditions: + return + attachment_query = db.query(AttachmentInstance).filter( + AttachmentInstance.tenant_id == tenant_id, + or_(*attachment_conditions), + ) + for attachment in _bounded_rows( + attachment_query.order_by(AttachmentInstance.id) + ): + append( # type: ignore[operator] + _record( + "campaign_attachment", + attachment.id, + "campaign_artifact", + attachment.filename, + { + "match_fields": _matching_fields( + attachment, subject_user_id, ("owner_user_id",) + ) + + ( + ["campaign.recipient"] + if attachment.campaign_id in campaign_ids + else [] + ), + "campaign_id": attachment.campaign_id, + "logical_name": attachment.logical_name, + "filename": attachment.filename, + "tags": _safe_value(attachment.tags), + }, + observed_at=attachment.updated_at, + immutable=attachment.campaign_id in campaign_ids, + retention_reason=( + "Attachments associated with recipient delivery follow Campaign artifact retention." + if attachment.campaign_id in campaign_ids + else None + ), + ) + ) + + +def _matching_jobs( + session: Session, + *, + tenant_id: str, + email: str | None, + job_id: str | None, + entry_id: str | None, +) -> list[CampaignJob]: + candidate_sets: list[set[str]] = [] + if email: + candidate_sets.append( + { + row[0] + for row in _bounded_rows( + session.query(CampaignJob.id) + .filter( + CampaignJob.tenant_id == tenant_id, + func.lower(CampaignJob.recipient_email) == email, + ) + .order_by(CampaignJob.id) + ) + } + ) + if job_id: + candidate_sets.append( + { + row[0] + for row in session.query(CampaignJob.id).filter( + CampaignJob.tenant_id == tenant_id, + CampaignJob.id == job_id, + ) + } + ) + if entry_id: + candidate_sets.append( + { + row[0] + for row in _bounded_rows( + session.query(CampaignJob.id) + .filter( + CampaignJob.tenant_id == tenant_id, + CampaignJob.entry_id == entry_id, + ) + .order_by(CampaignJob.id) + ) + } + ) + if not candidate_sets: + return [] + ids = set.intersection(*candidate_sets) + if not ids: + return [] + return _bounded_rows( + session.query(CampaignJob) + .filter(CampaignJob.tenant_id == tenant_id, CampaignJob.id.in_(ids)) + .order_by(CampaignJob.id) + ) + + +def _matching_versions( + session: Session, + *, + tenant_id: str, + subject_user_id: str | None, + email: str | None, + job_version_ids: set[str], + version_id: str | None, +) -> list[CampaignVersion]: + conditions = [] + if job_version_ids: + conditions.append(CampaignVersion.id.in_(job_version_ids)) + if version_id: + conditions.append(CampaignVersion.id == version_id) + if subject_user_id: + conditions.extend( + ( + CampaignVersion.locked_by_user_id == subject_user_id, + CampaignVersion.user_locked_by_user_id == subject_user_id, + CampaignVersion.archived_by_user_id == subject_user_id, + ) + ) + if email: + pattern = f"%{_escape_like(email)}%" + conditions.append( + func.lower(cast(CampaignVersion.raw_json, Text)).like(pattern, escape="\\") + ) + if not conditions: + return [] + candidates = _bounded_rows( + session.query(CampaignVersion) + .join(Campaign, Campaign.id == CampaignVersion.campaign_id) + .filter(Campaign.tenant_id == tenant_id, or_(*conditions)) + .order_by(CampaignVersion.id) + ) + return [ + row + for row in candidates + if row.id in job_version_ids + or row.id == version_id + or ( + subject_user_id is not None + and any( + getattr(row, field) == subject_user_id + for field in ( + "locked_by_user_id", + "user_locked_by_user_id", + "archived_by_user_id", + ) + ) + ) + or bool(_matching_entry_fragments(row.raw_json, email)) + ] + + +def _matching_campaigns( + session: Session, + *, + tenant_id: str, + subject_user_id: str | None, + campaign_ids: set[str], +) -> list[Campaign]: + conditions = [] + if campaign_ids: + conditions.append(Campaign.id.in_(campaign_ids)) + if subject_user_id: + conditions.extend( + ( + Campaign.created_by_user_id == subject_user_id, + Campaign.owner_user_id == subject_user_id, + ) + ) + if not conditions: + return [] + return _bounded_rows( + session.query(Campaign) + .filter(Campaign.tenant_id == tenant_id, or_(*conditions)) + .order_by(Campaign.id) + ) + + +def _matching_entry_fragments(value: object, email: str | None) -> list[object]: + if not email or not isinstance(value, Mapping): + return [] + entries_value = value.get("entries") + if isinstance(entries_value, Mapping): + entries_value = entries_value.get("inline") + if not isinstance(entries_value, list): + return [] + fragments: list[object] = [] + for index, entry in enumerate(entries_value): + paths = _matching_email_paths(entry, email=email) + if not paths: + continue + fragments.append( + { + "entry_index": index, + "entry_id": entry.get("id") if isinstance(entry, Mapping) else None, + "matched_fields": paths, + "data": _safe_value(entry, subject_email=email), + } + ) + return fragments + + +def _matching_recipient_fragment(value: object, email: str | None) -> object: + if not email or not isinstance(value, Mapping): + return {} + return _safe_value(value, subject_email=email) + + +def _matching_email_paths( + value: object, + *, + email: str, + path: str = "entry", + depth: int = 0, +) -> list[str]: + if depth >= 8: + return [] + matches: list[str] = [] + if isinstance(value, Mapping): + for key, item in list(value.items())[:128]: + child = f"{path}.{key}" + if str(key).casefold() == "email" and _normalized_email(item) == email: + matches.append(child) + else: + matches.extend( + _matching_email_paths( + item, + email=email, + path=child, + depth=depth + 1, + ) + ) + elif isinstance(value, list): + for index, item in enumerate(value[:256]): + matches.extend( + _matching_email_paths( + item, + email=email, + path=f"{path}[{index}]", + depth=depth + 1, + ) + ) + return matches[:256] + + +def _safe_value( + value: object, + *, + subject_email: str | None = None, + depth: int = 0, +) -> object: + if depth >= 6: + return "[depth-limited]" + if isinstance(value, Mapping): + item_email = _normalized_email(value.get("email")) + if subject_email and item_email and item_email != subject_email: + return None + result: dict[str, object] = {} + for key, item in list(value.items())[:128]: + normalized_key = str(key).casefold() + if ( + any(part in normalized_key for part in _SECRET_PARTS) + or normalized_key in _CONTENT_KEYS + or normalized_key.endswith("_path") + or normalized_key.endswith("_storage_key") + ): + continue + sanitized = _safe_value( + item, + subject_email=subject_email, + depth=depth + 1, + ) + if sanitized is not None: + result[str(key)] = sanitized + return result + if isinstance(value, list): + result = [] + for item in value[:256]: + sanitized = _safe_value( + item, + subject_email=subject_email, + depth=depth + 1, + ) + if sanitized is not None: + result.append(sanitized) + return result + if isinstance(value, str): + if subject_email and "@" in value and _normalized_email(value) != subject_email: + return None + return value[:2_000] + if value is None or isinstance(value, (bool, int, float)): + return value + return str(value)[:2_000] + + +def _version_is_evidence(version: CampaignVersion, matched_jobs: int) -> bool: + return bool( + matched_jobs + or version.workflow_state != "editing" + or version.published_at + or version.locked_at + or version.archived_at + or version.execution_snapshot_hash + ) + + +def _job_is_evidence(job: CampaignJob) -> bool: + return bool( + job.build_status != "pending" + or job.queue_status != "draft" + or job.send_status != "not_queued" + or job.postbox_status != "not_requested" + or job.print_status != "not_requested" + or job.imap_status != "not_requested" + or job.eml_sha256 + or job.eml_storage_key + or job.eml_local_path + ) + + +def _delete_mapping_profile( + session: Session, + *, + tenant_id: str, + subject_user_id: str, + action: DsarErasureActionRef, + request_id: str, +) -> DsarExecutionResultRef: + row = ( + session.query(RecipientImportMappingProfile) + .filter( + RecipientImportMappingProfile.id == action.resource_id, + RecipientImportMappingProfile.tenant_id == tenant_id, + ) + .with_for_update() + .one_or_none() + ) + if row is None: + return _result( + action, + "unchanged", + "The recipient import mapping was already absent.", + {"request_id": request_id}, + ) + if row.owner_user_id != subject_user_id: + return _blocked(action, "The mapping owner changed after planning.") + session.delete(row) + return _result( + action, + "executed", + "The personal recipient import mapping was deleted.", + {"request_id": request_id}, + ) + + +def _revoke_share( + session: Session, + *, + tenant_id: str, + subject_user_id: str, + action: DsarErasureActionRef, + request_id: str, +) -> DsarExecutionResultRef: + row = ( + session.query(CampaignShare) + .filter(CampaignShare.id == action.resource_id) + .with_for_update() + .one_or_none() + ) + if ( + row is None + or row.tenant_id != tenant_id + or row.target_type != "user" + or row.target_id != subject_user_id + ): + return _blocked( + action, "The subject-targeted Campaign share is no longer available." + ) + if row.revoked_at is not None: + return _result( + action, + "unchanged", + "The Campaign share was already revoked.", + {"request_id": request_id, "revoked_at": _iso(row.revoked_at)}, + ) + row.revoked_at = datetime.now(timezone.utc) + return _result( + action, + "executed", + "The subject-targeted Campaign share was revoked.", + {"request_id": request_id, "revoked_at": _iso(row.revoked_at)}, + ) + + +def _campaign_references(subject: DsarSubjectRef) -> dict[str, str]: + aliases = { + "campaign.campaign": "campaign", + "campaign.version": "version", + "campaign.job": "job", + "campaign.entry": "entry", + } + result: dict[str, str] = {} + for key, target in aliases.items(): + value = str(subject.external_references.get(key) or "").strip() + if value: + result[target] = value + return result + + +def _subject_user_id(subject: DsarSubjectRef) -> str | None: + candidates: list[str] = [] + if subject.membership_id: + candidates.append(subject.membership_id) + for key in ( + "campaign.user", + "campaign.membership", + "access.membership", + "membership_id", + ): + value = str(subject.external_references.get(key) or "").strip() + if value: + candidates.append(value) + normalized = {value.strip() for value in candidates if value.strip()} + return normalized.pop() if len(normalized) == 1 else None + + +def _subject_email(subject: DsarSubjectRef) -> str | None: + candidates = [] + if subject.email: + candidates.append(subject.email) + for key in ("campaign.email", "campaign.recipient_email"): + value = str(subject.external_references.get(key) or "").strip() + if value: + candidates.append(value) + normalized = { + email for value in candidates if (email := _normalized_email(value)) is not None + } + return normalized.pop() if len(normalized) == 1 else None + + +def _normalized_email(value: object) -> str | None: + if not isinstance(value, str): + return None + normalized = value.strip().casefold() + return normalized or None + + +def _matching_fields( + row: object, + subject_user_id: str | None, + fields: Sequence[str], +) -> list[str]: + if subject_user_id is None: + return [] + return [field for field in fields if getattr(row, field) == subject_user_id] + + +def _record( + resource_type: str, + resource_id: str, + category: str, + title: str, + data: Mapping[str, object], + *, + observed_at: datetime | None = None, + immutable: bool = False, + retention_reason: str | None = None, + source_path: str | None = None, +) -> DsarRecordRef: + return DsarRecordRef( + provider_id="campaigns", + module_id="campaigns", + resource_type=resource_type, + resource_id=resource_id, + category=category, + title=title, + data=data, + observed_at=observed_at, + immutable_evidence=immutable, + retention_reason=retention_reason, + source_path=source_path, + ) + + +def _action( + action_id: str, + kind: str, + record: DsarRecordRef, + title: str, + rationale: str, + *, + executable: bool, + irreversible: bool = False, + metadata: Mapping[str, object] | None = None, +) -> DsarErasureActionRef: + return DsarErasureActionRef( + action_id=action_id, + provider_id="campaigns", + module_id="campaigns", + kind=kind, # type: ignore[arg-type] + resource_type=record.resource_type, + resource_id=record.resource_id, + title=title, + rationale=rationale, + executable=executable, + irreversible=irreversible, + metadata=metadata or {}, + ) + + +def _result( + action: DsarErasureActionRef, + status: str, + summary: str, + evidence: Mapping[str, object] | None = None, +) -> DsarExecutionResultRef: + return DsarExecutionResultRef( + action_id=action.action_id, + status=status, # type: ignore[arg-type] + summary=summary, + evidence=evidence or {}, + ) + + +def _blocked(action: DsarErasureActionRef, summary: str) -> DsarExecutionResultRef: + return _result(action, "blocked", summary) + + +def _session(value: object) -> Session: + if not isinstance(value, Session): + raise TypeError("Campaign DSAR provider requires a SQLAlchemy session.") + return value + + +def _bounded_rows(query: object) -> list[object]: + rows = query.limit(_MAX_RECORDS + 1).all() # type: ignore[attr-defined] + if len(rows) > _MAX_RECORDS: + raise ValueError( + "Campaign DSAR match limit exceeded; narrow the subject selectors." + ) + return rows + + +def _escape_like(value: str) -> str: + return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + + +def _iso(value: datetime | None) -> str | None: + if value is None: + return None + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + return value.isoformat() + + +__all__ = ["CAMPAIGN_DSAR_CAPABILITY", "CampaignDsarProvider"] diff --git a/src/govoplan_campaign/backend/manifest.py b/src/govoplan_campaign/backend/manifest.py index ec12a0f..707bdec 100644 --- a/src/govoplan_campaign/backend/manifest.py +++ b/src/govoplan_campaign/backend/manifest.py @@ -65,11 +65,19 @@ from govoplan_campaign.backend.documentation import ( CAMPAIGN_USER_DOCUMENTATION, documentation_topics, ) +from govoplan_campaign.backend.dsar_provider import CAMPAIGN_DSAR_CAPABILITY from govoplan_campaign.backend.search_source import create_campaign_search_source register_campaign_change_tracking() +def _dsar_provider(context: ModuleContext) -> object: + del context + from govoplan_campaign.backend.dsar_provider import CampaignDsarProvider + + return CampaignDsarProvider() + + def _permission( scope: str, label: str, description: str, category: str ) -> PermissionDefinition: @@ -409,6 +417,7 @@ manifest = ModuleManifest( name=REPORT_PROVIDER_CAPABILITY_PREFIX + "campaigns", version="1.0.0", ), + ModuleInterfaceProvider(name=CAMPAIGN_DSAR_CAPABILITY, version="0.1.0"), ), requires_interfaces=( ModuleInterfaceRequirement( @@ -659,6 +668,65 @@ manifest = ModuleManifest( ), documentation=( *CAMPAIGN_USER_DOCUMENTATION, + DocumentationTopic( + id="campaigns.privacy.data-subject-requests", + title="Review Campaign data in a data-subject request", + summary="Collect recipient, version, delivery, report, and artifact metadata without rewriting immutable evidence.", + body=( + "Campaign's DSAR provider searches the effective tenant by normalized recipient email, direct membership references, and namespaced Campaign job, entry, version, or Campaign references. " + "It isolates matching inline-recipient fields and job metadata, and reports built versions, delivery attempts, Postbox and print outcomes, message-action corrections, recipient-specific report projections, generated-message digests, and attachment metadata. It does not export EML bytes, object or local paths, provider target snapshots, worker claims, idempotency material, secrets, credentials, or unrelated recipient addresses. " + "Built, locked, published, terminal, delivered, or corrected records remain retained with a reason and continue through Campaign's configured retention/redaction process. Draft recipient content and user-owned attachment content require coordinated manual review because the same data may occur in version JSON, jobs, and generated artifacts. The provider can idempotently delete a personal recipient-import mapping profile and revoke an active Campaign share aimed at the subject. It never rewrites delivered evidence or deletes generated artifacts directly. Campaign reports are derived projections rather than a separate personal-data store." + ), + layer="configured", + documentation_types=("admin",), + audience=("privacy_officer", "campaign_manager", "records_manager", "operator"), + order=42, + conditions=( + DocumentationCondition( + required_modules=("campaigns", "access"), + any_scopes=( + "access:privacy:read", + "access:privacy:manage", + "access:privacy:erase", + ), + ), + ), + links=( + DocumentationLink( + label="Data-subject requests", + href="/admin?section=tenant-data-subject-requests", + kind="runtime", + ), + DocumentationLink( + label="Campaign handbook", + href="govoplan-campaign/docs/CAMPAIGN_HANDBOOK.md", + kind="repository", + ), + ), + related_modules=("access", "audit", "files", "mail", "postbox", "reporting"), + metadata={ + "kind": "workflow", + "route": "/admin?section=tenant-data-subject-requests", + "screen": "Data-subject requests", + "help_contexts": ["admin.privacy.data-subject-requests"], + "prerequisites": [ + "The privacy request and recipient selectors have been independently authorized and corroborated.", + "The reviewer understands the effective Campaign retention policy and delivery-evidence obligations.", + ], + "steps": [ + "Run the Campaign provider search and review recipient, version, job, attempt, report-projection, and artifact dispositions.", + "Inspect matching draft content manually and keep every evidence retention reason with the case decision.", + "Execute only an approved user-owned mapping deletion or subject-targeted share revocation.", + "Use Campaign retention and artifact reconciliation for approved content redaction or expiry; do not mutate delivered evidence ad hoc.", + ], + "limitations": [ + "Generated EML bytes and attachment content are not embedded in the JSON export; authorized Campaign or Files review paths remain authoritative.", + "Draft recipient erasure is manual until a coordinated version/job/artifact rewrite contract can prove that no partial copy remains.", + ], + "outcome": "Campaign personal data receives an explicit retained, review, revoke, or delete disposition without weakening delivery evidence.", + "verification": "Confirm matching recipients are isolated, no locator or credential material appears, report counts derive from the same matched jobs, and repeated reversible actions are unchanged.", + }, + ), DocumentationTopic( id="campaigns.access.child-evidence", title="Explain access to Campaign child evidence", @@ -1313,6 +1381,7 @@ manifest = ModuleManifest( "govoplan_campaign.backend.reports.provider", fromlist=["CampaignAggregateReportProvider"], ).CampaignAggregateReportProvider(), + CAMPAIGN_DSAR_CAPABILITY: _dsar_provider, }, capability_documentation={ REPORT_PROVIDER_CAPABILITY_PREFIX + "campaigns": CapabilityDocumentation( @@ -1325,6 +1394,13 @@ manifest = ModuleManifest( documentation_types=("admin", "user"), audience=("user", "reporting_analyst", "privacy_officer"), ), + CAMPAIGN_DSAR_CAPABILITY: CapabilityDocumentation( + label="Campaign data-subject request provider", + summary="Finds isolated recipient and Campaign evidence metadata and classifies governed erasure actions.", + contract_version="0.1.0", + documentation_types=("admin",), + audience=("privacy_officer", "campaign_manager", "records_manager"), + ), }, operational_check_providers=( OperationalCheckProviderRegistration( diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..bf6718e --- /dev/null +++ b/tests/test_dsar_provider.py @@ -0,0 +1,705 @@ +from __future__ import annotations + +import unittest +from datetime import datetime, timezone + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from govoplan_access.backend.db.models import Account, Group, User +from govoplan_campaign.backend.db.models import ( + AttachmentBlob, + AttachmentInstance, + Campaign, + CampaignIssue, + CampaignJob, + CampaignMessageAction, + CampaignMessageActionAttempt, + CampaignSchedule, + CampaignShare, + CampaignVersion, + ImapAppendAttempt, + PostboxDeliveryAttempt, + PrintOutputAttempt, + RecipientImportMappingProfile, + SendAttempt, +) +from govoplan_campaign.backend.dsar_provider import ( + CAMPAIGN_DSAR_CAPABILITY, + CampaignDsarProvider, +) +from govoplan_campaign.backend.manifest import manifest +from govoplan_core.core.change_sequence import ChangeSequenceEntry +from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef +from govoplan_core.db.base import Base +from govoplan_core.privacy.dsar_workflow import ( + DataSubjectRequest, + create_data_subject_request, + execute_data_subject_erasure, + plan_data_subject_erasure, + search_data_subject_request, +) + + +class _Registry: + def __init__( + self, + provider: CampaignDsarProvider, + *, + campaign_active: bool = True, + ) -> None: + self.provider = provider + self.campaign_active = campaign_active + + def capability_names(self): + return (CAMPAIGN_DSAR_CAPABILITY,) + + def capability_owner(self, name): + self._assert_capability(name) + return "campaigns" + + def tenant_entitlement_resolver(self): + campaign_active = self.campaign_active + + class _Resolver: + @staticmethod + def resolve(session, tenant_id): + del session, tenant_id + return type( + "State", + (), + {"effective_modules": ("campaigns",) if campaign_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": "campaigns"})(),) + + @staticmethod + def _assert_capability(name: str) -> None: + if name != CAMPAIGN_DSAR_CAPABILITY: + raise KeyError(name) + + +class CampaignDsarProviderTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite:///:memory:", future=True) + Base.metadata.create_all( + bind=self.engine, + tables=[ + Account.__table__, + User.__table__, + Group.__table__, + ChangeSequenceEntry.__table__, + DataSubjectRequest.__table__, + Campaign.__table__, + CampaignShare.__table__, + CampaignVersion.__table__, + CampaignJob.__table__, + CampaignIssue.__table__, + AttachmentBlob.__table__, + AttachmentInstance.__table__, + SendAttempt.__table__, + CampaignMessageAction.__table__, + CampaignMessageActionAttempt.__table__, + ImapAppendAttempt.__table__, + PostboxDeliveryAttempt.__table__, + PrintOutputAttempt.__table__, + RecipientImportMappingProfile.__table__, + CampaignSchedule.__table__, + ], + ) + self.session = sessionmaker(bind=self.engine, future=True)() + now = datetime.now(timezone.utc) + self.account = Account( + id="account-1", + email="subject@example.test", + normalized_email="subject@example.test", + display_name="Subject", + ) + other_account = Account( + id="account-2", + email="other@example.test", + normalized_email="other@example.test", + display_name="Other", + ) + self.user = User( + id="membership-1", + tenant_id="tenant-1", + account_id=self.account.id, + email="subject@example.test", + display_name="Subject", + ) + self.other_user = User( + id="membership-2", + tenant_id="tenant-1", + account_id=other_account.id, + email="other@example.test", + display_name="Other", + ) + self.campaign = Campaign( + id="campaign-1", + tenant_id="tenant-1", + created_by_user_id=self.other_user.id, + owner_user_id=self.other_user.id, + external_id="privacy-notice", + name="Privacy notice", + status="active", + ) + self.version = CampaignVersion( + id="version-1", + campaign_id=self.campaign.id, + version_number=1, + workflow_state="built", + execution_snapshot_hash="a" * 64, + raw_json={ + "entries": { + "inline": [ + { + "id": "entry-subject", + "to": [ + { + "email": "subject@example.test", + "name": "Subject Person", + } + ], + "cc": [ + { + "email": "other@example.test", + "name": "Unrelated person", + } + ], + "body": "private-rendered-body-do-not-export", + "password": "inline-secret-do-not-export", + "case_reference": "CASE-SUBJECT-1", + }, + { + "id": "entry-other", + "to": [{"email": "other@example.test"}], + "private_value": "other-recipient-data-do-not-export", + }, + ] + } + }, + ) + self.draft_version = CampaignVersion( + id="version-draft", + campaign_id=self.campaign.id, + version_number=2, + workflow_state="editing", + raw_json={ + "entries": { + "inline": [ + { + "id": "entry-draft-subject", + "to": [{"email": "subject@example.test"}], + "case_reference": "CASE-DRAFT-1", + } + ] + } + }, + ) + self.job = CampaignJob( + id="job-subject", + tenant_id="tenant-1", + campaign_id=self.campaign.id, + campaign_version_id=self.version.id, + entry_index=0, + entry_id="entry-subject", + recipient_email="Subject@Example.Test", + subject="Your governed notice", + eml_storage_key="private/eml/key-do-not-export", + eml_local_path="/private/message-do-not-export.eml", + eml_size_bytes=512, + eml_sha256="b" * 64, + build_status="built", + validation_status="ready", + queue_status="completed", + send_status="smtp_accepted", + postbox_status="accepted", + print_status="accepted", + imap_status="appended", + attempt_count=1, + queued_at=now, + sent_at=now, + claim_token="job-claim-do-not-export", + resolved_recipients={ + "from": {"email": "sender@example.test"}, + "to": [{"email": "subject@example.test", "name": "Subject"}], + "cc": [{"email": "other@example.test", "name": "Other"}], + "legacy": ["other@example.test", "subject@example.test"], + }, + resolved_attachments=[ + {"storage_key": "resolved-attachment-key-do-not-export"} + ], + ) + other_job = CampaignJob( + id="job-other", + tenant_id="tenant-1", + campaign_id=self.campaign.id, + campaign_version_id=self.version.id, + entry_index=1, + entry_id="entry-other", + recipient_email="other@example.test", + subject="Other person's message", + build_status="built", + validation_status="ready", + ) + tenant_two_campaign = Campaign( + id="campaign-tenant-2", + tenant_id="tenant-2", + external_id="other-tenant", + name="Other tenant data do not export", + ) + tenant_two_version = CampaignVersion( + id="version-tenant-2", + campaign_id=tenant_two_campaign.id, + version_number=1, + raw_json={ + "entries": {"inline": [{"to": [{"email": "subject@example.test"}]}]} + }, + ) + tenant_two_job = CampaignJob( + id="job-tenant-2", + tenant_id="tenant-2", + campaign_id=tenant_two_campaign.id, + campaign_version_id=tenant_two_version.id, + entry_index=0, + recipient_email="subject@example.test", + ) + self.issue = CampaignIssue( + id="issue-1", + tenant_id="tenant-1", + campaign_id=self.campaign.id, + campaign_version_id=self.version.id, + job_id=self.job.id, + severity="warning", + code="delivery_warning", + message="issue-detail-do-not-export", + source="private-source-do-not-export", + behavior="review", + ) + send_attempt = SendAttempt( + id="send-attempt-1", + job_id=self.job.id, + attempt_number=1, + status="accepted", + claim_token="attempt-claim-do-not-export", + smtp_status_code=250, + smtp_response="smtp-response-do-not-export", + error_message="transport-detail-do-not-export", + started_at=now, + finished_at=now, + ) + postbox_attempt = PostboxDeliveryAttempt( + id="postbox-attempt-1", + tenant_id="tenant-1", + job_id=self.job.id, + target_key="target-key-do-not-export", + target_index=0, + attempt_number=1, + idempotency_key="postbox-idempotency-do-not-export", + status="accepted", + target_snapshot={"private": "snapshot-do-not-export"}, + provider_delivery_id="delivery-1", + provider_message_id="message-1", + postbox_id="postbox-1", + address="subject@example.test", + evidence={"private": "postbox-evidence-do-not-export"}, + started_at=now, + finished_at=now, + ) + print_attempt = PrintOutputAttempt( + id="print-attempt-1", + tenant_id="tenant-1", + job_id=self.job.id, + attempt_number=1, + idempotency_key="print-idempotency-do-not-export", + status="accepted", + render_id="render-1", + artifact_sha256="c" * 64, + evidence={"private": "print-evidence-do-not-export"}, + started_at=now, + finished_at=now, + ) + self.share = CampaignShare( + id="share-1", + tenant_id="tenant-1", + campaign_id=self.campaign.id, + target_type="user", + target_id=self.user.id, + permission="read", + created_by_user_id=self.other_user.id, + ) + self.profile = RecipientImportMappingProfile( + id="mapping-1", + tenant_id="tenant-1", + owner_user_id=self.user.id, + name="Subject mapping", + column_count=2, + headers=["email", "case_reference"], + normalized_headers=["email", "case_reference"], + ordered_header_fingerprint="d" * 64, + unordered_header_fingerprint="e" * 64, + delimiter=";", + header_rows=1, + quoted=True, + value_separators=",;|", + mappings=[ + {"header": "email", "field": "to.0.email"}, + {"secret": "profile-secret-do-not-export"}, + ], + ) + blob = AttachmentBlob( + id="blob-1", + tenant_id="tenant-1", + sha256="f" * 64, + size_bytes=42, + mime_type="application/pdf", + storage_bucket="private-bucket-do-not-export", + storage_key="private-attachment-key-do-not-export", + ) + attachment = AttachmentInstance( + id="attachment-1", + tenant_id="tenant-1", + owner_user_id=self.other_user.id, + campaign_id=self.campaign.id, + blob_id=blob.id, + logical_name="notice", + filename="notice.pdf", + tags=["notice"], + metadata_={"secret": "attachment-secret-do-not-export"}, + ) + schedule = CampaignSchedule( + id="schedule-1", + tenant_id="tenant-1", + campaign_id=self.campaign.id, + source_version_id=self.version.id, + created_by_user_id=self.other_user.id, + name="Recurring privacy notice", + delivery_mode="manual", + recurrence_kind="monthly", + starts_at=now, + next_fire_at=now, + max_occurrences=12, + source_snapshot={"private": "schedule-snapshot-do-not-export"}, + source_snapshot_hash="1" * 64, + ) + self.session.add_all( + [ + self.account, + other_account, + self.user, + self.other_user, + self.campaign, + self.version, + self.draft_version, + self.job, + other_job, + tenant_two_campaign, + tenant_two_version, + tenant_two_job, + self.issue, + send_attempt, + postbox_attempt, + print_attempt, + self.share, + self.profile, + blob, + attachment, + schedule, + ] + ) + self.session.commit() + self.provider = CampaignDsarProvider() + self.subject = DsarSubjectRef( + membership_id=self.user.id, + email="subject@example.test", + ) + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def test_manifest_publishes_protocol_conforming_provider(self) -> None: + provided_names = {item.name for item in manifest.provides_interfaces} + self.assertIn(CAMPAIGN_DSAR_CAPABILITY, provided_names) + provider = manifest.capability_factories[CAMPAIGN_DSAR_CAPABILITY](None) + self.assertIsInstance(provider, DsarProvider) + self.assertIn( + "campaigns.privacy.data-subject-requests", + {topic.id for topic in manifest.documentation}, + ) + + def test_search_is_tenant_scoped_minimized_and_recipient_specific(self) -> None: + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=self.subject, + ) + resource_types = {record.resource_type for record in records} + self.assertTrue( + { + "campaign", + "campaign_version", + "campaign_recipient_job", + "campaign_message_artifact", + "campaign_delivery_issue", + "campaign_send_attempt", + "campaign_postbox_attempt", + "campaign_print_attempt", + "campaign_report_projection", + "campaign_share", + "recipient_import_mapping_profile", + "campaign_schedule", + "campaign_attachment", + }.issubset(resource_types) + ) + report = next( + record + for record in records + if record.resource_type == "campaign_report_projection" + ) + self.assertEqual(1, report.data["matched_job_count"]) + artifact = next( + record + for record in records + if record.resource_type == "campaign_message_artifact" + ) + self.assertEqual("b" * 64, artifact.data["sha256"]) + self.assertEqual(512, artifact.data["size_bytes"]) + + serialized = repr([record.to_dict() for record in records]) + for hidden in ( + "job-tenant-2", + "Other tenant data do not export", + "job-other", + "other@example.test", + "Unrelated person", + "other-recipient-data-do-not-export", + "private-rendered-body-do-not-export", + "inline-secret-do-not-export", + "private/eml/key-do-not-export", + "/private/message-do-not-export.eml", + "resolved-attachment-key-do-not-export", + "job-claim-do-not-export", + "issue-detail-do-not-export", + "private-source-do-not-export", + "attempt-claim-do-not-export", + "smtp-response-do-not-export", + "transport-detail-do-not-export", + "target-key-do-not-export", + "postbox-idempotency-do-not-export", + "snapshot-do-not-export", + "postbox-evidence-do-not-export", + "print-idempotency-do-not-export", + "print-evidence-do-not-export", + "profile-secret-do-not-export", + "private-bucket-do-not-export", + "private-attachment-key-do-not-export", + "attachment-secret-do-not-export", + "schedule-snapshot-do-not-export", + ): + self.assertNotIn(hidden, serialized) + + def test_conflicting_email_references_fail_closed_for_recipient_data(self) -> None: + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + email="subject@example.test", + external_references={"campaign.email": "other@example.test"}, + ), + ) + + self.assertEqual((), records) + + def test_plan_retains_evidence_and_limits_execution_to_reversible_data( + self, + ) -> None: + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=self.subject, + ) + actions = self.provider.plan_erasure( + self.session, + tenant_id="tenant-1", + subject=self.subject, + records=records, + ) + + kinds = {action.kind for action in actions} + self.assertTrue({"retain", "manual_review", "revoke", "delete"}.issubset(kinds)) + self.assertTrue( + any( + action.action_id + == "campaigns:retain:campaign_recipient_job:job-subject" + for action in actions + ) + ) + self.assertTrue( + any( + action.action_id == "campaigns:review:campaign_version:version-draft" + for action in actions + ) + ) + executable_ids = {action.action_id for action in actions if action.executable} + self.assertEqual( + { + "campaigns:revoke:campaign_share:share-1", + "campaigns:delete:recipient_import_mapping_profile:mapping-1", + }, + executable_ids, + ) + + def test_execution_is_revalidated_tenant_bound_and_idempotent(self) -> None: + actions = self._executable_actions() + + wrong_tenant = self.provider.execute_erasure( + self.session, + tenant_id="tenant-2", + subject=self.subject, + actions=actions, + request_id="dsar-wrong-tenant", + ) + self.assertEqual({"blocked"}, {result.status for result in wrong_tenant}) + + first = self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=self.subject, + actions=actions, + request_id="dsar-1", + ) + self.assertEqual({"executed"}, {result.status for result in first}) + self.session.flush() + self.assertIsNotNone(self.share.revoked_at) + self.assertIsNone( + self.session.get(RecipientImportMappingProfile, self.profile.id) + ) + + repeated = self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=self.subject, + actions=actions, + request_id="dsar-1", + ) + self.assertEqual({"unchanged"}, {result.status for result in repeated}) + + def test_execution_blocks_when_mapping_owner_changed_after_planning(self) -> None: + delete_action = next( + action + for action in self._executable_actions() + if action.resource_type == "recipient_import_mapping_profile" + ) + self.profile.owner_user_id = self.other_user.id + self.session.flush() + + result = self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=self.subject, + actions=(delete_action,), + request_id="dsar-stale", + ) + + self.assertEqual("blocked", result[0].status) + self.assertIsNotNone( + self.session.get(RecipientImportMappingProfile, self.profile.id) + ) + + def test_core_workflow_discovers_active_provider_and_skips_it_when_disabled( + self, + ) -> None: + request = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-CAMPAIGN-1", + request_kind="access_and_erasure", + subject=self.subject, + purpose="Respond to an authorized privacy request.", + legal_basis="Article 15 and 17 GDPR", + due_at=None, + requested_by_account_id="privacy-officer", + ) + self.session.commit() + registry = _Registry(self.provider) + + search_data_subject_request( + self.session, + registry=registry, + row=request, + expected_revision=1, + ) + self.assertEqual("searched", request.status) + self.assertEqual(["campaigns"], request.coverage["covered_modules"]) + self.assertEqual([], request.coverage["modules_without_provider"]) + plan_data_subject_erasure( + self.session, + registry=registry, + row=request, + expected_revision=2, + ) + executable_ids = [ + action["action_id"] + for action in request.erasure_plan["actions"] + if action["executable"] + ] + execute_data_subject_erasure( + self.session, + registry=registry, + row=request, + expected_revision=3, + action_ids=executable_ids, + ) + self.assertEqual("completed", request.status) + + disabled = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-CAMPAIGN-DISABLED", + request_kind="access", + subject=self.subject, + purpose="Verify disabled-module coverage.", + legal_basis="Article 15 GDPR", + due_at=None, + requested_by_account_id="privacy-officer", + ) + search_data_subject_request( + self.session, + registry=_Registry(self.provider, campaign_active=False), + row=disabled, + expected_revision=1, + ) + + self.assertEqual(0, disabled.search_result["record_count"]) + self.assertEqual( + [CAMPAIGN_DSAR_CAPABILITY], + disabled.coverage["inactive_provider_capabilities"], + ) + + def _executable_actions(self): + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=self.subject, + ) + actions = self.provider.plan_erasure( + self.session, + tenant_id="tenant-1", + subject=self.subject, + records=records, + ) + return tuple(action for action in actions if action.executable) + + +if __name__ == "__main__": + unittest.main()