API Reference¶
Guard(policy, *, logger_sink=None, metrics=None, obligation_checker=None, role_resolver=None, relationship_checker=None, cache=None, cache_ttl=300, strict_types=False)
¶
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+.
clear_cache()
¶
Clear the decision cache if configured.
This is safe to call at any time. Errors are swallowed to avoid interfering with decision flow.
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.
decide(policyset, env)
¶
Evaluate a policy set with combining algorithm over its child policies.
DecisionLogger(*, sample_rate=1.0, redactions=None, logger_name='rbacx.audit', as_json=False, level=logging.INFO, redact_in_place=False, use_default_redactions=False, smart_sampling=False, category_sampling_rates=None, max_env_bytes=None)
¶
Bases: DecisionLogSink
Minimal, framework-agnostic audit logger for PDP decisions.
Backwards-compatible defaults
- No redactions are applied unless explicitly configured.
sample_ratecontrols probabilistic logging: 0.0 → drop all, 1.0 → log all.- Smart sampling is disabled by default.
- No env size limit by default.
Opt-in features
use_default_redactions=Trueenables DEFAULT_REDACTIONS whenredactionsis not provided.smart_sampling=Trueenables category-aware sampling (deny and permit-with-obligations can be forced to 1.0).max_env_bytestruncates the (redacted) env if the serialized JSON exceeds the threshold.
BasicObligationChecker
¶
Bases: ObligationChecker
Validate common obligations carried by a decision.
Design goals (documented for contributors):
- Fail-closed semantics preserved for legacy callers:
* If legacy string key decision is present and not equal to "permit" -> (False, None).
* If legacy key absent, derive effect from effect/allowed; any non-"permit" -> (False, None).
- Support obligations targeting the current effect (on: "permit" | "deny").
This allows, for example, an explicit http_challenge on deny to still surface a challenge.
- Do not mutate the incoming decision; return a (ok, challenge) tuple for the Guard to consume.
- Unknown obligation type is ignored (treated as advice/no-op).
Supported type values:
- require_mfa -> challenge "mfa"
- require_level (attrs.min) -> "step_up"
- http_challenge (attrs.scheme in Basic/Bearer/Digest) -> "http_basic" / "http_bearer" / "http_digest"; else "http_auth"
- require_consent (attrs.key or any consent) -> "consent"
- require_terms_accept -> "tos"
- require_captcha -> "captcha"
- require_reauth (attrs.max_age vs context.reauth_age_seconds) -> "reauth"
- require_age_verified -> "age_verification"
check(decision, context)
¶
Check obligations attached to a raw decision.
Parameters¶
decision: Mapping-like (dict). Legacy callers may pass string key decision ("permit"|"deny");
modern shape may include effect/allowed. obligations is a list of mappings.
context : Object whose .attrs is a dict (or context itself is a dict).
Returns¶
(ok, challenge): bool and optional machine-readable challenge string.
ObligationCheckResult(ok, challenge=None, reason=None)
dataclass
¶
Small DTO kept for backwards-compatibility if needed by contributors.
AbstractCache
¶
Bases: Protocol
Minimal cache interface for dependency inversion.
Implementations MUST be safe to call from multiple threads in-process or be clearly documented otherwise.
get should return None if a key doesn't exist or is expired. set may accept an optional TTL in seconds.
DefaultInMemoryCache(maxsize=2048)
¶
Bases: AbstractCache
Thread-safe in-memory LRU cache with optional per-key TTL.
Notes¶
- Uses time.monotonic() for TTL to avoid wall clock changes.
- Designed for single process scenarios. For multi-process/multi-host, inject a distributed cache implementation that conforms to AbstractCache.
- Values are stored as-is; callers are responsible for storing immutable or copy-safe data if necessary.
StaticRoleResolver(graph=None)
¶
Bases: RoleResolver
Simple in-memory role resolver with inheritance.
graph: {role: [parent_role, ...]} expand(['manager']) -> ['manager', 'employee', 'user', ...]
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
|
bool | None
|
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.
ComputedUserset(relation)
dataclass
¶
Follow another relation on the SAME object.
InMemoryRelationshipStore()
¶
Minimal tuple store with indexes by (resource, relation) and (subject, relation). Suitable for tests/dev. For production, implement the same interface on top of a DB.
LocalRelationshipChecker(store, *, rules=None, caveat_registry=None, max_depth=8, max_nodes=10000, deadline_ms=50)
¶
Bases: RelationshipChecker
In-process ReBAC implementation based on a userset-rewrite graph
- primitives: union (list), This, ComputedUserset, TupleToUserset
- safety limits: max_depth, max_nodes, deadline_ms
- conditional tuples via a caveat registry (predicate by name)
This()
dataclass
¶
Direct relation: subject --relation--> resource (aka 'this').
TupleToUserset(tupleset, computed_userset)
dataclass
¶
Traverse an object->object edge first (tupleset) and then evaluate a relation ('computed_userset') on the TARGET object.
OpenFGAChecker(config, *, client=None, async_client=None)
¶
Bases: RelationshipChecker
ReBAC provider backed by OpenFGA HTTP API.
- Uses /stores/{store_id}/check and /stores/{store_id}/batch-check.
- For conditions, forwards
context(OpenFGA merges persisted and request contexts). - If both clients are provided, AsyncClient takes precedence (methods return awaitables).
OpenFGAConfig(api_url, store_id, authorization_model_id=None, api_token=None, timeout_seconds=2.0)
dataclass
¶
Minimal configuration for OpenFGA HTTP client.
SpiceDBChecker(config, *, async_mode=False)
¶
Bases: RelationshipChecker
ReBAC provider backed by the SpiceDB/Authzed gRPC API.
- Uses CheckPermission; batch -> sequential calls (gRPC has no one-shot batch).
- Consistency: ZedToken (at_least_as_fresh) or fully_consistent.
- Caveats: pass context as google.protobuf.Struct.
SpiceDBConfig(endpoint, token=None, insecure=False, prefer_fully_consistent=False, timeout_seconds=2.0)
dataclass
¶
Minimal configuration for the SpiceDB/Authzed gRPC client.
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, *, client=None, session=None, config=None, client_extra=None, validate_schema=True, change_detector='etag', prefer_checksum='sha256')
¶
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.
HTTPPolicySource(url, *, headers=None, validate_schema=False)
¶
Bases: PolicySource
HTTP policy source using requests with ETag support.
Extra: rbacx[http]
RbacxMiddleware(app, *, guard, mode='enforce', build_env=None, add_headers=False)
¶
Framework-agnostic ASGI middleware.
Modes
- "inject": only injects guard into scope.
- "enforce": evaluates access for HTTP requests when build_env is provided.
Security
- Does not leak denial reasons in the response body.
- If
add_headers=True, attachesX-RBACX-*headers on deny.
require_access(guard, build_env, *, add_headers=False)
¶
Decorator for Flask view functions to enforce access.
RbacxDjangoMiddleware(get_response)
¶
Inject a Guard instance onto each Django request as request.rbacx_guard.
Config
- settings.RBACX_GUARD_FACTORY: dotted path to a zero-arg callable returning a Guard.
Notes
- Middleware init(get_response) runs once at startup; guard is created once.
- call(request) runs per-request; we attach the same guard to each request.
RBACXMiddleware(app, *, guard, build_env, add_headers=False)
¶
Bases: AbstractMiddleware
Litestar middleware that checks access using RBACX Guard.
- Prefers :class:
litestar.middleware.ASGIMiddleware(Litestar >= 2.15). - Falls back to :class:
litestar.middleware.AbstractMiddlewarewhen needed. - Uses :py:meth:
Guard.evaluate_async.
Decision object¶
Fields returned by Guard.evaluate*:
allowed: booleffect: "permit" | "deny"obligations: List[Dict[str, Any]]challenge: Optional[str]rule_id: Optional[str]policy_id: Optional[str]reason: Optional[str]
YAML policies¶
All built-in policy sources accept JSON and, with the optional rbacx[yaml] extra, YAML.
- File: detected by extension
.yaml/.yml. - HTTP: detected by
Content-Type(e.g.,application/yaml,application/x-yaml,text/yaml) or URL suffix. - S3: detected by key suffix
.yaml/.yml.
Internally YAML is parsed and validated against the same JSON Schema as JSON.