148 lines
5.5 KiB
Python
148 lines
5.5 KiB
Python
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,
|
|
)
|