HBAC Python Bindings
The hbac-rs crate provides Python bindings for FreeIPA Host-Based Access Control (HBAC) rule evaluation via PyO3, allowing you to evaluate HBAC policies from Python with native Rust performance.
Installation
Prerequisites
- Python 3.8 or higher
- Rust 1.70 or higher (for building from source)
- maturin for building
Building from Source
# Install maturin
pip install maturin
# Navigate to the hbac-python crate (Python bindings are in a separate crate)
cd crates/hbac-python
# Build and install in development mode
maturin develop
# Or build a release wheel
maturin build --release
Quick Start
from hbac_rs import HbacRuleBuilder, HbacRequest, HbacPolicy, Subject
# Create a policy
policy = HbacPolicy()
# Create a rule: allow admins to access all services on all hosts (fluent API)
rule = (HbacRuleBuilder("allow_admins")
.user_group("admins")
.host_category_all()
.service_category_all()
.enabled(True)
.build())
policy.add_rule(rule)
# Create a user with group membership
user = Subject("alice")
user.add_group("admins")
# Create a request
request = HbacRequest(user, "server.example.com", "sshd")
# Evaluate
result = policy.evaluate(request)
print(result.is_allowed) # True
API Reference
Subject
Represents a user attempting access.
Subject(name: str)
Parameters:
name: The user’s identifier
Methods:
add_group(group)
Add a group membership to the user.
group(str): Group name
Returns: None
Properties:
groups
Get all group memberships.
Type: list[str]
name
Get the user’s name.
Type: str
Example:
user = Subject("alice")
user.add_group("admins")
user.add_group("developers")
print(user.name) # alice
print(user.groups) # ['admins', 'developers']
HbacRuleBuilder
Builder for creating HBAC rules.
HbacRuleBuilder(name: str)
Parameters:
name: The rule identifier
Methods:
enabled(enabled) -> HbacRuleBuilder
Set whether the rule is enabled. Returns self for method chaining.
enabled(bool): Enable or disable the rule
Returns: HbacRuleBuilder - Returns self to enable fluent chaining
deny() -> HbacRuleBuilder
Mark this as a deny rule (default is allow). Returns self for method chaining.
Returns: HbacRuleBuilder - Returns self to enable fluent chaining
User Dimension
All methods return HbacRuleBuilder for fluent chaining:
user(user: str) -> HbacRuleBuilder: Add a specific useruser_group(group: str) -> HbacRuleBuilder: Add a user groupuser_category_all() -> HbacRuleBuilder: Match all users
Host Dimension
All methods return HbacRuleBuilder for fluent chaining:
host(host: str) -> HbacRuleBuilder: Add a specific hosthost_group(group: str) -> HbacRuleBuilder: Add a host grouphost_category_all() -> HbacRuleBuilder: Match all hosts
Service Dimension
All methods return HbacRuleBuilder for fluent chaining:
service(service: str) -> HbacRuleBuilder: Add a specific serviceservice_group(group: str) -> HbacRuleBuilder: Add a service groupservice_category_all() -> HbacRuleBuilder: Match all services
build()
Build the rule.
Returns: HbacRule
Raises: ValueError if the rule configuration is invalid
Example:
# Allow engineers to SSH to development servers (fluent API)
hbac_rule = (HbacRuleBuilder("dev_ssh_access")
.user_group("engineers")
.host_group("development")
.service("sshd")
.enabled(True)
.build())
HbacRule
An immutable, compiled HBAC rule produced by HbacRuleBuilder.build().
Properties:
name(str): The rule identifier.is_enabled(bool): Whether the rule is enabled.is_allow(bool): Whether this is an allow rule.is_deny(bool): Whether this is a deny rule.estimated_size(int): Estimated in-memory size of the rule, in bytes.
Methods:
to_json() -> str: Serialize the rule to JSON.HbacRule.from_json(json: str) -> HbacRule(static method): Deserialize a single rule from JSON.HbacRule.from_json_list(json: str) -> list[HbacRule](static method): Deserialize multiple rules from a JSON array.
Example:
json_str = hbac_rule.to_json()
restored = HbacRule.from_json(json_str)
print(restored.name, restored.is_allow, restored.estimated_size)
HbacRequest
Represents an access request to evaluate.
HbacRequest(user: Subject, targethost: str, service: str)
Parameters:
user: The user attempting accesstargethost: The target host being accessedservice: The service being accessed
Methods:
add_targethost_group(group)
Add a host group membership.
group(str): Host group name
Returns: None
add_service_group(group)
Add a service group membership.
group(str): Service group name
Returns: None
Properties:
targethost
Get the target host.
Type: str
service
Get the service.
Type: str
Example:
user = Subject("alice")
user.add_group("admins")
request = HbacRequest(user, "db-01.prod", "postgresql")
request.add_targethost_group("production")
request.add_targethost_group("databases")
request.add_service_group("database-services")
HbacPolicy
HBAC policy evaluation engine.
HbacPolicy()
# Or, with a maximum rule limit
HbacPolicy.with_max_rules(50)
Methods:
add_rule(rule) -> HbacPolicy
Add a rule to the policy. Returns self for method chaining (fluent API).
rule(HbacRule): The rule to add
Returns: HbacPolicy - Returns self to enable fluent chaining
Raises: ValueError if adding the rule fails
add_temporal_rule(rule) -> HbacPolicy
Add a TemporalHbacRule to the policy. Returns self for method chaining.
rule(TemporalHbacRule): The temporal rule to add
Returns: HbacPolicy
Raises: ValueError if adding the rule fails
evaluate(request)
Evaluate a request against the policy.
request(HbacRequest): The access request
Returns: HbacEvaluationResult
check_access(request)
Quick check if access is allowed.
request(HbacRequest): The access request
Returns: bool - True if allowed, False otherwise
check_access_at(request, timestamp) -> bool
Quick check if access is allowed at a specific millisecond timestamp, with
temporal rules evaluated against timestamp instead of the current
wall-clock time.
request(HbacRequest): The access requesttimestamp(int): Millisecond timestamp
Returns: bool
remove_rule(name) -> None
Remove a rule from the policy by name.
name(str): The rule identifier
Returns: None
enable_rule(name) -> None
Enable a previously disabled rule.
name(str): The rule identifier
Returns: None
disable_rule(name) -> None
Disable a rule without removing it.
name(str): The rule identifier
Returns: None
load_rules(rules) -> HbacPolicy
Bulk-load a list of rules. Returns self for method chaining.
rules(list[HbacRule]): Rules to add
Returns: HbacPolicy
load_rules_json(json) -> HbacPolicy
Bulk-load rules from a JSON array string. Returns self for method chaining.
json(str): JSON array of serializedHbacRuleobjects
Returns: HbacPolicy
Raises: ValueError if the JSON cannot be parsed or a rule fails to load
clear() -> None
Remove all rules from the policy.
Returns: None
Properties:
rules
Get all rules in the policy.
Type: list[HbacRule]
rule_count
Get the number of rules in the policy.
Type: int
max_rules
Get the configured rule limit. Returns 0 if there is no limit.
Type: int
stats
Get evaluation cache statistics.
Type: CacheStats
Methods (continued):
evaluate_at(request, timestamp) -> HbacEvaluationResult
Evaluate a request against the policy at a specific millisecond timestamp. Useful for testing temporal rules at a known point in time.
request(HbacRequest): The access requesttimestamp(int): Millisecond timestamp
Returns: HbacEvaluationResult
evaluate_detailed(request) -> HbacEvaluationResult
Evaluate a request with full rule tracking. Populates matched_rules and not_matched_rules on the result. Slightly slower than evaluate() due to tracking overhead.
request(HbacRequest): The access request
Returns: HbacEvaluationResult
evaluate_detailed_at(request, timestamp) -> HbacEvaluationResult
Evaluate at a specific millisecond timestamp with full rule tracking.
request(HbacRequest): The access requesttimestamp(int): Millisecond timestamp
Returns: HbacEvaluationResult
Example:
# Non-fluent style
policy = HbacPolicy()
policy.add_rule(rule1.build())
policy.add_rule(rule2.build())
# Fluent style - chain rule additions
policy = (HbacPolicy()
.add_rule(rule1.build())
.add_rule(rule2.build()))
request = HbacRequest(user, "server", "sshd")
if policy.check_access(request):
print("Access allowed")
# Bulk-load and inspect
policy2 = HbacPolicy().load_rules([rule1.build(), rule2.build()])
print(f"Policy has {policy2.rule_count} rules (limit: {policy2.max_rules or 'none'})")
# Detailed evaluation with rule tracking
result = policy.evaluate_detailed(request)
print("Matched:", result.matched_rules)
print("Not matched:", result.not_matched_rules)
# Cache statistics
stats = policy.stats
print(f"Memory: {stats.memory_bytes} bytes")
HbacEvaluationResult
Result of an HBAC evaluation.
Properties:
is_allowed
Check if access is allowed.
Type: bool
is_denied
Check if access is explicitly denied.
Type: bool
warnings
Return warning messages generated during evaluation. A non-empty list means the access decision may not reflect full policy intent — for example, when RBAC role resolution failed and the decision was made with degraded information.
Type: list[str]
has_warnings
Check whether any warnings were generated.
Type: bool
matched_rules
Return the names of rules that matched the request. Only populated when the request was evaluated via evaluate_detailed() or evaluate_detailed_at().
Type: list[str]
not_matched_rules
Return the names of rules that were candidates but did not match the request. Only populated when evaluated via evaluate_detailed() or evaluate_detailed_at().
Type: list[str]
Example:
result = policy.evaluate_detailed(request)
if result.is_allowed:
print("Access granted")
print("Matched rules:", result.matched_rules)
elif result.is_denied:
print("Access explicitly denied")
print("Candidates that did not match:", result.not_matched_rules)
if result.has_warnings:
print("Degraded decision — warnings:", result.warnings)
HbacRequestBuilder
Fluent builder for constructing HbacRequest objects. Use this instead of the HbacRequest constructor when you want to build requests incrementally.
HbacRequestBuilder()
Methods:
All methods return HbacRequestBuilder for fluent chaining:
user(subject: Subject) -> HbacRequestBuilder: Set the user (required)targethost(host: str) -> HbacRequestBuilder: Set the target host (required)service(svc: str) -> HbacRequestBuilder: Set the service (required)targethost_group(group: str) -> HbacRequestBuilder: Add a host group membershipservice_group(group: str) -> HbacRequestBuilder: Add a service group membership
build() -> HbacRequest
Build the request.
Returns: HbacRequest
Raises: ValueError if user, targethost, or service have not been set.
Example:
from hbac_rs import HbacRequestBuilder, Subject
user = Subject("alice")
user.add_group("admins")
request = (HbacRequestBuilder()
.user(user)
.targethost("server.example.com")
.service("sshd")
.targethost_group("production")
.service_group("secure-services")
.build())
TemporalHbacRule
Wraps an HbacRule with a time-based validity window. A temporal rule only participates in evaluation when the wall-clock time (or the timestamp passed to evaluate_at) falls within its window.
TemporalHbacRule(rule: HbacRule, valid_from: int | None = None, valid_until: int | None = None)
Parameters:
rule: The underlyingHbacRulevalid_from: Millisecond timestamp at which the rule becomes active (inclusive).Nonemeans active from the beginning of time.valid_until: Millisecond timestamp at which the rule expires (exclusive).Nonemeans never expires.
Raises: ValueError if valid_from > valid_until.
Static factory methods:
TemporalHbacRule.valid_until(rule, until: int) -> TemporalHbacRule: Rule active until the given timestamp.TemporalHbacRule.valid_from(rule, from_ts: int) -> TemporalHbacRule: Rule active from the given timestamp onwards.TemporalHbacRule.valid_for_duration(rule, duration_ms: int) -> TemporalHbacRule: Rule active forduration_msmilliseconds starting from now.
Properties:
is_currently_valid
Whether the rule is active at the current wall-clock time.
Type: bool
rule
The underlying HbacRule.
Type: HbacRule
not_before / not_after
Aliases for the rule’s valid_from / valid_until timestamps.
Type: int | None
Methods:
is_valid_at(timestamp: int) -> bool
Check whether the rule would be active at the given millisecond timestamp.
Returns: bool
Example:
import time
from hbac_rs import HbacRuleBuilder, TemporalHbacRule, HbacPolicy
base_rule = (HbacRuleBuilder("temp_access")
.user_group("contractors")
.host_group("staging")
.service("sshd")
.enabled(True)
.build())
now = time.time_ns() // 1_000_000
one_hour = 3600 * 1000
# Valid only for the next hour
temp_rule = TemporalHbacRule.valid_for_duration(base_rule, duration_ms=one_hour)
# Valid until a specific deadline
deadline_ms = now + 7 * 24 * 3600 * 1000 # 7 days from now
temp_rule2 = TemporalHbacRule(base_rule, valid_until=deadline_ms)
policy = HbacPolicy()
policy.add_temporal_rule(temp_rule)
policy.add_temporal_rule(temp_rule2)
print(temp_rule.is_currently_valid) # True
print(temp_rule.is_valid_at(now + one_hour + 1)) # False
HbacResource
Models a protected resource that requires a specific permission, derived
automatically as AtomicPermission(host, service). Useful when HBAC rules
are combined with permission-based access control.
Methods:
can_access(subject: Subject) -> bool
Check whether the subject has the required permission to access this resource.
Returns: bool
grant_access(subject: Subject) -> None
Grant the required permission to the subject (mutates the subject in place).
Returns: None
deny_access(subject: Subject) -> None
Explicitly deny the required permission for the subject (mutates the subject in place).
Returns: None
Properties:
required_permission
Return the AtomicPermission this resource requires — derived from the
resource’s host and service as AtomicPermission(host, service).
Type: AtomicPermission (hbac_rs’s own AtomicPermission class, not acls_rs.AtomicPermission)
HbacResourceBuilder
Fluent builder for HbacResource. There is no separate way to set the
required permission directly — it is always derived from host/service.
Example:
from hbac_rs import HbacResourceBuilder, Subject
resource = (HbacResourceBuilder()
.host("db")
.service("read")
.build())
user = Subject("alice")
resource.grant_access(user)
print(resource.can_access(user)) # True
print(resource.required_permission) # AtomicPermission('db', 'read')
CacheStats
Statistics about evaluation cache performance and memory usage, returned by HbacPolicy.stats.
Fields (read-only):
stats.rule_count # int — total number of rules in the policy
stats.memory_bytes # int — estimated memory usage in bytes
stats.hit_count # int — number of cache hits
stats.miss_count # int — number of cache misses
Example:
stats = policy.stats
total = stats.hit_count + stats.miss_count
hit_rate = stats.hit_count / total if total else 0.0
print(f"Cache hit rate: {hit_rate:.1%}")
print(f"Memory usage: {stats.memory_bytes} bytes")
Getting Timestamps
All temporal APIs accept millisecond Unix timestamps. Use Python’s time module:
import time
now = time.time_ns() // 1_000_000
future = now + 3600 * 1000 # one hour from now
Policy Composition
ComposedPolicy lets you combine an HbacPolicy (host-based access control) with an acls_rs.RbacPolicy (role-based permissions) into a single evaluation step.
CompositionMode
Determines how the HBAC and RBAC results are combined.
Static factory methods:
CompositionMode.and_mode()— Both HBAC and RBAC must allow (defense in depth).CompositionMode.or_mode()— Either HBAC or RBAC allowing is sufficient.CompositionMode.hbac_first()— HBAC is the primary policy; RBAC is consulted only when HBAC has no applicable rules.CompositionMode.rbac_first()— RBAC is the primary policy; HBAC is consulted only when RBAC has no applicable rules.
ComposedPolicy
from hbac_rs import ComposedPolicy, CompositionMode, HbacPolicy, HbacRequestBuilder, HbacRuleBuilder, Subject
import acls_rs
# Build HBAC policy
hbac = HbacPolicy()
hbac.add_rule(
HbacRuleBuilder("allow_devs")
.user_group("developers")
.host_group("staging")
.service("sshd")
.enabled(True)
.build()
)
# Build RBAC policy
rbac = acls_rs.RbacPolicy()
dev_role = acls_rs.Role("developers",
grants=[acls_rs.AtomicPermission("host", "ssh")],
denials=[])
rbac.add_role(dev_role)
# Compose — both must allow
mode = CompositionMode.and_mode()
policy = ComposedPolicy(hbac, rbac, mode)
# Evaluate
user = Subject("alice")
user.add_group("developers")
request = (HbacRequestBuilder()
.user(user)
.targethost("staging-01.example.com")
.service("sshd")
.targethost_group("staging")
.build())
result = policy.evaluate(request)
if result.has_warnings:
print("Degraded decision:", result.warnings)
if result.is_allowed:
print("Access granted")
Composition modes at a glance:
| Mode | Behaviour |
|---|---|
and_mode() | Both HBAC and RBAC must allow. Highest security. |
or_mode() | Either HBAC or RBAC allowing is sufficient. Useful during migration. |
hbac_first() | HBAC drives the decision; RBAC is the fallback. |
rbac_first() | RBAC drives the decision; HBAC is the fallback. |
Examples
Basic SSH Access Control
from hbac_rs import HbacRuleBuilder, HbacRequest, HbacPolicy, Subject
policy = HbacPolicy()
# Rule 1: Allow admins SSH access to all servers (fluent API)
admin_rule = (HbacRuleBuilder("admin_ssh")
.user_group("admins")
.host_category_all()
.service("sshd")
.enabled(True)
.build())
policy.add_rule(admin_rule)
# Rule 2: Allow developers SSH to dev servers only (fluent API)
dev_rule = (HbacRuleBuilder("dev_ssh")
.user_group("developers")
.host_group("development")
.service("sshd")
.enabled(True)
.build())
policy.add_rule(dev_rule)
# Test admin access
admin_user = Subject("bob")
admin_user.add_group("admins")
admin_request = HbacRequest(admin_user, "prod-server-01", "sshd")
print(policy.check_access(admin_request)) # True
# Test developer access to production (denied)
dev_user = Subject("alice")
dev_user.add_group("developers")
prod_request = HbacRequest(dev_user, "prod-server-01", "sshd")
print(policy.check_access(prod_request)) # False
# Test developer access to development (allowed)
dev_request = HbacRequest(dev_user, "dev-server-01", "sshd")
dev_request.add_targethost_group("development")
print(policy.check_access(dev_request)) # True
Multiple Services
from hbac_rs import HbacRuleBuilder, HbacRequest, HbacPolicy, Subject
policy = HbacPolicy()
# Allow database admins to access multiple database services (fluent API)
db_rule = (HbacRuleBuilder("database_access")
.user_group("dba")
.host_group("databases")
.service("postgresql")
.service("mysql")
.service("mongodb")
.enabled(True)
.build())
policy.add_rule(db_rule)
# Test PostgreSQL access
dba = Subject("charlie")
dba.add_group("dba")
pg_request = HbacRequest(dba, "db-pg-01", "postgresql")
pg_request.add_targethost_group("databases")
print(policy.check_access(pg_request)) # True
# Test MySQL access
mysql_request = HbacRequest(dba, "db-mysql-01", "mysql")
mysql_request.add_targethost_group("databases")
print(policy.check_access(mysql_request)) # True
Deny Rules
from hbac_rs import HbacRuleBuilder, HbacRequest, HbacPolicy, Subject
policy = HbacPolicy()
# Allow rule: engineers can access development (fluent API)
allow_rule = (HbacRuleBuilder("allow_dev")
.user_group("engineers")
.host_group("development")
.service_category_all()
.enabled(True)
.build())
policy.add_rule(allow_rule)
# Deny rule: suspended users cannot access anything (fluent API)
deny_rule = (HbacRuleBuilder("deny_suspended")
.user_group("suspended")
.host_category_all()
.service_category_all()
.deny()
.enabled(True)
.build())
policy.add_rule(deny_rule)
# Engineer with normal access
engineer = Subject("dave")
engineer.add_group("engineers")
request = HbacRequest(engineer, "dev-01", "sshd")
request.add_targethost_group("development")
print(policy.check_access(request)) # True
# Suspended engineer (deny takes precedence)
suspended = Subject("eve")
suspended.add_group("engineers")
suspended.add_group("suspended")
request = HbacRequest(suspended, "dev-01", "sshd")
request.add_targethost_group("development")
result = policy.evaluate(request)
print(result.is_denied) # True
Category Matching
from hbac_rs import HbacRuleBuilder, HbacRequest, HbacPolicy, Subject
policy = HbacPolicy()
# Emergency access: allow on-call engineers to access everything (fluent API)
emergency_rule = (HbacRuleBuilder("emergency_access")
.user_group("oncall")
.host_category_all()
.service_category_all()
.enabled(True)
.build())
policy.add_rule(emergency_rule)
# On-call engineer can access any service on any host
oncall = Subject("frank")
oncall.add_group("oncall")
# Access production database
request1 = HbacRequest(oncall, "prod-db-01", "postgresql")
print(policy.check_access(request1)) # True
# Access web server
request2 = HbacRequest(oncall, "web-server", "httpd")
print(policy.check_access(request2)) # True
Performance
The Python bindings use PyO3’s zero-cost abstractions with the single-threaded HbacPolicyLocal variant:
- Native speed: Rule evaluation runs at Rust performance
- Minimal overhead: Type conversions only at the Python/Rust boundary
- Optimized caching: 5-layer evaluation pipeline with LRU cache and Bloom filters
For high-throughput scenarios:
- Build your policy once and reuse it for multiple evaluations
- Add all rules before starting evaluations (triggers pipeline optimization)
- Consider batching if possible
See Also
- HBAC Rules - Detailed rule semantics
- HBAC Evaluation - Policy evaluation details
- HBAC Caching - Performance optimization
- Python Bindings for ACLS - Core ABAC bindings