markitect-tool/src/markitect_tool/policy/adapters.py

66 lines
1.8 KiB
Python
Raw Normal View History

"""Protocol boundaries for external authorization engines."""
from __future__ import annotations
from dataclasses import asdict, dataclass, field
from typing import Any, Protocol
from markitect_tool.policy.models import PolicyDecision
@dataclass(frozen=True)
class RelationshipPolicyRequest:
"""Relationship-based authorization request.
This maps cleanly to Zanzibar/OpenFGA/SpiceDB-style checks without binding
Markitect core to one service or tuple schema.
"""
subject: str
relation: str
object_id: str
namespace: str | None = None
context: dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> dict[str, Any]:
return _drop_empty(asdict(self))
class RelationshipPolicyAdapter(Protocol):
"""Adapter boundary for relationship authorization systems."""
def check(self, request: RelationshipPolicyRequest) -> PolicyDecision | dict[str, Any]:
"""Return a policy decision for a relationship check."""
@dataclass(frozen=True)
class RulePolicyRequest:
"""Attribute/rule policy evaluation request.
This can be mapped to OPA/Rego, Cedar, or local policy-as-data engines.
"""
subject: dict[str, Any]
action: str
object: dict[str, Any]
context: dict[str, Any] = field(default_factory=dict)
policy_id: str | None = None
def to_dict(self) -> dict[str, Any]:
return _drop_empty(asdict(self))
class RulePolicyAdapter(Protocol):
"""Adapter boundary for rule/attribute policy systems."""
def evaluate(self, request: RulePolicyRequest) -> PolicyDecision | dict[str, Any]:
"""Return a policy decision for a rule evaluation."""
def _drop_empty(data: dict[str, Any]) -> dict[str, Any]:
return {
key: value
for key, value in data.items()
if value not in (None, [], {}, "")
}