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

intermittent commit

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-14 13:22:12 +02:00
parent 664eb38ab9
commit 2511fbb5a8
5 changed files with 177 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -14,9 +14,9 @@ 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.schemas import RETENTION_POLICY_FIELD_KEYS
from govoplan_policy.backend.retention import (
PrivacyPolicyError,
RETENTION_POLICY_FIELD_KEYS,
apply_retention_policy,
effective_privacy_policy,
effective_privacy_policy_sources,

73
src/govoplan_policy/backend/api/v1/schemas.py
View File

@@ -1,43 +1,10 @@
from __future__ import annotations
from datetime import datetime
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator
from pydantic import BaseModel, ConfigDict, Field
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",
)
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
from govoplan_core.privacy.schemas import PrivacyRetentionPolicyItem, PrivacyRetentionPolicyPatchItem
class PolicySourceStepItem(BaseModel):
@@ -49,42 +16,6 @@ class PolicySourceStepItem(BaseModel):
policy: dict[str, Any] = Field(default_factory=dict)
class PrivacyRetentionPolicyItem(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("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 PrivacyRetentionPolicyPatchItem(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("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 PrivacyRetentionPolicyScopeRequest(BaseModel):
model_config = ConfigDict(extra="forbid")

183
src/govoplan_policy/backend/retention.py
View File

@@ -3,9 +3,9 @@ from __future__ import annotations
import json
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Literal
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator
from pydantic import field_validator
from sqlalchemy.orm import Session
from govoplan_core.admin.settings import get_system_settings
@@ -26,6 +26,14 @@ from govoplan_core.core.access import (
from govoplan_core.core.policy import PolicySourceStep, policy_source_step
from govoplan_core.core.runtime import get_registry
from govoplan_core.admin.models import SystemSettings
from govoplan_core.privacy.schemas import (
RETENTION_DAY_KEYS,
RETENTION_POLICY_FIELD_KEYS,
PrivacyRetentionPolicyItem,
PrivacyRetentionPolicyPatchItem,
default_allow_lower_level_limits,
normalize_allow_lower_level_limits,
)
from govoplan_core.settings import settings
from govoplan_core.tenancy.scope import Tenant
from govoplan_policy.backend.hierarchy import (
@@ -35,38 +43,8 @@ from govoplan_policy.backend.hierarchy import (
)
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",
@@ -91,18 +69,7 @@ AUDIT_DETAIL_REDACTION_KEYS = {
}
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)
class PrivacyRetentionPolicy(PrivacyRetentionPolicyItem):
@field_validator(
"raw_campaign_json_retention_days",
"generated_eml_retention_days",
@@ -120,21 +87,10 @@ class PrivacyRetentionPolicy(BaseModel):
@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)
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
class PrivacyRetentionPolicyPatch(PrivacyRetentionPolicyPatchItem):
@field_validator(*RETENTION_DAY_KEYS, mode="before")
@classmethod
def _blank_string_is_none(cls, value: Any) -> Any:
@@ -145,7 +101,7 @@ class PrivacyRetentionPolicyPatch(BaseModel):
@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)
return normalize_allow_lower_level_limits(value, fill_defaults=False)
class PrivacyPolicyError(RuntimeError):
@@ -524,47 +480,128 @@ def set_privacy_policy_for_scope(
) -> dict[str, Any]:
clean_scope = scope_type.strip().casefold()
if clean_scope == "system":
return _set_system_privacy_policy(session, policy)
patch = _privacy_policy_patch_payload(policy)
_validate_privacy_policy_patch_for_scope(
session,
tenant_id=tenant_id,
scope_type=clean_scope,
scope_id=scope_id,
patch=patch,
)
return _set_scoped_privacy_policy(
session,
tenant_id=tenant_id,
scope_type=clean_scope,
scope_id=scope_id,
patch=patch,
)
def _set_system_privacy_policy(session: Session, policy: dict[str, Any] | None) -> dict[str, Any]:
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":
def _privacy_policy_patch_payload(policy: dict[str, Any] | None) -> dict[str, Any]:
return PrivacyRetentionPolicyPatch.model_validate(policy or {}).model_dump(mode="json", exclude_none=True)
def _validate_privacy_policy_patch_for_scope(
session: Session,
*,
tenant_id: str,
scope_type: str,
scope_id: str | None,
patch: dict[str, Any],
) -> None:
parent_scope_id = _parent_privacy_policy_scope_id(tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id)
parent = parent_privacy_policy(session, tenant_id=tenant_id, scope_type=scope_type, scope_id=parent_scope_id)
_validate_privacy_patch_against_parent(parent, patch)
def _parent_privacy_policy_scope_id(*, tenant_id: str, scope_type: str, scope_id: str | None) -> str | None:
if scope_id:
return scope_id
if scope_type == "tenant":
return tenant_id
return None
def _set_scoped_privacy_policy(
session: Session,
*,
tenant_id: str,
scope_type: str,
scope_id: str | None,
patch: dict[str, Any],
) -> dict[str, Any]:
if scope_type == "tenant":
return _set_tenant_privacy_policy(session, tenant_id=tenant_id, scope_id=scope_id, patch=patch)
clean_scope_id = _required_privacy_policy_scope_id(scope_type, scope_id)
if scope_type == "user":
return _set_user_privacy_policy(session, tenant_id=tenant_id, user_id=clean_scope_id, patch=patch)
if scope_type == "group":
return _set_group_privacy_policy(session, tenant_id=tenant_id, group_id=clean_scope_id, patch=patch)
if scope_type == "campaign":
return _set_campaign_privacy_policy(session, tenant_id=tenant_id, campaign_id=clean_scope_id, patch=patch)
raise PrivacyPolicyError("Privacy policy scope must be system, tenant, user, group or campaign")
def _required_privacy_policy_scope_id(scope_type: str, scope_id: str | None) -> str:
if scope_id:
return scope_id
raise PrivacyPolicyError(f"{scope_type.capitalize()} privacy policy requires scope_id")
def _set_tenant_privacy_policy(
session: Session,
*,
tenant_id: str,
scope_id: str | None,
patch: dict[str, Any],
) -> dict[str, Any]:
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)
def _set_user_privacy_policy(session: Session, *, tenant_id: str, user_id: str, patch: dict[str, Any]) -> dict[str, Any]:
current_settings = _user_settings(session, user_id=user_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:
settings_payload = _set_settings_privacy_policy(current_settings, patch)
if _set_user_settings(session, user_id=user_id, tenant_id=tenant_id, settings_payload=settings_payload) 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)
def _set_group_privacy_policy(session: Session, *, tenant_id: str, group_id: str, patch: dict[str, Any]) -> dict[str, Any]:
current_settings = _group_settings(session, group_id=group_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:
settings_payload = _set_settings_privacy_policy(current_settings, patch)
if _set_group_settings(session, group_id=group_id, tenant_id=tenant_id, settings_payload=settings_payload) is None:
raise PrivacyPolicyError("Group privacy policy not found")
return patch
if clean_scope == "campaign":
def _set_campaign_privacy_policy(session: Session, *, tenant_id: str, campaign_id: str, patch: dict[str, Any]) -> dict[str, Any]:
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)
campaign = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=campaign_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)
provider.set_campaign_settings(session, tenant_id=tenant_id, campaign_id=campaign_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 simulate_privacy_policy_change(

38
tests/test_policy_hierarchy.py
View File

@@ -2,6 +2,14 @@ from __future__ import annotations
import unittest
from govoplan_policy.backend.retention import (
PrivacyRetentionPolicy,
PrivacyRetentionPolicyPatch,
PrivacyPolicyError,
_parent_privacy_policy_scope_id,
_privacy_policy_patch_payload,
_required_privacy_policy_scope_id,
)
from govoplan_policy.backend.hierarchy import (
PolicyRestrictionRule,
simulate_hierarchical_policy_change,
@@ -10,6 +18,36 @@ from govoplan_policy.backend.hierarchy import (
class PolicyHierarchyTests(unittest.TestCase):
def test_privacy_policy_parent_scope_ids_follow_scope_precedence(self) -> None:
self.assertEqual(
"tenant-1",
_parent_privacy_policy_scope_id(tenant_id="tenant-1", scope_type="tenant", scope_id=None),
)
self.assertIsNone(_parent_privacy_policy_scope_id(tenant_id="tenant-1", scope_type="user", scope_id=None))
self.assertIsNone(_parent_privacy_policy_scope_id(tenant_id="tenant-1", scope_type="group", scope_id=None))
self.assertEqual(
"campaign-1",
_parent_privacy_policy_scope_id(tenant_id="tenant-1", scope_type="campaign", scope_id="campaign-1"),
)
def test_privacy_policy_patch_payload_validates_and_removes_unset_fields(self) -> None:
payload = _privacy_policy_patch_payload({"audit_detail_level": "minimal", "generated_eml_retention_days": None})
self.assertEqual({"audit_detail_level": "minimal"}, payload)
def test_privacy_policy_models_extend_shared_schema_contract(self) -> None:
policy = PrivacyRetentionPolicy.model_validate({"generated_eml_retention_days": ""})
patch = PrivacyRetentionPolicyPatch.model_validate({"allow_lower_level_limits": {"audit_detail_level": False}})
self.assertIsNone(policy.generated_eml_retention_days)
self.assertEqual({"audit_detail_level": False}, patch.allow_lower_level_limits)
def test_required_privacy_policy_scope_id_reports_missing_scope(self) -> None:
with self.assertRaises(PrivacyPolicyError) as captured:
_required_privacy_policy_scope_id("group", None)
self.assertEqual("Group privacy policy requires scope_id", str(captured.exception))
def test_parent_locks_block_lower_level_field_changes_and_limit_reenable(self) -> None:
issues = validate_hierarchical_policy_patch(
parent_policy={"retention_days": 30},

2
tests/test_policy_module_contract.py
View File

@@ -13,7 +13,7 @@ class PolicyModuleContractTests(unittest.TestCase):
project = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8"))["project"]
dependencies = tuple(project["dependencies"])
self.assertIn("govoplan-core>=0.1.6", dependencies)
self.assertTrue(any(item.startswith("govoplan-core>=") for item in dependencies))
self.assertFalse(any(item.startswith("govoplan-access") for item in dependencies))
def test_policy_source_does_not_import_access_implementation(self) -> None: