fastapi-casbin-acl 0.2.0

FastAPI permissions infrastructure dependency with RBAC+ABAC support using Casbin

FastAPI Casbin ACL

A robust, production-ready FastAPI infrastructure dependency for permissions management, providing RBAC and ABAC support via Casbin.

Features

• Zero Business Logic Intrusion: Decoupled from your business logic.
• Strong Constraints, Few Conventions: Enforces a consistent permission model (RBAC + ABAC + Permission RBAC).
• Pluggable Authentication: Works with any authentication system (JWT, OAuth2, Session) via dependency injection.
• Casbin Lifecycle Management: Centralized management of the Casbin Enforcer.
• Async Support: Fully compatible with FastAPI's async nature.
• Multiple Model Support: Support for multiple permission models (RBAC, ABAC, Permission RBAC, custom) with per-route model selection.
• Permission RBAC Model: Advanced RBAC model with permission abstraction layer, supporting API-to-Permission mapping using route names (api_name).
• Model Registry: Built-in model registry with automatic registration of built-in models.
• Multi-Worker Synchronization: Redis-based policy synchronization for multi-process deployments (e.g., gunicorn), ensuring all workers have consistent policy state in real-time.
• CLI Tools: Command-line tools for policy management, API synchronization, and policy explanation.
• Policy Management API: Built-in REST API for managing policies, roles, permissions, and API mappings.
• Frontend Integration: Example frontend interface for dynamic policy management and visualization.

Installation

pip install fastapi-casbin-acl

Quick Start

1. Initialize

In your startup logic (e.g., main.py):

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi_casbin_acl.enforcer import acl
from fastapi_casbin_acl.config import ACLConfig
from fastapi_casbin_acl.adapter import SQLModelAdapter
from sqlalchemy.ext.asyncio import AsyncSession

# Example: Using SQLModel adapter
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize database connection
    # ... your database setup ...
    
    # Initialize Casbin ACL
    adapter = SQLModelAdapter(AsyncSessionLocal)
    config = ACLConfig(
        policy_router_enable=True,  # Optional: Enable policy management API
        # redis_url="redis://localhost:6379/0",  # Optional: Enable multi-worker sync
    )
    await acl.init(adapter=adapter, config=config, app=app)
    
    # Note: When app is provided, the system automatically builds an api_name mapping table
    # This table maps route names to route paths for debugging and validation
    # If app is not provided, a warning will be issued and api_name-based matching won't be available
    
    # Initialize policies (example for Permission RBAC)
    # await init_policies()
    
    yield
    
    # Cleanup
    await acl.shutdown()  # Stop watcher if Redis is configured
    # ... your cleanup logic ...

app = FastAPI(lifespan=lifespan)

Multiple Models Support:

# Initialize multiple models at once
await acl.init(
    adapter=adapter,
    models=["rbac", "abac"],  # Initialize both models
    config=config
)

# Or add models at runtime
await acl.init_model("custom_model", adapter=adapter)

⚠️ Important for Alembic Users: If you're using Alembic for database migrations, you need to manually import CasbinRule in your alembic/env.py file to ensure Alembic can detect and generate migration scripts for the Casbin policy table:

# In alembic/env.py
from fastapi_casbin_acl.adapter.orm import CasbinRule

# Make sure to import all your application models as well
# from your_app.models import User, Order, ...

target_metadata = SQLModel.metadata

This ensures that CasbinRule is registered in SQLModel's metadata, allowing Alembic to automatically generate migration scripts for the t_casbin_rule table.

2. Define Subject & Resource

Define how to get the current user (subject) and how to resolve resources (for ABAC).

from fastapi import Request, Depends
from sqlmodel import select
from sqlalchemy.ext.asyncio import AsyncSession

# 1. Subject Provider (Authentication)
async def get_current_user(
    request: Request,
    session: AsyncSession = Depends(get_session)
):
    """Get current user from request (e.g., JWT token, session)"""
    user_id = request.headers.get("X-User-ID")
    if not user_id:
        return None
    
    # Query user from database
    # ... your user query logic ...
    return user

async def get_subject_from_user(user = Depends(get_current_user)) -> str | None:
    """Extract subject (user ID) for permission checking"""
    if user is None:
        return None
    return str(user.id)  # Return user ID as string

