Python Bindings: ABAC

The abac-rs crate provides Python bindings for generic Attribute-Based Access Control (ABAC) evaluation. This allows you to use the high-performance Rust ABAC engine from Python applications.

Installation

# Using maturin (recommended for development)
cd crates/abac-python
maturin develop

# Or build and install a wheel
maturin build --release
pip install target/wheels/abac_rs-*.whl

See the Installation Guide for detailed setup instructions.

Quick Start

from abac_rs import AbacRuleBuilder, AbacRequest, AbacPolicy, AttributeType

# Create a policy
policy = AbacPolicy()

# Add a rule: engineers can read production databases (using fluent API)
rule = (AbacRuleBuilder("allow-engineers-prod-read")
    .dimension_values("user", [AttributeType.string("group:engineers")])
    .dimension_values("resource", [AttributeType.string("prod:db-01")])
    .dimension_values("action", [AttributeType.string("read")])
    .enabled(True)
    .build())
policy.add_rule(rule)

# Evaluate a request
request = AbacRequest()
request.add_attribute("user", AttributeType.string("alice"), 
                     [AttributeType.string("group:engineers")])
request.add_attribute("resource", AttributeType.string("prod:db-01"), [])
request.add_attribute("action", AttributeType.string("read"), [])

if policy.evaluate(request):
    print("Access granted!")

API Reference

AttributeType

Represents a typed attribute value that can be used in ABAC rules and requests.

Methods

AttributeType.string(value: str) -> AttributeType

Create a string attribute.

attr = AttributeType.string("group:engineers")

AttributeType.integer(value: int) -> AttributeType

Create an integer attribute.

attr = AttributeType.integer(42)

AttributeType.float(value: float) -> AttributeType

Create a floating-point attribute.

attr = AttributeType.float(3.14)

AttributeType.ip_addr(value: str) -> AttributeType

Create an IP address attribute.

attr = AttributeType.ip_addr("192.168.1.1")

AttributeType.ip_cidr(value: str) -> AttributeType

Create an IP CIDR range attribute.

attr = AttributeType.ip_cidr("10.0.0.0/8")

AbacRuleBuilder

Builder for creating ABAC rules.

Constructor

builder = AbacRuleBuilder(name: str)

Methods

dimension_values(dimension: str, values: list[AttributeType]) -> AbacRuleBuilder

Add specific attribute values for a dimension. Returns self for method chaining.

builder.dimension_values("user", [
    AttributeType.string("group:engineers"),
    AttributeType.string("group:seniors")
])

dimension_all(dimension: str) -> AbacRuleBuilder

Mark a dimension to match any value (wildcard). Returns self for method chaining.

builder.dimension_all("user")  # Matches any user

enabled(is_enabled: bool) -> AbacRuleBuilder

Enable or disable the rule (default: disabled). Returns self for method chaining.

builder.enabled(True)

deny() -> AbacRuleBuilder

Mark this as a deny rule (default: allow rule). Returns self for method chaining.

builder.deny()

build() -> AbacRule

Build the final rule.

rule = builder.build()

AbacRule

Represents a compiled ABAC rule. Created via AbacRuleBuilder.build().

Methods

name() -> str

Get the rule name.

print(rule.name())  # "allow-engineers-prod-read"

is_enabled() -> bool

Check if the rule is enabled.

if rule.is_enabled():
    print("Rule is active")

is_deny() -> bool

Check if this is a deny rule.

if rule.is_deny():
    print("This rule denies access")

dimension_names() -> list[str]

Get all dimension names in this rule.

dims = rule.dimension_names()  # ["user", "resource", "action"]

get_dimension(dimension: str) -> list[AttributeType] | None

Get the specific values for a dimension. Returns None if the dimension uses wildcard (match all).

users = rule.get_dimension("user")
if users:
    print(f"Rule matches users: {users}")

dimension_is_all(dimension: str) -> bool

Check if a dimension is set to match all values (wildcard).

if rule.dimension_is_all("resource"):
    print("Rule applies to all resources")

to_json() -> str

Serialize the rule to JSON format.

json_str = rule.to_json()
# Save to file or transfer over network

AbacRule.from_json(json: str) -> AbacRule (static method)

Deserialize a rule from JSON.

rule = AbacRule.from_json(json_str)
print(rule.name())

AbacRequest

