add-ideas/govoplan-policy
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Sync GovOPlaN module state

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-10 21:57:27 +02:00
parent 47cabafa0e
commit bebb8a6e73
5 changed files with 1005 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

253
.gitignore vendored
View File

@@ -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

10
docs/POLICY_DECISION_PROVENANCE.md
View File

@@ -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:

2
src/govoplan_policy/backend/api/v1/routes.py
View File

@@ -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,

11
src/govoplan_policy/backend/manifest.py
View File

@@ -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,
},
)

730
src/govoplan_policy/backend/retention.py Normal file
View File

@@ -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] = "<redacted>"
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)