# 2. Resource Getter (Optional, for ABAC)
class Order:
    def __init__(self, id, owner_id):
        self.id = id
        self.owner_id = owner_id
    
    def get_owner_sub(self) -> str | None:
        """Return owner ID for ABAC ownership check"""
        return str(self.owner_id) if self.owner_id else None

async def get_order_resource(request: Request) -> Order | None:
    """Fetch resource from database based on path parameters"""
    order_id = request.path_params.get("id")
    if not order_id:
        return None
    
    # Query order from database
    # ... your order query logic ...
    return Order(id=order_id, owner_id="user_123")

3. Understanding API Names

In the Permission RBAC model, the system uses API names (route names) instead of API paths for permission mapping. This provides a more stable and maintainable approach.

What is an API Name?

• Each FastAPI route has a name attribute that uniquely identifies the route
• By default, FastAPI automatically sets the route name to the function name
• You can also explicitly set the route name using the name parameter

Example:

# Route name is automatically set to "get_user" (function name)
@app.get("/users/{id}")
async def get_user(id: int):
    return {"user": id}

# Explicitly set route name
@app.get("/users/{id}", name="fetch_user_by_id")
async def get_user(id: int):
    return {"user": id}

# Route name is automatically set to "list_users"
@app.get("/users")
async def list_users():
    return {"users": [...]}

Why use API Names?

• Stability: Route names remain constant even when paths change
• Clarity: Function names are more meaningful than paths
• Maintainability: Easier to manage permissions when refactoring routes
• Automatic: No need to manually configure path patterns

Important Notes:

• Routes without a name attribute cannot be used for permission checking
• Route names must be unique within your application
• The system will raise an error during initialization if duplicate route names are found

4. Protect Routes

Use permission_required in your route dependencies. The function now uses the Permission RBAC model by default, which provides a cleaner abstraction layer.

Permission RBAC (Recommended)

The Permission RBAC model introduces a permission abstraction layer between roles and APIs:

• g: User -> Role mapping (e.g., g, user_1, admin)
• g2: api_name -> Permission mapping (e.g., g2, get_user, user_management)
• p: Role -> Permission -> Action policy (e.g., p, admin, user_management, read)

API Name: Each FastAPI route has a unique name attribute (automatically set to the function name if not explicitly provided). The system uses this api_name to map routes to permissions, providing a cleaner and more maintainable approach than path-based matching.

from fastapi_casbin_acl.dependency import permission_required

@app.get("/users")
async def list_users(
    _: None = Depends(permission_required(
        get_subject=get_subject_from_user,
        action="read"
    ))
):
    return {"users": [...]}

@app.post("/orders")
async def create_order(
    _: None = Depends(permission_required(
        get_subject=get_subject_from_user,
        action="write"
    ))
):
    return {"order": "created"}

How it works:

1. The system extracts the route's name attribute (api_name) from the request
2. Looks up the corresponding permission via g2 mapping (e.g., get_user -> user_management)
3. Checks if the user's role has the required permission and action via p policy
4. Grants or denies access accordingly

Benefits:

• Cleaner abstraction: APIs are mapped to permissions via route names, not directly to roles
• Route name based: Uses FastAPI's route name attribute, which is more stable than paths
• Easier management: Change API paths without updating role policies (route names remain constant)
• Better organization: Group related APIs under the same permission
• Automatic mapping: Route names are automatically available from FastAPI, no manual path configuration needed

Legacy Models (RBAC & ABAC)

For backward compatibility, you can still use the legacy RBAC and ABAC models by specifying them explicitly. However, the Permission RBAC model is recommended for new projects.

RBAC (Interface Level):

from fastapi_casbin_acl.dependency import permission_required

@app.get(
    "/dashboard",
    dependencies=[
        Depends(permission_required(
            get_subject=get_subject_from_user,
            action="read",
            model="rbac"  # Use RBAC model
        ))
    ]
)
async def dashboard():
    return {"data": "..."}

ABAC (Data Level):

Enforces that the user is the owner of the resource OR has the admin role.

@app.get(
    "/orders/{id}",
    dependencies=[
        Depends(permission_required(
            get_subject=get_subject_from_user,
            resource=get_order_resource,  # Inject resource getter
            action="read",
            model="abac"  # Use ABAC model for ownership check
        ))
    ]
)
async def get_order(id: str):
    return {"order": id}

Model Management

Built-in Models

The package provides three built-in models that are automatically registered:

