diff --git a/README.md b/README.md index 54e3741..e8d3963 100644 --- a/README.md +++ b/README.md @@ -9,3 +9,8 @@ implementation while route ownership is separated before model migration. Policy decision and provenance payloads use the shared kernel DTOs documented in [docs/POLICY_DECISION_PROVENANCE.md](docs/POLICY_DECISION_PROVENANCE.md) and `/mnt/DATA/git/govoplan-core/docs/POLICY_CONTRACTS.md`. + +Hierarchical policy evaluation, delegation ceilings, and write simulations are +implemented in `govoplan_policy.backend.hierarchy`. Privacy retention uses that +shared helper and exposes `/api/v1/admin/privacy-retention/policies/{scope}/simulate` +for preflight checks before saving lower-level policy changes. diff --git a/docs/POLICY_DECISION_PROVENANCE.md b/docs/POLICY_DECISION_PROVENANCE.md index 9b44f1b..c64c7d9 100644 --- a/docs/POLICY_DECISION_PROVENANCE.md +++ b/docs/POLICY_DECISION_PROVENANCE.md @@ -9,6 +9,13 @@ Privacy retention implementation lives in this module at 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. + +Reusable hierarchical policy validation lives in +`govoplan_policy.backend.hierarchy`. Policy families should use that helper for +parent locks, lower-level override ceilings, "more restrictive only" checks, +and read-only simulations before destructive or limiting changes are saved. +Domain modules keep their own policy fields and restriction rules, but the +decision shape and simulation payload stay consistent. When retention needs audit-log storage behavior, it requests the `audit.retention` capability; it does not import audit module tables or providers directly. @@ -47,6 +54,18 @@ GET /api/v1/admin/privacy-retention/policies/{scope_type}/explain The response contains `decision`, `effective_policy`, `parent_policy`, `effective_policy_sources`, `parent_policy_sources`, and `blocked_fields`. + +Retention policy also exposes a write-preflight endpoint: + +```text +POST /api/v1/admin/privacy-retention/policies/{scope}/simulate +``` + +The request body is the same as the write endpoint. The response contains a +`simulation` object with `allowed`, `changed_fields`, `issues`, +`before_policy`, `requested_policy`, and a shared `PolicyDecision` payload. +The endpoint never writes policy state and is intended for UI validation before +operators attempt destructive or limiting changes. Clients can use `blocked_fields` to disable controls before a save attempt. ## UI Expectations diff --git a/pyproject.toml b/pyproject.toml index cfa7029..412aed1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,6 @@ requires-python = ">=3.12" authors = [{ name = "GovOPlaN" }] dependencies = [ "govoplan-core>=0.1.6", - "govoplan-access>=0.1.6", ] [tool.setuptools.packages.find] @@ -22,4 +21,3 @@ govoplan_policy = ["py.typed"] [project.entry-points."govoplan.modules"] policy = "govoplan_policy.backend.manifest:get_manifest" - diff --git a/src/govoplan_policy/backend/api/v1/routes.py b/src/govoplan_policy/backend/api/v1/routes.py index 686662f..21d4ff0 100644 --- a/src/govoplan_policy/backend/api/v1/routes.py +++ b/src/govoplan_policy/backend/api/v1/routes.py @@ -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 [] diff --git a/src/govoplan_policy/backend/api/v1/schemas.py b/src/govoplan_policy/backend/api/v1/schemas.py index 039d5ea..1f4834b 100644 --- a/src/govoplan_policy/backend/api/v1/schemas.py +++ b/src/govoplan_policy/backend/api/v1/schemas.py @@ -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") diff --git a/src/govoplan_policy/backend/hierarchy.py b/src/govoplan_policy/backend/hierarchy.py new file mode 100644 index 0000000..21f20e1 --- /dev/null +++ b/src/govoplan_policy/backend/hierarchy.py @@ -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, + ) diff --git a/src/govoplan_policy/backend/retention.py b/src/govoplan_policy/backend/retention.py index b60f63e..d493584 100644 --- a/src/govoplan_policy/backend/retention.py +++ b/src/govoplan_policy/backend/retention.py @@ -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) diff --git a/tests/test_policy_hierarchy.py b/tests/test_policy_hierarchy.py new file mode 100644 index 0000000..49b7030 --- /dev/null +++ b/tests/test_policy_hierarchy.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import unittest + +from govoplan_policy.backend.hierarchy import ( + PolicyRestrictionRule, + simulate_hierarchical_policy_change, + validate_hierarchical_policy_patch, +) + + +class PolicyHierarchyTests(unittest.TestCase): + def test_parent_locks_block_lower_level_field_changes_and_limit_reenable(self) -> None: + issues = validate_hierarchical_policy_patch( + parent_policy={"retention_days": 30}, + patch={"retention_days": 20, "allow_lower_level_limits": {"retention_days": True}}, + field_keys=("retention_days",), + parent_allow_lower_level_limits={"retention_days": False}, + ) + + self.assertEqual(["field_locked_by_parent", "lower_level_limit_locked_by_parent"], [issue.code for issue in issues]) + self.assertEqual(["retention_days", "retention_days"], [issue.field for issue in issues]) + + def test_restriction_rules_block_less_restrictive_patch(self) -> None: + issues = validate_hierarchical_policy_patch( + parent_policy={"retention_days": 30}, + patch={"retention_days": 45}, + field_keys=("retention_days",), + parent_allow_lower_level_limits={"retention_days": True}, + restriction_rules=( + PolicyRestrictionRule( + field="retention_days", + is_more_restrictive_or_equal=lambda parent, requested: int(requested) <= int(parent), + less_restrictive_message="retention_days cannot be widened", + ), + ), + ) + + self.assertEqual(1, len(issues)) + self.assertEqual("less_restrictive_than_parent", issues[0].code) + self.assertEqual("retention_days cannot be widened", issues[0].message) + + def test_policy_simulation_returns_decision_payload(self) -> None: + simulation = simulate_hierarchical_policy_change( + parent_policy={"retention_days": 30}, + current_policy={"retention_days": 30}, + patch={"retention_days": 45}, + field_keys=("retention_days",), + parent_allow_lower_level_limits={"retention_days": True}, + restriction_rules=( + PolicyRestrictionRule( + field="retention_days", + is_more_restrictive_or_equal=lambda parent, requested: int(requested) <= int(parent), + less_restrictive_message="retention_days cannot be widened", + ), + ), + ) + + self.assertFalse(simulation.allowed) + self.assertEqual(("retention_days",), simulation.changed_fields) + self.assertIsNotNone(simulation.decision) + self.assertEqual(("retention_days",), simulation.decision.requirements) + self.assertEqual("Requested policy change is blocked by parent policy constraints.", simulation.decision.reason) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_policy_module_contract.py b/tests/test_policy_module_contract.py new file mode 100644 index 0000000..d15e711 --- /dev/null +++ b/tests/test_policy_module_contract.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +import pathlib +import tomllib +import unittest + + +ROOT = pathlib.Path(__file__).resolve().parents[1] + + +class PolicyModuleContractTests(unittest.TestCase): + def test_policy_package_does_not_hard_require_access(self) -> None: + 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.assertFalse(any(item.startswith("govoplan-access") for item in dependencies)) + + def test_policy_source_does_not_import_access_implementation(self) -> None: + offenders: list[str] = [] + for path in (ROOT / "src" / "govoplan_policy").rglob("*.py"): + source = path.read_text(encoding="utf-8") + if "govoplan_access" in source: + offenders.append(str(path.relative_to(ROOT))) + + self.assertEqual([], offenders) + + +if __name__ == "__main__": + unittest.main()