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

groups()

Get all group memberships.

Returns: list[str]

name()

Get the user’s name.

Returns: 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 user
• user_group(group: str) -> HbacRuleBuilder: Add a user group
• user_category_all() -> HbacRuleBuilder: Match all users

Host Dimension

All methods return HbacRuleBuilder for fluent chaining:

• host(host: str) -> HbacRuleBuilder: Add a specific host
• host_group(group: str) -> HbacRuleBuilder: Add a host group
• host_category_all() -> HbacRuleBuilder: Match all hosts

Service Dimension

All methods return HbacRuleBuilder for fluent chaining:

• service(service: str) -> HbacRuleBuilder: Add a specific service
• service_group(group: str) -> HbacRuleBuilder: Add a service group
• service_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())

HbacRequest

Represents an access request to evaluate.

HbacRequest(user: Subject, targethost: str, service: str)

Parameters:

• user: The user attempting access
• targethost: The target host being accessed
• service: 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

targethost()

Get the target host.

Returns: str

service()

Get the service.

Returns: 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()

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

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

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

rules() -> list[HbacRule]

Get all rules in the policy.

Returns: list[HbacRule]

rule_count() -> int

Get the number of rules in the policy.

Returns: int

clear() -> None

Remove all rules from the policy.

Returns: None

max_rules() -> int

Get the configured rule limit. Returns 0 if there is no limit.

Returns: int

stats() -> CacheStats

Get evaluation cache statistics.

Returns: CacheStats

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 request
• timestamp (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 request
• timestamp (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.

Methods:

is_allowed()

Check if access is allowed.

Returns: bool

is_denied()

Check if access is explicitly denied.

Returns: bool

warnings() -> list[str]

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.

Returns: list[str]

has_warnings() -> bool

Check whether any warnings were generated.

Returns: bool

matched_rules() -> list[str]

Return the names of rules that matched the request. Only populated when the request was evaluated via evaluate_detailed() or evaluate_detailed_at().

Returns: list[str]

not_matched_rules() -> list[str]

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().

Returns: 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 membership
• service_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 underlying HbacRule
• valid_from: Millisecond timestamp at which the rule becomes active (inclusive). None means active from the beginning of time.
• valid_until: Millisecond timestamp at which the rule expires (exclusive). None means 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 for duration_ms milliseconds starting from now.

Methods:

is_currently_active() -> bool

Check whether the rule is active at the current wall-clock time.

Returns: bool

is_active_at(timestamp: int) -> bool

Check whether the rule would be active at the given millisecond timestamp.

Returns: bool

Example:

from hbac_rs import HbacRuleBuilder, TemporalHbacRule, HbacPolicy, current_timestamp_millis

base_rule = (HbacRuleBuilder("temp_access")
    .user_group("contractors")
    .host_group("staging")
    .service("sshd")
    .enabled(True)
    .build())

now = current_timestamp_millis()
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_active())    # True
print(temp_rule.is_active_at(now + one_hour + 1))  # False

HbacResource

Models a protected resource that requires a specific AtomicPermission from acls_rs. 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) -> bool

Grant the required permission to the subject.

Returns: bool

deny_access(subject: Subject) -> bool

Explicitly deny the required permission for the subject.

Returns: bool

required_permission() -> AtomicPermission

Return the AtomicPermission this resource requires.

Returns: AtomicPermission (from acls_rs)

HbacResourceBuilder

Fluent builder for HbacResource.

Example:

from acls_rs import AtomicPermission
from hbac_rs import HbacResourceBuilder, Subject

resource = (HbacResourceBuilder()
    .required_permission(AtomicPermission("db", "read"))
    .build())

user = Subject("alice")
resource.grant_access(user)
print(resource.can_access(user))   # True

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")

current_timestamp_millis

Utility function that returns the current wall-clock time as a millisecond Unix timestamp. Use this with evaluate_at(), TemporalHbacRule, and TemporalPermission.

from hbac_rs import current_timestamp_millis

now = current_timestamp_millis()
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:

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