• permission_rbac: Permission-based RBAC (recommended) - Introduces permission abstraction layer
  • g: User -> Role mapping
  • g2: api_name -> Permission mapping (uses FastAPI route name)
  • p: Role -> Permission -> Action policy
• rbac: Role-Based Access Control (3 args: sub, obj, act)
• abac: Attribute-Based Access Control (4 args: sub, obj, act, owner)

Model Registry

You can register custom models using the ModelRegistry:

from fastapi_casbin_acl.registry import model_registry

# Register a custom model
model_registry.register("custom_model", "/path/to/custom.conf")

# List all registered models
models = model_registry.list_models()  # ['rbac', 'abac', 'custom_model']

# Get model path
path = model_registry.get_path("abac")

Using Different Models per Route

You can use different permission models for different routes:

# Route 1: Use RBAC model
@app.get("/public")
async def public_endpoint(
    _=Depends(permission_required(
        get_subject=get_subject_from_user,
        action="read",
        model="rbac"
    ))
):
    return {"message": "public"}

# Route 2: Use ABAC model
@app.get("/orders/{id}")
async def order_endpoint(
    _=Depends(permission_required(
        get_subject=get_subject_from_user,
        resource=get_order_resource,
        action="read",
        model="abac"
    ))
):
    return {"order": "..."}

Configuration

You can customize the ACL behavior by passing a config object during initialization.

from fastapi_casbin_acl.config import ACLConfig

config = ACLConfig(
    external_model_path="/path/to/custom.conf",  # Register external model
    admin_role="superuser",  # Change default admin role from 'admin'
    policy_router_enable=True,  # Enable policy management REST API
    policy_router_prefix="/api/policies",  # Custom prefix for policy router
    get_subject=get_subject_from_user,  # Required if policy_router_enable=True
    redis_url="redis://localhost:6379/0",  # Optional: Multi-worker sync
    policy_sync_channel="fastapi_casbin:policy_update",  # Optional: Custom channel
)

await acl.init(adapter=adapter, config=config)

If external_model_path is provided, it will be automatically registered with the name "external".

Redis Configuration (Multi-Worker Synchronization)

For multi-process deployments (e.g., gunicorn with multiple workers), you can enable Redis-based policy synchronization to ensure all workers have consistent policy state:

config = ACLConfig(
    redis_url="redis://localhost:6379/0",  # Redis connection URL
    policy_sync_channel="fastapi_casbin:policy_update",  # Optional: Custom channel name
)

await acl.init(adapter=adapter, config=config, app=app)

How it works:

• When redis_url is provided, a background watcher automatically subscribes to a Redis Pub/Sub channel
• When any worker updates policies (via API or CLI), it publishes a notification to Redis
• All other workers receive the notification and automatically reload their policies from the database
• This ensures all workers have the latest policy state within milliseconds

Benefits:

• Real-time synchronization: Policy changes are propagated to all workers immediately
• No manual reload: Workers automatically reload policies when notified
• Production-ready: Essential for multi-worker deployments to avoid permission inconsistencies

Note: If you're running a single-process application (e.g., development with uvicorn --reload), Redis configuration is optional. However, it's recommended for production deployments with multiple workers.

Command Line Interface

The package provides CLI tools for managing permissions, synchronizing APIs, and explaining policies.

Initialize Permissions

Use the init_permission command to grant admin role to a user and add policy management permissions:

uv run fastapi-casbin-acl init_permission <subject_id> <app_path>

Parameters:

• subject_id: The subject ID (user ID) to grant admin role to
• app_path: FastAPI application path in the format module:app, e.g., main:app or examples.main:app

Example:

# Grant admin role to user with ID "1" in the examples app
uv run fastapi-casbin-acl init_permission 1 examples.main:app

What it does:

1. Loads the FastAPI application from the specified path
2. Executes the application's lifespan startup process to initialize ACL
3. Adds policy management permission for the admin role: admin -> policy_management -> write
4. Assigns the admin role to the specified user: <subject_id> -> admin
5. Saves the policies to the database
6. If Redis is configured, notifies all workers to reload policies (for multi-worker deployments)

Requirements:

• The FastAPI application must have a lifespan parameter configured
• The lifespan must call await acl.init() to initialize the ACL system
• The application must be importable from the specified module path

Note: The command will automatically handle the application lifecycle (startup and shutdown) to ensure proper initialization and cleanup.

