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

Add hierarchical policy simulation helpers

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-11 00:46:09 +02:00
parent bebb8a6e73
commit 423720229b
9 changed files with 394 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -24,6 +24,7 @@ from govoplan_policy.backend.retention import (
parent_privacy_policy,
parent_privacy_policy_sources,
set_privacy_policy_for_scope,
simulate_privacy_policy_change,
)
from .schemas import (
@@ -31,6 +32,7 @@ from .schemas import (
PrivacyRetentionPolicyItem,
PrivacyRetentionPolicyScopeRequest,
PrivacyRetentionPolicyScopeResponse,
PrivacyRetentionPolicySimulationResponse,
RetentionRunRequest,
RetentionRunResponse,
)
@@ -200,6 +202,32 @@ def explain_privacy_retention_policy(
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
@router.post("/privacy-retention/policies/{scope_type}/simulate", response_model=PrivacyRetentionPolicySimulationResponse)
def simulate_privacy_retention_policy(
scope_type: str,
payload: PrivacyRetentionPolicyScopeRequest,
scope_id: str | None = Query(default=None),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
):
clean_scope = scope_type.strip().casefold()
_require_privacy_policy_write(principal, clean_scope)
try:
return PrivacyRetentionPolicySimulationResponse(
scope_type=clean_scope,
scope_id=scope_id,
simulation=simulate_privacy_policy_change(
session,
tenant_id=principal.tenant_id,
scope_type=clean_scope,
scope_id=scope_id,
policy=payload.policy.model_dump(mode="json", exclude_none=True),
),
)
except PrivacyPolicyError as exc:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
def _blocked_privacy_retention_fields(parent) -> list[str]:
if parent is None:
return []

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

@@ -121,6 +121,12 @@ class PrivacyRetentionPolicyExplainResponse(BaseModel):
blocked_fields: list[str] = Field(default_factory=list)
class PrivacyRetentionPolicySimulationResponse(BaseModel):
scope_type: Literal["system", "tenant", "user", "group", "campaign"]
scope_id: str | None = None
simulation: dict[str, Any]
class RetentionRunRequest(BaseModel):
model_config = ConfigDict(extra="forbid")

147
src/govoplan_policy/backend/hierarchy.py Normal file
View File

@@ -0,0 +1,147 @@
from __future__ import annotations
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass, field
from typing import Any, Literal
from govoplan_core.core.policy import PolicyDecision, PolicySourceStep
PolicyIssueSeverity = Literal["blocker", "warning", "info"]
RestrictionCheck = Callable[[Any, Any], bool]
@dataclass(frozen=True, slots=True)
class PolicyValidationIssue:
code: str
message: str
field: str | None = None
severity: PolicyIssueSeverity = "blocker"
def to_dict(self) -> dict[str, object]:
payload: dict[str, object] = {
"severity": self.severity,
"code": self.code,
"message": self.message,
}
if self.field is not None:
payload["field"] = self.field
return payload
@dataclass(frozen=True, slots=True)
class PolicyRestrictionRule:
field: str
is_more_restrictive_or_equal: RestrictionCheck
less_restrictive_message: str
@dataclass(frozen=True, slots=True)
class PolicySimulationResult:
allowed: bool
changed_fields: tuple[str, ...]
issues: tuple[PolicyValidationIssue, ...] = ()
before_policy: Mapping[str, Any] = field(default_factory=dict)
requested_policy: Mapping[str, Any] = field(default_factory=dict)
decision: PolicyDecision | None = None
def to_dict(self) -> dict[str, object]:
return {
"allowed": self.allowed,
"changed_fields": list(self.changed_fields),
"issues": [issue.to_dict() for issue in self.issues],
"before_policy": dict(self.before_policy),
"requested_policy": dict(self.requested_policy),
"decision": self.decision.to_dict() if self.decision is not None else None,
}
def validate_hierarchical_policy_patch(
*,
parent_policy: Mapping[str, Any],
patch: Mapping[str, Any],
field_keys: Sequence[str],
parent_allow_lower_level_limits: Mapping[str, bool],
restriction_rules: Sequence[PolicyRestrictionRule] = (),
locked_field_message: Callable[[str], str] | None = None,
relock_message: Callable[[str], str] | None = None,
) -> tuple[PolicyValidationIssue, ...]:
issues: list[PolicyValidationIssue] = []
lock_message = locked_field_message or (lambda field: f"{field} is locked by the parent policy.")
reenable_message = relock_message or (lambda field: f"{field} limiting cannot be re-enabled below a parent policy lock.")
for field in field_keys:
if field in patch and patch.get(field) is not None and not parent_allow_lower_level_limits.get(field, True):
issues.append(PolicyValidationIssue(
code="field_locked_by_parent",
field=field,
message=lock_message(field),
))
patch_allow = patch.get("allow_lower_level_limits")
if isinstance(patch_allow, Mapping):
for field, allowed in patch_allow.items():
clean_field = str(field)
if bool(allowed) and not parent_allow_lower_level_limits.get(clean_field, True):
issues.append(PolicyValidationIssue(
code="lower_level_limit_locked_by_parent",
field=clean_field,
message=reenable_message(clean_field),
))
for rule in restriction_rules:
if rule.field not in patch or patch.get(rule.field) is None:
continue
parent_value = parent_policy.get(rule.field)
requested_value = patch.get(rule.field)
if not rule.is_more_restrictive_or_equal(parent_value, requested_value):
issues.append(PolicyValidationIssue(
code="less_restrictive_than_parent",
field=rule.field,
message=rule.less_restrictive_message,
))
return tuple(issues)
def simulate_hierarchical_policy_change(
*,
parent_policy: Mapping[str, Any],
current_policy: Mapping[str, Any],
patch: Mapping[str, Any],
field_keys: Sequence[str],
parent_allow_lower_level_limits: Mapping[str, bool],
restriction_rules: Sequence[PolicyRestrictionRule] = (),
source_path: Sequence[PolicySourceStep] = (),
locked_field_message: Callable[[str], str] | None = None,
relock_message: Callable[[str], str] | None = None,
) -> PolicySimulationResult:
issues = validate_hierarchical_policy_patch(
parent_policy=parent_policy,
patch=patch,
field_keys=field_keys,
parent_allow_lower_level_limits=parent_allow_lower_level_limits,
restriction_rules=restriction_rules,
locked_field_message=locked_field_message,
relock_message=relock_message,
)
changed_fields = tuple(
field
for field in (*field_keys, "allow_lower_level_limits")
if field in patch and current_policy.get(field) != patch.get(field)
)
allowed = not any(issue.severity == "blocker" for issue in issues)
decision = PolicyDecision(
allowed=allowed,
reason=None if allowed else "Requested policy change is blocked by parent policy constraints.",
source_path=tuple(source_path),
requirements=tuple(issue.field or issue.code for issue in issues if issue.severity == "blocker"),
details={"issues": [issue.to_dict() for issue in issues], "changed_fields": list(changed_fields)},
)
return PolicySimulationResult(
allowed=allowed,
changed_fields=changed_fields,
issues=issues,
before_policy=dict(current_policy),
requested_policy=dict(patch),
decision=decision,
)

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

@@ -23,11 +23,16 @@ from govoplan_core.core.access import (
AuditRecordRef,
AuditRetentionProvider,
)
from govoplan_core.core.policy import policy_source_step
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.settings import settings
from govoplan_core.tenancy.scope import Tenant
from govoplan_policy.backend.hierarchy import (
PolicyRestrictionRule,
simulate_hierarchical_policy_change,
validate_hierarchical_policy_patch,
)
PRIVACY_POLICY_SETTINGS_KEY = "privacy_retention_policy"
RETENTION_DAY_KEYS = (
@@ -277,26 +282,46 @@ def _merge_privacy_policy(parent: PrivacyRetentionPolicy, patch: dict[str, Any])
return PrivacyRetentionPolicy.model_validate(payload)
def _parent_allow_lower_level_limits(parent_payload: dict[str, Any]) -> dict[str, bool]:
return {**default_allow_lower_level_limits(), **(parent_payload.get("allow_lower_level_limits") or {})}
def _privacy_restriction_rules() -> tuple[PolicyRestrictionRule, ...]:
return (
PolicyRestrictionRule(
field="store_raw_campaign_json",
is_more_restrictive_or_equal=lambda parent_value, requested_value: not (requested_value is True and parent_value is False),
less_restrictive_message="Raw campaign JSON storage cannot be re-enabled below a parent policy that disables it.",
),
*(
PolicyRestrictionRule(
field=key,
is_more_restrictive_or_equal=lambda parent_value, requested_value: parent_value is None or int(requested_value) <= int(parent_value),
less_restrictive_message=f"{key} cannot be less restrictive than the parent retention policy.",
)
for key in RETENTION_DAY_KEYS
),
PolicyRestrictionRule(
field="audit_detail_level",
is_more_restrictive_or_equal=lambda parent_value, requested_value: AUDIT_DETAIL_LEVEL_ORDER[str(requested_value)] >= AUDIT_DETAIL_LEVEL_ORDER[str(parent_value or "full")],
less_restrictive_message="Audit detail level cannot be less restrictive than the parent retention policy.",
),
)
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.")
issues = validate_hierarchical_policy_patch(
parent_policy=parent_payload,
patch=patch,
field_keys=RETENTION_POLICY_FIELD_KEYS,
parent_allow_lower_level_limits=_parent_allow_lower_level_limits(parent_payload),
restriction_rules=_privacy_restriction_rules(),
locked_field_message=lambda key: f"{key} is locked by the parent retention policy.",
relock_message=lambda key: f"{key} limiting cannot be re-enabled below a parent retention policy lock.",
)
if issues:
raise PrivacyPolicyError(issues[0].message)
def _set_settings_privacy_policy(settings_payload: dict[str, Any] | None, policy: dict[str, Any]) -> dict[str, Any]:
@@ -542,6 +567,51 @@ def set_privacy_policy_for_scope(
raise PrivacyPolicyError("Privacy policy scope must be system, tenant, user, group or campaign")
def simulate_privacy_policy_change(
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()
patch = PrivacyRetentionPolicyPatch.model_validate(policy or {}).model_dump(mode="json", exclude_none=True)
current = get_privacy_policy_for_scope(session, tenant_id=tenant_id, scope_type=clean_scope, scope_id=scope_id)
if clean_scope == "system":
parent_payload = PrivacyRetentionPolicy.model_validate(current).model_dump(mode="json")
parent_allow = default_allow_lower_level_limits()
parent_sources = effective_privacy_policy_sources(session)
else:
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),
)
parent_payload = parent.model_dump(mode="json")
parent_allow = _parent_allow_lower_level_limits(parent_payload)
parent_sources = parent_privacy_policy_sources(
session,
tenant_id=tenant_id,
scope_type=clean_scope,
scope_id=scope_id or (tenant_id if clean_scope == "tenant" else None),
)
simulation = simulate_hierarchical_policy_change(
parent_policy=parent_payload,
current_policy=current,
patch=patch,
field_keys=RETENTION_POLICY_FIELD_KEYS,
parent_allow_lower_level_limits=parent_allow,
restriction_rules=() if clean_scope == "system" else _privacy_restriction_rules(),
source_path=[PolicySourceStep.from_mapping(source) for source in parent_sources],
locked_field_message=lambda key: f"{key} is locked by the parent retention policy.",
relock_message=lambda key: f"{key} limiting cannot be re-enabled below a parent retention policy lock.",
)
return simulation.to_dict()
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":
@@ -723,6 +793,9 @@ class SqlPrivacyRetentionService:
def set_privacy_policy_for_scope(self, *args: Any, **kwargs: Any) -> Any:
return set_privacy_policy_for_scope(*args, **kwargs)
def simulate_privacy_policy_change(self, *args: Any, **kwargs: Any) -> Any:
return simulate_privacy_policy_change(*args, **kwargs)
def sanitize_audit_details_for_policy(self, *args: Any, **kwargs: Any) -> Any:
return sanitize_audit_details_for_policy(*args, **kwargs)