diff --git a/CHANGELOG.md b/CHANGELOG.md index f00f662..324e168 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,42 @@ All notable changes to this project will be documented in this file. This project adheres to [Semantic Versioning 2.0.0](https://semver.org/spec/v2.0.0.html). The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). +## 1.18.0 — 2026-04-12 + +**Added** + +* **Executable obligation handlers** — `Guard.register_obligation_handler(type, handler)` + lets callers register sync or async callbacks that are invoked automatically + after a `permit` decision carrying the matching obligation type. + + ```python + from rbacx.core.engine import ObligationNotMetError + + def check_mfa(decision, context): + if not context.attrs.get("mfa"): + raise ObligationNotMetError("MFA required", challenge="mfa") + + guard.register_obligation_handler("require_mfa", check_mfa) + ``` + + Key properties: + + * Handlers run only for `permit` decisions, in obligation order. + * `ObligationNotMetError` flips the decision to `deny` with + `reason="obligation_failed"`; the optional `challenge` attribute is + propagated to `Decision.challenge`. + * Any other exception is logged and also flips to `deny` (fail-closed). + * The first failing handler short-circuits — subsequent handlers are skipped. + * Conditional obligations (`condition` field) are respected — the handler is + skipped when the condition evaluates to `False`. + * Unregistered obligation types remain in `Decision.obligations` for manual + handling (backward-compatible). + * Registering a handler for an existing type replaces the previous one. + +* **`rbacx.core.engine.ObligationNotMetError`** — new public exception + raised by obligation handlers to signal that an obligation was not met. + Accepts an optional `challenge` keyword argument. + ## 1.17.0 — 2026-04-05 **Added** diff --git a/docs/api.md b/docs/api.md index f8645bb..ca81781 100644 --- a/docs/api.md +++ b/docs/api.md @@ -104,10 +104,26 @@ --- -## Decision explanation / trace +## Decision object + +Fields returned by all `Guard.evaluate*` methods: + +| Field | Type | Description | +|---|---|---| +| `allowed` | `bool` | Whether access is granted | +| `effect` | `str` | Declared effect: `"permit"` or `"deny"` | +| `obligations` | `List[Dict]` | Obligations from the matched rule | +| `challenge` | `str \| None` | Machine-readable auth challenge (e.g. `"mfa"`) | +| `rule_id` | `str \| None` | ID of the matched rule | +| `policy_id` | `str \| None` | ID of the matched policy (policy sets only) | +| `reason` | `str \| None` | Why the decision was made — see [Reasons](reasons.md) | +| `trace` | `List[RuleTrace] \| None` | Per-rule evaluation log; `None` unless `explain=True` | -Pass `explain=True` to any evaluation method to get a per-rule evaluation log -attached to the returned `Decision`. +--- + +## Decision trace (`explain=True`) + +Pass `explain=True` to any evaluation method to get a per-rule evaluation log: ```python d = guard.evaluate_sync(subject, action, resource, context, explain=True) @@ -117,88 +133,54 @@ for entry in d.trace: print(f" rule {entry.rule_id!r} [{entry.effect}] → {status}") ``` -When `explain=False` (the default) `Decision.trace` is `None` — there is no -overhead on the hot path. - -**`RuleTrace` fields** - -| Field | Type | Description | -|---|---|---| -| `rule_id` | `str` | The `id` field of the rule as declared in the policy | -| `effect` | `str` | Declared effect: `"permit"` or `"deny"` | -| `matched` | `bool` | `True` when the rule fully matched; `False` when skipped | -| `skip_reason` | `str \| None` | Why the rule was skipped, or `None` when `matched=True` | - -Possible `skip_reason` values: `"action_mismatch"`, `"resource_mismatch"`, -`"condition_mismatch"`, `"condition_type_mismatch"`, `"condition_depth_exceeded"`. - -**Algorithm-specific trace behaviour** - -* `deny-overrides` — trace includes every rule up to and including the first - matching deny (the loop breaks there). When only permits fire, all rules - are present. -* `permit-overrides` — trace up to and including the first matching permit. -* `first-applicable` — trace up to and including the first match; subsequent - rules are absent. -* No match — every rule appears in the trace with `matched=False`. +When `explain=False` (default) `Decision.trace` is `None` — zero overhead on the hot path. `explain=True` is supported on all four evaluation methods: ```python -# Single request -d = guard.evaluate_sync(..., explain=True) -d = await guard.evaluate_async(..., explain=True) - -# Batch — explain applies to every request in the batch +d = guard.evaluate_sync(..., explain=True) +d = await guard.evaluate_async(..., explain=True) decisions = guard.evaluate_batch_sync([...], explain=True) decisions = await guard.evaluate_batch_async([...], explain=True) ``` -`RuleTrace` is importable directly from the root package: +**`RuleTrace` fields** + +| Field | Type | Description | +|---|---|---| +| `rule_id` | `str` | Rule `id` as declared in the policy | +| `effect` | `str` | `"permit"` or `"deny"` | +| `matched` | `bool` | `True` when the rule fully matched | +| `skip_reason` | `str \| None` | Why the rule was skipped; `None` when `matched=True` | + +Possible `skip_reason` values: `"action_mismatch"`, `"resource_mismatch"`, +`"condition_mismatch"`, `"condition_type_mismatch"`, `"condition_depth_exceeded"`. ```python -from rbacx import RuleTrace +from rbacx import RuleTrace # importable from root package ``` --- ## Batch evaluation -`Guard` exposes two methods for evaluating multiple access requests in a single -call — useful for populating UIs that need to know which buttons/tabs/actions -to show for a given user. +Evaluate multiple access requests in one call — useful for UI state checks +(which buttons/actions to show for a given user). ```python -from rbacx import Guard, Subject, Action, Resource, Context - -guard = Guard(policy) -subject = Subject(id="u1", roles=["editor"]) -resource = Resource(type="document", id="doc-42") -ctx = Context(attrs={"mfa": True}) - -# Async (preferred in ASGI applications) decisions = await guard.evaluate_batch_async([ (subject, Action("read"), resource, ctx), (subject, Action("write"), resource, ctx), (subject, Action("delete"), resource, ctx), -]) +], timeout=2.0) -# Sync (works everywhere, including inside a running event loop) -decisions = guard.evaluate_batch_sync([ - (subject, Action("read"), resource, ctx), - (subject, Action("write"), resource, ctx), - (subject, Action("delete"), resource, ctx), -]) - -for action_name, decision in zip(["read", "write", "delete"], decisions): - print(action_name, "→", "allow" if decision.allowed else "deny") +decisions = guard.evaluate_batch_sync([...]) ``` -**Signature** +**Signatures** ```python async def evaluate_batch_async( - self, requests: Sequence[tuple[Subject, Action, Resource, Context | None]], *, explain: bool = False, @@ -206,7 +188,6 @@ async def evaluate_batch_async( ) -> list[Decision]: ... def evaluate_batch_sync( - self, requests: Sequence[tuple[Subject, Action, Resource, Context | None]], *, explain: bool = False, @@ -216,40 +197,48 @@ def evaluate_batch_sync( **Guarantees** -* Results are returned in the **same order** as the input sequence. -* Requests are evaluated **concurrently** via `asyncio.gather` — wall-clock - time grows with the slowest single request rather than the total count. -* An **empty** input list returns `[]` immediately without any evaluation. -* `timeout` (seconds) bounds the total wall-clock time for the batch. - `asyncio.TimeoutError` is raised if the deadline is exceeded. ``None`` - (default) means no deadline. -* `Context` may be `None` for any individual request. -* All DI hooks (metrics, logger, obligation checker, role resolver, cache) are - invoked **per request**, exactly as with `evaluate_async` / `evaluate_sync`. -* If any individual request raises an exception the entire batch propagates - that exception (**fail-fast** semantics, consistent with `asyncio.gather`). +* Results are in the **same order** as the input. +* Requests run **concurrently** via `asyncio.gather` — total time equals the slowest check. +* Empty input returns `[]` immediately. +* `timeout` bounds total wall-clock time; raises `asyncio.TimeoutError` on expiry. +* All DI hooks (metrics, logger, cache, obligations, handlers) apply per request. --- -## Decision object +## Executable obligation handlers + +Register handlers that `Guard` calls automatically after a `permit` decision: + +```python +from rbacx.core.engine import ObligationNotMetError + +def check_mfa(decision, context): + if not context.attrs.get("mfa"): + raise ObligationNotMetError("MFA required", challenge="mfa") + +guard.register_obligation_handler("require_mfa", check_mfa) +``` + +**Signature** + +```python +guard.register_obligation_handler(obligation_type: str, handler: Callable) -> None +``` + +Handler signature: `(decision: Decision, context: Context) -> None` — sync or async. -Fields returned by `Guard.evaluate*`: +`ObligationNotMetError(message="", *, challenge=None)` — flips the decision to +`deny` with `reason="obligation_failed"`. `challenge` is propagated to +`Decision.challenge`. Any other exception also causes deny (fail-closed). -* `allowed: bool` -* `effect: "permit" | "deny"` -* `obligations: List[Dict[str, Any]]` -* `challenge: Optional[str]` -* `rule_id: Optional[str]` -* `policy_id: Optional[str]` -* `reason: Optional[str]` -* `trace: Optional[List[RuleTrace]]` — populated when `explain=True`; `None` by default +See [Obligations](../r/docs/obligations.md) for full details and behaviour. --- ## `require_batch_access` (FastAPI) FastAPI dependency that evaluates multiple `(action, resource_type)` pairs in -one `evaluate_batch_async` call and returns a `list[Decision]`. +one batch call and returns `list[Decision]`: ```python from rbacx.adapters.fastapi import require_batch_access diff --git a/docs/obligations.md b/docs/obligations.md index e6d363d..be320e0 100644 --- a/docs/obligations.md +++ b/docs/obligations.md @@ -299,6 +299,48 @@ blocking all access. Obligations without a `condition` field behave exactly as before. +## Executable obligation handlers + +Instead of inspecting `Decision.obligations` and dispatching handlers manually, +register them directly on the `Guard` instance: + +```python +from rbacx.core.engine import ObligationNotMetError + +def check_mfa(decision, context): + if not context.attrs.get("mfa"): + raise ObligationNotMetError("MFA not satisfied", challenge="mfa") + +async def check_geo(decision, context): + if context.attrs.get("geo") not in ("EU", "US"): + raise ObligationNotMetError("Geo not allowed", challenge="geo") + +guard.register_obligation_handler("require_mfa", check_mfa) +guard.register_obligation_handler("require_geo", check_geo) +``` + +Handler signature: `(decision: Decision, context: Context) -> None` (sync or async). + +**Behaviour:** + +* Handlers are called **only for `permit` decisions**, in the order obligations + appear in the matched rule. +* Raising :class:`~rbacx.core.engine.ObligationNotMetError` flips the + decision to `deny` with `reason="obligation_failed"`. Set `challenge=` to + propagate a machine-readable hint (e.g. `"mfa"`) to `Decision.challenge`. +* Any other exception is logged and also flips to `deny` (fail-closed). +* When the first handler raises, subsequent handlers are **skipped** (fail-fast). +* Registering a handler for a type that already has one **replaces** the + previous handler. +* Conditional obligations (`condition` field) are respected — the handler is + skipped when the condition evaluates to `False`. +* Unregistered obligation types are left in `Decision.obligations` for manual + handling — backward-compatible with the existing approach. + +**`ObligationNotMetError`** is importable from `rbacx.core.engine`. + +--- + ## Notes & best practices * Keep obligation handlers **pure** (no I/O) and quick; they run in the request path. diff --git a/pyproject.toml b/pyproject.toml index 3538e25..8922bfe 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "rbacx" -version = "1.17.0" +version = "1.18.0" description = "RBAC/ABAC policy engine for Python with policy sets, condition DSL, and hot reload" readme = "README.md" requires-python = ">=3.10" diff --git a/src/rbacx/core/engine.py b/src/rbacx/core/engine.py index 004db98..2225edd 100644 --- a/src/rbacx/core/engine.py +++ b/src/rbacx/core/engine.py @@ -39,6 +39,34 @@ def _now() -> float: return time.perf_counter() +class ObligationNotMetError(Exception): + """Raised by an obligation handler when the obligation cannot be satisfied. + + Raising this exception from a handler registered via + :meth:`Guard.register_obligation_handler` causes the engine to flip the + decision to ``deny`` with ``reason="obligation_failed"``. + + Args: + message: human-readable description of why the obligation was not met. + challenge: optional machine-readable challenge string (e.g. ``"mfa"``, + ``"step_up"``). When set, the value is propagated to + :attr:`~rbacx.core.decision.Decision.challenge` so the PEP can + issue the appropriate response header (e.g. ``WWW-Authenticate``). + + Example:: + + from rbacx.core.engine import ObligationNotMetError + + def require_mfa(decision, context): + if not context.attrs.get("mfa"): + raise ObligationNotMetError("MFA required", challenge="mfa") + """ + + def __init__(self, message: str = "", *, challenge: str | None = None) -> None: + super().__init__(message) + self.challenge: str | None = challenge + + class Guard: """Policy evaluation engine. @@ -85,6 +113,9 @@ def __init__( self._compiled: Callable[[dict[str, Any]], dict[str, Any]] | None = None self.strict_types: bool = bool(strict_types) self.relationship_checker = relationship_checker + # Registry of executable obligation handlers. + # Keys are obligation type strings; values are sync or async callables. + self._obligation_handlers: dict[str, Any] = {} # Guards atomic replacement of policy / etag / compiled function. # RLock allows re-entrant acquisition: set_policy -> _recompute_etag -> clear_cache # can all hold the lock in the same thread without deadlocking. @@ -94,6 +125,55 @@ def __init__( # ---------------------------------------------------------------- set/update + def register_obligation_handler( + self, + obligation_type: str, + handler: Any, + ) -> None: + """Register an executable handler for a specific obligation type. + + When ``Guard`` produces a ``permit`` decision that carries an obligation + of the given *obligation_type*, it automatically calls the registered + *handler* instead of (or in addition to) returning the obligation in + ``Decision.obligations``. + + If the handler raises :class:`~rbacx.core.engine.ObligationNotMetError` + the decision is flipped to ``deny`` with ``reason="obligation_failed"`` + and the exception's ``challenge`` attribute (if set) is propagated to + ``Decision.challenge``. Any other exception is treated as a handler + error: it is logged and the decision is also flipped to ``deny`` + (fail-closed semantics). + + Registering a handler for a type that already has one **replaces** the + previous handler. + + Handler signature:: + + def handler(decision: Decision, context: Context) -> None: + # raise ObligationNotMetError on failure + ... + + # async handlers are also supported: + async def handler(decision: Decision, context: Context) -> None: + ... + + Args: + obligation_type: the ``type`` string of the obligation as it + appears in the policy (e.g. ``"require_mfa"``). + handler: sync or async callable conforming to the signature above. + + Example:: + + from rbacx.core.engine import ObligationNotMetError + + def check_mfa(decision, context): + if not context.attrs.get("mfa"): + raise ObligationNotMetError("MFA not satisfied", challenge="mfa") + + guard.register_obligation_handler("require_mfa", check_mfa) + """ + self._obligation_handlers[obligation_type] = handler + def set_policy(self, policy: dict[str, Any]) -> None: """Replace policy/policyset. @@ -295,6 +375,65 @@ async def _evaluate_core_async( trace=trace, ) + # Executable obligation handlers — invoked for permit decisions only. + # Handlers receive the fully-built Decision so they can inspect all fields. + # ObligationNotMetError flips the decision to deny; any other exception + # is logged and also flips to deny (fail-closed). + # + # Conditional obligations: if an obligation carries a ``condition`` field + # we evaluate it here (same logic as BasicObligationChecker) — skip the + # handler when the condition is False or raises. + if d.allowed and self._obligation_handlers: + from .policy import ( # noqa: PLC0415 (deferred to avoid circular import) + ConditionDepthError, + ConditionTypeError, + eval_condition, + ) + + for ob in raw.get("obligations") or []: + ob_type = (ob or {}).get("type") + handler = self._obligation_handlers.get(ob_type) if ob_type else None + if handler is None: + continue + # Respect conditional obligations — skip when condition is False/error + ob_condition = (ob or {}).get("condition") + if ob_condition is not None: + try: + if not eval_condition(ob_condition, env): + continue + except (ConditionTypeError, ConditionDepthError): + continue # fail-safe: skip handler on condition error + try: + await maybe_await(handler(d, context)) + except ObligationNotMetError as exc: + d = Decision( + allowed=False, + effect="deny", + obligations=d.obligations, + challenge=exc.challenge if exc.challenge is not None else d.challenge, + rule_id=d.rule_id, + policy_id=d.policy_id, + reason="obligation_failed", + trace=d.trace, + ) + break + except Exception: + logger.exception( + "RBACX: obligation handler %r raised unexpected error (fail-closed)", + ob_type, + ) + d = Decision( + allowed=False, + effect="deny", + obligations=d.obligations, + challenge=d.challenge, + rule_id=d.rule_id, + policy_id=d.policy_id, + reason="obligation_failed", + trace=d.trace, + ) + break + # metrics (do not use return values; conditionally await) if self.metrics is not None: labels = {"decision": d.effect} diff --git a/tests/unit/engine/test_executable_obligations.py b/tests/unit/engine/test_executable_obligations.py new file mode 100644 index 0000000..85739e7 --- /dev/null +++ b/tests/unit/engine/test_executable_obligations.py @@ -0,0 +1,511 @@ +"""Unit tests for executable obligation handlers (Guard.register_obligation_handler).""" + +import pytest + +from rbacx import Action, Context, Guard, Resource, Subject +from rbacx.core.engine import ObligationNotMetError + +# --------------------------------------------------------------------------- +# Shared fixtures +# --------------------------------------------------------------------------- + +_POLICY_WITH_MFA = { + "algorithm": "deny-overrides", + "rules": [ + { + "id": "r-permit", + "effect": "permit", + "actions": ["read"], + "resource": {"type": "doc"}, + "obligations": [{"type": "require_mfa", "on": "permit"}], + } + ], +} + +_POLICY_NO_OBLIGATIONS = { + "algorithm": "deny-overrides", + "rules": [ + {"id": "r-permit", "effect": "permit", "actions": ["read"], "resource": {"type": "doc"}}, + ], +} + +_POLICY_DENY = { + "algorithm": "deny-overrides", + "rules": [ + {"id": "r-deny", "effect": "deny", "actions": ["read"], "resource": {"type": "doc"}}, + ], +} + +_POLICY_TWO_OBLIGATIONS = { + "algorithm": "deny-overrides", + "rules": [ + { + "id": "r-permit", + "effect": "permit", + "actions": ["read"], + "resource": {"type": "doc"}, + "obligations": [ + {"type": "require_mfa", "on": "permit"}, + {"type": "require_geo", "on": "permit"}, + ], + } + ], +} + +_S = Subject(id="u1") +_R = Resource(type="doc", id="d1") +_CTX = Context() + + +# --------------------------------------------------------------------------- +# No handler registered — backward compatible +# --------------------------------------------------------------------------- + + +def test_no_handler_registered_permit_unchanged(): + """Without any registered handler permit decisions are unaffected.""" + g = Guard(_POLICY_WITH_MFA) + d = g.evaluate_sync(_S, Action("read"), _R, _CTX) + # BasicObligationChecker passes (mfa not in context → denied by checker) + # but no executable handler is registered so obligations fall through + assert d.obligations == [{"type": "require_mfa", "on": "permit"}] + + +def test_no_handler_permit_no_obligations(): + """Rule with no obligations: no handler is ever consulted.""" + g = Guard(_POLICY_NO_OBLIGATIONS) + d = g.evaluate_sync(_S, Action("read"), _R, _CTX) + assert d.allowed is True + + +# --------------------------------------------------------------------------- +# Handler registered — passes (does not raise) +# --------------------------------------------------------------------------- + + +def test_handler_passes_permit_preserved(): + """Handler that does not raise leaves the decision as permit.""" + g = Guard(_POLICY_WITH_MFA) + called = [] + + def ok_handler(decision, context): + called.append(True) + + # Bypass BasicObligationChecker by providing mfa=True + g.register_obligation_handler("require_mfa", ok_handler) + ctx = Context(attrs={"mfa": True}) + d = g.evaluate_sync(_S, Action("read"), _R, ctx) + assert d.allowed is True + assert called == [True] + + +@pytest.mark.asyncio +async def test_async_handler_passes_permit_preserved(): + """Async handler that does not raise leaves the decision as permit.""" + g = Guard(_POLICY_WITH_MFA) + called = [] + + async def ok_handler(decision, context): + called.append(True) + + g.register_obligation_handler("require_mfa", ok_handler) + ctx = Context(attrs={"mfa": True}) + d = await g.evaluate_async(_S, Action("read"), _R, ctx) + assert d.allowed is True + assert called == [True] + + +# --------------------------------------------------------------------------- +# Handler raises ObligationNotMetError +# --------------------------------------------------------------------------- + + +def test_handler_raises_obligation_not_met_flips_to_deny(): + """ObligationNotMetError from handler flips decision to deny.""" + g = Guard(_POLICY_WITH_MFA) + + def failing_handler(decision, context): + raise ObligationNotMetError("MFA required") + + g.register_obligation_handler("require_mfa", failing_handler) + ctx = Context(attrs={"mfa": True}) # bypass BasicObligationChecker + d = g.evaluate_sync(_S, Action("read"), _R, ctx) + assert d.allowed is False + assert d.reason == "obligation_failed" + + +def test_handler_raises_with_challenge_propagated(): + """challenge from ObligationNotMetError is propagated to Decision.""" + g = Guard(_POLICY_WITH_MFA) + + def failing_handler(decision, context): + raise ObligationNotMetError("MFA required", challenge="mfa") + + g.register_obligation_handler("require_mfa", failing_handler) + ctx = Context(attrs={"mfa": True}) + d = g.evaluate_sync(_S, Action("read"), _R, ctx) + assert d.allowed is False + assert d.challenge == "mfa" + assert d.reason == "obligation_failed" + + +def test_handler_raises_without_challenge_preserves_existing(): + """When ObligationNotMetError has no challenge, existing challenge is kept.""" + g = Guard(_POLICY_WITH_MFA) + + def failing_handler(decision, context): + raise ObligationNotMetError("failed") # no challenge + + g.register_obligation_handler("require_mfa", failing_handler) + ctx = Context(attrs={"mfa": True}) + d = g.evaluate_sync(_S, Action("read"), _R, ctx) + assert d.allowed is False + assert d.challenge is None # no challenge from either checker or handler + + +@pytest.mark.asyncio +async def test_async_handler_raises_obligation_not_met(): + """Async handler that raises ObligationNotMetError flips to deny.""" + g = Guard(_POLICY_WITH_MFA) + + async def failing_handler(decision, context): + raise ObligationNotMetError("MFA", challenge="mfa") + + g.register_obligation_handler("require_mfa", failing_handler) + ctx = Context(attrs={"mfa": True}) + d = await g.evaluate_async(_S, Action("read"), _R, ctx) + assert d.allowed is False + assert d.challenge == "mfa" + + +# --------------------------------------------------------------------------- +# Handler raises unexpected exception — fail-closed +# --------------------------------------------------------------------------- + + +def test_handler_unexpected_exception_fail_closed(): + """Non-ObligationNotMetError exception is logged and causes deny (fail-closed).""" + g = Guard(_POLICY_WITH_MFA) + + def broken_handler(decision, context): + raise RuntimeError("handler crashed") + + g.register_obligation_handler("require_mfa", broken_handler) + ctx = Context(attrs={"mfa": True}) + d = g.evaluate_sync(_S, Action("read"), _R, ctx) + assert d.allowed is False + assert d.reason == "obligation_failed" + + +# --------------------------------------------------------------------------- +# Deny decisions — handlers never called +# --------------------------------------------------------------------------- + + +def test_handler_not_called_on_deny(): + """Handlers are only called for permit decisions.""" + g = Guard(_POLICY_DENY) + called = [] + + def handler(decision, context): + called.append(True) + + g.register_obligation_handler("require_mfa", handler) + d = g.evaluate_sync(_S, Action("read"), _R, _CTX) + assert d.allowed is False + assert called == [] + + +# --------------------------------------------------------------------------- +# Multiple obligations / handlers +# --------------------------------------------------------------------------- + + +def test_multiple_handlers_all_called(): + """All handlers for obligations present in the decision are called.""" + g = Guard(_POLICY_TWO_OBLIGATIONS) + called = [] + + def mfa_handler(decision, context): + called.append("mfa") + + def geo_handler(decision, context): + called.append("geo") + + g.register_obligation_handler("require_mfa", mfa_handler) + g.register_obligation_handler("require_geo", geo_handler) + ctx = Context(attrs={"mfa": True}) + d = g.evaluate_sync(_S, Action("read"), _R, ctx) + assert d.allowed is True + assert "mfa" in called and "geo" in called + + +def test_first_handler_fails_second_not_called(): + """When the first handler raises, subsequent handlers are skipped (fail-fast).""" + g = Guard(_POLICY_TWO_OBLIGATIONS) + called = [] + + def mfa_handler(decision, context): + called.append("mfa") + raise ObligationNotMetError("mfa", challenge="mfa") + + def geo_handler(decision, context): + called.append("geo") + + g.register_obligation_handler("require_mfa", mfa_handler) + g.register_obligation_handler("require_geo", geo_handler) + ctx = Context(attrs={"mfa": True}) + d = g.evaluate_sync(_S, Action("read"), _R, ctx) + assert d.allowed is False + assert called == ["mfa"] # geo handler was NOT called + + +def test_unregistered_obligation_type_ignored(): + """Obligations with no registered handler are left in Decision.obligations.""" + g = Guard(_POLICY_WITH_MFA) + # Register handler for a DIFFERENT type + g.register_obligation_handler("require_geo", lambda d, c: None) + ctx = Context(attrs={"mfa": True}) + d = g.evaluate_sync(_S, Action("read"), _R, ctx) + # require_mfa has no handler — BasicObligationChecker already passed (mfa=True) + assert d.allowed is True + + +# --------------------------------------------------------------------------- +# register_obligation_handler replaces existing +# --------------------------------------------------------------------------- + + +def test_register_replaces_previous_handler(): + """Registering for an existing type replaces the previous handler.""" + g = Guard(_POLICY_WITH_MFA) + called = [] + + def first(decision, context): + called.append("first") + + def second(decision, context): + called.append("second") + + g.register_obligation_handler("require_mfa", first) + g.register_obligation_handler("require_mfa", second) # replaces first + ctx = Context(attrs={"mfa": True}) + g.evaluate_sync(_S, Action("read"), _R, ctx) + assert called == ["second"] + + +# --------------------------------------------------------------------------- +# Handler receives correct Decision fields +# --------------------------------------------------------------------------- + + +def test_handler_receives_decision_with_correct_fields(): + """Handler is called with the fully-built Decision object.""" + g = Guard(_POLICY_WITH_MFA) + received = {} + + def inspector(decision, context): + received["allowed"] = decision.allowed + received["rule_id"] = decision.rule_id + received["obligations"] = decision.obligations + + g.register_obligation_handler("require_mfa", inspector) + ctx = Context(attrs={"mfa": True}) + g.evaluate_sync(_S, Action("read"), _R, ctx) + assert received["allowed"] is True + assert received["rule_id"] == "r-permit" + assert any(o.get("type") == "require_mfa" for o in received["obligations"]) + + +# --------------------------------------------------------------------------- +# Cache interaction — handlers still called on cache hit +# --------------------------------------------------------------------------- + + +def test_handler_called_on_cache_hit(): + """Handlers are executed even when the raw decision is served from cache.""" + from rbacx.core.cache import DefaultInMemoryCache + + g = Guard(_POLICY_WITH_MFA, cache=DefaultInMemoryCache()) + called = [] + + def handler(decision, context): + called.append(True) + + g.register_obligation_handler("require_mfa", handler) + ctx = Context(attrs={"mfa": True}) + + g.evaluate_sync(_S, Action("read"), _R, ctx) # populates cache + g.evaluate_sync(_S, Action("read"), _R, ctx) # cache hit + + assert len(called) == 2 + + +# --------------------------------------------------------------------------- +# Conditional obligation + handler +# --------------------------------------------------------------------------- + + +def test_handler_respects_conditional_obligation(): + """Handler is only called when the obligation's condition evaluates to True. + + Uses mfa=True in context so BasicObligationChecker passes for both + resources; the handler is then gated only by the obligation's own condition. + """ + policy = { + "algorithm": "deny-overrides", + "rules": [ + { + "id": "r1", + "effect": "permit", + "actions": ["read"], + "resource": {"type": "doc"}, + "obligations": [ + { + "type": "require_mfa", + "on": "permit", + "condition": {"==": [{"attr": "resource.attrs.sensitivity"}, "high"]}, + } + ], + } + ], + } + g = Guard(policy) + called = [] + # mfa=True so BasicObligationChecker does not block; handler gated by condition + ctx = Context(attrs={"mfa": True}) + + def handler(decision, context): + called.append(True) + + g.register_obligation_handler("require_mfa", handler) + + # condition False (low) → handler NOT called + d_low = g.evaluate_sync( + _S, Action("read"), Resource(type="doc", attrs={"sensitivity": "low"}), ctx + ) + assert d_low.allowed is True + assert called == [] + + # condition True (high) → handler called + d_high = g.evaluate_sync( + _S, Action("read"), Resource(type="doc", attrs={"sensitivity": "high"}), ctx + ) + assert d_high.allowed is True + assert called == [True] + + +# --------------------------------------------------------------------------- +# Batch evaluate +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_handler_called_per_request_in_batch(): + """evaluate_batch_async calls handlers for each request independently.""" + g = Guard(_POLICY_WITH_MFA) + called = [] + + async def handler(decision, context): + called.append(True) + + g.register_obligation_handler("require_mfa", handler) + ctx = Context(attrs={"mfa": True}) + + results = await g.evaluate_batch_async( + [ + (_S, Action("read"), _R, ctx), + (_S, Action("read"), _R, ctx), + (_S, Action("read"), _R, ctx), + ] + ) + assert all(d.allowed for d in results) + assert len(called) == 3 + + +# --------------------------------------------------------------------------- +# ObligationNotMetError attributes +# --------------------------------------------------------------------------- + + +def test_obligation_not_met_error_defaults(): + exc = ObligationNotMetError() + assert exc.challenge is None + assert str(exc) == "" + + +def test_obligation_not_met_error_with_challenge(): + exc = ObligationNotMetError("need mfa", challenge="mfa") + assert exc.challenge == "mfa" + assert str(exc) == "need mfa" + + +# --------------------------------------------------------------------------- +# Coverage: condition evaluation errors in handler gating (lines 401-402) +# --------------------------------------------------------------------------- + + +def test_handler_skipped_on_condition_type_error(): + """When the obligation condition raises ConditionTypeError the handler + is skipped (fail-safe) and the decision remains permit (lines 401-402).""" + policy = { + "algorithm": "deny-overrides", + "rules": [ + { + "id": "r1", + "effect": "permit", + "actions": ["read"], + "resource": {"type": "doc"}, + "obligations": [ + { + "type": "require_mfa", + "on": "permit", + # str > int → ConditionTypeError + "condition": {">": [{"attr": "resource.attrs.name"}, 42]}, + } + ], + } + ], + } + g = Guard(policy) + called = [] + g.register_obligation_handler("require_mfa", lambda d, c: called.append(True)) + + d = g.evaluate_sync( + _S, + Action("read"), + Resource(type="doc", attrs={"name": "report"}), + _CTX, + ) + assert d.allowed is True + assert called == [] # handler skipped due to condition error + + +def test_handler_skipped_on_condition_depth_exceeded(): + """When the obligation condition raises ConditionDepthError the handler + is skipped (fail-safe) and the decision remains permit (lines 401-402).""" + from rbacx.core.policy import MAX_CONDITION_DEPTH + + deep: dict = {"==": [1, 1]} + for _ in range(MAX_CONDITION_DEPTH + 2): + deep = {"and": [deep]} + + policy = { + "algorithm": "deny-overrides", + "rules": [ + { + "id": "r1", + "effect": "permit", + "actions": ["read"], + "resource": {"type": "doc"}, + "obligations": [{"type": "require_mfa", "on": "permit", "condition": deep}], + } + ], + } + g = Guard(policy) + called = [] + g.register_obligation_handler("require_mfa", lambda d, c: called.append(True)) + + d = g.evaluate_sync(_S, Action("read"), _R, _CTX) + assert d.allowed is True + assert called == [] # handler skipped due to depth error