generated from coulomb/repo-seed
Access controlled knowledge gateway functionality
This commit is contained in:
65
src/markitect_tool/policy/adapters.py
Normal file
65
src/markitect_tool/policy/adapters.py
Normal file
@@ -0,0 +1,65 @@
|
||||
"""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, [], {}, "")
|
||||
}
|
||||
Reference in New Issue
Block a user