Represents an access request with multiple dimensions and attributes.

Constructor

request = AbacRequest()

Methods

add_attribute(dimension: str, primary: AttributeType, secondary: list[AttributeType]) -> None

Add an attribute to a dimension. The primary attribute is the main value, and secondary attributes are additional values (like group memberships).

# User with group memberships
request.add_attribute(
    "user",
    AttributeType.string("alice"),
    [AttributeType.string("group:engineers"), AttributeType.string("group:seniors")]
)

# Resource with no additional attributes
request.add_attribute(
    "resource",
    AttributeType.string("prod:db-01"),
    []
)

# Action
request.add_attribute(
    "action",
    AttributeType.string("read"),
    []
)

dimensions() -> list[str]

Get all dimension names in the request.

dims = request.dimensions()  # ["user", "resource", "action"]

get_value(dimension: str) -> AttributeType | None

Get the primary value for a dimension.

user = request.get_value("user")
if user:
    print(f"Request from user: {user}")

get_groups(dimension: str) -> list[AttributeType]

Get the secondary values (groups/memberships) for a dimension.

groups = request.get_groups("user")  # ["group:engineers", "group:seniors"]

has_dimension(dimension: str) -> bool

Check if a dimension exists in the request.

if request.has_dimension("user"):
    print("User dimension is set")

AbacPolicy

The main policy engine that evaluates requests against rules.

Constructors

# Create empty policy with default settings
policy = AbacPolicy()

# Create policy with custom cache size
policy = AbacPolicy.with_cache_size(2048)

# Create policy with custom rule capacity and cache size
policy = AbacPolicy.with_capacity(rules_capacity=100, cache_size=1000)

# Create policy with maximum rule limit
policy = AbacPolicy.with_max_rules(50)

Methods

add_rule(rule: AbacRule) -> AbacPolicy

Add a single rule to the policy. Returns self for method chaining (fluent API).

# Non-fluent style
policy.add_rule(rule)

# Fluent style - chain multiple rules
policy = (AbacPolicy()
    .add_rule(rule1)
    .add_rule(rule2)
    .add_rule(rule3))

load_rules(rules: list[AbacRule]) -> AbacPolicy

Add multiple rules in one batch (more efficient than individual add_rule() calls). Returns self for method chaining.

rules = [rule1, rule2, rule3]

# Non-fluent style
policy.load_rules(rules)

# Fluent style
policy = AbacPolicy().load_rules(rules)

rule_count() -> int

Get the number of rules in the policy.

print(f"Policy has {policy.rule_count()} rules")

stats() -> CacheStats

Get policy statistics including cache performance metrics.

stats = policy.stats()
print(f"Cache fill rate: {stats.cache_fill_rate():.2%}")
print(f"Memory usage: {stats.memory_bytes} bytes")

enable_rule(name: str) -> bool

Enable a rule by name. Returns True if the rule was found and enabled.

if policy.enable_rule("allow-engineers"):
    print("Rule enabled")

disable_rule(name: str) -> bool

Disable a rule by name. Returns True if the rule was found and disabled.

if policy.disable_rule("deny-contractors"):
    print("Rule disabled")

remove_rule(name: str) -> bool

Remove a rule from the policy. Returns True if the rule was found and removed.

if policy.remove_rule("old-rule"):
    print("Rule removed")

clear() -> None

Remove all rules from the policy.

policy.clear()

get_rule(name: str) -> AbacRule | None

Get a rule by name.

rule = policy.get_rule("allow-engineers")
if rule:
    print(f"Found rule: {rule.name()}")

has_rule(name: str) -> bool

Check if a rule exists in the policy.

if policy.has_rule("allow-engineers"):
    print("Rule exists")

rules() -> list[AbacRule]

Get all rules in the policy.

for rule in policy.rules():
    print(f"Rule: {rule.name()}")

export_rules() -> str

Export all rules to JSON format.

json_data = policy.export_rules()
with open('policy.json', 'w') as f:
    f.write(json_data)

import_rules(json: str) -> None

Import rules from JSON. Rules are loaded in batch using load_rules().

with open('policy.json') as f:
    json_data = f.read()
policy.import_rules(json_data)

evaluate(request: AbacRequest) -> Decision

Evaluate a request against the policy.

decision = policy.evaluate(request)
if decision.is_allowed():
    print("Access granted")

