From bebb8a6e7336f59caf41b592d3ae03dbbd4c11aa Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 10 Jul 2026 21:57:27 +0200 Subject: [PATCH] Sync GovOPlaN module state --- .gitignore | 253 +++++++ docs/POLICY_DECISION_PROVENANCE.md | 10 + src/govoplan_policy/backend/api/v1/routes.py | 2 +- src/govoplan_policy/backend/manifest.py | 11 + src/govoplan_policy/backend/retention.py | 730 +++++++++++++++++++ 5 files changed, 1005 insertions(+), 1 deletion(-) create mode 100644 src/govoplan_policy/backend/retention.py diff --git a/.gitignore b/.gitignore index 5fd4061..2a33867 100644 --- a/.gitignore +++ b/.gitignore @@ -21,3 +21,256 @@ webui/.module-test-build/ webui/.policy-test-build/ webui/.template-preview-test-build/ webui/.import-test-build/ + +# GovOPlaN shared ignore rules from govoplan-core +# ---> Node +# Logs +logs +*.log +npm-debug.log* +yarn-debug.log* +yarn-error.log* +lerna-debug.log* +.pnpm-debug.log* +# Diagnostic reports (https://nodejs.org/api/report.html) +report.[0-9]*.[0-9]*.[0-9]*.[0-9]*.json +# Runtime data +pids +*.pid +*.seed +*.pid.lock +# Directory for instrumented libs generated by jscoverage/JSCover +lib-cov +# Coverage directory used by tools like istanbul +coverage +*.lcov +# nyc test coverage +.nyc_output +# Grunt intermediate storage (https://gruntjs.com/creating-plugins#storing-task-files) +.grunt +# Bower dependency directory (https://bower.io/) +bower_components +# node-waf configuration +.lock-wscript +# Compiled binary addons (https://nodejs.org/api/addons.html) +build/Release +# Dependency directories +jspm_packages/ +# Snowpack dependency directory (https://snowpack.dev/) +web_modules/ +# TypeScript cache +# Optional npm cache directory +.npm +# Optional eslint cache +.eslintcache +# Optional stylelint cache +.stylelintcache +# Microbundle cache +.rpt2_cache/ +.rts2_cache_cjs/ +.rts2_cache_es/ +.rts2_cache_umd/ +# Optional REPL history +.node_repl_history +# Output of 'npm pack' +*.tgz +# Yarn Integrity file +.yarn-integrity +# dotenv environment variable files +.env +.env.development.local +.env.test.local +.env.production.local +.env.local +# parcel-bundler cache (https://parceljs.org/) +.cache +.parcel-cache +# Next.js build output +.next +out +# Nuxt.js build / generate output +.nuxt +dist +# Gatsby files +.cache/ +# Comment in the public line in if your project uses Gatsby and not Next.js +# https://nextjs.org/blog/next-9-1#public-directory-support +# public +# vuepress build output +.vuepress/dist +# vuepress v2.x temp and cache directory +.temp +.cache +# vitepress build output +**/.vitepress/dist +# vitepress cache directory +**/.vitepress/cache +# Docusaurus cache and generated files +.docusaurus +# Serverless directories +.serverless/ +# FuseBox cache +.fusebox/ +# DynamoDB Local files +.dynamodb/ +# TernJS port file +.tern-port +# Stores VSCode versions used for testing VSCode extensions +.vscode-test +# yarn v2 +.yarn/cache +.yarn/unplugged +.yarn/build-state.yml +.yarn/install-state.gz +.pnp.* +# Local WebUI test/build scratch directories +# ---> Python +# Byte-compiled / optimized / DLL files +*$py.class +# C extensions +*.so +# Distribution / packaging +.Python +develop-eggs/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +.installed.cfg +*.egg +MANIFEST +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec +# Installer logs +pip-log.txt +pip-delete-this-directory.txt +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +cover/ +# Translations +*.mo +*.pot +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal +# Flask stuff: +instance/ +.webassets-cache +# Scrapy stuff: +.scrapy +# Sphinx documentation +docs/_build/ +# PyBuilder +.pybuilder/ +target/ +# Jupyter Notebook +.ipynb_checkpoints +# IPython +profile_default/ +ipython_config.py +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock +# UV +# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +#uv.lock +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +#poetry.lock +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +#pdm.lock +# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it +# in version control. +# https://pdm.fming.dev/latest/usage/project/#working-with-version-control +.pdm.toml +.pdm-python +.pdm-build/ +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ +# Celery stuff +celerybeat-schedule +celerybeat.pid +# SageMath parsed files +*.sage.py +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ +# Spyder project settings +.spyderproject +.spyproject +# Rope project settings +.ropeproject +# mkdocs documentation +/site +# mypy +.dmypy.json +dmypy.json +# Pyre type checker +.pyre/ +# pytype static type analyzer +.pytype/ +# Cython debug symbols +cython_debug/ +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +#.idea/ +# Ruff stuff: +# PyPI configuration file +.pypirc +# ---> VisualStudioCode +.vscode/* +!.vscode/settings.json +!.vscode/tasks.json +!.vscode/launch.json +!.vscode/extensions.json +!.vscode/*.code-snippets +# Local History for Visual Studio Code +.history/ +# Built Visual Studio Code Extensions +*.vsix +*.db +# GovOPlaN local runtime state +runtime/ +# GovOPlaN WebUI test output diff --git a/docs/POLICY_DECISION_PROVENANCE.md b/docs/POLICY_DECISION_PROVENANCE.md index f8d7db6..9b44f1b 100644 --- a/docs/POLICY_DECISION_PROVENANCE.md +++ b/docs/POLICY_DECISION_PROVENANCE.md @@ -3,6 +3,16 @@ `govoplan-policy` owns policy and retention route contributions. Core keeps the small shared DTOs that let policy decisions look the same across modules. +Privacy retention implementation lives in this module at +`govoplan_policy.backend.retention`. It is also exposed as the +`policy.privacyRetention` capability so older core compatibility imports can +dispatch to the active policy module without owning policy logic in core. Core +does not import this implementation as a hidden fallback when policy is +disabled. +When retention needs audit-log storage behavior, it requests the +`audit.retention` capability; it does not import audit module tables or +providers directly. + ## Backend DTOs Use `govoplan_core.core.policy.PolicyDecision` for explainable policy results: diff --git a/src/govoplan_policy/backend/api/v1/routes.py b/src/govoplan_policy/backend/api/v1/routes.py index 17fab1a..686662f 100644 --- a/src/govoplan_policy/backend/api/v1/routes.py +++ b/src/govoplan_policy/backend/api/v1/routes.py @@ -14,7 +14,7 @@ from govoplan_core.core.configuration_control import ( ) from govoplan_core.core.policy import PolicyDecision, PolicySourceStep from govoplan_core.db.session import get_session -from govoplan_core.privacy.retention import ( +from govoplan_policy.backend.retention import ( PrivacyPolicyError, RETENTION_POLICY_FIELD_KEYS, apply_retention_policy, diff --git a/src/govoplan_policy/backend/manifest.py b/src/govoplan_policy/backend/manifest.py index 32752f0..c7a91cd 100644 --- a/src/govoplan_policy/backend/manifest.py +++ b/src/govoplan_policy/backend/manifest.py @@ -1,6 +1,7 @@ from __future__ import annotations from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER +from govoplan_core.core.policy import CAPABILITY_POLICY_PRIVACY_RETENTION from govoplan_core.core.modules import ModuleContext, ModuleManifest @@ -11,12 +12,22 @@ def _route_factory(context: ModuleContext): return router +def _privacy_retention_service(context: ModuleContext) -> object: + del context + from govoplan_policy.backend.retention import SqlPrivacyRetentionService + + return SqlPrivacyRetentionService() + + manifest = ModuleManifest( id="policy", name="Policy", version="0.1.6", required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR), route_factory=_route_factory, + capability_factories={ + CAPABILITY_POLICY_PRIVACY_RETENTION: _privacy_retention_service, + }, ) diff --git a/src/govoplan_policy/backend/retention.py b/src/govoplan_policy/backend/retention.py new file mode 100644 index 0000000..b60f63e --- /dev/null +++ b/src/govoplan_policy/backend/retention.py @@ -0,0 +1,730 @@ +from __future__ import annotations + +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.settings import get_system_settings +from govoplan_core.core.campaigns import ( + CAPABILITY_CAMPAIGNS_POLICY_CONTEXT, + CAPABILITY_CAMPAIGNS_RETENTION, + CampaignPolicyContext, + CampaignPolicyContextProvider, + CampaignRetentionProvider, +) +from govoplan_core.core.access import ( + CAPABILITY_ACCESS_ADMINISTRATION, + CAPABILITY_AUDIT_RETENTION, + AccessAdministration, + AuditRecordRef, + AuditRetentionProvider, +) +from govoplan_core.core.policy import policy_source_step +from govoplan_core.core.runtime import get_registry +from govoplan_core.admin.models import SystemSettings +from govoplan_core.settings import settings +from govoplan_core.tenancy.scope import Tenant + +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 + +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_policy_provider() -> CampaignPolicyContextProvider | None: + registry = get_registry() + if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(CAPABILITY_CAMPAIGNS_POLICY_CONTEXT): + return None + capability = registry.require_capability(CAPABILITY_CAMPAIGNS_POLICY_CONTEXT) + if not isinstance(capability, CampaignPolicyContextProvider): + raise PrivacyPolicyError("Campaign policy context capability is invalid") + return capability + + +def _campaign_retention_provider() -> CampaignRetentionProvider | None: + registry = get_registry() + if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(CAPABILITY_CAMPAIGNS_RETENTION): + return None + capability = registry.require_capability(CAPABILITY_CAMPAIGNS_RETENTION) + if not isinstance(capability, CampaignRetentionProvider): + raise PrivacyPolicyError("Campaign retention capability is invalid") + return capability + + +def _access_administration() -> AccessAdministration | None: + registry = get_registry() + if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(CAPABILITY_ACCESS_ADMINISTRATION): + return None + capability = registry.require_capability(CAPABILITY_ACCESS_ADMINISTRATION) + if not isinstance(capability, AccessAdministration): + raise PrivacyPolicyError("Access administration capability is invalid") + return capability + + +def _audit_retention_provider() -> AuditRetentionProvider | None: + registry = get_registry() + if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(CAPABILITY_AUDIT_RETENTION): + return None + + capability = registry.require_capability(CAPABILITY_AUDIT_RETENTION) + if not isinstance(capability, AuditRetentionProvider): + raise PrivacyPolicyError("Audit retention capability is invalid") + return capability + + +def _user_settings(session: Session, *, user_id: str, tenant_id: str) -> dict[str, Any] | None: + provider = _access_administration() + if provider is None: + return None + payload = provider.user_settings(session, user_id, tenant_id=tenant_id) + return dict(payload) if payload is not None else None + + +def _set_user_settings(session: Session, *, user_id: str, tenant_id: str, settings_payload: dict[str, Any]) -> dict[str, Any] | None: + provider = _access_administration() + if provider is None: + return None + payload = provider.set_user_settings(session, user_id, tenant_id=tenant_id, settings=settings_payload) + return dict(payload) if payload is not None else None + + +def _group_settings(session: Session, *, group_id: str, tenant_id: str) -> dict[str, Any] | None: + provider = _access_administration() + if provider is None: + return None + payload = provider.group_settings(session, group_id, tenant_id=tenant_id) + return dict(payload) if payload is not None else None + + +def _set_group_settings(session: Session, *, group_id: str, tenant_id: str, settings_payload: dict[str, Any]) -> dict[str, Any] | None: + provider = _access_administration() + if provider is None: + return None + payload = provider.set_group_settings(session, group_id, tenant_id=tenant_id, settings=settings_payload) + return dict(payload) if payload is not None else None + + +def _campaign_policy_context(session: Session, *, campaign_id: str, tenant_id: str | None = None) -> CampaignPolicyContext: + provider = _campaign_policy_provider() + if provider is None: + raise PrivacyPolicyError("Campaign module is not installed") + context = provider.get_campaign_policy_context(session, tenant_id=tenant_id, campaign_id=campaign_id) + if context is None: + raise PrivacyPolicyError("Campaign not found for privacy policy") + return context + + +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 _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: CampaignPolicyContext | None = None + if campaign_id: + campaign = _campaign_policy_context(session, campaign_id=campaign_id) + 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_settings = _user_settings(session, user_id=owner_user_id, tenant_id=tenant_id) if tenant_id else None + if user_settings is not None: + policy = _merge_privacy_policy(policy, _privacy_policy_patch_from_settings(user_settings)) + if owner_group_id: + group_settings = _group_settings(session, group_id=owner_group_id, tenant_id=tenant_id) if tenant_id else None + if group_settings is not None: + policy = _merge_privacy_policy(policy, _privacy_policy_patch_from_settings(group_settings)) + if campaign is not None: + policy = _merge_privacy_policy(policy, _privacy_policy_patch_from_settings(dict(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_policy_context(session, tenant_id=tenant_id, campaign_id=scope_id) + if campaign.owner_user_id: + user_settings = _user_settings(session, user_id=campaign.owner_user_id, tenant_id=tenant_id) + if user_settings is not None: + policy = _merge_privacy_policy(policy, _privacy_policy_patch_from_settings(user_settings)) + if campaign.owner_group_id: + group_settings = _group_settings(session, group_id=campaign.owner_group_id, tenant_id=tenant_id) + if group_settings is not None: + policy = _merge_privacy_policy(policy, _privacy_policy_patch_from_settings(group_settings)) + 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 policy_source_step( + scope_type, + label, + scope_id, + 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: CampaignPolicyContext | None = None + if campaign_id: + campaign = _campaign_policy_context(session, campaign_id=campaign_id) + 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_settings = _user_settings(session, user_id=owner_user_id, tenant_id=tenant_id) if tenant_id else None + if user_settings is not None: + sources.append(_retention_policy_source_step("user", "Owner user", owner_user_id, _privacy_policy_patch_from_settings(user_settings))) + if owner_group_id: + group_settings = _group_settings(session, group_id=owner_group_id, tenant_id=tenant_id) if tenant_id else None + if group_settings is not None: + sources.append(_retention_policy_source_step("group", "Owner group", owner_group_id, _privacy_policy_patch_from_settings(group_settings))) + if campaign is not None: + sources.append(_retention_policy_source_step("campaign", "Campaign", campaign.id, _privacy_policy_patch_from_settings(dict(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_policy_context(session, tenant_id=tenant_id, campaign_id=scope_id) + if campaign.owner_user_id: + user_settings = _user_settings(session, user_id=campaign.owner_user_id, tenant_id=tenant_id) + if user_settings is not None: + sources.append(_retention_policy_source_step("user", "Owner user", campaign.owner_user_id, _privacy_policy_patch_from_settings(user_settings))) + if campaign.owner_group_id: + group_settings = _group_settings(session, group_id=campaign.owner_group_id, tenant_id=tenant_id) + if group_settings is not None: + sources.append(_retention_policy_source_step("group", "Owner group", campaign.owner_group_id, _privacy_policy_patch_from_settings(group_settings))) + 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_settings = _user_settings(session, user_id=scope_id, tenant_id=tenant_id) + if user_settings is None: + raise PrivacyPolicyError("User privacy policy not found") + return _privacy_policy_patch_from_settings(user_settings) + if clean_scope == "group": + group_settings = _group_settings(session, group_id=scope_id, tenant_id=tenant_id) + if group_settings is None: + raise PrivacyPolicyError("Group privacy policy not found") + return _privacy_policy_patch_from_settings(group_settings) + if clean_scope == "campaign": + campaign = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=scope_id) + return _privacy_policy_patch_from_settings(dict(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": + current_settings = _user_settings(session, user_id=scope_id, tenant_id=tenant_id) + if current_settings is None: + raise PrivacyPolicyError("User privacy policy not found") + if _set_user_settings(session, user_id=scope_id, tenant_id=tenant_id, settings_payload=_set_settings_privacy_policy(current_settings, patch)) is None: + raise PrivacyPolicyError("User privacy policy not found") + return patch + if clean_scope == "group": + current_settings = _group_settings(session, group_id=scope_id, tenant_id=tenant_id) + if current_settings is None: + raise PrivacyPolicyError("Group privacy policy not found") + if _set_group_settings(session, group_id=scope_id, tenant_id=tenant_id, settings_payload=_set_settings_privacy_policy(current_settings, patch)) is None: + raise PrivacyPolicyError("Group privacy policy not found") + return patch + if clean_scope == "campaign": + provider = _campaign_policy_provider() + if provider is None: + raise PrivacyPolicyError("Campaign module is not installed") + campaign = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=scope_id) + settings_payload = _set_settings_privacy_policy(dict(campaign.settings or {}), patch) + try: + provider.set_campaign_settings(session, tenant_id=tenant_id, campaign_id=scope_id, settings=settings_payload) + except ValueError as exc: + raise PrivacyPolicyError("Campaign privacy policy not found") from exc + 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 _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) + policy_cache: dict[str, PrivacyRetentionPolicy] = {} + campaign_retention = _campaign_retention_provider() + campaign_counts = { + "raw_campaign_json": {"eligible": 0, "redacted": 0, "skipped_not_final": 0, "already_redacted": 0}, + "generated_eml": {"eligible": 0, "metadata_cleared": 0, "files_deleted": 0, "files_missing": 0, "skipped_not_final": 0}, + "stored_report_detail": {"eligible_versions": 0, "summaries_redacted": 0, "already_redacted": 0}, + } + if campaign_retention is not None: + campaign_counts.update( + { + key: dict(value) + for key, value in campaign_retention.apply_retention( + session, + dry_run=dry_run, + now=now, + policy_for_campaign_id=lambda campaign_id: _campaign_policy_for_id(session, campaign_id, policy_cache), + ).items() + } + ) + counts = { + "raw_campaign_json": campaign_counts["raw_campaign_json"], + "generated_eml": campaign_counts["generated_eml"], + "stored_report_detail": campaign_counts["stored_report_detail"], + "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_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: AuditRecordRef, campaign_cache: dict[str, PrivacyRetentionPolicy], tenant_cache: dict[str, PrivacyRetentionPolicy]) -> PrivacyRetentionPolicy: + if item.object_type == "campaign" and item.object_id: + provider = _campaign_policy_provider() + if provider is not None: + context = provider.get_campaign_policy_context(session, campaign_id=str(item.object_id)) + if context is not None: + return _campaign_policy_for_id(session, context.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]: + provider = _audit_retention_provider() + if provider is None: + return {"eligible": 0, "redacted": 0, "already_redacted": 0} + campaign_cache: dict[str, PrivacyRetentionPolicy] = {} + tenant_cache: dict[str, PrivacyRetentionPolicy] = {} + return dict(provider.apply_detail_retention( + session, + dry_run=dry_run, + now=now, + policy_for_record=lambda item: _privacy_policy_for_audit_item(session, item, campaign_cache, tenant_cache), + )) + + +class SqlPrivacyRetentionService: + def privacy_policy_from_settings(self, *args: Any, **kwargs: Any) -> Any: + return privacy_policy_from_settings(*args, **kwargs) + + def privacy_policy_from_session(self, *args: Any, **kwargs: Any) -> Any: + return privacy_policy_from_session(*args, **kwargs) + + def set_privacy_policy(self, *args: Any, **kwargs: Any) -> Any: + return set_privacy_policy(*args, **kwargs) + + def effective_privacy_policy(self, *args: Any, **kwargs: Any) -> Any: + return effective_privacy_policy(*args, **kwargs) + + def parent_privacy_policy(self, *args: Any, **kwargs: Any) -> Any: + return parent_privacy_policy(*args, **kwargs) + + def effective_privacy_policy_sources(self, *args: Any, **kwargs: Any) -> Any: + return effective_privacy_policy_sources(*args, **kwargs) + + def parent_privacy_policy_sources(self, *args: Any, **kwargs: Any) -> Any: + return parent_privacy_policy_sources(*args, **kwargs) + + def get_privacy_policy_for_scope(self, *args: Any, **kwargs: Any) -> Any: + return get_privacy_policy_for_scope(*args, **kwargs) + + def set_privacy_policy_for_scope(self, *args: Any, **kwargs: Any) -> Any: + return set_privacy_policy_for_scope(*args, **kwargs) + + def sanitize_audit_details_for_policy(self, *args: Any, **kwargs: Any) -> Any: + return sanitize_audit_details_for_policy(*args, **kwargs) + + def apply_retention_policy(self, *args: Any, **kwargs: Any) -> Any: + return apply_retention_policy(*args, **kwargs)