Keyboard shortcuts

Press or to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

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

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

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 request
  • timestamp (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 serialized HbacRule objects

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

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

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:

ModeBehaviour
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