Skip to main content

Casbin Authorization Module

info

The Casbin module provides policy-based access control (RBAC, ABAC) for Fluvius domains.

Introduction​

The Casbin module integrates with the Casbin authorization library to provide:

  • Role-Based Access Control (RBAC)
  • Attribute-Based Access Control (ABAC)
  • Policy management
  • Command and query authorization
  • Domain-level policy enforcement

Quick Start​

Configure Policy Manager​

from fluvius.casbin import PolicyManager

policymgr = PolicyManager(
    model_path='path/to/model.conf',
    policy_adapter=policy_adapter
)

Enable in Domain​

from fluvius.domain import Domain
from fluvius.casbin import PolicyManager

class UserDomain(Domain):
    __policymgr__ = PolicyManager(...)
    __aggregate__ = UserAggregate

Define Policies​

Policies are defined in CSV or database:

p, alice, user, create
p, alice, user, read
p, bob, user, read
g, alice, admin

Check Authorization​

Authorization is automatically checked when processing commands:

# Policy is checked automatically
response = await domain.process_command(command)

Policy Models​

RBAC Model​

[request_definition]
r = sub, obj, act

[policy_definition]
p = sub, obj, act

[role_definition]
g = _, _

[policy_effect]
e = some(where (p.eft == allow))

[matchers]
m = g(r.sub, p.sub) && r.obj == p.obj && r.act == p.act

ABAC Model​

[request_definition]
r = sub, obj, act

[policy_definition]
p = sub, obj, act, eft

[policy_effect]
e = some(where (p.eft == allow))

[matchers]
m = r.sub == p.sub && r.obj == p.obj && r.act == p.act

Integration with Domains​

Domains automatically check policies:

# Policy is checked before command processing
# If policy check fails, ForbiddenError is raised
try:
    response = await domain.process_command(command)
except ForbiddenError:
    # Handle authorization failure
    pass

Next Steps​