Evaluation logic:

• If any deny rule matches, access is denied
• If any allow rule matches (and no deny rules match), access is granted
• If no rules match, access is denied (default deny)

evaluate_at(request: AbacRequest, timestamp: int) -> Decision

Evaluate a request at a specific millisecond timestamp. Temporal rules are checked against timestamp instead of the current wall-clock time. Useful for testing, auditing, or pre-computing decisions for a known future point.

from abac_rs import current_timestamp_millis

future = current_timestamp_millis() + 3600 * 1000  # 1 hour from now
decision = policy.evaluate_at(request, future)
if decision.is_allowed():
    print("Access will be granted in one hour")

max_rules() -> int

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

print(f"Rule limit: {policy.max_rules() or 'none'}")

Use AbacPolicy.with_max_rules(n) to create a policy with a rule cap:

policy = AbacPolicy.with_max_rules(50)
print(policy.max_rules())  # 50

AbacPolicyThreadSafe

Thread-safe variant of AbacPolicy using Mutex for concurrent access from multiple Python threads.

API: Same as AbacPolicy - all constructors and methods are identical.

from threading import Thread
from abac_rs import AbacPolicyThreadSafe, AbacRequest, AttributeType

# Create thread-safe policy
policy = AbacPolicyThreadSafe()

# Use from multiple threads
def worker(policy, user_id):
    request = AbacRequest()
    request.add_attribute("user", AttributeType.string(f"user{user_id}"), [])
    decision = policy.evaluate(request)
    print(f"User {user_id}: {decision}")