Sync APIs

Use the sync_apis command to automatically scan all registered routes and create API-to-Permission mappings:

uv run fastapi-casbin-acl sync_apis <app_path> [--permission-prefix PREFIX]

Parameters:

• app_path: FastAPI application path in the format module:app
• --permission-prefix: Optional prefix for permission names (default: "api")

Example:

# Sync all routes from the examples app
uv run fastapi-casbin-acl sync_apis examples.main:app --permission-prefix api

What it does:

1. Loads the FastAPI application and initializes ACL
2. Scans all registered routes (excluding docs and static files)
3. For each route, uses the route's name attribute (api_name) to create a g2 mapping: api_name -> permission
4. Generates permission names based on the api_name (e.g., get_user -> api_get_user if prefix is "api")
5. Only adds new mappings (does not update or delete existing ones)
6. Saves all mappings to the database

Note: Routes without a name attribute will be skipped. FastAPI automatically assigns route names based on the function name, but you can also explicitly set them using the name parameter in route decorators.

Output example:

🔍 正在扫描路由...
   找到 12 个路由
📝 正在添加 API-Permission 映射...
   ✅ get_user -> api_get_user
   ✅ create_user -> api_create_user
   ✅ list_orders -> api_list_orders
   ...
🎉 API 同步完成！
   新增: 12
   跳过: 0

Explain Policies

Use the explain command to understand which policies match a specific URI and subject:

uv run fastapi-casbin-acl explain --uri <uri> --subject <subject_id> [--method METHOD] <app_path>

Parameters:

• --uri: API URI path to check (e.g., /api/users/1)
• --subject: Subject ID (user ID) to check permissions for
• --method: Optional HTTP method (e.g., GET, POST). If not provided, matches all methods
• app_path: FastAPI application path in the format module:app

Example:

# Explain permissions for user "1" accessing GET /api/users/1
uv run fastapi-casbin-acl explain --uri "/api/users/1" --subject "1" --method "GET" examples.main:app

Output example:

📋 策略解释结果
Subject: 1
Method: GET
URI: /api/users/1

Route Matching:
  URI: /api/users/1
  Method: GET
  Route Path: /api/users/{id}
  API Name: get_user

Permission Mapping:
  get_user -> user_management

Role Chain:
  1 -> admin

Matched Policies:
  p, admin, user_management, read  [ALLOW]

What it shows:

• The matched route and its api_name
• The permission mapped to the api_name (via g2)
• The role chain for the subject (via g)
• All matching policies that grant access (via p)

Multi-Worker Policy Synchronization

When deploying FastAPI applications with multiple workers (e.g., using gunicorn), each worker process maintains its own in-memory policy cache. Without synchronization, when one worker updates permissions, other workers continue using stale policies, leading to inconsistent permission checks.

The Problem

sequenceDiagram
    participant Client
    participant Worker1 as Worker-1
    participant Worker2 as Worker-2
    participant DB as Database

    Note over Worker1,Worker2: 启动时各自加载策略到内存
    
    Client->>Worker1: POST /policies (添加权限)
    Worker1->>DB: 写入新策略
    Worker1->>Worker1: 更新内存缓存
    Worker1-->>Client: 200 OK
    
    Note over Worker2: 内存中仍是旧策略
    
    Client->>Worker2: GET /orders/1 (请求资源)
    Worker2->>Worker2: 使用旧策略检查权限
    Worker2-->>Client: 403 Forbidden (权限拒绝)

The Solution: Redis Pub/Sub

This package provides Redis-based policy synchronization using Pub/Sub messaging:

sequenceDiagram
    participant Client
    participant Worker1 as Worker-1
    participant Worker2 as Worker-2
    participant Redis
    participant DB as Database

    Note over Worker1,Worker2: 启动时订阅 Redis channel
    
    Client->>Worker1: POST /policies (添加权限)
    Worker1->>DB: 写入新策略
    Worker1->>Worker1: 更新内存缓存
    Worker1->>Redis: PUBLISH policy_update
    Worker1-->>Client: 200 OK
    
    Redis-->>Worker2: 收到更新通知
    Worker2->>DB: 重新加载策略
    Worker2->>Worker2: 更新内存缓存
    
    Client->>Worker2: GET /orders/1 (请求资源)
    Worker2->>Worker2: 使用新策略检查权限
    Worker2-->>Client: 200 OK

