Files
kontextual-engine/src/kontextual_engine/core/policy.py

93 lines
2.8 KiB
Python

"""Policy decision primitives for permission-aware operations."""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
from .primitives import compact_dict, new_id, utc_now
class PolicyEffect(str, Enum):
ALLOW = "allow"
DENY = "deny"
REDACT = "redact"
REQUIRE_REVIEW = "require_review"
DRY_RUN_ONLY = "dry_run_only"
FAIL_CLOSED = "fail_closed"
@dataclass(frozen=True)
class PolicyDecision:
effect: PolicyEffect
subject_id: str
action: str
resource: str
reason: str = ""
decision_id: str = field(default_factory=lambda: new_id("policy"))
obligations: dict[str, Any] = field(default_factory=dict)
context: dict[str, Any] = field(default_factory=dict)
decided_at: str = field(default_factory=lambda: utc_now().isoformat())
@classmethod
def allow(cls, subject_id: str, action: str, resource: str, **kwargs: Any) -> "PolicyDecision":
return cls(PolicyEffect.ALLOW, subject_id, action, resource, **kwargs)
@classmethod
def deny(
cls,
subject_id: str,
action: str,
resource: str,
*,
reason: str,
**kwargs: Any,
) -> "PolicyDecision":
return cls(PolicyEffect.DENY, subject_id, action, resource, reason=reason, **kwargs)
@classmethod
def fail_closed(
cls,
subject_id: str,
action: str,
resource: str,
*,
reason: str = "Permission context is missing or ambiguous",
**kwargs: Any,
) -> "PolicyDecision":
return cls(PolicyEffect.FAIL_CLOSED, subject_id, action, resource, reason=reason, **kwargs)
@property
def allowed(self) -> bool:
return self.effect == PolicyEffect.ALLOW
def to_dict(self) -> dict[str, Any]:
return compact_dict(
{
"decision_id": self.decision_id,
"effect": self.effect.value,
"subject_id": self.subject_id,
"action": self.action,
"resource": self.resource,
"reason": self.reason,
"obligations": dict(self.obligations),
"context": dict(self.context),
"decided_at": self.decided_at,
}
)
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "PolicyDecision":
return cls(
decision_id=data["decision_id"],
effect=PolicyEffect(data["effect"]),
subject_id=data["subject_id"],
action=data["action"],
resource=data["resource"],
reason=data.get("reason", ""),
obligations=dict(data.get("obligations", {})),
context=dict(data.get("context", {})),
decided_at=data["decided_at"],
)