threads = [Thread(target=worker, args=(policy, i)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

TemporalAbacRule

Temporal rules add time-based validity to ABAC rules.

Constructors

# Rule valid between specific timestamps
temporal = TemporalAbacRule(rule, valid_from=1000, valid_until=2000)

# Rule valid until a specific time
temporal = TemporalAbacRule.valid_until(rule, until_timestamp)

# Rule valid for a duration from now
temporal = TemporalAbacRule.valid_for_duration(rule, duration_ms=3600000)  # 1 hour

# Rule valid from a specific time onwards
temporal = TemporalAbacRule.valid_from(rule, from_timestamp)

Methods

is_currently_valid() -> bool

Check if the rule is valid at the current time.

if temporal.is_currently_valid():
    print("Rule is active")

is_valid_at(timestamp: int) -> bool

Check if the rule is valid at a specific timestamp.

if temporal.is_valid_at(future_time):
    print("Rule will be valid")

inner() -> AbacRule

Get the underlying ABAC rule.

rule = temporal.inner()
print(rule.name())

Helper Function

current_timestamp_millis() -> int

Get the current timestamp in milliseconds.

from abac_rs import current_timestamp_millis

now = current_timestamp_millis()
future = now + 3600000  # 1 hour from now

Policy Composition

ComposedPolicy allows you to combine ABAC and RBAC policies for hybrid access control.

Important: Import acls_rs and hbac_rs before using ComposedPolicy:

import acls_rs  # Import first
import hbac_rs   # Import first
from abac_rs import ComposedPolicy, CompositionMode

CompositionMode

Determines how ABAC and RBAC results are combined.

Static Methods

CompositionMode.and_mode() -> CompositionMode

Both ABAC and RBAC must allow. Use for defense in depth.

from abac_rs import CompositionMode
mode = CompositionMode.and_mode()

CompositionMode.or_mode() -> CompositionMode

Either ABAC or RBAC allowing is sufficient.

mode = CompositionMode.or_mode()

CompositionMode.abac_first() -> CompositionMode

ABAC takes precedence. If ABAC has matching rules, use that decision. Otherwise fall back to RBAC.

mode = CompositionMode.abac_first()

CompositionMode.rbac_first() -> CompositionMode

RBAC takes precedence. If user has roles with relevant permissions, use that decision. Otherwise fall back to RBAC.

mode = CompositionMode.rbac_first()

ComposedPolicy

Unified policy combining ABAC and RBAC evaluation.

Constructor

from acls_rs import RbacPolicy, Role, AtomicPermission
from abac_rs import ComposedPolicy, CompositionMode, AbacRuleBuilder, AttributeType
from hbac_rs import Subject

# Create ABAC rules
abac_rules = []
rule = (AbacRuleBuilder("allow-engineers-prod-read")
    .dimension_values("user", [AttributeType.string("group:engineers")])
    .dimension_values("resource", [AttributeType.string("database")])
    .dimension_values("action", [AttributeType.string("read")])
    .enabled(True)
    .build())
abac_rules.append(rule)

# Create RBAC policy
rbac = RbacPolicy()
engineer = Role("engineer", 
    grants=[AtomicPermission("database", "read"), 
            AtomicPermission("database", "write")],
    denials=[])
rbac.add_role(engineer)

# Compose policies with an RbacPolicy instance
policy = ComposedPolicy(abac_rules, rbac, CompositionMode.and_mode())

Methods

evaluate(request: AbacRequest, subject: Subject) -> Decision

Evaluate a request against both ABAC and RBAC policies.

from hbac_rs import Subject
from abac_rs import AbacRequest, AttributeType

# Create subject with group memberships
user = Subject("alice")
user.add_group("engineer")

# Create request
request = AbacRequest()
request.add_attribute("user", AttributeType.string("alice"),
                     [AttributeType.string("group:engineers")])
request.add_attribute("resource", AttributeType.string("database"), [])
request.add_attribute("action", AttributeType.string("read"), [])

# Evaluate against composed policy
decision = policy.evaluate(request, user)
if decision.is_allowed():
    print("Access granted by both ABAC and RBAC")

RBAC Permission Mapping:

ComposedPolicy extracts resource and action attributes from the ABAC request to form the RBAC permission:

• resource attribute → permission namespace
• action attribute → permission action

For example, resource="database" and action="read" maps to checking the database:read permission in RBAC.

If these attributes are missing, RBAC evaluation is skipped and only the ABAC result is used.

mode() -> CompositionMode

Get the current composition mode.

current_mode = policy.mode()

set_mode(mode: CompositionMode) -> None

Change the composition mode at runtime.

policy.set_mode(CompositionMode.or_mode())

rbac_policy() -> RbacPolicy

Get a reference to the RBAC policy (read-only).

rbac = policy.rbac_policy()
num_roles = len(rbac.roles())  # Count roles

invalidate_permission_cache() -> None

Invalidate the RBAC permission cache. Call this after modifying the RBAC policy externally.

policy.invalidate_permission_cache()

Complete Example

import acls_rs  # Import first
import hbac_rs   # Import first

from abac_rs import (
    ComposedPolicy, CompositionMode, AbacRuleBuilder, 
    AbacRequest, AttributeType
)
from acls_rs import RbacPolicy, Role, AtomicPermission
from hbac_rs import Subject

# ABAC: attribute-based rules for fine-grained control
abac_rules = []

# Engineers can read production databases
rule1 = (AbacRuleBuilder("allow-engineers-prod-read")
    .dimension_values("user", [AttributeType.string("group:engineers")])
    .dimension_values("resource", [AttributeType.string("database")])
    .dimension_values("action", [AttributeType.string("read")])
    .enabled(True)
    .build())
abac_rules.append(rule1)

# RBAC: role-based permissions for organizational hierarchy
rbac = RbacPolicy()

engineer_role = Role("engineer",
    grants=[
        AtomicPermission("database", "read"),
        AtomicPermission("database", "write"),
    ],
    denials=[])
rbac.add_role(engineer_role)

dba_role = Role("dba",
    grants=[
        AtomicPermission("database", "read"),
        AtomicPermission("database", "write"),
        AtomicPermission("database", "delete"),
    ],
    denials=[])
rbac.add_role(dba_role)

# Compose with AND mode: both must agree
policy = ComposedPolicy(abac_rules, rbac, CompositionMode.and_mode())

# Test 1: Engineer reading production database
alice = Subject("alice")
alice.add_group("engineer")

request1 = AbacRequest()
request1.add_attribute("user", AttributeType.string("alice"),
                      [AttributeType.string("group:engineers")])
request1.add_attribute("resource", AttributeType.string("database"), [])
request1.add_attribute("action", AttributeType.string("read"), [])

decision1 = policy.evaluate(request1, alice)
assert decision1.is_allowed()  # ABAC allows (rule matches) AND RBAC allows (role grants permission)

# Test 2: Engineer trying to delete (RBAC denies, no ABAC rule)
request2 = AbacRequest()
request2.add_attribute("user", AttributeType.string("alice"),
                      [AttributeType.string("group:engineers")])
request2.add_attribute("resource", AttributeType.string("database"), [])
request2.add_attribute("action", AttributeType.string("delete"), [])

decision2 = policy.evaluate(request2, alice)
assert decision2.is_denied()  # ABAC denies (no matching rule) AND RBAC denies (engineer lacks delete)

# Test 3: DBA deleting (has RBAC permission but no ABAC rule)
bob = Subject("bob")
bob.add_group("dba")

request3 = AbacRequest()
request3.add_attribute("user", AttributeType.string("bob"),
                      [AttributeType.string("group:dba")])
request3.add_attribute("resource", AttributeType.string("database"), [])
request3.add_attribute("action", AttributeType.string("delete"), [])

decision3 = policy.evaluate(request3, bob)
assert decision3.is_denied()  # ABAC denies (no rule for DBA delete), so AND mode denies

Use Cases

Defense in Depth (AND mode):

• Both ABAC and RBAC must approve
• Fine-grained attribute checks + role verification
• Maximum security for critical resources

Convenience First (OR mode):

• Either ABAC or RBAC allowing is sufficient
• Useful during migration from one model to another
• Fallback mechanism when one policy is incomplete

Hybrid (ABAC First / RBAC First):

• Use primary policy where it has coverage
• Fall back to secondary policy otherwise
• Gradual migration from RBAC → ABAC or vice versa

CacheStats

Statistics about policy cache performance and memory usage.

Important: The ABAC policy uses a multi-layer optimization strategy. When faster optimization layers are available (deny_index for deny-only policies, compiled evaluator for certain rule patterns), the LRU cache is bypassed entirely. This means cache_entries may remain 0 even under heavy load — this is expected and indicates optimal performance.

Fields (read-only)

stats.rule_count          # Total number of rules
stats.enabled_rule_count  # Number of enabled rules
stats.cache_capacity      # Maximum cache entries
stats.cache_entries       # Current cache entries (may be 0 if optimized away)
stats.memory_bytes        # Estimated memory usage

Methods

cache_fill_rate() -> float

Get the cache utilization as a ratio (0.0 to 1.0).

Note: Returns 0.0 when optimization layers bypass caching.

fill_rate = stats.cache_fill_rate()
print(f"Cache is {fill_rate * 100:.1f}% full")

Decision

Result of a policy evaluation.

Methods

is_allowed() -> bool

Check if access is allowed.

if decision.is_allowed():
    print("Access granted")

is_denied() -> bool

Check if access is denied.

if decision.is_denied():
    print("Access denied")

Error Handling

The abac-rs Python bindings provide specific exception types for different error scenarios:

PolicyError

Raised when policy operations fail (e.g., too many rules).

from abac_rs import AbacPolicy, PolicyError

try:
    policy = AbacPolicy.with_max_rules(10)
    # Try to load more than max
    policy.load_rules(many_rules)
except PolicyError as e:
    print(f"Policy error: {e}")

RequestError

Raised when request operations fail (e.g., too many attributes).

from abac_rs import AbacRequest, AttributeType, RequestError

try:
    request = AbacRequest()
    # Try to add too many attributes (max 100)
    for i in range(150):
        request.add_attribute(f"dim{i}", AttributeType.string("value"), [])
except RequestError as e:
    print(f"Request error: {e}")

TemporalError

Raised when temporal rule creation fails (e.g., invalid time window).

from abac_rs import TemporalAbacRule, TemporalError

try:
    # Invalid: from >= until
    temporal = TemporalAbacRule(rule, valid_from=2000, valid_until=1000)
except TemporalError as e:
    print(f"Temporal error: {e}")

Examples

Database Access Control

from abac_rs import AbacRuleBuilder, AbacRequest, AbacPolicy, AttributeType

policy = AbacPolicy()

# Rule: Engineers can read production databases (fluent API)
rule1 = (AbacRuleBuilder("allow-engineers-prod-read")
    .dimension_values("user", [AttributeType.string("group:engineers")])
    .dimension_values("resource", [
        AttributeType.string("prod:db-01"),
        AttributeType.string("prod:db-02")
    ])
    .dimension_values("action", [AttributeType.string("read")])
    .enabled(True)
    .build())
policy.add_rule(rule1)

# Rule: DBAs have full access (fluent API)
rule2 = (AbacRuleBuilder("allow-dba-all")
    .dimension_values("user", [AttributeType.string("group:dba")])
    .dimension_values("resource", [
        AttributeType.string("prod:db-01"),
        AttributeType.string("prod:db-02")
    ])
    .dimension_values("action", [
        AttributeType.string("read"),
        AttributeType.string("write"),
        AttributeType.string("delete")
    ])
    .enabled(True)
    .build())
policy.add_rule(rule2)

# Rule: Deny junior engineers from modifying production (fluent API)
rule3 = (AbacRuleBuilder("deny-junior-prod-write")
    .dimension_values("user", [AttributeType.string("group:junior")])
    .dimension_values("resource", [
        AttributeType.string("prod:db-01"),
        AttributeType.string("prod:db-02")
    ])
    .dimension_values("action", [
        AttributeType.string("write"),
        AttributeType.string("delete")
    ])
    .deny()
    .enabled(True)
    .build())
policy.add_rule(rule3)

# Test: Engineer reading production (allowed)
request = AbacRequest()
request.add_attribute("user", AttributeType.string("alice"),
                     [AttributeType.string("group:engineers")])
request.add_attribute("resource", AttributeType.string("prod:db-01"), [])
request.add_attribute("action", AttributeType.string("read"), [])
assert policy.evaluate(request) == True

# Test: Junior engineer trying to delete (denied by explicit deny rule)
request2 = AbacRequest()
request2.add_attribute("user", AttributeType.string("charlie"),
                      [AttributeType.string("group:junior"),
                       AttributeType.string("group:engineers")])
request2.add_attribute("resource", AttributeType.string("prod:db-01"), [])
request2.add_attribute("action", AttributeType.string("delete"), [])
assert policy.evaluate(request2) == False  # Deny rule takes precedence

Multi-Type Attributes

ABAC supports multiple attribute types:

from abac_rs import AttributeType

# String attributes (most common)
group = AttributeType.string("group:engineers")
resource = AttributeType.string("prod:db-01")

# Integer attributes (for numeric IDs, priorities, etc.)
priority = AttributeType.integer(5)
user_id = AttributeType.integer(1234)

# Float attributes (for weights, scores, etc.)
score = AttributeType.float(0.95)
threshold = AttributeType.float(3.14)

# IP addresses
client_ip = AttributeType.ip_addr("192.168.1.100")

# IP CIDR ranges
internal_network = AttributeType.ip_cidr("10.0.0.0/8")

Custom Dimensions

ABAC is not limited to user/resource/action. You can use any dimension names:

from abac_rs import AbacRuleBuilder, AbacRequest, AbacPolicy, AttributeType

policy = AbacPolicy()

# Rule with custom dimensions (fluent API)
rule = (AbacRuleBuilder("time-based-access")
    .dimension_values("user", [AttributeType.string("group:operators")])
    .dimension_values("operation", [AttributeType.string("deploy")])
    .dimension_values("environment", [AttributeType.string("staging")])
    .dimension_values("time", [AttributeType.string("business-hours")])
    .enabled(True)
    .build())
policy.add_rule(rule)

# Request with custom dimensions
request = AbacRequest()
request.add_attribute("user", AttributeType.string("bob"),
                     [AttributeType.string("group:operators")])
request.add_attribute("operation", AttributeType.string("deploy"), [])
request.add_attribute("environment", AttributeType.string("staging"), [])
request.add_attribute("time", AttributeType.string("business-hours"), [])

if policy.evaluate(request):
    print("Deployment authorized")

Custom Matchers

Custom matchers allow you to implement domain-specific matching logic beyond the default exact matching. This enables predicates, ranges, CIDR matching, and other custom semantics.

Matcher Protocol

A custom matcher is any Python callable (function, lambda, or class with __call__) that takes three arguments and returns a boolean:

def my_matcher(rule_value, request_value, request_groups):
    """
    Args:
        rule_value: List of AttributeType from rule, or None for wildcard
        request_value: AttributeType from request  
        request_groups: List of AttributeType group memberships
    
    Returns:
        bool: True if request matches rule requirement
    """
    # Custom matching logic here
    return matches

Registering Matchers

Register custom matchers for specific dimensions before adding rules:

from abac_rs import AbacPolicy, AttributeType, AbacRuleBuilder

policy = AbacPolicy()

# Define a range matcher for numeric attributes
def range_matcher(rule_value, request_value, request_groups):
    if rule_value is None:
        return True  # Wildcard matches everything
    
    # Expect rule to contain [min, max] bounds
    if len(rule_value) != 2:
        return False
    
    # Extract values (simplified - production code needs type checking)
    min_val = rule_value[0]  # AttributeType
    max_val = rule_value[1]
    req_val = request_value
    
    # Custom range check logic here
    return True  # Simplified

# Register matcher for 'priority' dimension
policy.register_matcher("priority", range_matcher)

# Now add rules using that dimension
rule = (
    AbacRuleBuilder("priority-access")
    .dimension_values("priority", [
        AttributeType.integer(1),    # min
        AttributeType.integer(10)    # max
    ])
    .dimension_values("action", [AttributeType.string("access")])
    .enabled(True)
    .build()
)
policy.add_rule(rule)

CIDR Matcher Example

import ipaddress

def cidr_matcher(rule_value, request_value, request_groups):
    """Match IP addresses against CIDR ranges."""
    if rule_value is None:
        return True
    
    # Parse request IP (simplified)
    request_ip = ipaddress.ip_address("192.168.1.100")
    
    # Check if IP is in any CIDR range
    for cidr in rule_value:
        network = ipaddress.ip_network("192.168.0.0/16")
        if request_ip in network:
            return True
    
    return False

policy.register_matcher("client_ip", cidr_matcher)

Predicate Matcher Example

import re

def regex_matcher(rule_value, request_value, request_groups):
    """Match strings against regex patterns."""
    if rule_value is None:
        return True
    
    # Get request string value (simplified)
    request_str = str(request_value)
    
    # Check against all patterns
    for pattern in rule_value:
        pattern_str = str(pattern)
        if re.match(pattern_str, request_str):
            return True
    
    return False

policy.register_matcher("resource_path", regex_matcher)

Important Notes

1. Register before adding rules: Matchers must be registered before adding rules that use them
2. Performance: Custom matchers are slower than exact matching and disable Bloom filter optimization
3. Error handling: Matcher exceptions are caught and treated as non-matches (deny by default)
4. Thread safety: Matchers must be thread-safe if using AbacPolicyThreadSafe
5. GIL constraints: Matchers are called from Rust code with GIL held, so keep them fast

Complete Example

See crates/abac-python/examples/custom_matchers.py for a complete working example with range, CIDR, and predicate matchers.

Performance

The Python bindings use the same high-performance Rust engine as the native library:

• Zero-copy evaluation: Minimal overhead for crossing the Python/Rust boundary
• Compiled policies: Rules are pre-compiled for fast evaluation
• Bitmap indexing: Deny rules use bitmap-based fast path
• JIT optimization: Optional JIT compilation for maximum performance

For high-throughput applications, create the policy once and reuse it for multiple evaluations:

# Initialize policy once (expensive)
policy = AbacPolicy()
for rule in rules:
    policy.add_rule(rule)

# Evaluate many requests (fast)
for request in requests:
    result = policy.evaluate(request)

Differences from Rust API

The Python bindings closely follow the Rust API but with some Pythonic adjustments:

• Builder pattern: Methods return self for fluent chaining, just like the Rust API
• Type constructors: Use static methods like AttributeType.string() instead of enum variants
• No lifetime parameters: Memory management is automatic via Python GC
• Thread safety: Policies use single-threaded variants (safe with Python GIL)

Examples Directory

See crates/abac-python/examples/ for complete working examples:

# Basic ABAC usage
python3 crates/abac-python/examples/basic_usage.py

# Advanced features (IP types, wildcards, introspection)
python3 crates/abac-python/examples/advanced_features.py

# Temporal rules and thread-safe policies
python3 crates/abac-python/examples/temporal_and_threading.py

# JSON serialization and persistence
python3 crates/abac-python/examples/serialization.py

# Custom matcher functions
python3 crates/abac-python/examples/custom_matchers.py

See Also

• ABAC Rust Documentation - Native Rust API
• ABAC Policy Composition - Composing ABAC with RBAC in Rust
• acls-rs Python Bindings - Foundation library Python API
• hbac-rs Python Bindings - HBAC-specific Python API
• Installation Guide - Setup instructions