"""Python policy lifecycle objects modeled after ``dogwood_language``.
The high-level SDK follows the Rust workflow:
``ServiceSchema + PolicySchema -> ParsedPolicySet -> LoweredPolicySet -> Validator``.
When a non-empty Cedar action schema is supplied, operations delegate to the
native PyO3 binding and therefore to Rust ``dogwood_language``. The pure-Python
path is a temporary schema-less fallback for simple examples.
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from typing import Any
from .errors import ParseError
from . import native
from .values import Decision, Diagnostics, DogwoodRuleRef, Response
[docs]
@dataclass(frozen=True)
class ServiceSchema:
"""Dogwood service schema inputs.
Rust mapping: ``dogwood_language::ServiceSchema``. In Rust this is built
with ``ServiceSchema::defaults()`` or ``ServiceSchema::builder()`` and can
contain an event-schema DSL, provider declarations, and macro source.
In dogwood-py, ``event_schema`` is passed through to the native binding as
``ServiceSchema::builder().event_schema_str(...).build()``. ``macros`` and
``providers`` are reserved for future binding support.
"""
event_schema: str | None = None
macros: str | None = None
providers: dict[str, Any] | None = None
[docs]
@classmethod
def defaults(cls) -> "ServiceSchema":
"""Return the default Dogwood service schema.
Rust mapping: ``ServiceSchema::defaults()``.
"""
return cls()
[docs]
@dataclass(frozen=True)
class PolicySchema:
"""Cedar action schema text used to lower Dogwood policies.
Rust mapping: ``dogwood_language::PolicySchema`` constructed with
``PolicySchema::from_cedarschema_str``.
"""
source: str
[docs]
@classmethod
def from_cedarschema_str(cls, source: str) -> "PolicySchema":
"""Create a policy schema from Cedar ``.cedarschema`` source.
Rust mapping: ``PolicySchema::from_cedarschema_str(source)``.
"""
return cls(source)
[docs]
@dataclass(frozen=True)
class ParsedPolicy:
"""Schema-free parsed policy summary.
Rust mapping: one policy inside ``dogwood_language::ParsedPolicySet``. The
Python fallback stores only a lightweight summary; native schema-backed
parsing/lowering is handled by Rust.
"""
id: str
effect: str
action: str | None
when: tuple[str, ...] = ()
unless: tuple[str, ...] = ()
temporal: tuple[str, ...] = ()
index: int = 0
[docs]
def uses_temporal(self) -> bool:
return bool(self.temporal)
[docs]
def uses_providers(self) -> bool:
return False
[docs]
@dataclass(frozen=True)
class ParsedPolicySet:
"""Dogwood policy source after schema-free parse.
Rust mapping: ``dogwood_language::ParsedPolicySet``. The Rust split is
``ParsedPolicySet::parse(source, &service_schema)`` followed later by
``ParsedPolicySet::lower(&policy_schema)`` when the Cedar action schema is
available.
"""
source: str
service_schema: ServiceSchema
_policies: tuple[ParsedPolicy, ...]
[docs]
@classmethod
def parse(cls, source: str, service_schema: ServiceSchema | None = None) -> "ParsedPolicySet":
"""Parse policy source without a Cedar action schema.
Rust mapping: ``ParsedPolicySet::parse(source, &service_schema)``.
"""
service_schema = service_schema or ServiceSchema.defaults()
return cls(source, service_schema, tuple(_parse_policies(source)))
[docs]
def lower(self, policy_schema: PolicySchema) -> "LoweredPolicySet":
"""Lower this parsed policy set against a Cedar action schema.
Rust mapping: ``ParsedPolicySet::lower(&policy_schema)``.
"""
return LoweredPolicySet(self, policy_schema)
[docs]
def lower_with_distincter(self, policy_schema: PolicySchema, distincter: str) -> "LoweredPolicySet":
"""Lower with a caller-supplied Cedar identifier namespace.
Rust mapping: ``ParsedPolicySet::lower_with_distincter``. The current
Python fallback uses ``distincter`` for generated fallback Cedar ids;
native distincter support is not yet exposed through PyO3.
"""
if not re.match(r"^[_A-Za-z][_A-Za-z0-9]*$", distincter):
raise ParseError(f"invalid distincter {distincter!r}")
return LoweredPolicySet(self, policy_schema, distincter=distincter)
[docs]
def policy_count(self) -> int:
return len(self._policies)
[docs]
def policies(self) -> tuple[ParsedPolicy, ...]:
return self._policies
[docs]
@dataclass(frozen=True)
class ValidationResult:
"""Validation findings for a lowered policy set.
Rust mapping: ``dogwood_language::ValidationResult``. Errors make
``validation_passed`` false; warnings are reported separately.
"""
errors: tuple[str, ...] = ()
warnings: tuple[str, ...] = ()
[docs]
def validation_passed(self) -> bool:
"""Return true when validation has no errors.
Rust mapping: ``ValidationResult::validation_passed()``.
"""
return not self.errors
[docs]
class Validator:
"""Validate a lowered Dogwood policy set.
Rust mapping: ``dogwood_language::Validator``. Unlike Cedar's validator,
Dogwood's Rust ``Validator::new()`` takes no schema because the effective
augmented schema travels on ``LoweredPolicySet``.
"""
[docs]
def validate(self, policies: "LoweredPolicySet") -> ValidationResult:
"""Validate policies and return accumulated errors/warnings.
Rust mapping: ``Validator::new().validate(&policies)``.
"""
if policies.policy_schema.source.strip():
native.require_available()
result = native.validate_policy(
policies.source,
policies.policy_schema.source,
policies.parsed.service_schema.event_schema,
)
return ValidationResult(tuple(result["errors"]), tuple(result["warnings"]))
errors: list[str] = []
if not policies.parsed._policies:
errors.append("policy set is empty")
for policy in policies.parsed._policies:
if policy.effect not in {"permit", "forbid"}:
errors.append(f"policy {policy.id}: unsupported effect {policy.effect}")
return ValidationResult(tuple(errors))
[docs]
@dataclass
class LoweredPolicySet:
"""Policy set lowered against Dogwood service and Cedar action schemas.
Rust mapping: ``dogwood_language::LoweredPolicySet``. For schema-backed
workflows, construction delegates to Rust ``LoweredPolicySet::from_str``.
The lowered set is schema-bound; validation and Cedar export use the same
schemas it was lowered against.
"""
parsed: ParsedPolicySet
policy_schema: PolicySchema
distincter: str = "policy"
cedar_policies: str = field(init=False)
source: str = field(init=False)
def __post_init__(self) -> None:
object.__setattr__(self, "source", self.parsed.source)
cedar = self._render_cedar()
if self.policy_schema.source.strip():
native.require_available()
cedar = native.lower_to_cedar(
self.parsed.source,
self.policy_schema.source,
self.parsed.service_schema.event_schema,
)
object.__setattr__(self, "cedar_policies", cedar)
[docs]
@classmethod
def from_str(
cls,
source: str,
service_schema: ServiceSchema | None = None,
policy_schema: PolicySchema | None = None,
) -> "LoweredPolicySet":
"""Parse and lower policy source in one step.
Rust mapping: ``LoweredPolicySet::from_str(source, &service_schema,
&policy_schema)``. This is the fused parse/lower form.
"""
parsed = ParsedPolicySet.parse(source, service_schema or ServiceSchema.defaults())
return parsed.lower(policy_schema or PolicySchema(""))
[docs]
def as_cedar(self) -> str:
"""Return lowered Cedar policy text.
Rust mapping: ``LoweredPolicySet::as_cedar()`` rendered to text.
"""
return self.cedar_policies
[docs]
def cedar_schema(self) -> str:
"""Return the augmented Cedar schema text.
Rust mapping: ``LoweredPolicySet::cedar_schema_str()``.
"""
if self.policy_schema.source.strip():
native.require_available()
return native.cedar_schema(
self.source,
self.policy_schema.source,
self.parsed.service_schema.event_schema,
)
return self.policy_schema.source
[docs]
def is_self_contained_cedar(self) -> bool:
"""Return whether exported Cedar is self-contained.
Rust mapping: ``LoweredPolicySet::is_self_contained_cedar()``. Native
support is not fully exposed yet; the Python fallback approximates this
by checking whether parsed policies contain temporal clauses.
"""
return not any(policy.temporal for policy in self.parsed._policies)
[docs]
def decide(self, event: Any, history: list[Any]) -> Response:
matched_forbid: list[DogwoodRuleRef] = []
matched_permit: list[DogwoodRuleRef] = []
errors: list[str] = []
for policy in self.parsed._policies:
try:
if _policy_matches(policy, event, history):
ref = DogwoodRuleRef(policy.index, f"{self.distincter}{policy.index}")
if policy.effect == "forbid":
matched_forbid.append(ref)
else:
matched_permit.append(ref)
except Exception as exc:
errors.append(f"policy {policy.id}: {exc}")
if errors or matched_forbid:
return Response(Decision.DENY, Diagnostics(tuple(matched_forbid), tuple(errors)))
if matched_permit:
return Response(Decision.ALLOW, Diagnostics(tuple(matched_permit), ()))
return Response(Decision.DENY, Diagnostics())
def _render_cedar(self) -> str:
lines = []
for policy in self.parsed._policies:
clauses = list(policy.when)
clauses.extend(f"!({expr})" for expr in policy.unless)
if policy.temporal:
clauses.extend(f"context.{self.distincter}_{policy.index}__temporal_{i}" for i, _ in enumerate(policy.temporal))
guard = " && ".join(clauses) if clauses else "true"
action = f"action == {policy.action}" if policy.action else "action"
lines.append(
f'@id("{policy.id}")\n{policy.effect}(principal, {action}, resource) when {{ {guard} }};'
)
return "\n".join(lines)
def _parse_policies(source: str) -> list[ParsedPolicy]:
clean = re.sub(r"//.*", "", source)
chunks = [chunk.strip() for chunk in clean.split(";") if chunk.strip()]
policies: list[ParsedPolicy] = []
for index, chunk in enumerate(chunks):
policy_id = f"policy{index}"
id_match = re.search(r'@id\s*\(\s*"([^"]+)"\s*\)', chunk)
if id_match:
policy_id = id_match.group(1)
chunk = chunk[: id_match.start()] + chunk[id_match.end() :]
effect_match = re.search(r"\b(permit|forbid)\s*\((.*?)\)", chunk, re.S)
if not effect_match:
raise ParseError(f"could not parse policy {index}")
effect = effect_match.group(1)
scope = " ".join(effect_match.group(2).split())
action_match = re.search(r"action\s*==\s*([^,\)]+)", scope)
action = action_match.group(1).strip() if action_match else None
tail = chunk[effect_match.end() :]
when: list[str] = []
unless: list[str] = []
temporal: list[str] = []
for kind, body in _extract_clauses(tail):
if kind == "when temporal":
temporal.append(body)
elif kind == "when guardrail":
when.append(body)
elif kind == "when":
when.append(body)
elif kind == "unless":
unless.append(body)
policies.append(ParsedPolicy(policy_id, effect, action, tuple(when), tuple(unless), tuple(temporal), index))
return policies
def _extract_clauses(text: str) -> list[tuple[str, str]]:
clauses: list[tuple[str, str]] = []
i = 0
while i < len(text):
match = re.search(r"\b(when\s+temporal|when\s+guardrail|when|unless)\s*\{", text[i:])
if not match:
break
kind = " ".join(match.group(1).split())
start = i + match.end()
depth = 1
j = start
in_string = False
while j < len(text):
ch = text[j]
if ch == '"' and (j == 0 or text[j - 1] != "\\"):
in_string = not in_string
elif not in_string and ch == "{":
depth += 1
elif not in_string and ch == "}":
depth -= 1
if depth == 0:
clauses.append((kind, text[start:j].strip()))
i = j + 1
break
j += 1
else:
raise ParseError("unclosed policy clause")
return clauses
def _policy_matches(policy: ParsedPolicy, event: Any, history: list[Any]) -> bool:
if policy.action and _normalize_action(policy.action) != event.action:
return False
for expr in policy.when:
if not _eval_expr(expr, event):
return False
for expr in policy.unless:
if _eval_expr(expr, event):
return False
for expr in policy.temporal:
if not _eval_temporal(expr, event, history):
return False
return True
def _eval_temporal(expr: str, event: Any, history: list[Any]) -> bool:
match = re.search(
r"formerly(?:\s+within\s+(\d+)([smhd]))?\s+(.+?)::([A-Za-z_]\w*)\s*(?:\{(.*)\})?\s*$",
" ".join(expr.split()),
)
if not match:
raise ValueError(f"unsupported temporal expression: {expr}")
amount, unit, action, kind, pins = match.groups()
window = None if amount is None else int(amount) * {"s": 1, "m": 60, "h": 3600, "d": 86400}[unit]
wanted_action = _normalize_action(action)
pin_pairs = _parse_pins(pins or "")
for past in reversed(history):
if past.action != wanted_action or past.kind != kind:
continue
if window is not None and event.timestamp() - past.timestamp() > window:
continue
if all(past.field_path(left.split(".")) == _resolve_ref(right, event) for left, right in pin_pairs):
return True
return False
def _parse_pins(text: str) -> list[tuple[str, str]]:
if not text.strip():
return []
out = []
for item in _split_top_level(text, ","):
if ":" not in item:
continue
left, right = item.split(":", 1)
out.append((left.strip(), right.strip()))
return out
def _eval_expr(expr: str, event: Any) -> bool:
expr = expr.strip()
if not expr:
return True
or_parts = _split_operator(expr, "||")
if len(or_parts) > 1:
return any(_eval_expr(part, event) for part in or_parts)
and_parts = _split_operator(expr, "&&")
if len(and_parts) > 1:
return all(_eval_expr(part, event) for part in and_parts)
if expr.startswith("!"):
return not _eval_expr(expr[1:].strip(), event)
if expr.startswith("(") and expr.endswith(")"):
return _eval_expr(expr[1:-1], event)
for op in ("<=", ">=", "==", "!=", "<", ">"):
parts = _split_operator(expr, op)
if len(parts) == 2:
left = _resolve_ref(parts[0].strip(), event)
right = _literal(parts[1].strip(), event)
return _compare(left, op, right)
value = _literal(expr, event)
return bool(value)
def _resolve_ref(text: str, event: Any) -> Any:
text = text.strip()
if text.startswith("context."):
return event.request_context_path(text.removeprefix("context.").split("."))
if text.startswith("principal."):
entity = event.scope_principal
return getattr(entity, text.removeprefix("principal."), None) if entity else None
if text.startswith("resource."):
entity = event.scope_resource
return getattr(entity, text.removeprefix("resource."), None) if entity else None
return _literal(text, event)
def _literal(text: str, event: Any) -> Any:
text = text.strip()
if text.startswith("context.") or text.startswith("principal.") or text.startswith("resource."):
return _resolve_ref(text, event)
if text == "true":
return True
if text == "false":
return False
if text.startswith('"') and text.endswith('"'):
return text[1:-1]
if re.match(r"^-?\d+$", text):
return int(text)
if re.match(r"^-?\d+\.\d+$", text):
return float(text)
return _normalize_action(text)
def _compare(left: Any, op: str, right: Any) -> bool:
if op == "==":
return left == right
if op == "!=":
return left != right
if left is None or right is None:
return False
if op == "<":
return left < right
if op == ">":
return left > right
if op == "<=":
return left <= right
if op == ">=":
return left >= right
raise ValueError(op)
def _split_operator(expr: str, op: str) -> list[str]:
return _split_top_level(expr, op)
def _split_top_level(expr: str, sep: str) -> list[str]:
parts: list[str] = []
depth = 0
in_string = False
start = 0
i = 0
while i < len(expr):
ch = expr[i]
if ch == '"' and (i == 0 or expr[i - 1] != "\\"):
in_string = not in_string
elif not in_string:
if ch in "({[":
depth += 1
elif ch in ")}]":
depth -= 1
elif depth == 0 and expr.startswith(sep, i):
parts.append(expr[start:i].strip())
i += len(sep)
start = i
continue
i += 1
parts.append(expr[start:].strip())
return parts
def _normalize_action(action: str) -> str:
return action.strip().replace('::"', "::").replace('"', "")