Version: 2.0
Last Updated: January 2026
Status: Production Ready
Purple8’s RBAC (Role-Based Access Control) and IAM (Identity & Access Management) system provides enterprise-grade security for the Chat and Builder products.
| Feature | Description |
|---|---|
| Role Hierarchy | 6-level hierarchy from Viewer to Super Admin |
| Fine-Grained Permissions | 60+ permissions across Chat, Builder, KB, Admin |
| Multi-Tenancy | Organization-based isolation |
| API Key Auth | Scoped API keys with rate limiting |
| JWT Tokens | Short-lived access + refresh tokens |
| Audit Logging | Complete audit trail of all actions |
| Resource-Level Access | Per-project, per-conversation controls |
| Async Database | High-concurrency async SQLAlchemy |
┌─────────────────────────────────────────────────────────────────┐
│ Frontend (Vue.js) │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ │
│ │ Login Page │ │ Admin Views │ │ Product Views (Chat/ │ │
│ │ │ │ │ │ Builder) │ │
│ └──────┬──────┘ └──────┬──────┘ └───────────┬─────────────┘ │
│ │ │ │ │
│ └────────────────┼──────────────────────┘ │
│ │ │
│ ┌─────▼─────┐ │
│ │ Pinia │ │
│ │ Auth Store│ │
│ └─────┬─────┘ │
└──────────────────────────┼──────────────────────────────────────┘
│ HTTP/WebSocket
▼
┌─────────────────────────────────────────────────────────────────┐
│ API Gateway (FastAPI) │
│ ┌─────────────────────────────────────────────────────────────┐│
│ │ Auth Middleware ││
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌─────────────┐ ││
│ │ │JWT Auth │ │API Key │ │Permission│ │Rate Limiter │ ││
│ │ │ │ │Auth │ │Checker │ │ │ ││
│ │ └──────────┘ └──────────┘ └──────────┘ └─────────────┘ ││
│ └─────────────────────────────────────────────────────────────┘│
│ │ │
│ ┌───────────────────────────▼──────────────────────────────┐ │
│ │ Auth Service │ │
│ │ ┌────────────┐ ┌────────────┐ ┌────────────────────┐ │ │
│ │ │User Repo │ │Role Repo │ │Permission Repo │ │ │
│ │ │ │ │ │ │ │ │ │
│ │ └────────────┘ └────────────┘ └────────────────────┘ │ │
│ │ ┌────────────┐ ┌────────────┐ ┌────────────────────┐ │ │
│ │ │API Key Repo│ │Session Repo│ │Audit Log Repo │ │ │
│ │ │ │ │ │ │ │ │ │
│ │ └────────────┘ └────────────┘ └────────────────────┘ │ │
│ └───────────────────────────────────────────────────────────┘ │
└──────────────────────────┬──────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ PostgreSQL Database │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌───────────┐ │
│ │ users │ │ roles │ │permissions│ │api_keys │ │audit_logs │ │
│ └─────────┘ └─────────┘ └─────────┘ └─────────┘ └───────────┘ │
└─────────────────────────────────────────────────────────────────┘
shared/
├── auth/
│ ├── __init__.py # Public exports (70+ symbols)
│ ├── models.py # Pydantic models (Permission enum, Role, etc.)
│ ├── permissions.py # Permission checker, role hierarchy
│ ├── middleware.py # FastAPI auth middleware, JWT handling
│ ├── decorators.py # @check_permission, @check_admin, etc.
│ ├── repository.py # Async SQLAlchemy CRUD (950 lines)
│ └── dependencies.py # FastAPI Depends() injection
├── database/
│ ├── models/
│ │ └── auth.py # SQLAlchemy ORM models (12 tables)
│ ├── migrations/
│ │ └── 002_create_auth_tables.sql
│ └── async_connection.py # Async database session
frontend-vue/
├── src/
│ ├── store/
│ │ └── adminStore.js # Pinia auth store
│ ├── views/admin/
│ │ ├── AdminLayout.vue
│ │ ├── LoginPage.vue
│ │ ├── UserManagement.vue
│ │ ├── RoleManagement.vue
│ │ ├── APIKeyManagement.vue
│ │ └── AuditLogViewer.vue
│ └── router/index.js # Auth route guards
from fastapi import Depends
from shared.auth import (
AuthContext,
Permission,
require_permission,
get_current_user,
require_admin,
)
# Basic authentication (any logged-in user)
@router.get("/api/chat/history")
async def get_chat_history(
auth: AuthContext = Depends(get_current_user)
):
return {"user_id": str(auth.user_id)}
# Require specific permission
@router.post("/api/project/create")
async def create_project(
auth: AuthContext = Depends(require_permission(Permission.PROJECT_CREATE))
):
return {"created_by": str(auth.user_id)}
# Require admin role
@router.delete("/api/user/{user_id}")
async def delete_user(
user_id: str,
auth: AuthContext = Depends(require_admin)
):
return {"deleted": user_id}
from shared.auth import check_permission, check_admin, Permission
@router.post("/api/kb/upload")
@check_permission(Permission.KB_UPLOAD)
async def upload_document(
auth: AuthContext = Depends(get_current_user)
):
pass
@router.delete("/api/system/cache")
@check_admin()
async def clear_cache(
auth: AuthContext = Depends(get_current_user)
):
pass
from shared.auth import AuthService, get_auth_service
@router.post("/api/admin/users")
async def create_user(
request: CreateUserRequest,
auth: AuthService = Depends(get_auth_service)
):
# Create user
user = await auth.users.create(
email=request.email,
username=request.username,
password=request.password,
role_codes=["user"]
)
# Log to audit
await auth.audit.log_action(
user_id=current_user.id,
action="user.created",
resource_type="user",
resource_id=str(user.id)
)
return {"id": str(user.id)}
<template>
<div>
<!-- Show button only if user has permission -->
<button v-if="canCreateProject" @click="createProject">
New Project
</button>
<!-- Admin-only section -->
<AdminPanel v-if="isAdmin" />
</div>
</template>
<script setup>
import { computed } from 'vue'
import { useAdminStore } from '@/store/adminStore'
const adminStore = useAdminStore()
const canCreateProject = computed(() =>
adminStore.hasPermission('project:create')
)
const isAdmin = computed(() =>
adminStore.hasRole('admin') || adminStore.hasRole('super_admin')
)
</script>
Purple8 uses a 6-level role hierarchy where higher roles inherit all permissions from lower roles.
Level 100: Super Admin ─────────────────────────────────────────┐
│ │
│ • Full system access │
│ • Organization management │
│ • User management across all orgs │
│ • System configuration │
│ │
Level 80: Admin ────────────────────────────────────────────────┤
│ │
│ • All Team Lead permissions │
│ • User management (within org) │
│ • Role assignment │
│ • API key management │
│ • View audit logs │
│ │
Level 60: Team Lead ────────────────────────────────────────────┤
│ │
│ • All Power User permissions │
│ • Team management │
│ • Project settings │
│ • Approve deployments │
│ │
Level 40: Power User ───────────────────────────────────────────┤
│ │
│ • All User permissions │
│ • Create projects │
│ • Upload to knowledge base │
│ • Premium model access │
│ • Export data │
│ │
Level 20: User ─────────────────────────────────────────────────┤
│ │
│ • All Viewer permissions │
│ • Create conversations │
│ • Execute pipelines │
│ • Search knowledge base │
│ │
Level 10: Viewer ───────────────────────────────────────────────┘
• Read-only access
• View conversations (shared)
• View projects (shared)
• View knowledge base
| Role | Code | Level | Description |
|---|---|---|---|
| Super Admin | super_admin |
100 | Full platform control |
| Admin | admin |
80 | Organization administrator |
| Team Lead | team_lead |
60 | Team manager with deployment rights |
| Power User | power_user |
40 | Advanced user with create rights |
| User | user |
20 | Standard user |
| Viewer | viewer |
10 | Read-only access |
# Conversations
CHAT_VIEW = "chat:view" # View chat interface
CHAT_CREATE = "chat:create" # Create conversations
CHAT_DELETE = "chat:delete" # Delete own conversations
CHAT_DELETE_ANY = "chat:delete:any" # Delete any conversation
CHAT_EXPORT = "chat:export" # Export conversations
CHAT_SHARE = "chat:share" # Share conversations
# Models
CHAT_MODEL_FREE = "chat:model:free" # Use free tier models
CHAT_MODEL_BALANCED = "chat:model:balanced" # Use balanced tier
CHAT_MODEL_PREMIUM = "chat:model:premium" # Use premium models
CHAT_MODEL_SELECT = "chat:model:select" # Manual model selection
# Advanced
CHAT_MULTIMODAL = "chat:multimodal" # Image/document analysis
CHAT_STREAMING = "chat:streaming" # Real-time streaming
CHAT_HISTORY = "chat:history" # View history
CHAT_CONTEXT = "chat:context" # Custom context
CHAT_SYSTEM_PROMPT = "chat:system_prompt" # Custom system prompts
# Projects
PROJECT_VIEW = "project:view" # View projects
PROJECT_CREATE = "project:create" # Create projects
PROJECT_EDIT = "project:edit" # Edit projects
PROJECT_DELETE = "project:delete" # Delete projects
PROJECT_SETTINGS = "project:settings" # Manage settings
# Pipeline
PIPELINE_VIEW = "pipeline:view" # View pipelines
PIPELINE_CREATE = "pipeline:create" # Create pipelines
PIPELINE_EXECUTE = "pipeline:execute" # Run pipelines
PIPELINE_CANCEL = "pipeline:cancel" # Cancel running pipelines
# Agents
AGENT_VIEW = "agent:view" # View agents
AGENT_CONFIGURE = "agent:configure" # Configure agents
AGENT_CUSTOM = "agent:custom" # Create custom agents
# Deployment
DEPLOY_VIEW = "deploy:view" # View deployments
DEPLOY_EXECUTE = "deploy:execute" # Execute deployments
DEPLOY_APPROVE = "deploy:approve" # Approve deployments
KB_VIEW = "kb:view" # View knowledge base
KB_SEARCH = "kb:search" # Search documents
KB_UPLOAD = "kb:upload" # Upload documents
KB_DELETE = "kb:delete" # Delete documents
KB_INDEX = "kb:index" # Manage indexes
KB_CONFIGURE = "kb:configure" # Configure KB settings
KB_EXPORT = "kb:export" # Export documents
KB_SHARE = "kb:share" # Share KB access
KB_EMBED = "kb:embed" # Generate embeddings
KB_RAG = "kb:rag" # Use RAG features
# User Management
ADMIN_USER_VIEW = "admin:user:view"
ADMIN_USER_CREATE = "admin:user:create"
ADMIN_USER_EDIT = "admin:user:edit"
ADMIN_USER_DELETE = "admin:user:delete"
ADMIN_USER_ROLES = "admin:user:roles"
# Role Management
ADMIN_ROLE_VIEW = "admin:role:view"
ADMIN_ROLE_CREATE = "admin:role:create"
ADMIN_ROLE_EDIT = "admin:role:edit"
ADMIN_ROLE_DELETE = "admin:role:delete"
# API Keys
ADMIN_APIKEY_VIEW = "admin:apikey:view"
ADMIN_APIKEY_CREATE = "admin:apikey:create"
ADMIN_APIKEY_REVOKE = "admin:apikey:revoke"
# Audit & System
ADMIN_AUDIT_VIEW = "admin:audit:view"
ADMIN_AUDIT_EXPORT = "admin:audit:export"
ADMIN_SYSTEM_CONFIG = "admin:system:config"
ADMIN_SYSTEM_HEALTH = "admin:system:health"
ADMIN_SYSTEM_BACKUP = "admin:system:backup"
# Organization
ADMIN_ORG_VIEW = "admin:org:view"
ADMIN_ORG_CREATE = "admin:org:create"
ADMIN_ORG_EDIT = "admin:org:edit"
from shared.auth import PermissionChecker, Permission
checker = PermissionChecker()
# Check single permission
if checker.has_permission(user_permissions, Permission.PROJECT_CREATE):
# Allow action
# Check any of multiple permissions
if checker.has_any_permission(user_permissions, [
Permission.ADMIN_USER_VIEW,
Permission.ADMIN_USER_EDIT
]):
# Allow action
# Check all permissions required
if checker.has_all_permissions(user_permissions, [
Permission.KB_UPLOAD,
Permission.KB_INDEX
]):
# Allow action
from shared.auth import get_auth_service, AuthService
from fastapi import Depends
@router.post("/api/admin/organizations")
async def create_organization(
name: str,
auth: AuthService = Depends(get_auth_service)
):
org = await auth.organizations.create(
name=name,
slug=name.lower().replace(" ", "-"),
settings={"max_users": 100, "features": ["chat", "builder"]}
)
# Log audit
await auth.audit.log_action(
user_id=current_user.id,
action="org.created",
resource_type="organization",
resource_id=str(org.id),
details={"name": name}
)
return {"id": str(org.id), "name": org.name}
@router.post("/api/admin/users")
async def create_user(
request: CreateUserRequest,
auth: AuthService = Depends(get_auth_service)
):
# Create user with initial role
user = await auth.users.create(
email=request.email,
username=request.username,
password=request.password,
full_name=request.full_name,
organization_id=request.organization_id,
role_codes=["user"], # Default role
)
# Log audit
await auth.audit.log_action(
user_id=current_user.id,
action="user.created",
resource_type="user",
resource_id=str(user.id),
details={"email": request.email}
)
return {"id": str(user.id), "email": user.email}
@router.post("/api/admin/users/{user_id}/roles")
async def assign_role(
user_id: UUID,
role_code: str,
auth: AuthService = Depends(get_auth_service)
):
# Get role by code
role = await auth.roles.get_by_code(role_code)
if not role:
raise HTTPException(404, f"Role '{role_code}' not found")
# Assign role to user
await auth.users.assign_role(
user_id=user_id,
role_id=role.id,
assigned_by=current_user.id
)
return {"status": "role_assigned"}
@router.post("/api/admin/users/{user_id}/api-keys")
async def create_api_key(
user_id: UUID,
request: CreateAPIKeyRequest,
auth: AuthService = Depends(get_auth_service)
):
api_key, full_key = await auth.api_keys.create(
name=request.name,
user_id=user_id,
organization_id=request.organization_id,
scopes=request.scopes, # e.g., ["chat:read", "chat:write"]
expires_in_days=request.expires_in_days,
rate_limit_per_minute=request.rate_limit_per_minute,
)
# Return full key only once - it cannot be retrieved later!
return {
"id": str(api_key.id),
"name": api_key.name,
"key": full_key, # pk8_xxxx... - store securely!
"prefix": api_key.key_prefix,
"expires_at": api_key.expires_at,
}
User Frontend API Gateway Database
│ │ │ │
│─── Click Sign Up ─────▶│ │ │
│ │─── POST /register ──▶│ │
│ │ │─── Create user ───▶│
│ │ │ (role=viewer) │
│ │ │◀── User created ───│
│ │ │─── Gen verify ────▶│
│ │ │ token │
│◀── Verification email ─│◀─────────────────────│ │
│ │ │ │
│─── Click verify link ─▶│ │ │
│ │─── POST /verify ────▶│ │
│ │ │─── Mark verified ─▶│
│ │ │─── Upgrade role ──▶│
│ │ │ (viewer→user) │
│ │◀── JWT tokens ───────│ │
│◀── Redirect to app ────│ │ │
# Login to get tokens
curl -X POST http://localhost:8000/api/auth/login \
-H "Content-Type: application/json" \
-d '{"email": "user@example.com", "password": "secret"}'
# Response
{
"access_token": "eyJhbGciOiJIUzI1NiIs...",
"refresh_token": "eyJhbGciOiJIUzI1NiIs...",
"token_type": "bearer",
"expires_in": 3600
}
# Use access token
curl http://localhost:8000/api/chat/history \
-H "Authorization: Bearer eyJhbGciOiJIUzI1NiIs..."
# Use API key in header
curl http://localhost:8000/api/chat/query \
-H "X-API-Key: pk8_live_abc123..."
# Or in query parameter
curl "http://localhost:8000/api/chat/query?api_key=pk8_live_abc123..."
curl -X POST http://localhost:8000/api/auth/refresh \
-H "Content-Type: application/json" \
-d '{"refresh_token": "eyJhbGciOiJIUzI1NiIs..."}'
{
"sub": "550e8400-e29b-41d4-a716-446655440000",
"email": "user@example.com",
"org_id": "org_123",
"roles": ["user", "power_user"],
"permissions": ["chat:view", "chat:create", "project:view"],
"type": "access",
"iat": 1704067200,
"exp": 1704070800
}
The adminStore.js provides reactive state management for authentication:
// store/adminStore.js
import { defineStore } from 'pinia'
export const useAdminStore = defineStore('admin', {
state: () => ({
user: null,
tokens: {
access: localStorage.getItem('access_token'),
refresh: localStorage.getItem('refresh_token')
},
permissions: [],
roles: []
}),
getters: {
isAuthenticated: (state) => !!state.tokens.access,
hasPermission: (state) => (permission) => state.permissions.includes(permission),
hasRole: (state) => (role) => state.roles.includes(role),
isAdmin: (state) => state.roles.some(r => ['admin', 'super_admin'].includes(r))
},
actions: {
async login(email, password) { /* ... */ },
async logout() { /* ... */ },
async refreshToken() { /* ... */ }
}
})
// plugins/axios.js
import axios from 'axios'
import { useAdminStore } from '@/store/adminStore'
const api = axios.create({ baseURL: 'http://localhost:8000' })
// Add auth header
api.interceptors.request.use((config) => {
const store = useAdminStore()
if (store.tokens.access) {
config.headers.Authorization = `Bearer ${store.tokens.access}`
}
return config
})
// Handle 401 with refresh
api.interceptors.response.use(
(response) => response,
async (error) => {
if (error.response?.status === 401) {
const store = useAdminStore()
try {
await store.refreshToken()
error.config.headers.Authorization = `Bearer ${store.tokens.access}`
return api.request(error.config)
} catch {
store.logout()
window.location.href = '/admin/login'
}
}
return Promise.reject(error)
}
)
// router/index.js
router.beforeEach((to, from, next) => {
const store = useAdminStore()
if (to.meta.requiresAuth && !store.isAuthenticated) {
return next({ path: '/admin/login', query: { redirect: to.fullPath } })
}
if (to.meta.requiredPermission && !store.hasPermission(to.meta.requiredPermission)) {
return next({ path: '/403' })
}
next()
})
| Table | Description |
|---|---|
organizations |
Multi-tenant organizations |
departments |
Organization departments |
teams |
Department teams |
users |
User accounts |
roles |
Role definitions |
permissions |
Permission definitions |
user_roles |
User-role assignments (N:M) |
role_permissions |
Role-permission mappings (N:M) |
api_keys |
API key credentials |
sessions |
Active login sessions |
audit_logs |
Activity audit trail |
resource_access |
Resource-level permissions |
# Apply migrations
psql -h localhost -U 8 -d 8 -f shared/database/migrations/002_create_auth_tables.sql
# Verify tables
psql -c "\dt" -U 8 -d 8
from shared.auth import get_auth_service
async def seed_roles():
async with get_auth_service() as auth:
# Seed permissions from enum
await auth.permissions.seed_permissions()
# Create built-in roles
for role_data in BUILT_IN_ROLES:
await auth.roles.create(**role_data)
# List users
GET /api/admin/users?page=1&limit=20&search=john
# Get user details
GET /api/admin/users/{user_id}
# Create user
POST /api/admin/users
{
"email": "newuser@example.com",
"username": "newuser",
"password": "SecurePass123!",
"full_name": "New User",
"organization_id": "uuid",
"role_codes": ["user"]
}
# Update user
PATCH /api/admin/users/{user_id}
{ "full_name": "Updated Name", "is_active": true }
# Delete user (soft delete)
DELETE /api/admin/users/{user_id}
# Assign role
POST /api/admin/users/{user_id}/roles
{ "role_code": "power_user" }
# Revoke role
DELETE /api/admin/users/{user_id}/roles/{role_code}
# List roles
GET /api/admin/roles
# Create custom role
POST /api/admin/roles
{
"code": "content_manager",
"name": "Content Manager",
"description": "Manages knowledge base content",
"level": 35,
"permission_codes": ["kb:view", "kb:upload", "kb:delete", "kb:index"]
}
# Update role permissions
PATCH /api/admin/roles/{role_id}
{ "permission_codes": ["kb:view", "kb:upload", "kb:delete"] }
# Get recent logs
GET /api/admin/audit?limit=100
# Filter by user
GET /api/admin/audit?user_id={user_id}
# Filter by action
GET /api/admin/audit?action=user.login
# Filter by date range
GET /api/admin/audit?start_date=2026-01-01&end_date=2026-01-31
# Export audit logs
GET /api/admin/audit/export?format=csv
# Before (v1)
from gateway.routers.purple8 import router
from shared.auth import create_access_token
# After (v2)
from gateway.routers.purple8_auth import router
from shared.auth import (
AuthService,
get_auth_service,
require_permission,
Permission,
)
# services/gateway/main.py
# Before
app.include_router(purple8_router)
app.include_router(pipeline_router)
# After
app.include_router(purple8_auth_router) # RBAC enabled
app.include_router(pipeline_auth_router) # RBAC enabled
app.include_router(admin_rbac_router) # Admin APIs
# Run new migration
psql -U 8 -d 8 -f shared/database/migrations/002_create_auth_tables.sql
# Migrate existing users (if any)
python scripts/migrate_users_to_rbac.py
// Update store import
import { useAdminStore } from '@/store/adminStore'
// Update route guards
router.beforeEach((to, from, next) => {
const store = useAdminStore()
// ... new guard logic
})
import jwt
from datetime import datetime
token = "eyJhbGciOiJIUzI1NiIs..."
payload = jwt.decode(token, options={"verify_signature": False})
exp = datetime.fromtimestamp(payload['exp'])
print(f"Token expires: {exp}")
Solution: Implement token refresh in frontend interceptor.
from shared.auth import get_permissions_for_role
user_roles = ["user", "power_user"]
all_permissions = set()
for role in user_roles:
all_permissions.update(get_permissions_for_role(role))
print(f"User permissions: {all_permissions}")
print(f"Has project:create: {'project:create' in all_permissions}")
import asyncio
from shared.database.async_connection import get_async_db_connection
async def test():
db = get_async_db_connection()
healthy = await db.health_check()
print(f"Database healthy: {healthy}")
asyncio.run(test())
# Verify API key format (should be 40+ chars)
# Valid: pk8_live_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
# Check key in database
psql -c "SELECT id, name, is_active, expires_at FROM api_keys WHERE key_prefix = 'pk8_live_xx';"
# Enable debug logging in .env
LOG_LEVEL=DEBUG
SQL_ECHO=true
curl http://localhost:8000/api/health/auth
# Response
{
"status": "healthy",
"database": "connected",
"active_sessions": 42,
"api_keys_active": 15
}
# JWT Configuration
JWT_SECRET=your-secret-key-change-in-production
JWT_ALGORITHM=HS256
JWT_ACCESS_EXPIRATION_HOURS=1
JWT_REFRESH_EXPIRATION_DAYS=7
# Database
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
POSTGRES_DB=8
POSTGRES_USER=8
POSTGRES_PASSWORD=your-password
# Optional
BCRYPT_ROUNDS=12
RATE_LIMIT_DEFAULT=60
/docs/http://localhost:8000/docs