from __future__ import annotations import copy import hashlib import json from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Literal from pydantic import BaseModel, ConfigDict, Field, field_validator from sqlalchemy.orm import Session from govoplan_core.admin.governance import get_system_settings from govoplan_core.core.optional import reraise_unless_missing_package from govoplan_core.db.models import AuditLog, Group, SystemSettings, Tenant, User from govoplan_core.settings import settings PRIVACY_POLICY_SETTINGS_KEY = "privacy_retention_policy" RETENTION_DAY_KEYS = ( "raw_campaign_json_retention_days", "generated_eml_retention_days", "stored_report_detail_retention_days", "mock_mailbox_retention_days", "audit_detail_retention_days", ) RETENTION_POLICY_FIELD_KEYS = ( "store_raw_campaign_json", *RETENTION_DAY_KEYS, "audit_detail_level", ) AUDIT_DETAIL_LEVEL_ORDER = {"full": 0, "redacted": 1, "minimal": 2} def default_allow_lower_level_limits() -> dict[str, bool]: return {key: True for key in RETENTION_POLICY_FIELD_KEYS} def _normalize_allow_lower_level_limits(value: Any, *, fill_defaults: bool) -> dict[str, bool] | None: if value in (None, ""): return default_allow_lower_level_limits() if fill_defaults else None if not isinstance(value, dict): raise ValueError("allow_lower_level_limits must be an object") normalized = default_allow_lower_level_limits() if fill_defaults else {} for key, allowed in value.items(): clean_key = str(key) if clean_key not in RETENTION_POLICY_FIELD_KEYS: raise ValueError(f"Unknown retention policy field: {clean_key}") normalized[clean_key] = bool(allowed) return normalized FINAL_VERSION_STATES = { "completed", "partially_completed", "outcome_unknown", "failed", "cancelled", "archived", } FINAL_EML_SEND_STATUSES = { "smtp_accepted", "sent", "failed_permanent", "cancelled", "outcome_unknown", } AUDIT_DETAIL_REDACTION_KEYS = { "attachments", "bcc", "body", "campaign_json", "cc", "content", "entries", "html", "imap", "message", "password", "raw_eml", "raw_json", "recipients", "secret", "smtp", "template", "text", "token", "to", } class PrivacyRetentionPolicy(BaseModel): model_config = ConfigDict(extra="forbid") store_raw_campaign_json: bool = True raw_campaign_json_retention_days: int | None = Field(default=None, ge=0) generated_eml_retention_days: int | None = Field(default=None, ge=0) stored_report_detail_retention_days: int | None = Field(default=None, ge=0) mock_mailbox_retention_days: int | None = Field(default=None, ge=0) audit_detail_retention_days: int | None = Field(default=None, ge=0) audit_detail_level: Literal["full", "redacted", "minimal"] = "full" allow_lower_level_limits: dict[str, bool] = Field(default_factory=default_allow_lower_level_limits) @field_validator( "raw_campaign_json_retention_days", "generated_eml_retention_days", "stored_report_detail_retention_days", "mock_mailbox_retention_days", "audit_detail_retention_days", mode="before", ) @classmethod def _blank_string_is_none(cls, value: Any) -> Any: if value == "": return None return value @field_validator("allow_lower_level_limits", mode="before") @classmethod def _normalize_allow_lower_level_limits(cls, value: Any) -> Any: return _normalize_allow_lower_level_limits(value, fill_defaults=True) class PrivacyRetentionPolicyPatch(BaseModel): model_config = ConfigDict(extra="forbid") store_raw_campaign_json: bool | None = None raw_campaign_json_retention_days: int | None = Field(default=None, ge=0) generated_eml_retention_days: int | None = Field(default=None, ge=0) stored_report_detail_retention_days: int | None = Field(default=None, ge=0) mock_mailbox_retention_days: int | None = Field(default=None, ge=0) audit_detail_retention_days: int | None = Field(default=None, ge=0) audit_detail_level: Literal["full", "redacted", "minimal"] | None = None allow_lower_level_limits: dict[str, bool] | None = None @field_validator(*RETENTION_DAY_KEYS, mode="before") @classmethod def _blank_string_is_none(cls, value: Any) -> Any: if value == "": return None return value @field_validator("allow_lower_level_limits", mode="before") @classmethod def _normalize_allow_lower_level_limits(cls, value: Any) -> Any: return _normalize_allow_lower_level_limits(value, fill_defaults=False) class PrivacyPolicyError(RuntimeError): pass def _campaign_models(): try: from govoplan_campaign.backend.db.models import Campaign, CampaignJob, CampaignVersion, JobImapStatus, JobQueueStatus except ModuleNotFoundError as exc: reraise_unless_missing_package(exc, "govoplan_campaign") raise PrivacyPolicyError("Campaign module is not installed") from exc return Campaign, CampaignJob, CampaignVersion, JobImapStatus, JobQueueStatus def _campaign_models_or_none(): try: return _campaign_models() except PrivacyPolicyError: return None def _utcnow() -> datetime: return datetime.now(timezone.utc) def _cutoff(days: int | None, *, now: datetime) -> datetime | None: if days is None: return None return now - timedelta(days=days) def _json_sha256(value: Any) -> str: payload = json.dumps(value, sort_keys=True, ensure_ascii=False, default=str).encode("utf-8") return hashlib.sha256(payload).hexdigest() def _privacy_policy_data(settings_payload: dict[str, Any] | None) -> dict[str, Any]: if not isinstance(settings_payload, dict): return {} data = settings_payload.get(PRIVACY_POLICY_SETTINGS_KEY, {}) return data if isinstance(data, dict) else {} def _privacy_policy_patch_from_settings(settings_payload: dict[str, Any] | None) -> dict[str, Any]: data = _privacy_policy_data(settings_payload) if not data: return {} return PrivacyRetentionPolicyPatch.model_validate(data).model_dump(mode="json", exclude_none=True) def _merge_privacy_policy(parent: PrivacyRetentionPolicy, patch: dict[str, Any]) -> PrivacyRetentionPolicy: payload = parent.model_dump(mode="json") parent_allow = {**default_allow_lower_level_limits(), **(payload.get("allow_lower_level_limits") or {})} if parent_allow["store_raw_campaign_json"] and patch.get("store_raw_campaign_json") is False: payload["store_raw_campaign_json"] = False for key in RETENTION_DAY_KEYS: value = patch.get(key) if value is None or not parent_allow[key]: continue current = payload.get(key) payload[key] = int(value) if current is None else min(int(current), int(value)) detail_level = patch.get("audit_detail_level") if parent_allow["audit_detail_level"] and detail_level and AUDIT_DETAIL_LEVEL_ORDER[detail_level] > AUDIT_DETAIL_LEVEL_ORDER[payload["audit_detail_level"]]: payload["audit_detail_level"] = detail_level patch_allow = patch.get("allow_lower_level_limits") or {} payload["allow_lower_level_limits"] = { key: parent_allow[key] and bool(patch_allow.get(key, parent_allow[key])) for key in RETENTION_POLICY_FIELD_KEYS } return PrivacyRetentionPolicy.model_validate(payload) def _validate_privacy_patch_against_parent(parent: PrivacyRetentionPolicy, patch: dict[str, Any]) -> None: parent_payload = parent.model_dump(mode="json") parent_allow = {**default_allow_lower_level_limits(), **(parent_payload.get("allow_lower_level_limits") or {})} for key in RETENTION_POLICY_FIELD_KEYS: if key in patch and patch.get(key) is not None and not parent_allow[key]: raise PrivacyPolicyError(f"{key} is locked by the parent retention policy.") patch_allow = patch.get("allow_lower_level_limits") or {} for key, allowed in patch_allow.items(): if allowed and not parent_allow[key]: raise PrivacyPolicyError(f"{key} limiting cannot be re-enabled below a parent retention policy lock.") if patch.get("store_raw_campaign_json") is True and not parent.store_raw_campaign_json: raise PrivacyPolicyError("Raw campaign JSON storage cannot be re-enabled below a parent policy that disables it.") for key in RETENTION_DAY_KEYS: value = patch.get(key) parent_value = parent_payload.get(key) if value is not None and parent_value is not None and int(value) > int(parent_value): raise PrivacyPolicyError(f"{key} cannot be less restrictive than the parent retention policy.") detail_level = patch.get("audit_detail_level") if detail_level and AUDIT_DETAIL_LEVEL_ORDER[detail_level] < AUDIT_DETAIL_LEVEL_ORDER[parent.audit_detail_level]: raise PrivacyPolicyError("Audit detail level cannot be less restrictive than the parent retention policy.") def _set_settings_privacy_policy(settings_payload: dict[str, Any] | None, policy: dict[str, Any]) -> dict[str, Any]: payload = dict(settings_payload or {}) payload[PRIVACY_POLICY_SETTINGS_KEY] = policy return payload def privacy_policy_from_settings(item: SystemSettings) -> PrivacyRetentionPolicy: return PrivacyRetentionPolicy.model_validate(_privacy_policy_data(item.settings or {})) def privacy_policy_from_session(session: Session) -> PrivacyRetentionPolicy: return privacy_policy_from_settings(get_system_settings(session)) def set_privacy_policy(item: SystemSettings, policy: PrivacyRetentionPolicy | dict[str, Any]) -> PrivacyRetentionPolicy: validated = policy if isinstance(policy, PrivacyRetentionPolicy) else PrivacyRetentionPolicy.model_validate(policy) item.settings = _set_settings_privacy_policy(item.settings, validated.model_dump(mode="json")) return validated def effective_privacy_policy( session: Session, *, tenant_id: str | None = None, owner_user_id: str | None = None, owner_group_id: str | None = None, campaign_id: str | None = None, ) -> PrivacyRetentionPolicy: policy = privacy_policy_from_session(session) campaign: Any | None = None if campaign_id: Campaign, _, _, _, _ = _campaign_models() campaign = session.get(Campaign, campaign_id) if campaign is None: raise PrivacyPolicyError("Campaign not found for privacy policy") tenant_id = campaign.tenant_id owner_user_id = campaign.owner_user_id owner_group_id = campaign.owner_group_id if tenant_id: tenant = session.get(Tenant, tenant_id) if tenant is None: raise PrivacyPolicyError("Tenant not found for privacy policy") policy = _merge_privacy_policy(policy, _privacy_policy_patch_from_settings(tenant.settings or {})) if owner_user_id: user = session.get(User, owner_user_id) if user and (not tenant_id or user.tenant_id == tenant_id): policy = _merge_privacy_policy(policy, _privacy_policy_patch_from_settings(user.settings or {})) if owner_group_id: group = session.get(Group, owner_group_id) if group and (not tenant_id or group.tenant_id == tenant_id): policy = _merge_privacy_policy(policy, _privacy_policy_patch_from_settings(group.settings or {})) if campaign is not None: policy = _merge_privacy_policy(policy, _privacy_policy_patch_from_settings(campaign.settings or {})) return policy def parent_privacy_policy(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None = None) -> PrivacyRetentionPolicy: clean_scope = scope_type.strip().casefold() policy = privacy_policy_from_session(session) if clean_scope == "tenant": return policy tenant = session.get(Tenant, tenant_id) if tenant is None: raise PrivacyPolicyError("Tenant not found for privacy policy") policy = _merge_privacy_policy(policy, _privacy_policy_patch_from_settings(tenant.settings or {})) if clean_scope in {"user", "group"}: return policy if clean_scope != "campaign" or not scope_id: return policy Campaign, _, _, _, _ = _campaign_models() campaign = session.get(Campaign, scope_id) if campaign is None or campaign.tenant_id != tenant_id: raise PrivacyPolicyError("Campaign not found for privacy policy") if campaign.owner_user_id: user = session.get(User, campaign.owner_user_id) if user and user.tenant_id == tenant_id: policy = _merge_privacy_policy(policy, _privacy_policy_patch_from_settings(user.settings or {})) if campaign.owner_group_id: group = session.get(Group, campaign.owner_group_id) if group and group.tenant_id == tenant_id: policy = _merge_privacy_policy(policy, _privacy_policy_patch_from_settings(group.settings or {})) return policy def _retention_policy_source_fields(patch: dict[str, Any], *, baseline: bool = False) -> list[str]: fields: list[str] = [] for key in RETENTION_POLICY_FIELD_KEYS: if key in patch and patch.get(key) is not None: fields.append(key) if isinstance(patch.get("allow_lower_level_limits"), dict) and patch["allow_lower_level_limits"]: fields.append("allow_lower_level_limits") if baseline and not fields: fields.append("defaults") return fields def _retention_policy_source_step(scope_type: str, label: str, scope_id: str | None, patch: dict[str, Any], *, baseline: bool = False) -> dict[str, Any]: source_policy = PrivacyRetentionPolicy.model_validate(patch).model_dump(mode="json") if baseline else dict(patch) return { "scope_type": scope_type, "scope_id": scope_id, "label": label, "applied_fields": _retention_policy_source_fields(patch, baseline=baseline), "policy": source_policy, } def effective_privacy_policy_sources( session: Session, *, tenant_id: str | None = None, owner_user_id: str | None = None, owner_group_id: str | None = None, campaign_id: str | None = None, ) -> list[dict[str, Any]]: system_settings = get_system_settings(session) sources = [_retention_policy_source_step("system", "System", None, _privacy_policy_patch_from_settings(system_settings.settings or {}), baseline=True)] campaign: Any | None = None if campaign_id: Campaign, _, _, _, _ = _campaign_models() campaign = session.get(Campaign, campaign_id) if campaign is None: raise PrivacyPolicyError("Campaign not found for privacy policy") tenant_id = campaign.tenant_id owner_user_id = campaign.owner_user_id owner_group_id = campaign.owner_group_id if tenant_id: tenant = session.get(Tenant, tenant_id) if tenant is None: raise PrivacyPolicyError("Tenant not found for privacy policy") sources.append(_retention_policy_source_step("tenant", "Tenant", tenant.id, _privacy_policy_patch_from_settings(tenant.settings or {}))) if owner_user_id: user = session.get(User, owner_user_id) if user and (not tenant_id or user.tenant_id == tenant_id): sources.append(_retention_policy_source_step("user", "Owner user", user.id, _privacy_policy_patch_from_settings(user.settings or {}))) if owner_group_id: group = session.get(Group, owner_group_id) if group and (not tenant_id or group.tenant_id == tenant_id): sources.append(_retention_policy_source_step("group", "Owner group", group.id, _privacy_policy_patch_from_settings(group.settings or {}))) if campaign is not None: sources.append(_retention_policy_source_step("campaign", "Campaign", campaign.id, _privacy_policy_patch_from_settings(campaign.settings or {}))) return sources def parent_privacy_policy_sources(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None = None) -> list[dict[str, Any]]: clean_scope = scope_type.strip().casefold() if clean_scope == "system": return [] system_settings = get_system_settings(session) sources = [_retention_policy_source_step("system", "System", None, _privacy_policy_patch_from_settings(system_settings.settings or {}), baseline=True)] if clean_scope == "tenant": return sources tenant = session.get(Tenant, tenant_id) if tenant is None: raise PrivacyPolicyError("Tenant not found for privacy policy") sources.append(_retention_policy_source_step("tenant", "Tenant", tenant.id, _privacy_policy_patch_from_settings(tenant.settings or {}))) if clean_scope in {"user", "group"}: return sources if clean_scope != "campaign" or not scope_id: return sources Campaign, _, _, _, _ = _campaign_models() campaign = session.get(Campaign, scope_id) if campaign is None or campaign.tenant_id != tenant_id: raise PrivacyPolicyError("Campaign not found for privacy policy") if campaign.owner_user_id: user = session.get(User, campaign.owner_user_id) if user and user.tenant_id == tenant_id: sources.append(_retention_policy_source_step("user", "Owner user", user.id, _privacy_policy_patch_from_settings(user.settings or {}))) if campaign.owner_group_id: group = session.get(Group, campaign.owner_group_id) if group and group.tenant_id == tenant_id: sources.append(_retention_policy_source_step("group", "Owner group", group.id, _privacy_policy_patch_from_settings(group.settings or {}))) return sources def get_privacy_policy_for_scope(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None = None) -> dict[str, Any]: clean_scope = scope_type.strip().casefold() if clean_scope == "system": return privacy_policy_from_session(session).model_dump(mode="json") if clean_scope == "tenant": tenant = session.get(Tenant, scope_id or tenant_id) if tenant is None or tenant.id != tenant_id: raise PrivacyPolicyError("Tenant privacy policy not found") return _privacy_policy_patch_from_settings(tenant.settings or {}) if not scope_id: raise PrivacyPolicyError(f"{clean_scope.capitalize()} privacy policy requires scope_id") if clean_scope == "user": user = session.get(User, scope_id) if user is None or user.tenant_id != tenant_id: raise PrivacyPolicyError("User privacy policy not found") return _privacy_policy_patch_from_settings(user.settings or {}) if clean_scope == "group": group = session.get(Group, scope_id) if group is None or group.tenant_id != tenant_id: raise PrivacyPolicyError("Group privacy policy not found") return _privacy_policy_patch_from_settings(group.settings or {}) if clean_scope == "campaign": Campaign, _, _, _, _ = _campaign_models() campaign = session.get(Campaign, scope_id) if campaign is None or campaign.tenant_id != tenant_id: raise PrivacyPolicyError("Campaign privacy policy not found") return _privacy_policy_patch_from_settings(campaign.settings or {}) raise PrivacyPolicyError("Privacy policy scope must be system, tenant, user, group or campaign") def set_privacy_policy_for_scope( session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None = None, policy: dict[str, Any] | None = None, ) -> dict[str, Any]: clean_scope = scope_type.strip().casefold() if clean_scope == "system": item = get_system_settings(session) validated = set_privacy_policy(item, PrivacyRetentionPolicy.model_validate(policy or {})) session.add(item) return validated.model_dump(mode="json") patch = PrivacyRetentionPolicyPatch.model_validate(policy or {}).model_dump(mode="json", exclude_none=True) _validate_privacy_patch_against_parent(parent_privacy_policy(session, tenant_id=tenant_id, scope_type=clean_scope, scope_id=scope_id or (tenant_id if clean_scope == "tenant" else None)), patch) if clean_scope == "tenant": tenant = session.get(Tenant, scope_id or tenant_id) if tenant is None or tenant.id != tenant_id: raise PrivacyPolicyError("Tenant privacy policy not found") tenant.settings = _set_settings_privacy_policy(tenant.settings, patch) session.add(tenant) return patch if not scope_id: raise PrivacyPolicyError(f"{clean_scope.capitalize()} privacy policy requires scope_id") if clean_scope == "user": user = session.get(User, scope_id) if user is None or user.tenant_id != tenant_id: raise PrivacyPolicyError("User privacy policy not found") user.settings = _set_settings_privacy_policy(user.settings, patch) session.add(user) return patch if clean_scope == "group": group = session.get(Group, scope_id) if group is None or group.tenant_id != tenant_id: raise PrivacyPolicyError("Group privacy policy not found") group.settings = _set_settings_privacy_policy(group.settings, patch) session.add(group) return patch if clean_scope == "campaign": Campaign, _, _, _, _ = _campaign_models() campaign = session.get(Campaign, scope_id) if campaign is None or campaign.tenant_id != tenant_id: raise PrivacyPolicyError("Campaign privacy policy not found") campaign.settings = _set_settings_privacy_policy(campaign.settings, patch) session.add(campaign) return patch raise PrivacyPolicyError("Privacy policy scope must be system, tenant, user, group or campaign") def sanitize_audit_details_for_policy(session: Session, details: dict[str, Any]) -> dict[str, Any]: policy = privacy_policy_from_session(session) if policy.audit_detail_level == "full": return details if policy.audit_detail_level == "minimal": return { "_privacy": {"detail_level": "minimal"}, "keys": sorted(str(key) for key in details.keys()), } return _redact_audit_value(details) def _redact_audit_value(value: Any) -> Any: if isinstance(value, dict): redacted: dict[str, Any] = {} for key, item in value.items(): if str(key).lower() in AUDIT_DETAIL_REDACTION_KEYS: redacted[key] = "" else: redacted[key] = _redact_audit_value(item) return redacted if isinstance(value, list): return [_redact_audit_value(item) for item in value] return value def _is_before_cutoff(value: datetime | None, cutoff: datetime | None) -> bool: if value is None or cutoff is None: return False candidate = value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value return candidate < cutoff def _system_cutoffs(policy: PrivacyRetentionPolicy, *, now: datetime) -> dict[str, datetime | None]: return { "raw_campaign_json": _cutoff(0 if not policy.store_raw_campaign_json else policy.raw_campaign_json_retention_days, now=now), "generated_eml": _cutoff(policy.generated_eml_retention_days, now=now), "stored_report_detail": _cutoff(policy.stored_report_detail_retention_days, now=now), "mock_mailbox": _cutoff(policy.mock_mailbox_retention_days, now=now), "audit_detail": _cutoff(policy.audit_detail_retention_days, now=now), } def apply_retention_policy(session: Session, *, dry_run: bool = True) -> dict[str, Any]: now = _utcnow() policy = privacy_policy_from_session(session) cutoffs = _system_cutoffs(policy, now=now) counts = { "raw_campaign_json": _apply_raw_json_retention(session, dry_run=dry_run, now=now), "generated_eml": _apply_eml_retention(session, dry_run=dry_run, now=now), "stored_report_detail": _apply_report_detail_retention(session, dry_run=dry_run, now=now), "mock_mailbox": _apply_mock_mailbox_retention(cutoffs["mock_mailbox"], dry_run=dry_run), "audit_detail": _apply_audit_detail_retention(session, dry_run=dry_run, now=now), } return { "dry_run": dry_run, "policy": policy.model_dump(mode="json"), "cutoffs": {key: value.isoformat() if value else None for key, value in cutoffs.items()}, "effective_policy_scope": "per-object", "counts": counts, } def _campaign_policy_for_id(session: Session, campaign_id: str | None, cache: dict[str, PrivacyRetentionPolicy]) -> PrivacyRetentionPolicy: if not campaign_id: return privacy_policy_from_session(session) if campaign_id not in cache: cache[campaign_id] = effective_privacy_policy(session, campaign_id=campaign_id) return cache[campaign_id] def _apply_raw_json_retention(session: Session, *, dry_run: bool, now: datetime) -> dict[str, int]: result = {"eligible": 0, "redacted": 0, "skipped_not_final": 0, "already_redacted": 0} models = _campaign_models_or_none() if models is None: return result _, _, CampaignVersion, _, _ = models policy_cache: dict[str, PrivacyRetentionPolicy] = {} versions = ( session.query(CampaignVersion) .order_by(CampaignVersion.updated_at.asc()) .all() ) for version in versions: policy = _campaign_policy_for_id(session, version.campaign_id, policy_cache) cutoff = _cutoff(0 if not policy.store_raw_campaign_json else policy.raw_campaign_json_retention_days, now=now) if not _is_before_cutoff(version.updated_at, cutoff): continue raw_json = version.raw_json if isinstance(version.raw_json, dict) else {} if raw_json.get("_retention", {}).get("raw_json_redacted"): result["already_redacted"] += 1 continue if version.workflow_state not in FINAL_VERSION_STATES: result["skipped_not_final"] += 1 continue result["eligible"] += 1 if dry_run: continue snapshot = version.execution_snapshot if isinstance(version.execution_snapshot, dict) else {} version.raw_json = { "version": version.schema_version or "1.0", "_retention": { "raw_json_redacted": True, "redacted_at": now.isoformat(), "original_sha256": snapshot.get("campaign_json_sha256") or _json_sha256(raw_json), }, } session.add(version) result["redacted"] += 1 return result def _apply_eml_retention(session: Session, *, dry_run: bool, now: datetime) -> dict[str, int]: result = {"eligible": 0, "metadata_cleared": 0, "files_deleted": 0, "files_missing": 0, "skipped_not_final": 0} models = _campaign_models_or_none() if models is None: return result _, CampaignJob, _, JobImapStatus, JobQueueStatus = models policy_cache: dict[str, PrivacyRetentionPolicy] = {} jobs = ( session.query(CampaignJob) .filter((CampaignJob.eml_local_path.is_not(None)) | (CampaignJob.eml_storage_key.is_not(None))) .order_by(CampaignJob.updated_at.asc()) .all() ) for job in jobs: policy = _campaign_policy_for_id(session, job.campaign_id, policy_cache) cutoff = _cutoff(policy.generated_eml_retention_days, now=now) if not _is_before_cutoff(job.updated_at, cutoff): continue if job.queue_status in {JobQueueStatus.QUEUED.value, JobQueueStatus.SENDING.value} or job.send_status not in FINAL_EML_SEND_STATUSES: result["skipped_not_final"] += 1 continue if job.imap_status == JobImapStatus.PENDING.value: result["skipped_not_final"] += 1 continue result["eligible"] += 1 if dry_run: continue if job.eml_local_path: path = Path(job.eml_local_path) if path.exists(): path.unlink() result["files_deleted"] += 1 else: result["files_missing"] += 1 job.eml_local_path = None job.eml_storage_key = None session.add(job) result["metadata_cleared"] += 1 return result def _apply_report_detail_retention(session: Session, *, dry_run: bool, now: datetime) -> dict[str, int]: result = {"eligible_versions": 0, "summaries_redacted": 0, "already_redacted": 0} models = _campaign_models_or_none() if models is None: return result _, _, CampaignVersion, _, _ = models policy_cache: dict[str, PrivacyRetentionPolicy] = {} versions = ( session.query(CampaignVersion) .order_by(CampaignVersion.updated_at.asc()) .all() ) for version in versions: policy = _campaign_policy_for_id(session, version.campaign_id, policy_cache) cutoff = _cutoff(policy.stored_report_detail_retention_days, now=now) if not _is_before_cutoff(version.updated_at, cutoff): continue next_validation, validation_changed = _redacted_report_summary(version.validation_summary, now=now) next_build, build_changed = _redacted_report_summary(version.build_summary, now=now) if not validation_changed and not build_changed: if _summary_was_redacted(version.validation_summary) or _summary_was_redacted(version.build_summary): result["already_redacted"] += 1 continue result["eligible_versions"] += 1 if dry_run: continue version.validation_summary = next_validation version.build_summary = next_build session.add(version) result["summaries_redacted"] += int(validation_changed) + int(build_changed) return result def _summary_was_redacted(summary: Any) -> bool: return isinstance(summary, dict) and bool(summary.get("_retention", {}).get("report_detail_redacted")) def _redacted_report_summary(summary: Any, *, now: datetime) -> tuple[dict[str, Any] | None, bool]: if not isinstance(summary, dict): return summary, False if _summary_was_redacted(summary): return summary, False detail_keys = {"messages", "issues", "recent_failures", "jobs"} if not any(key in summary for key in detail_keys): return summary, False next_summary = copy.deepcopy(summary) for key in detail_keys: next_summary.pop(key, None) retention = dict(next_summary.get("_retention") or {}) retention.update({"report_detail_redacted": True, "redacted_at": now.isoformat()}) next_summary["_retention"] = retention return next_summary, True def _apply_mock_mailbox_retention(cutoff: datetime | None, *, dry_run: bool) -> dict[str, int]: result = {"eligible_records": 0, "json_deleted": 0, "eml_deleted": 0} if cutoff is None: return result message_dir = Path(settings.mock_mailbox_dir) / "messages" if not message_dir.exists(): return result for path in message_dir.glob("*.json"): try: record = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError: continue created_at = _parse_datetime(record.get("created_at")) if created_at and created_at >= cutoff: continue result["eligible_records"] += 1 if dry_run: continue raw_filename = record.get("raw_filename") if raw_filename: raw_path = message_dir / str(raw_filename) if raw_path.exists(): raw_path.unlink() result["eml_deleted"] += 1 if path.exists(): path.unlink() result["json_deleted"] += 1 return result def _parse_datetime(value: Any) -> datetime | None: if not value: return None try: parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) except ValueError: return None if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed def _privacy_policy_for_audit_item(session: Session, item: AuditLog, campaign_cache: dict[str, PrivacyRetentionPolicy], tenant_cache: dict[str, PrivacyRetentionPolicy]) -> PrivacyRetentionPolicy: if item.object_type == "campaign" and item.object_id: models = _campaign_models_or_none() if models is not None: Campaign, _, _, _, _ = models campaign = session.get(Campaign, str(item.object_id)) if campaign is not None: return _campaign_policy_for_id(session, campaign.id, campaign_cache) if item.tenant_id: if item.tenant_id not in tenant_cache: tenant_cache[item.tenant_id] = effective_privacy_policy(session, tenant_id=item.tenant_id) return tenant_cache[item.tenant_id] return privacy_policy_from_session(session) def _apply_audit_detail_retention(session: Session, *, dry_run: bool, now: datetime) -> dict[str, int]: result = {"eligible": 0, "redacted": 0, "already_redacted": 0} campaign_cache: dict[str, PrivacyRetentionPolicy] = {} tenant_cache: dict[str, PrivacyRetentionPolicy] = {} items = ( session.query(AuditLog) .order_by(AuditLog.created_at.asc()) .all() ) for item in items: policy = _privacy_policy_for_audit_item(session, item, campaign_cache, tenant_cache) cutoff = _cutoff(policy.audit_detail_retention_days, now=now) if not _is_before_cutoff(item.created_at, cutoff): continue details = item.details if isinstance(item.details, dict) else {} if details.get("_retention", {}).get("audit_detail_redacted"): result["already_redacted"] += 1 continue result["eligible"] += 1 if dry_run: continue item.details = { "_retention": { "audit_detail_redacted": True, "redacted_at": now.isoformat(), "original_detail_sha256": _json_sha256(details), } } session.add(item) result["redacted"] += 1 return result