How It Works

1. Initialization: When redis_url is configured, each worker automatically:

   • Connects to Redis
   • Subscribes to the policy update channel (default: fastapi_casbin:policy_update)
   • Starts a background listener task

2. Policy Updates: When any worker updates policies (via API or CLI):

   • The change is saved to the database
   • A notification is published to Redis
   • The worker's own cache is updated

3. Synchronization: All other workers:

   • Receive the notification via Redis Pub/Sub
   • Automatically reload policies from the database
   • Update their in-memory cache

4. Shutdown: When the application shuts down:

   • The watcher is stopped
   • Redis connections are closed
   • Background tasks are cancelled

Configuration

Enable synchronization by providing a Redis URL in your configuration:

from fastapi_casbin_acl.config import ACLConfig

config = ACLConfig(
    redis_url="redis://localhost:6379/0",  # Enable synchronization
    policy_sync_channel="fastapi_casbin:policy_update",  # Optional: Custom channel
)

await acl.init(adapter=adapter, config=config, app=app)

Manual Notification

You can also manually notify workers to reload policies:

# Notify all workers to reload all models
await acl.notify_policy_update()

# Notify all workers to reload a specific model
await acl.notify_policy_update(model_name="abac")

Automatic Notification

Policy updates via the built-in policy router automatically trigger notifications. The CLI tool also sends notifications after policy changes when Redis is configured.

Performance Considerations

• Latency: Policy synchronization typically completes within milliseconds
• Overhead: Minimal - only publishes a small notification message
• Reliability: Uses Redis Pub/Sub, which is highly reliable and widely used in production
• Scalability: Works with any number of workers

Working with Enforcers

Get Enforcer for Specific Model

# Get enforcer for a specific model
enforcer = acl.get_enforcer("abac")

# Add policies
await enforcer.add_policy("admin", "/api/users/*", "read")
await enforcer.add_grouping_policy("user_1", "admin")

# Save policies
await acl.save_policy("abac")  # Save specific model
await acl.save_policy()        # Save all models

Check Initialized Models

# Check if a model is initialized
if acl.is_model_initialized("abac"):
    enforcer = acl.get_enforcer("abac")

# List all initialized models
models = acl.list_initialized_models()  # ['rbac', 'abac']

Exception Handling

The dependency raises Unauthorized (401) or Forbidden (403) exceptions. You should handle them in your application.

from fastapi.responses import JSONResponse
from fastapi_casbin_acl.exceptions import Unauthorized, Forbidden

@app.exception_handler(Unauthorized)
async def unauthorized_handler(request: Request, exc: Unauthorized):
    return JSONResponse(
        status_code=401,
        content={"detail": "Unauthorized: Please provide valid authentication"}
    )

@app.exception_handler(Forbidden)
async def forbidden_handler(request: Request, exc: Forbidden):
    return JSONResponse(
        status_code=403,
        content={"detail": "Permission Denied"}
    )

Built-in Models

ABAC Model (Default)

The package comes with a built-in Casbin model optimized for RBAC + ABAC:

[request_definition]
r = sub, obj, act, owner

[policy_definition]
p = sub, obj, act

[role_definition]
g = _, _

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

[matchers]
m = (p.sub == "" || p.sub == "*" || g(r.sub, p.sub)) && keyMatch2(r.obj, p.obj) && r.act == p.act && (r.owner == "" || r.sub == r.owner || g(r.sub, "admin"))

