Skip to content

API Reference

Guard(policy, *, logger_sink=None, metrics=None, obligation_checker=None, role_resolver=None)

Policy evaluation engine.

Holds a policy or a policy set and evaluates access decisions.

Design
  • Single async core _evaluate_core_async (one source of truth).
  • Sync API wraps the async core; if a loop is already running, uses a helper thread.
  • DI (resolver/obligations/metrics/logger) can be sync or async; both supported via _maybe_await.
  • CPU-bound evaluation is offloaded to a thread via asyncio.to_thread.
  • On init we ensure a current event loop exists in this thread so legacy tests using asyncio.get_event_loop().run_until_complete(...) don’t crash on Python 3.12+.

evaluate_async(subject, action, resource, context=None) async

True async API for ASGI frameworks.

evaluate_sync(subject, action, resource, context=None)

Synchronous wrapper for the async core. - If no running loop in this thread: use asyncio.run(...) - If a loop is running: run the async core in a helper thread with its own loop.

set_policy(policy)

Replace policy/policyset.

update_policy(policy)

Alias kept for backward-compatibility.

ConditionTypeError

Bases: Exception

Raised when a condition compares incompatible types.

eval_condition(cond, env)

Evaluate condition dict safely. On type mismatches, raise ConditionTypeError.

resolve(token, env)

Resolve a token; supports {"attr": "a.b.c"} lookups in env.

decide(policyset, env)

Evaluate a policy set with combining algorithm over its child policies.

RoleResolver

Bases: Protocol

expand(roles)

Return roles including inherited/derived ones.

FilePolicySource(path, *, validate_schema=False, include_mtime_in_etag=False, chunk_size=512 * 1024)

Bases: PolicySource

Policy source backed by a local JSON file.

ETag semantics
  • By default, ETag = SHA-256 of file content.
  • If include_mtime_in_etag=True, the ETag also includes mtime (ns), so a simple "touch" (metadata-only change) will trigger a reload.

The class caches the last SHA by (size, mtime_ns) to avoid unnecessary hashing.

HTTPPolicySource(url, *, headers=None)

Bases: PolicySource

HTTP policy source using requests with ETag support. Extra: rbacx[http]

S3PolicySource(url, *, validate_schema=False, change_detector='etag', prefer_checksum='sha256', session=None, botocore_config=None, client_params=None)

Bases: PolicySource

Policy source backed by Amazon S3.

Change detection strategies (choose one via change_detector): - "etag" : HeadObject ETag (default). - "version_id" : HeadObject VersionId (requires bucket versioning). - "checksum" : GetObjectAttributes(..., ObjectAttributes=['Checksum']) if available.

Networking defaults are production-friendly (timeouts + retries) and can be overridden via a custom botocore Config or client parameters.

etag()

Return a stable identifier of the object according to change_detector.

load()

Fetch the object via GetObject and parse JSON; optionally validate schema.

atomic_write(path, data, *, encoding='utf-8')

Write data atomically to path using a temp file + os.replace().

Policy package (hot reloader, etc.).

HotReloader(guard, source, *, initial_load=False, poll_interval=5.0, backoff_min=2.0, backoff_max=30.0, jitter_ratio=0.15, thread_daemon=True)

Unified, production-grade policy reloader.

Features
  • ETag-first logic: call source.etag() and only load/apply when it changes.
  • Error suppression with exponential backoff + jitter to avoid log/IO storms.
  • Optional background polling loop with clean start/stop.
  • Backwards-compatible one-shot API aliases: refresh_if_needed()/poll_once().
Notes
  • If source.etag() returns None, we will attempt to load() and let the source decide.
  • Guard.set_policy(policy) is called only after a successful load().
  • This class is thread-safe for concurrent check_and_reload() calls.

Parameters:

Name Type Description Default
initial_load bool

Controls startup behavior. - False (default): prime ETag at construction time; the first check will NO-OP unless the policy changes. (Backwards-compatible with previous versions.) - True: do not prime ETag; the first check will load the current policy.

 False

check_and_reload(*, force=False)

Perform a single reload check (sync wrapper over the async core).

Parameters:

Name Type Description Default
force bool

If True, load/apply the policy regardless of ETag state.

 False

Returns:

Type Description
bool

True if a new policy was loaded and applied; otherwise False.

check_and_reload_async(*, force=False) async

Async-aware reload check
  • supports sync/async PolicySource.etag()/load() via _maybe_await
  • never holds the thread lock while awaiting

start(interval=None, *, initial_load=None, force_initial=False)

Start the background polling thread.

Parameters:

Name Type Description Default
interval float | None

seconds between checks; if None, uses self.poll_interval (or 5.0 fallback).

 None
initial_load Optional[bool]

override constructor's initial_load just for this start(). If True, perform a synchronous load/check before starting the thread. If False, skip any initial load. If None, inherit the constructor setting.

 None
force_initial bool

if True and an initial load is requested, bypass the ETag check for that initial load (equivalent to check_and_reload(force=True)).

 False

stop(timeout=1.0)

Signal the polling thread to stop and optionally wait for it.

FilePolicySource(path, *, validate_schema=False, include_mtime_in_etag=False, chunk_size=512 * 1024)

Bases: PolicySource

Policy source backed by a local JSON file.

ETag semantics
  • By default, ETag = SHA-256 of file content.
  • If include_mtime_in_etag=True, the ETag also includes mtime (ns), so a simple "touch" (metadata-only change) will trigger a reload.

The class caches the last SHA by (size, mtime_ns) to avoid unnecessary hashing.

atomic_write(path, data, *, encoding='utf-8')

Write data atomically to path using a temp file + os.replace().

S3PolicySource(url, *, validate_schema=False, change_detector='etag', prefer_checksum='sha256', session=None, botocore_config=None, client_params=None)

Bases: PolicySource

Policy source backed by Amazon S3.

Change detection strategies (choose one via change_detector): - "etag" : HeadObject ETag (default). - "version_id" : HeadObject VersionId (requires bucket versioning). - "checksum" : GetObjectAttributes(..., ObjectAttributes=['Checksum']) if available.

Networking defaults are production-friendly (timeouts + retries) and can be overridden via a custom botocore Config or client parameters.

etag()

Return a stable identifier of the object according to change_detector.

load()

Fetch the object via GetObject and parse JSON; optionally validate schema.

HTTPPolicySource(url, *, headers=None)

Bases: PolicySource

HTTP policy source using requests with ETag support. Extra: rbacx[http]

require_access(guard, build_env, *, add_headers=False)

Decorator for Flask view functions to enforce access.

RBACXMiddleware(app, *, guard, build_env)

Bases: AbstractMiddleware

Litestar middleware that checks access using RBACX Guard.

Configure with a function build_env(scope) -> (Subject, Action, Resource, Context).

Decision object

Fields returned by Guard.evaluate*:

  • allowed: bool
  • effect: "permit" | "deny"
  • obligations: List[Dict[str, Any]]
  • challenge: Optional[str]
  • rule_id: Optional[str]
  • policy_id: Optional[str]
  • reason: Optional[str]