Skip to main content

Auth & Security

info

Fluvius Framework provides built-in support for authentication and authorization using Casbin.

Authentication​

Basic Authentication​

Authenticate users:

from fluvius.fastapi.authentication import AuthMiddleware

app = FluviusFastAPI()
app.add_middleware(AuthMiddleware)

JWT Authentication​

Use JWT tokens:

from fluvius.fastapi.authentication import JWTAuth

auth = JWTAuth(secret_key="your-secret-key")

@app.post("/login")
async def login(username: str, password: str):
    # Validate credentials
    user = await validate_user(username, password)
    token = auth.create_token(user.id)
    return {"token": token}

API Key Authentication​

Use API keys:

from fluvius.fastapi.authentication import APIKeyAuth

auth = APIKeyAuth()

@app.post("/api/endpoint")
@auth.required
async def protected_endpoint():
    # Requires API key
    pass

Authorization with Casbin​

Casbin Integration​

Fluvius integrates with Casbin for policy-based access control:

from fluvius.casbin import CasbinEnforcer

enforcer = CasbinEnforcer(
    model_path="path/to/model.conf",
    policy_path="path/to/policy.csv"
)

Policy Model​

Define policy model:

# model.conf
[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

Policy Rules​

Define policy rules:

# policy.csv
p, admin, user, create
p, admin, user, read
p, admin, user, update
p, admin, user, delete
p, user, user, read
p, user, user, update

Enforcing Policies​

Enforce policies in aggregates:

from fluvius.domain import Aggregate, action
from fluvius.casbin import require_permission

class UserAggregate(Aggregate):
    @action(evt_key='user-created', resources=['user'])
    @require_permission('user', 'create')
    async def create_user(self, name: str, email: str):
        # Only users with 'user:create' permission can call this
        pass

Resource-Based Authorization​

Resource Authorization​

Authorize based on resources:

@action(evt_key='user-updated', resources=['user'])
@require_permission('user', 'update')
async def update_user(self, name: str):
    # Requires 'user:update' permission
    pass

Resource Ownership​

Check resource ownership:

@action(evt_key='user-updated', resources=['user'])
@require_permission('user', 'update')
async def update_user(self, name: str):
    # Check if user owns the resource
    if not self.domain.context.user_id == self._state.owner_id:
        raise PermissionError("Not authorized")
    pass

Role-Based Access Control (RBAC)​

Roles​

Define roles:

# Assign roles
enforcer.add_grouping_policy("alice", "admin")
enforcer.add_grouping_policy("bob", "user")

Role Permissions​

Define role permissions:

# Admin can do everything
enforcer.add_policy("admin", "user", "create")
enforcer.add_policy("admin", "user", "read")
enforcer.add_policy("admin", "user", "update")
enforcer.add_policy("admin", "user", "delete")

# User can only read and update
enforcer.add_policy("user", "user", "read")
enforcer.add_policy("user", "user", "update")

Attribute-Based Access Control (ABAC)​

Attributes​

Use attributes for authorization:

# Check attributes
enforcer.add_policy("admin", "user", "create", "department == 'IT'")

Dynamic Policies​

Create dynamic policies:

# Policy based on resource attributes
if resource.department == user.department:
    enforcer.add_policy(user.id, resource.type, "read")

Security Best Practices​

1. Validate Input​

Always validate input:

from pydantic import BaseModel, EmailStr

class CreateUserRequest(BaseModel):
    name: str
    email: EmailStr  # Validated email

@action(evt_key='user-created')
async def create_user(self, name: str, email: str):
    # Input is validated by Pydantic
    pass

2. Use HTTPS​

Always use HTTPS in production:

# Use HTTPS
app = FluviusFastAPI()
# Configure SSL certificates

3. Secure Secrets​

Store secrets securely:

import os

# Use environment variables
secret_key = os.getenv("SECRET_KEY")

# Never commit secrets
# Use .env file (gitignored)

4. Rate Limiting​

Implement rate limiting:

from slowapi import Limiter

limiter = Limiter(key_func=get_remote_address)

@app.post("/api/users")
@limiter.limit("10/minute")
async def create_user():
    pass

5. Input Sanitization​

Sanitize user input:

import html

@action(evt_key='user-created')
async def create_user(self, name: str, email: str):
    # Sanitize input
    name = html.escape(name)
    email = html.escape(email)
    pass

Security Headers​

CORS Configuration​

Configure CORS:

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

Security Headers​

Add security headers:

@app.middleware("http")
async def add_security_headers(request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    return response

Audit Logging​

Audit Events​

Log security events:

@action(evt_key='user-created')
async def create_user(self, name: str, email: str):
    # Event is automatically logged
    # Include audit information
    logger.info(f"User created by {self.domain.context.user_id}")
    pass

Access Logging​

Log access attempts:

@middleware
async def access_logging_middleware(request, call_next):
    logger.info(f"Access: {request.method} {request.url} by {request.user}")
    response = await call_next(request)
    return response

Next Steps​