• RBAC: When owner is empty (""), matches sub, obj, act against policy p. Supports wildcard policies (p, *, /path/*, action).
• ABAC: When owner is provided, checks if r.sub == r.owner (Resource Ownership). Requires a matching policy for the resource path.
• Admin Override: If sub has role admin (via g, sub, admin), access is allowed regardless of ownership.

Permission RBAC Model (Recommended)

The Permission RBAC model introduces a permission abstraction layer, providing better organization and HTTP method awareness:

[request_definition]
r = sub, obj, act

[policy_definition]
p = sub, obj, act

[role_definition]
g = _, _        # user -> role
g2 = _, _       # api_name -> permission

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

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

How it works:

1. g (User-Role mapping): g, user_1, admin - User 1 has admin role
2. g2 (API-Permission mapping): g2, get_user, user_management - Route named get_user requires user_management permission
3. p (Role-Permission-Action policy): p, admin, user_management, read - Admin role has read action on user_management permission

API Name: The system uses FastAPI's route name attribute (api_name) for permission mapping. Route names are automatically set to the function name, but you can also explicitly set them:

@app.get("/users/{id}", name="get_user")
async def get_user(id: int):
    return {"user": id}

Example policies:

# User-Role bindings
await enforcer.add_grouping_policy("1", "admin")
await enforcer.add_grouping_policy("2", "user")

# API-Permission mappings (using route names)
await enforcer.add_named_grouping_policy("g2", "get_user", "user_management")
await enforcer.add_named_grouping_policy("g2", "create_user", "user_management")
await enforcer.add_named_grouping_policy("g2", "list_orders", "order_management")
await enforcer.add_named_grouping_policy("g2", "delete_order", "order_management")

# Role-Permission-Action policies
await enforcer.add_policy("admin", "user_management", "read")
await enforcer.add_policy("admin", "user_management", "write")
await enforcer.add_policy("user", "order_management", "read")
await enforcer.add_policy("user", "order_management", "write")

Benefits:

• Permission abstraction: APIs are grouped under permissions, making management easier
• Route name based: Uses stable route names instead of paths, making refactoring easier
• Flexible organization: Change API paths without updating role policies (route names remain constant)
• Clear separation: Roles, Permissions, and APIs are clearly separated
• Automatic detection: Route names are automatically available from FastAPI

Initialization Example:

from fastapi_casbin_acl.enforcer import acl

# Get the Permission RBAC enforcer
enforcer = acl.get_enforcer("permission_rbac")

# 1. Define user-role bindings (g)
await enforcer.add_grouping_policy("1", "admin")  # User 1 is admin
await enforcer.add_grouping_policy("2", "user")   # User 2 is user

# 2. Define API-Permission mappings (g2) - using route names
await enforcer.add_named_grouping_policy("g2", "get_user", "user_management")
await enforcer.add_named_grouping_policy("g2", "create_user", "user_management")
await enforcer.add_named_grouping_policy("g2", "list_orders", "order_management")
await enforcer.add_named_grouping_policy("g2", "create_order", "order_management")
await enforcer.add_named_grouping_policy("g2", "delete_order", "order_management")

# 3. Define role-permission-action policies (p)
await enforcer.add_policy("admin", "user_management", "read")
await enforcer.add_policy("admin", "user_management", "write")
await enforcer.add_policy("admin", "order_management", "read")
await enforcer.add_policy("admin", "order_management", "write")
await enforcer.add_policy("admin", "order_management", "delete")
await enforcer.add_policy("user", "order_management", "read")
await enforcer.add_policy("user", "order_management", "write")

# Save all policies to database
await acl.save_policy()

RBAC Model

The RBAC model is simpler and only supports role-based access control:

[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 = (p.sub == "" || p.sub == "*" || g(r.sub, p.sub)) && keyMatch2(r.obj, p.obj) && r.act == p.act

Policy Management API

The package provides a built-in REST API for managing policies, roles, permissions, and API mappings. Enable it by setting policy_router_enable=True in your config:

config = ACLConfig(
    policy_router_enable=True,  # Enable policy management API
    policy_router_prefix="/api/policies",  # Optional: Custom prefix
    get_subject=get_subject_from_user,  # Required: Function to get current user
)

Available Endpoints

Policies:

• GET /api/policies - List all policies
• POST /api/policies - Create a policy
• DELETE /api/policies - Delete a policy
• GET /api/policies/list - Paginated policy list with filters

Role Bindings:

• GET /api/policies/role_bindings - List all role bindings
• GET /api/policies/roles/{user} - Get roles for a user
• POST /api/policies/roles - Create a role binding
• DELETE /api/policies/roles - Delete a role binding

Permissions:

• GET /api/policies/permissions - List all permissions

API Mappings:

• GET /api/policies/permission_mappings - List all API-Permission mappings (based on route table)
• GET /api/policies/routes - List all registered route names (api_names)
• GET /api/policies/permission_groups - List permission groups (permission -> api_names mapping)
• PUT /api/policies/permission_groups/{permission} - Update a permission group (set api_names for a permission)
• DELETE /api/policies/permission_groups/{permission} - Delete a permission group (remove all mappings for a permission)

Example Usage

# Create a policy
POST /api/policies
{
    "sub": "admin",
    "obj": "user_management",
    "act": "read"
}

# Create a role binding
POST /api/policies/roles
{
    "user": "1",
    "role": "admin"
}

# Create an API-Permission mapping
POST /api/policies/permission_mappings
{
    "api_name": "get_user",
    "permission": "user_management"
}

# Update a permission group (set multiple api_names for a permission)
PUT /api/policies/permission_groups/user_management
{
    "api_names": ["get_user", "create_user", "update_user", "delete_user"]
}

Complete Example

See the examples/ directory for a complete working example with:

• User and Order management
• Permission RBAC permission checks
• Database integration with SQLModel
• Frontend interface with dynamic policy management
• Policy visualization and adjustment

To run the example:

cd examples
python run.py
# or
uvicorn main:app --reload

Then visit http://localhost:8000 to access the frontend interface.

Initialization:

# 1. Grant admin role to user with ID "1"
uv run fastapi-casbin-acl init_permission 1 examples.main:app

# 2. Sync all API routes to create permission mappings
uv run fastapi-casbin-acl sync_apis examples.main:app --permission-prefix api

# 3. Explain permissions for debugging
uv run fastapi-casbin-acl explain --uri "/api/users/1" --subject "1" --method "GET" examples.main:app

Frontend Features:

• User and Order management with permission checks
• Dynamic policy management interface
• Real-time policy visualization
• Create/delete policies, role bindings, and API mappings
• Tabbed interface for different policy types

API Reference

permission_required

Factory function that creates a permission dependency. Uses the Permission RBAC model by default.

Parameters:

• get_subject: Callable that returns the subject (user ID/username) for permission checking
• action: The action being performed (e.g., "read", "write", "delete")

Returns: FastAPI dependency function

How it works (Permission RBAC model):

1. Extracts the route's name attribute (api_name) from the request
2. Looks up the permission via g2 mapping: api_name -> permission
3. Checks if the user's role (via g) has the required permission and action (via p)
4. Raises Forbidden if no mapping is found or permission is denied

Note: The route must have a name attribute. FastAPI automatically assigns route names based on the function name, but you can also explicitly set them using the name parameter in route decorators.

Example:

@app.get("/users")
async def list_users(
    _: None = Depends(permission_required(
        get_subject=get_subject_from_user,
        action="read"
    ))
):
    return {"users": [...]}

Note: For legacy RBAC/ABAC models, you can still specify the model parameter, but Permission RBAC is recommended for new projects.

ACLConfig

Configuration class for ACL system.

Fields:

• external_model_path: Optional path to external model file (registered as "external")
• admin_role: Role name that bypasses ownership checks (default: "admin")
• policy_router_enable: Enable built-in policy management REST API (default: False)
• policy_router_prefix: Custom prefix for policy router (default: "/casbin_policies")
• get_subject: Function to get current user subject (required if policy_router_enable=True)
• redis_url: Optional Redis connection URL for multi-worker policy synchronization (e.g., "redis://localhost:6379/0")
• policy_sync_channel: Redis Pub/Sub channel name for policy updates (default: "fastapi_casbin:policy_update")

ModelRegistry

Registry for managing Casbin permission models.

Methods:

• register(name, path): Register a new model
• unregister(name): Unregister a model
• get_path(name): Get path to model file
• is_registered(name): Check if model is registered
• list_models(): List all registered model names

AsyncEnforcerManager

Singleton manager for multiple Casbin AsyncEnforcer instances.

Methods:

• init(adapter, models=None, config=None, app=None): Initialize enforcers
  • If app is provided, automatically builds an api_name mapping table from all registered routes
  • Validates route names for uniqueness (raises error if duplicates found)
  • If app is not provided, issues a warning (api_name-based matching requires app)
• get_enforcer(model_name): Get enforcer for specific model
• enforce(model_name, *args): Execute enforce with specific model
• init_model(model_name, adapter=None): Initialize a model at runtime
• is_model_initialized(model_name): Check if model is initialized
• list_initialized_models(): List all initialized models
• load_policy(model_name=None): Reload policies from database
• save_policy(model_name=None): Save policies to database
• notify_policy_update(model_name=None): Notify all workers to reload policies (requires Redis)
• get_api_name_map(): Get the api_name to route_path mapping table (for debugging)
• shutdown(): Stop watcher and clean up resources

License

MIT