| Name | abs-auth-rbac-core JSON |
| Version |
0.1.19
JSON |
| download |
| home_page | None |
| Summary | RBAC and Auth core utilities including JWT token management. |
| upload_time | 2025-08-06 10:19:10 |
| maintainer | None |
| docs_url | None |
| author | AutoBridgeSystems |
| requires_python | <4.0,>=3.13 |
| license | MIT |
| keywords |
|
| VCS |
|
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# ABS Auth RBAC Core
A comprehensive authentication and Role-Based Access Control (RBAC) package for FastAPI applications. This package provides robust JWT-based authentication and flexible role-based permission management using Casbin with Redis support for real-time policy updates.
## Features
- **JWT-based Authentication**: Secure token-based authentication with customizable expiration
- **Password Hashing**: Secure password storage using bcrypt
- **Role-Based Access Control (RBAC)**: Flexible permission management using Casbin
- **Real-time Policy Updates**: Redis integration for live policy synchronization
- **User-Role Management**: Dynamic role assignment and revocation
- **Permission Enforcement**: Decorator-based permission checking
- **Middleware Integration**: Seamless FastAPI middleware integration
- **Comprehensive Error Handling**: Built-in exception handling for security scenarios
## Installation
```bash
pip install abs-auth-rbac-core
```
## Quick Start
### 1. Basic Setup
```python
from abs_auth_rbac_core.auth.jwt_functions import JWTFunctions
from abs_auth_rbac_core.rbac import RBACService
import os
# Initialize JWT functions
jwt_functions = JWTFunctions(
secret_key=os.getenv("JWT_SECRET_KEY"),
algorithm=os.getenv("JWT_ALGORITHM", "HS256"),
expire_minutes=int(os.getenv("JWT_EXPIRE_MINUTES", "60"))
)
# Initialize RBAC service with database session
rbac_service = RBACService(
session=your_db_session,
redis_config=RedisWatcherSchema(
host=os.getenv("REDIS_HOST"),
port=int(os.getenv("REDIS_PORT", "6379")),
channel=os.getenv("REDIS_CHANNEL", "casbin_policy_updates"),
password=os.getenv("REDIS_PASSWORD"),
ssl=os.getenv("REDIS_SSL", "false").lower() == "true"
)
)
```
### 2. Authentication Setup
The auth middleware can be implemented in two ways:
#### Option 1: Using the Package Middleware (Recommended)
The package middleware automatically handles JWT token validation and sets the user in the request state:
```python
from abs_auth_rbac_core.auth.middleware import auth_middleware
# Create authentication middleware
auth_middleware = auth_middleware(
db_session=your_db_session,
jwt_secret_key=os.getenv("JWT_SECRET_KEY"),
jwt_algorithm=os.getenv("JWT_ALGORITHM", "HS256")
)
# Apply to specific routers (recommended approach)
app.include_router(
protected_router,
dependencies=[Depends(auth_middleware)]
)
# Public routes (no middleware)
app.include_router(public_router)
```
**How it works:**
1. The middleware validates the JWT token from the Authorization header
2. Extracts the user UUID from the token payload
3. Fetches the user from the database using the UUID
4. **Sets the user object in `request.state.user`**
5. Returns the user object for use in route handlers
**Accessing the user in routes:**
```python
@router.get("/profile")
async def get_profile(request: Request):
# User is automatically available in request.state.user
user = request.state.user
return {"user_id": user.uuid, "email": user.email}
```
#### Option 2: Custom Authentication Function
```python
from abs_auth_rbac_core.auth import JWTFunctions
from fastapi import Security, HTTPAuthorizationCredentials
from fastapi.security import HTTPBearer
from abs_exception_core.exceptions import UnauthorizedError
# Create security scheme
security = HTTPBearer(auto_error=False)
jwt_functions = JWTFunctions(
secret_key=os.getenv("JWT_SECRET_KEY"),
algorithm=os.getenv("JWT_ALGORITHM", "HS256"),
expire_minutes=int(os.getenv("JWT_EXPIRE_MINUTES", "60"))
)
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Security(security),
) -> Dict:
try:
if not credentials:
raise UnauthorizedError(detail="No authorization token provided")
token = credentials.credentials
# Remove 'Bearer ' prefix if present
if token.lower().startswith("bearer "):
token = token[7:]
decoded_token = jwt_functions.decode_jwt(token)
if not decoded_token:
raise UnauthorizedError(detail="Invalid or expired token")
return decoded_token
except Exception as e:
raise UnauthorizedError(detail=str(e))
# Use in individual routes
@app.get("/protected")
async def protected_route(current_user: dict = Depends(get_current_user)):
return {"message": f"Hello {current_user.get('name')}"}
```
### 3. RBAC Operations
```python
# Create a role with permissions
role = rbac_service.create_role(
name="admin",
description="Administrator role with full access",
permission_ids=["permission_uuid1", "permission_uuid2"]
)
# Assign roles to user
rbac_service.bulk_assign_roles_to_user(
user_uuid="user_uuid",
role_uuids=["role_uuid1", "role_uuid2"]
)
# Check user permissions
has_permission = rbac_service.check_permission(
user_uuid="user_uuid",
resource="USER_MANAGEMENT",
action="VIEW",
module="USER_MANAGEMENT"
)
# Get user permissions
user_permissions = rbac_service.get_user_permissions(user_uuid="user_uuid")
user_roles = rbac_service.get_user_roles(user_uuid="user_uuid")
```
## Core Components
### Authentication (`auth/`)
- `jwt_functions.py`: JWT token management and password hashing
- `middleware.py`: Authentication middleware for FastAPI
- `auth_functions.py`: Core authentication functions
### RBAC (`rbac/`)
- `service.py`: Main RBAC service with role and permission management
- `decorator.py`: Decorators for permission checking
### Models (`models/`)
- `user.py`: User model
- `roles.py`: Role model
- `permissions.py`: Permission model
- `user_role.py`: User-Role association model
- `role_permission.py`: Role-Permission association model
- `rbac_model.py`: Base RBAC model
- `base_model.py`: Base model with common fields
### Utilities (`util/`)
- `permission_constants.py`: Predefined permission constants and enums
## Complete Implementation Example
### 1. Dependency Injection Setup
```python
from dependency_injector import containers, providers
from abs_auth_rbac_core.auth.middleware import auth_middleware
from abs_auth_rbac_core.rbac import RBACService
from abs_auth_rbac_core.util.permission_constants import (
PermissionAction,
PermissionModule,
PermissionResource
)
class Container(containers.DeclarativeContainer):
# Configure wiring for dependency injection
wiring_config = containers.WiringConfiguration(
modules=[
"src.api.auth_route",
"src.api.endpoints.rbac.permission_route",
"src.api.endpoints.rbac.role_route",
"src.api.endpoints.rbac.users_route",
# Add other modules that need dependency injection
]
)
# Database session provider
db_session = providers.Factory(your_db_session_factory)
# RBAC service provider
rbac_service = providers.Singleton(
RBACService,
session=db_session,
redis_config=RedisWatcherSchema(
host=os.getenv("REDIS_HOST"),
port=int(os.getenv("REDIS_PORT", "6379")),
channel=os.getenv("REDIS_CHANNEL", "casbin_policy_updates"),
password=os.getenv("REDIS_PASSWORD"),
ssl=os.getenv("REDIS_SSL", "false").lower() == "true"
)
)
# Auth middleware provider
get_auth_middleware = providers.Factory(
auth_middleware,
db_session=db_session,
jwt_secret_key=os.getenv("JWT_SECRET_KEY"),
jwt_algorithm=os.getenv("JWT_ALGORITHM", "HS256")
)
# Initialize container
container = Container()
app.container = container
```
### 2. Complete Application Setup
```python
from fastapi import FastAPI, Depends
from dependency_injector.wiring import Provide, inject
from src.core.container import Container
@singleton
class CreateApp:
def __init__(self):
self.container = Container()
self.db = self.container.db()
# Get the auth middleware factory
self.auth_middleware = self.container.get_auth_middleware()
self.app = FastAPI(
title="Your Service",
description="Service Description",
version="0.0.1"
)
# Apply CORS middleware
self.app.add_middleware(
CORSMiddleware,
allow_origins=[str(origin) for origin in configs.BACKEND_CORS_ORIGINS],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Public routes (no authentication required)
self.app.include_router(auth_router, tags=["Auth"])
self.app.include_router(public_router_v1)
# Protected routes (authentication required)
self.app.include_router(
router_v1,
dependencies=[Depends(self.auth_middleware)]
)
# Register exception handlers
register_exception_handlers(self.app)
# Initialize application
application = CreateApp()
app = application.app
```
### 3. Route Implementation with Permissions
```python
from fastapi import APIRouter, Depends, Request
from dependency_injector.wiring import Provide, inject
from abs_auth_rbac_core.rbac import rbac_require_permission
from abs_auth_rbac_core.util.permission_constants import (
PermissionAction,
PermissionModule,
PermissionResource
)
# Protected router (requires authentication)
router = APIRouter(prefix="/users")
# Public router (no authentication required)
public_router = APIRouter(prefix="/users")
# Public route example (no authentication or permissions required)
@public_router.post("/all", response_model=FindUserResult)
@inject
async def get_user_list(
request: Request,
find_query: FindUser = Body(...),
rbac_service: RBACService = Depends(Provide[Container.rbac_service]),
service: UserService = Depends(Provide[Container.user_service]),
):
"""Get the list of users with filtering, sorting and pagination"""
find_query.searchable_fields = find_query.searchable_fields or ["name"]
users = service.get_list(schema=find_query)
return users
# Protected route with permission check
@router.get("/{user_id}", response_model=UserProfile)
@inject
@rbac_require_permission(
f"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.VIEW.value}"
)
async def get_user(
user_id: int,
request: Request,
service: UserService = Depends(Provide[Container.user_service]),
rbac_service: RBACService = Depends(Provide[Container.rbac_service]),
):
"""Get user profile with permissions and roles"""
return service.get_user_profile("id", user_id, rbac_service)
```
**How the RBAC decorator works:**
1. The `@rbac_require_permission` decorator automatically extracts the user from `request.state.user`
2. Gets the user's UUID: `current_user_uuid = request.state.user.uuid`
3. Checks if the user has the required permissions using the RBAC service
4. If permission is denied, raises `PermissionDeniedError`
5. If permission is granted, the route handler executes normally
**Important:** The `request: Request` parameter is required in routes that use the `@rbac_require_permission` decorator because it needs access to `request.state.user`.
### Authentication Flow Overview
```
1. Client sends request with Authorization header
Authorization: Bearer <jwt_token>
2. Auth middleware intercepts the request
├── Validates JWT token
├── Extracts user UUID from token
├── Fetches user from database
└── Sets user in request.state.user
3. RBAC decorator checks permissions
├── Gets user UUID from request.state.user
├── Checks permissions against Casbin policies
└── Allows/denies access based on permissions
4. Route handler executes (if permissions granted)
└── Can access user via request.state.user
```
### Accessing Current User
#### In Routes with RBAC Decorator
```python
@router.get("/my-profile")
@inject
@rbac_require_permission(
f"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.VIEW.value}"
)
async def get_my_profile(
request: Request,
rbac_service: RBACService = Depends(Provide[Container.rbac_service]),
):
# Access current user from request state
current_user = request.state.user
return service.get_user_profile("uuid", current_user.uuid, rbac_service)
```
#### In Routes with Custom Auth Function
```python
@router.get("/profile")
async def get_profile(current_user: dict = Depends(get_current_user)):
# current_user is the decoded JWT payload
return {"user_id": current_user["uuid"], "email": current_user["email"]}
```
#### In Service Methods
```python
def some_service_method(self, user_uuid: str, rbac_service: RBACService):
# Get user permissions
permissions = rbac_service.get_user_permissions(user_uuid=user_uuid)
# Get user roles
roles = rbac_service.get_user_roles(user_uuid=user_uuid)
return {"permissions": permissions, "roles": roles}
```
@router.post("/create")
@inject
@rbac_require_permission(
f"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.CREATE.value}"
)
async def create_user(
user: CreateUserWithRoles,
request: Request,
rbac_service: RBACService = Depends(Provide[Container.rbac_service]),
):
"""Create a new user with roles"""
new_user = service.add_user(user)
# Assign roles if provided
if user.role_uuids and len(user.role_uuids) > 0 and new_user.uuid:
rbac_service.bulk_assign_roles_to_user(
user_uuid=new_user.uuid,
role_uuids=user.role_uuids,
)
return new_user
@router.patch("/{user_id}")
@inject
@rbac_require_permission(
f"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.EDIT.value}"
)
async def update_user(
user_id: int,
user: UpdateUserWithRoles,
request: Request,
rbac_service: RBACService = Depends(Provide[Container.rbac_service]),
):
"""Update user with new attributes and roles"""
return service.patch_user(user_id, user, rbac_service)
@router.delete("/{user_id}")
@inject
@rbac_require_permission(
f"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.DELETE.value}"
)
async def delete_user(
user_id: int,
request: Request,
rbac_service: RBACService = Depends(Provide[Container.rbac_service]),
):
"""Delete user"""
return service.remove_user(user_id, rbac_service)
```
### 3. Role and Permission Management
```python
@router.get("/roles")
@inject
@rbac_require_permission(
f"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.ROLE_MANAGEMENT.value}:{PermissionAction.VIEW.value}"
)
async def get_roles(
request: Request,
rbac_service: RBACService = Depends(Provide[Container.rbac_service]),
):
"""Get all roles"""
return rbac_service.list_roles()
@router.post("/roles")
@inject
@rbac_require_permission(
f"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.ROLE_MANAGEMENT.value}:{PermissionAction.CREATE.value}"
)
async def create_role(
role: CreateRoleSchema,
request: Request,
rbac_service: RBACService = Depends(Provide[Container.rbac_service]),
):
"""Create a new role with permissions"""
return rbac_service.create_role(
name=role.name,
description=role.description,
permission_ids=role.permission_ids
)
@router.get("/user-permissions/{user_uuid}")
@inject
@rbac_require_permission([
f"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.VIEW.value}",
f"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.ROLE_MANAGEMENT.value}:{PermissionAction.VIEW.value}"
])
async def get_user_permissions(
user_uuid: str,
request: Request,
rbac_service: RBACService = Depends(Provide[Container.rbac_service]),
):
"""Get all permissions for a user"""
return rbac_service.get_user_permissions(user_uuid)
```
### 4. User Profile with Permissions
```python
def get_user_profile(self, attr: str, value: any, rbac_service: RBACService) -> UserProfile:
"""Get user profile with permissions and roles"""
user = self.user_repository.read_by_attr(attr, value, eager=True)
# Get user permissions and roles
permissions = rbac_service.get_user_permissions(user_uuid=user.uuid)
user_permissions = rbac_service.get_user_only_permissions(user_uuid=user.uuid)
roles = rbac_service.get_user_roles(user_uuid=user.uuid)
# Convert roles to response models
role_models = [UserRoleResponse.model_validate(role) for role in roles]
return UserProfile(
id=user.id,
uuid=user.uuid,
email=user.email,
name=user.name,
is_active=user.is_active,
last_login_at=user.last_login_at,
permissions=permissions,
user_permissions=user_permissions,
roles=role_models,
)
```
## Permission System
### Permission Format
Permissions follow the format: `module:resource:action`
- **Module**: The system module (e.g., `USER_MANAGEMENT`, `EMAIL_PROCESS`)
- **Resource**: The specific resource within the module (e.g., `USER_MANAGEMENT`, `ROLE_MANAGEMENT`)
- **Action**: The action being performed (e.g., `VIEW`, `CREATE`, `EDIT`, `DELETE`)
### Using Permission Constants
```python
from abs_auth_rbac_core.util.permission_constants import (
PermissionAction,
PermissionModule,
PermissionResource,
PermissionConstants
)
# Using enums
permission_string = f"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.VIEW.value}"
# Using predefined constants
user_view_permission = PermissionConstants.RBAC_USER_MANAGEMENT_VIEW
permission_string = f"{user_view_permission.module}:{user_view_permission.resource}:{user_view_permission.action}"
```
### Multiple Permissions
```python
@rbac_require_permission([
f"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.VIEW.value}",
f"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.ROLE_MANAGEMENT.value}:{PermissionAction.VIEW.value}"
])
async def get_user_with_roles():
# User needs both permissions to access this endpoint
pass
```
## Configuration
### Environment Variables
```bash
# JWT Configuration
JWT_SECRET_KEY=your-secret-key
JWT_ALGORITHM=HS256
JWT_ACCESS_TOKEN_EXPIRE_MINUTES=60
JWT_REFRESH_TOKEN_EXPIRE_MINUTES=1440
# Redis Configuration (for real-time policy updates)
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_PASSWORD=your-redis-password
REDIS_CHANNEL=casbin_policy_updates
REDIS_SSL=false
# Database Configuration
DATABASE_URI=postgresql://user:password@localhost/dbname
```
### Casbin Policy Configuration
The package uses a default policy configuration that supports:
- Role-based access control
- Resource-based permissions
- Module-based organization
- Super admin bypass
Policy format: `[role] [resource] [action] [module]`
## Error Handling
The package includes comprehensive error handling:
```python
from abs_exception_core.exceptions import (
UnauthorizedError,
PermissionDeniedError,
ValidationError,
DuplicatedError,
NotFoundError
)
# Handle authentication errors
try:
user = await auth_middleware(request)
except UnauthorizedError as e:
return {"error": "Authentication failed", "detail": str(e)}
# Handle permission errors
try:
# Protected operation
pass
except PermissionDeniedError as e:
return {"error": "Permission denied", "detail": str(e)}
```
## Best Practices
### 1. Security
- Always use environment variables for sensitive data
- Implement proper password policies
- Regularly rotate JWT secret keys
- Use HTTPS in production
- Implement rate limiting for authentication endpoints
### 2. Permission Design
- Use descriptive permission names
- Group related permissions by module
- Implement least privilege principle
- Document permission requirements
### 3. Performance
- Use Redis for real-time policy updates
- Implement caching for frequently accessed permissions
- Optimize database queries with eager loading
- Monitor policy enforcement performance
### 4. Maintenance
- Regularly audit user permissions
- Implement permission cleanup for inactive users
- Monitor and log security events
- Keep dependencies updated
### 5. Testing
- Test all permission combinations
- Mock external dependencies
- Test error scenarios
- Implement integration tests
## Monitoring and Logging
```python
import logging
from abs_utils.logger import setup_logger
logger = setup_logger(__name__)
# Log authentication events
logger.info(f"User {user_uuid} authenticated successfully")
# Log permission checks
logger.info(f"Permission check: {user_uuid} -> {resource}:{action}:{module}")
# Log role assignments
logger.info(f"Roles assigned to user {user_uuid}: {role_uuids}")
```
## Migration and Deployment
### Database Migrations
Ensure your database has the required tables:
- `users`
- `roles`
- `permissions`
- `user_roles`
- `role_permissions`
- `gov_casbin_rules`
### Redis Setup
For real-time policy updates, configure Redis:
```bash
# Install Redis
sudo apt-get install redis-server
# Configure Redis
redis-cli config set requirepass your-password
redis-cli config set notify-keyspace-events KEA
```
### Health Checks
```python
@app.get("/health")
async def health_check():
return {
"status": "healthy",
"rbac_watcher": rbac_service.is_watcher_active(),
"policy_count": rbac_service.get_policy_count()
}
```
## Troubleshooting
### Common Issues
1. **Authentication Fails**
- Check JWT secret key configuration
- Verify token expiration settings
- Ensure user exists in database
2. **Permission Denied**
- Verify user has required roles
- Check role-permission assignments
- Validate permission format
3. **Redis Connection Issues**
- Check Redis server status
- Verify connection parameters
- Ensure Redis supports pub/sub
4. **Policy Not Updating**
- Check Redis watcher configuration
- Verify policy format
- Monitor Redis logs
## License
This project is licensed under the MIT License - see the LICENSE file for details.
Raw data
{
"_id": null,
"home_page": null,
"name": "abs-auth-rbac-core",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.13",
"maintainer_email": null,
"keywords": null,
"author": "AutoBridgeSystems",
"author_email": "info@autobridgesystems.com",
"download_url": "https://files.pythonhosted.org/packages/8f/6f/f49c63ed07cbc11a3534c1bd989a675a160f8856056e314e767cc9cf07b2/abs_auth_rbac_core-0.1.19.tar.gz",
"platform": null,
"description": "# ABS Auth RBAC Core\n\nA comprehensive authentication and Role-Based Access Control (RBAC) package for FastAPI applications. This package provides robust JWT-based authentication and flexible role-based permission management using Casbin with Redis support for real-time policy updates.\n\n## Features\n\n- **JWT-based Authentication**: Secure token-based authentication with customizable expiration\n- **Password Hashing**: Secure password storage using bcrypt\n- **Role-Based Access Control (RBAC)**: Flexible permission management using Casbin\n- **Real-time Policy Updates**: Redis integration for live policy synchronization\n- **User-Role Management**: Dynamic role assignment and revocation\n- **Permission Enforcement**: Decorator-based permission checking\n- **Middleware Integration**: Seamless FastAPI middleware integration\n- **Comprehensive Error Handling**: Built-in exception handling for security scenarios\n\n## Installation\n\n```bash\npip install abs-auth-rbac-core\n```\n\n## Quick Start\n\n### 1. Basic Setup\n\n```python\nfrom abs_auth_rbac_core.auth.jwt_functions import JWTFunctions\nfrom abs_auth_rbac_core.rbac import RBACService\nimport os\n\n# Initialize JWT functions\njwt_functions = JWTFunctions(\n secret_key=os.getenv(\"JWT_SECRET_KEY\"),\n algorithm=os.getenv(\"JWT_ALGORITHM\", \"HS256\"),\n expire_minutes=int(os.getenv(\"JWT_EXPIRE_MINUTES\", \"60\"))\n)\n\n# Initialize RBAC service with database session\nrbac_service = RBACService(\n session=your_db_session,\n redis_config=RedisWatcherSchema(\n host=os.getenv(\"REDIS_HOST\"),\n port=int(os.getenv(\"REDIS_PORT\", \"6379\")),\n channel=os.getenv(\"REDIS_CHANNEL\", \"casbin_policy_updates\"),\n password=os.getenv(\"REDIS_PASSWORD\"),\n ssl=os.getenv(\"REDIS_SSL\", \"false\").lower() == \"true\"\n )\n)\n```\n\n### 2. Authentication Setup\n\nThe auth middleware can be implemented in two ways:\n\n#### Option 1: Using the Package Middleware (Recommended)\n\nThe package middleware automatically handles JWT token validation and sets the user in the request state:\n\n```python\nfrom abs_auth_rbac_core.auth.middleware import auth_middleware\n\n# Create authentication middleware\nauth_middleware = auth_middleware(\n db_session=your_db_session,\n jwt_secret_key=os.getenv(\"JWT_SECRET_KEY\"),\n jwt_algorithm=os.getenv(\"JWT_ALGORITHM\", \"HS256\")\n)\n\n# Apply to specific routers (recommended approach)\napp.include_router(\n protected_router,\n dependencies=[Depends(auth_middleware)]\n)\n\n# Public routes (no middleware)\napp.include_router(public_router)\n```\n\n**How it works:**\n1. The middleware validates the JWT token from the Authorization header\n2. Extracts the user UUID from the token payload\n3. Fetches the user from the database using the UUID\n4. **Sets the user object in `request.state.user`**\n5. Returns the user object for use in route handlers\n\n**Accessing the user in routes:**\n```python\n@router.get(\"/profile\")\nasync def get_profile(request: Request):\n # User is automatically available in request.state.user\n user = request.state.user\n return {\"user_id\": user.uuid, \"email\": user.email}\n```\n\n#### Option 2: Custom Authentication Function\n\n```python\nfrom abs_auth_rbac_core.auth import JWTFunctions\nfrom fastapi import Security, HTTPAuthorizationCredentials\nfrom fastapi.security import HTTPBearer\nfrom abs_exception_core.exceptions import UnauthorizedError\n\n# Create security scheme\nsecurity = HTTPBearer(auto_error=False)\njwt_functions = JWTFunctions(\n secret_key=os.getenv(\"JWT_SECRET_KEY\"),\n algorithm=os.getenv(\"JWT_ALGORITHM\", \"HS256\"),\n expire_minutes=int(os.getenv(\"JWT_EXPIRE_MINUTES\", \"60\"))\n)\n\nasync def get_current_user(\n credentials: HTTPAuthorizationCredentials = Security(security),\n) -> Dict:\n try:\n if not credentials:\n raise UnauthorizedError(detail=\"No authorization token provided\")\n\n token = credentials.credentials\n # Remove 'Bearer ' prefix if present\n if token.lower().startswith(\"bearer \"):\n token = token[7:]\n\n decoded_token = jwt_functions.decode_jwt(token)\n if not decoded_token:\n raise UnauthorizedError(detail=\"Invalid or expired token\")\n\n return decoded_token\n except Exception as e:\n raise UnauthorizedError(detail=str(e))\n\n# Use in individual routes\n@app.get(\"/protected\")\nasync def protected_route(current_user: dict = Depends(get_current_user)):\n return {\"message\": f\"Hello {current_user.get('name')}\"}\n```\n\n### 3. RBAC Operations\n\n```python\n# Create a role with permissions\nrole = rbac_service.create_role(\n name=\"admin\",\n description=\"Administrator role with full access\",\n permission_ids=[\"permission_uuid1\", \"permission_uuid2\"]\n)\n\n# Assign roles to user\nrbac_service.bulk_assign_roles_to_user(\n user_uuid=\"user_uuid\",\n role_uuids=[\"role_uuid1\", \"role_uuid2\"]\n)\n\n# Check user permissions\nhas_permission = rbac_service.check_permission(\n user_uuid=\"user_uuid\",\n resource=\"USER_MANAGEMENT\",\n action=\"VIEW\",\n module=\"USER_MANAGEMENT\"\n)\n\n# Get user permissions\nuser_permissions = rbac_service.get_user_permissions(user_uuid=\"user_uuid\")\nuser_roles = rbac_service.get_user_roles(user_uuid=\"user_uuid\")\n```\n\n## Core Components\n\n### Authentication (`auth/`)\n- `jwt_functions.py`: JWT token management and password hashing\n- `middleware.py`: Authentication middleware for FastAPI\n- `auth_functions.py`: Core authentication functions\n\n### RBAC (`rbac/`)\n- `service.py`: Main RBAC service with role and permission management\n- `decorator.py`: Decorators for permission checking\n\n### Models (`models/`)\n- `user.py`: User model\n- `roles.py`: Role model\n- `permissions.py`: Permission model\n- `user_role.py`: User-Role association model\n- `role_permission.py`: Role-Permission association model\n- `rbac_model.py`: Base RBAC model\n- `base_model.py`: Base model with common fields\n\n### Utilities (`util/`)\n- `permission_constants.py`: Predefined permission constants and enums\n\n## Complete Implementation Example\n\n### 1. Dependency Injection Setup\n\n```python\nfrom dependency_injector import containers, providers\nfrom abs_auth_rbac_core.auth.middleware import auth_middleware\nfrom abs_auth_rbac_core.rbac import RBACService\nfrom abs_auth_rbac_core.util.permission_constants import (\n PermissionAction,\n PermissionModule,\n PermissionResource\n)\n\nclass Container(containers.DeclarativeContainer):\n # Configure wiring for dependency injection\n wiring_config = containers.WiringConfiguration(\n modules=[\n \"src.api.auth_route\",\n \"src.api.endpoints.rbac.permission_route\",\n \"src.api.endpoints.rbac.role_route\",\n \"src.api.endpoints.rbac.users_route\",\n # Add other modules that need dependency injection\n ]\n )\n \n # Database session provider\n db_session = providers.Factory(your_db_session_factory)\n \n # RBAC service provider\n rbac_service = providers.Singleton(\n RBACService,\n session=db_session,\n redis_config=RedisWatcherSchema(\n host=os.getenv(\"REDIS_HOST\"),\n port=int(os.getenv(\"REDIS_PORT\", \"6379\")),\n channel=os.getenv(\"REDIS_CHANNEL\", \"casbin_policy_updates\"),\n password=os.getenv(\"REDIS_PASSWORD\"),\n ssl=os.getenv(\"REDIS_SSL\", \"false\").lower() == \"true\"\n )\n )\n \n # Auth middleware provider\n get_auth_middleware = providers.Factory(\n auth_middleware,\n db_session=db_session,\n jwt_secret_key=os.getenv(\"JWT_SECRET_KEY\"),\n jwt_algorithm=os.getenv(\"JWT_ALGORITHM\", \"HS256\")\n )\n\n# Initialize container\ncontainer = Container()\napp.container = container\n```\n\n### 2. Complete Application Setup\n\n```python\nfrom fastapi import FastAPI, Depends\nfrom dependency_injector.wiring import Provide, inject\nfrom src.core.container import Container\n\n@singleton\nclass CreateApp:\n def __init__(self):\n self.container = Container()\n self.db = self.container.db()\n # Get the auth middleware factory\n self.auth_middleware = self.container.get_auth_middleware()\n\n self.app = FastAPI(\n title=\"Your Service\",\n description=\"Service Description\", \n version=\"0.0.1\"\n )\n \n # Apply CORS middleware\n self.app.add_middleware(\n CORSMiddleware,\n allow_origins=[str(origin) for origin in configs.BACKEND_CORS_ORIGINS],\n allow_credentials=True,\n allow_methods=[\"*\"],\n allow_headers=[\"*\"],\n )\n\n # Public routes (no authentication required)\n self.app.include_router(auth_router, tags=[\"Auth\"])\n self.app.include_router(public_router_v1)\n \n # Protected routes (authentication required)\n self.app.include_router(\n router_v1,\n dependencies=[Depends(self.auth_middleware)]\n )\n \n # Register exception handlers\n register_exception_handlers(self.app)\n\n# Initialize application\napplication = CreateApp()\napp = application.app\n```\n\n### 3. Route Implementation with Permissions\n\n```python\nfrom fastapi import APIRouter, Depends, Request\nfrom dependency_injector.wiring import Provide, inject\nfrom abs_auth_rbac_core.rbac import rbac_require_permission\nfrom abs_auth_rbac_core.util.permission_constants import (\n PermissionAction,\n PermissionModule,\n PermissionResource\n)\n\n# Protected router (requires authentication)\nrouter = APIRouter(prefix=\"/users\")\n\n# Public router (no authentication required)\npublic_router = APIRouter(prefix=\"/users\")\n\n# Public route example (no authentication or permissions required)\n@public_router.post(\"/all\", response_model=FindUserResult)\n@inject\nasync def get_user_list(\n request: Request,\n find_query: FindUser = Body(...),\n rbac_service: RBACService = Depends(Provide[Container.rbac_service]),\n service: UserService = Depends(Provide[Container.user_service]),\n):\n \"\"\"Get the list of users with filtering, sorting and pagination\"\"\"\n find_query.searchable_fields = find_query.searchable_fields or [\"name\"]\n users = service.get_list(schema=find_query)\n return users\n\n# Protected route with permission check\n@router.get(\"/{user_id}\", response_model=UserProfile)\n@inject\n@rbac_require_permission(\n f\"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.VIEW.value}\"\n)\nasync def get_user(\n user_id: int,\n request: Request,\n service: UserService = Depends(Provide[Container.user_service]),\n rbac_service: RBACService = Depends(Provide[Container.rbac_service]),\n):\n \"\"\"Get user profile with permissions and roles\"\"\"\n return service.get_user_profile(\"id\", user_id, rbac_service)\n```\n\n**How the RBAC decorator works:**\n1. The `@rbac_require_permission` decorator automatically extracts the user from `request.state.user`\n2. Gets the user's UUID: `current_user_uuid = request.state.user.uuid`\n3. Checks if the user has the required permissions using the RBAC service\n4. If permission is denied, raises `PermissionDeniedError`\n5. If permission is granted, the route handler executes normally\n\n**Important:** The `request: Request` parameter is required in routes that use the `@rbac_require_permission` decorator because it needs access to `request.state.user`.\n\n### Authentication Flow Overview\n\n```\n1. Client sends request with Authorization header\n Authorization: Bearer <jwt_token>\n\n2. Auth middleware intercepts the request\n \u251c\u2500\u2500 Validates JWT token\n \u251c\u2500\u2500 Extracts user UUID from token\n \u251c\u2500\u2500 Fetches user from database\n \u2514\u2500\u2500 Sets user in request.state.user\n\n3. RBAC decorator checks permissions\n \u251c\u2500\u2500 Gets user UUID from request.state.user\n \u251c\u2500\u2500 Checks permissions against Casbin policies\n \u2514\u2500\u2500 Allows/denies access based on permissions\n\n4. Route handler executes (if permissions granted)\n \u2514\u2500\u2500 Can access user via request.state.user\n```\n\n### Accessing Current User\n\n#### In Routes with RBAC Decorator\n```python\n@router.get(\"/my-profile\")\n@inject\n@rbac_require_permission(\n f\"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.VIEW.value}\"\n)\nasync def get_my_profile(\n request: Request,\n rbac_service: RBACService = Depends(Provide[Container.rbac_service]),\n):\n # Access current user from request state\n current_user = request.state.user\n return service.get_user_profile(\"uuid\", current_user.uuid, rbac_service)\n```\n\n#### In Routes with Custom Auth Function\n```python\n@router.get(\"/profile\")\nasync def get_profile(current_user: dict = Depends(get_current_user)):\n # current_user is the decoded JWT payload\n return {\"user_id\": current_user[\"uuid\"], \"email\": current_user[\"email\"]}\n```\n\n#### In Service Methods\n```python\ndef some_service_method(self, user_uuid: str, rbac_service: RBACService):\n # Get user permissions\n permissions = rbac_service.get_user_permissions(user_uuid=user_uuid)\n # Get user roles\n roles = rbac_service.get_user_roles(user_uuid=user_uuid)\n return {\"permissions\": permissions, \"roles\": roles}\n```\n\n@router.post(\"/create\")\n@inject\n@rbac_require_permission(\n f\"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.CREATE.value}\"\n)\nasync def create_user(\n user: CreateUserWithRoles,\n request: Request,\n rbac_service: RBACService = Depends(Provide[Container.rbac_service]),\n):\n \"\"\"Create a new user with roles\"\"\"\n new_user = service.add_user(user)\n \n # Assign roles if provided\n if user.role_uuids and len(user.role_uuids) > 0 and new_user.uuid:\n rbac_service.bulk_assign_roles_to_user(\n user_uuid=new_user.uuid,\n role_uuids=user.role_uuids,\n )\n return new_user\n\n@router.patch(\"/{user_id}\")\n@inject\n@rbac_require_permission(\n f\"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.EDIT.value}\"\n)\nasync def update_user(\n user_id: int,\n user: UpdateUserWithRoles,\n request: Request,\n rbac_service: RBACService = Depends(Provide[Container.rbac_service]),\n):\n \"\"\"Update user with new attributes and roles\"\"\"\n return service.patch_user(user_id, user, rbac_service)\n\n@router.delete(\"/{user_id}\")\n@inject\n@rbac_require_permission(\n f\"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.DELETE.value}\"\n)\nasync def delete_user(\n user_id: int,\n request: Request,\n rbac_service: RBACService = Depends(Provide[Container.rbac_service]),\n):\n \"\"\"Delete user\"\"\"\n return service.remove_user(user_id, rbac_service)\n```\n\n### 3. Role and Permission Management\n\n```python\n@router.get(\"/roles\")\n@inject\n@rbac_require_permission(\n f\"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.ROLE_MANAGEMENT.value}:{PermissionAction.VIEW.value}\"\n)\nasync def get_roles(\n request: Request,\n rbac_service: RBACService = Depends(Provide[Container.rbac_service]),\n):\n \"\"\"Get all roles\"\"\"\n return rbac_service.list_roles()\n\n@router.post(\"/roles\")\n@inject\n@rbac_require_permission(\n f\"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.ROLE_MANAGEMENT.value}:{PermissionAction.CREATE.value}\"\n)\nasync def create_role(\n role: CreateRoleSchema,\n request: Request,\n rbac_service: RBACService = Depends(Provide[Container.rbac_service]),\n):\n \"\"\"Create a new role with permissions\"\"\"\n return rbac_service.create_role(\n name=role.name,\n description=role.description,\n permission_ids=role.permission_ids\n )\n\n@router.get(\"/user-permissions/{user_uuid}\")\n@inject\n@rbac_require_permission([\n f\"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.VIEW.value}\",\n f\"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.ROLE_MANAGEMENT.value}:{PermissionAction.VIEW.value}\"\n])\nasync def get_user_permissions(\n user_uuid: str,\n request: Request,\n rbac_service: RBACService = Depends(Provide[Container.rbac_service]),\n):\n \"\"\"Get all permissions for a user\"\"\"\n return rbac_service.get_user_permissions(user_uuid)\n```\n\n### 4. User Profile with Permissions\n\n```python\ndef get_user_profile(self, attr: str, value: any, rbac_service: RBACService) -> UserProfile:\n \"\"\"Get user profile with permissions and roles\"\"\"\n user = self.user_repository.read_by_attr(attr, value, eager=True)\n \n # Get user permissions and roles\n permissions = rbac_service.get_user_permissions(user_uuid=user.uuid)\n user_permissions = rbac_service.get_user_only_permissions(user_uuid=user.uuid)\n roles = rbac_service.get_user_roles(user_uuid=user.uuid)\n \n # Convert roles to response models\n role_models = [UserRoleResponse.model_validate(role) for role in roles]\n \n return UserProfile(\n id=user.id,\n uuid=user.uuid,\n email=user.email,\n name=user.name,\n is_active=user.is_active,\n last_login_at=user.last_login_at,\n permissions=permissions,\n user_permissions=user_permissions,\n roles=role_models,\n )\n```\n\n## Permission System\n\n### Permission Format\nPermissions follow the format: `module:resource:action`\n\n- **Module**: The system module (e.g., `USER_MANAGEMENT`, `EMAIL_PROCESS`)\n- **Resource**: The specific resource within the module (e.g., `USER_MANAGEMENT`, `ROLE_MANAGEMENT`)\n- **Action**: The action being performed (e.g., `VIEW`, `CREATE`, `EDIT`, `DELETE`)\n\n### Using Permission Constants\n\n```python\nfrom abs_auth_rbac_core.util.permission_constants import (\n PermissionAction,\n PermissionModule,\n PermissionResource,\n PermissionConstants\n)\n\n# Using enums\npermission_string = f\"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.VIEW.value}\"\n\n# Using predefined constants\nuser_view_permission = PermissionConstants.RBAC_USER_MANAGEMENT_VIEW\npermission_string = f\"{user_view_permission.module}:{user_view_permission.resource}:{user_view_permission.action}\"\n```\n\n### Multiple Permissions\n\n```python\n@rbac_require_permission([\n f\"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.USER_MANAGEMENT.value}:{PermissionAction.VIEW.value}\",\n f\"{PermissionModule.USER_MANAGEMENT.value}:{PermissionResource.ROLE_MANAGEMENT.value}:{PermissionAction.VIEW.value}\"\n])\nasync def get_user_with_roles():\n # User needs both permissions to access this endpoint\n pass\n```\n\n## Configuration\n\n### Environment Variables\n\n```bash\n# JWT Configuration\nJWT_SECRET_KEY=your-secret-key\nJWT_ALGORITHM=HS256\nJWT_ACCESS_TOKEN_EXPIRE_MINUTES=60\nJWT_REFRESH_TOKEN_EXPIRE_MINUTES=1440\n\n# Redis Configuration (for real-time policy updates)\nREDIS_HOST=localhost\nREDIS_PORT=6379\nREDIS_PASSWORD=your-redis-password\nREDIS_CHANNEL=casbin_policy_updates\nREDIS_SSL=false\n\n# Database Configuration\nDATABASE_URI=postgresql://user:password@localhost/dbname\n```\n\n### Casbin Policy Configuration\n\nThe package uses a default policy configuration that supports:\n- Role-based access control\n- Resource-based permissions\n- Module-based organization\n- Super admin bypass\n\nPolicy format: `[role] [resource] [action] [module]`\n\n## Error Handling\n\nThe package includes comprehensive error handling:\n\n```python\nfrom abs_exception_core.exceptions import (\n UnauthorizedError,\n PermissionDeniedError,\n ValidationError,\n DuplicatedError,\n NotFoundError\n)\n\n# Handle authentication errors\ntry:\n user = await auth_middleware(request)\nexcept UnauthorizedError as e:\n return {\"error\": \"Authentication failed\", \"detail\": str(e)}\n\n# Handle permission errors\ntry:\n # Protected operation\n pass\nexcept PermissionDeniedError as e:\n return {\"error\": \"Permission denied\", \"detail\": str(e)}\n```\n\n## Best Practices\n\n### 1. Security\n- Always use environment variables for sensitive data\n- Implement proper password policies\n- Regularly rotate JWT secret keys\n- Use HTTPS in production\n- Implement rate limiting for authentication endpoints\n\n### 2. Permission Design\n- Use descriptive permission names\n- Group related permissions by module\n- Implement least privilege principle\n- Document permission requirements\n\n### 3. Performance\n- Use Redis for real-time policy updates\n- Implement caching for frequently accessed permissions\n- Optimize database queries with eager loading\n- Monitor policy enforcement performance\n\n### 4. Maintenance\n- Regularly audit user permissions\n- Implement permission cleanup for inactive users\n- Monitor and log security events\n- Keep dependencies updated\n\n### 5. Testing\n- Test all permission combinations\n- Mock external dependencies\n- Test error scenarios\n- Implement integration tests\n\n## Monitoring and Logging\n\n```python\nimport logging\nfrom abs_utils.logger import setup_logger\n\nlogger = setup_logger(__name__)\n\n# Log authentication events\nlogger.info(f\"User {user_uuid} authenticated successfully\")\n\n# Log permission checks\nlogger.info(f\"Permission check: {user_uuid} -> {resource}:{action}:{module}\")\n\n# Log role assignments\nlogger.info(f\"Roles assigned to user {user_uuid}: {role_uuids}\")\n```\n\n## Migration and Deployment\n\n### Database Migrations\nEnsure your database has the required tables:\n- `users`\n- `roles`\n- `permissions`\n- `user_roles`\n- `role_permissions`\n- `gov_casbin_rules`\n\n### Redis Setup\nFor real-time policy updates, configure Redis:\n```bash\n# Install Redis\nsudo apt-get install redis-server\n\n# Configure Redis\nredis-cli config set requirepass your-password\nredis-cli config set notify-keyspace-events KEA\n```\n\n### Health Checks\n```python\n@app.get(\"/health\")\nasync def health_check():\n return {\n \"status\": \"healthy\",\n \"rbac_watcher\": rbac_service.is_watcher_active(),\n \"policy_count\": rbac_service.get_policy_count()\n }\n```\n\n## Troubleshooting\n\n### Common Issues\n\n1. **Authentication Fails**\n - Check JWT secret key configuration\n - Verify token expiration settings\n - Ensure user exists in database\n\n2. **Permission Denied**\n - Verify user has required roles\n - Check role-permission assignments\n - Validate permission format\n\n3. **Redis Connection Issues**\n - Check Redis server status\n - Verify connection parameters\n - Ensure Redis supports pub/sub\n\n4. **Policy Not Updating**\n - Check Redis watcher configuration\n - Verify policy format\n - Monitor Redis logs\n\n## License\n\nThis project is licensed under the MIT License - see the LICENSE file for details.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "RBAC and Auth core utilities including JWT token management.",
"version": "0.1.19",
"project_urls": null,
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "932e9c1a9212de32b0955ee8a924b2fd4aa85e9c115b3af2aa0026730a9b175b",
"md5": "382a73c6aae03311756500eb7a4310ee",
"sha256": "d790605e60d8b8c0031e1e19f6347ed3207fa0c59d48bec1054c6ddba3b99f86"
},
"downloads": -1,
"filename": "abs_auth_rbac_core-0.1.19-py3-none-any.whl",
"has_sig": false,
"md5_digest": "382a73c6aae03311756500eb7a4310ee",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.13",
"size": 36214,
"upload_time": "2025-08-06T10:19:08",
"upload_time_iso_8601": "2025-08-06T10:19:08.974446Z",
"url": "https://files.pythonhosted.org/packages/93/2e/9c1a9212de32b0955ee8a924b2fd4aa85e9c115b3af2aa0026730a9b175b/abs_auth_rbac_core-0.1.19-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "8f6ff49c63ed07cbc11a3534c1bd989a675a160f8856056e314e767cc9cf07b2",
"md5": "adeab707f7be0e9bf314385bdefe0adf",
"sha256": "c1e453a01cfc847d8d138e50525f0dbdd5229126cfa1418b171b88958c8e72da"
},
"downloads": -1,
"filename": "abs_auth_rbac_core-0.1.19.tar.gz",
"has_sig": false,
"md5_digest": "adeab707f7be0e9bf314385bdefe0adf",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.13",
"size": 34680,
"upload_time": "2025-08-06T10:19:10",
"upload_time_iso_8601": "2025-08-06T10:19:10.605762Z",
"url": "https://files.pythonhosted.org/packages/8f/6f/f49c63ed07cbc11a3534c1bd989a675a160f8856056e314e767cc9cf07b2/abs_auth_rbac_core-0.1.19.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-06 10:19:10",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "abs-auth-rbac-core"
}
