Building An RBAC System With FastAPI & Cloudflare Workers
Hey guys! Today, we're diving deep into building a Role-Based Access Control (RBAC) system using FastAPI and Cloudflare Workers. This is super important for securing your applications and ensuring that only authorized users can access specific resources. We'll explore how to design the system, implement it with FastAPI, and deploy it using Cloudflare Workers, making use of D1 Database and KV storage for optimal performance. So, buckle up and let's get started!
Understanding RBAC and Its Importance
Before we jump into the implementation, let's quickly recap what RBAC is and why it's so crucial. RBAC is an access control mechanism that restricts system access to authorized users based on their roles within an organization. Instead of assigning permissions directly to users, RBAC assigns permissions to roles and then assigns users to those roles. This makes managing permissions much easier, especially in large and complex systems. Imagine you have a system with hundreds or even thousands of users. Managing individual permissions would be a nightmare! With RBAC, you can simply assign users to roles like "admin," "editor," or "viewer," and each role has a predefined set of permissions. This not only simplifies management but also enhances security by minimizing the risk of misconfiguration or unauthorized access.
In our context, we'll be building an RBAC system where users have permissions based on a RESOURCE:PERMISSION format. For example, a user might have the COMPANIES:CREATE permission, allowing them to create new company records, or the LISTS:READ permission, granting them access to read lists. The most critical part of our system will be the access-check route, which the Marketing API will use to verify if a user has the necessary permissions to access a particular resource. This ensures that our marketing applications only perform actions that users are explicitly authorized to do.
Designing the System Architecture
Our RBAC system will be built using FastAPI, a modern, fast (high-performance), web framework for building APIs with Python. We'll also leverage Cloudflare Workers, a serverless execution environment that allows us to deploy our API to Cloudflare's global network. This ensures low latency and high availability. To store our data, we'll use Cloudflare's D1 Database and KV storage. D1 is a serverless SQL database, while KV storage is a key-value store that offers extremely fast read operations.
The key architectural decisions we need to make are:
- Where to store the RBAC data: We'll need to decide which data should reside in D1 and which should go into KV storage. Generally, data that requires complex queries and relationships is better suited for D1, while data that needs to be accessed very quickly is a good candidate for KV storage.
- How to structure our API endpoints: We need to define the API endpoints for managing roles, permissions, and user assignments, as well as the crucial
access-checkendpoint. - How to optimize the
access-checkroute: This route will be called frequently, so it's essential to make it as efficient as possible. We'll likely use KV storage to cache permission data for faster lookups.
Database Schema
Let's take a look at the database schema we'll be using, which is defined in the migrations/0001_seed.sql file. This schema will likely include tables for:
users: To store user information.roles: To store role information (e.g., admin, editor).permissions: To store permission information (e.g.,user:create,article:publish).role_permissions: A many-to-many relationship table linking roles to permissions.user_roles: A many-to-many relationship table linking users to roles.
This schema allows us to define roles with specific permissions and then assign users to those roles. When a user tries to access a resource, we can efficiently query the database to check if they have the necessary permissions through their assigned roles.
API Endpoints: A Detailed Overview
Now, let's dive into the API endpoints we'll be building. These endpoints will allow us to manage roles, permissions, and user assignments, as well as perform access checks. Understanding these endpoints is crucial for building a functional and secure RBAC system. We will cover GET, POST, PUT, and DELETE methods for various resources like roles and permissions, detailing their request and response structures.
Managing Roles
Roles are the cornerstone of our RBAC system. They define sets of permissions that can be assigned to users. Our API will provide endpoints to manage roles, allowing us to create, read, update, and delete them.
-
GET /roles: This endpoint will retrieve a list of all roles in the system. The response will be an array of role objects, each containing an
id,name,description, and a list ofpermissions.[ { "id": 1, "name": "admin", "description": "System administrator", "permissions": ["user:create", "user:delete"] } ] -
POST /roles: This endpoint will create a new role. The request body will contain the
nameanddescriptionof the role. The response will be the newly created role object, including itsid.Request:
{ "name": "editor", "description": "Can edit and publish content" }Response:
{ "id": 2, "name": "editor", "description": "Can edit and publish content" } -
**PUT /roles/id}
in the URL is the ID of the role to update. The request body will contain the fields to update, such as thedescription`. The response will be the updated role object.Request:
{ "description": "Editor with limited access" }Response:
{ "id": 2, "name": "editor", "description": "Editor with limited access" } -
**DELETE /roles/id}` in the URL is the ID of the role to delete. The response will be a success message.
Response:
{ "message": "Role deleted successfully" }
Managing Permissions
Permissions define specific actions that users can perform within the system. Like roles, our API will provide endpoints to manage permissions.
-
GET /permissions: This endpoint will retrieve a list of all permissions in the system. The response will be an array of permission objects, each containing an
id,name, anddescription.[ { "id": 1, "name": "user:create", "description": "Create a new user" } ] -
POST /permissions: This endpoint will create a new permission. The request body will contain the
nameanddescriptionof the permission. The response will be the newly created permission object, including itsid.Request:
{ "name": "article:publish", "description": "Publish an article" }Response:
{ "id": 5, "name": "article:publish", "description": "Publish an article" } -
**PUT /permissions/id}
in the URL is the ID of the permission to update. The request body will contain the fields to update, such as thedescription`. The response will be the updated permission object.Request:
{ "description": "Publish articles to public feed" }Response:
{ "id": 5, "name": "article:publish", "description": "Publish articles to public feed" } -
**DELETE /permissions/id}` in the URL is the ID of the permission to delete. The response will be a success message.
{ "message": "Permission deleted successfully" }
Assigning Permissions to Roles
This set of endpoints allows us to manage the relationship between roles and permissions.
-
**POST /roles/role_id}/permissions
in the URL is the ID of the role. The request body will contain an array ofpermission_idsto assign to the role. The response will include therole_idand the list of assignedpermission_ids`.Request:
{ "permission_ids": [1, 2, 3] }Response:
{ "role_id": 1, "assigned_permissions": [1, 2, 3] } -
**GET /roles/role_id}/permissions
in the URL is the ID of the role. The response will include therole_id` and an array of permission objects assigned to the role.{ "role_id": 1, "permissions": [ { "id": 1, "name": "user:create" }, { "id": 2, "name": "user:delete" } ] }
Assigning Roles to Users
These endpoints manage the relationship between users and roles.
-
**POST /users/user_id}/roles
in the URL is the ID of the user. The request body will contain an array ofrole_idsto assign to the user. The response will include theuser_idand the list of assignedrole_ids`.Request:
{ "role_ids": [1, 2] }Response:
{ "user_id": 10, "assigned_roles": [1, 2] } -
**GET /users/user_id}/roles
in the URL is the ID of the user. The response will include theuser_id` and an array of role objects assigned to the user.{ "user_id": 10, "roles": [ { "id": 1, "name": "admin" }, { "id": 2, "name": "editor" } ] }
Retrieving User Permissions
This endpoint retrieves the effective permissions for a user, considering all their assigned roles.
-
**GET /users/user_id}/permissions
in the URL is the ID of the user. The response will include theuser_id` and an array of permission names that the user has.{ "user_id": 10, "permissions": [ "user:create", "user:delete", "article:publish" ] }
The Crucial Access Check Endpoint
This is the most important endpoint in our system. It's used to check if a user has a specific permission.
-
POST /access/check: This endpoint will check if a user has a specific permission. The request body will contain the
user_idand thepermissionto check. The response will indicate whether the user is allowed or not.Request:
{ "user_id": 10, "permission": "article:publish" }Response:
{ "allowed": true }
Implementing the RBAC System with FastAPI
Now that we have a clear understanding of the API endpoints, let's talk about how to implement them using FastAPI. We'll need to define our data models, create database connections, and implement the logic for each endpoint. This section will delve into the practical steps of building the system, including setting up FastAPI, defining models, connecting to databases, and implementing the endpoint logic. Remember, the goal is to create a robust and efficient RBAC system that can handle a large number of users and requests.
Setting up FastAPI
First, we need to set up a FastAPI project. This involves creating a new directory, installing FastAPI and its dependencies, and creating a basic application structure. We'll also need to install a database driver for D1, such as aiosqlite, and any other necessary libraries.
# Create a new directory for the project
mkdir rbac-system
cd rbac-system
# Create a virtual environment
python3 -m venv venv
source venv/bin/activate
# Install FastAPI and dependencies
pip install fastapi uvicorn aiosqlite
# Create the main application file
touch main.py
In main.py, we'll create a basic FastAPI application:
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_root():
return {"message": "Hello, world!"}
We can then run the application using Uvicorn:
uvicorn main:app --reload
This will start the FastAPI application, and you can access it in your browser at http://localhost:8000.
Defining Data Models
Next, we need to define our data models using Pydantic, a data validation and settings management library. Pydantic allows us to define the structure of our data and ensures that incoming data conforms to our expectations. We'll create models for users, roles, permissions, and the relationships between them.
from pydantic import BaseModel
from typing import List, Optional
class Permission(BaseModel):
id: int
name: str
description: str
class Role(BaseModel):
id: int
name: str
description: str
permissions: Optional[List[Permission]] = None
class User(BaseModel):
id: int
username: str
email: str
roles: Optional[List[Role]] = None
class RoleCreate(BaseModel):
name: str
description: str
class PermissionCreate(BaseModel):
name: str
description: str
class RoleUpdate(BaseModel):
description: str
class PermissionUpdate(BaseModel):
description: str
class AssignPermissionsRequest(BaseModel):
permission_ids: List[int]
class AssignRolesRequest(BaseModel):
role_ids: List[int]
class AccessCheckRequest(BaseModel):
user_id: int
permission: str
class AccessCheckResponse(BaseModel):
allowed: bool
These models define the structure of our data and will be used for request and response bodies in our API endpoints.
Connecting to D1 Database
Now, let's connect to our D1 database. We'll use aiosqlite to interact with the database asynchronously. We'll create a database connection pool and define functions to execute queries.
import aiosqlite
DATABASE_URL = "rbac.db" # Replace with your D1 database URL
async def create_database_connection():
async with aiosqlite.connect(DATABASE_URL) as db:
await db.execute("""
CREATE TABLE IF NOT EXISTS permissions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
description TEXT
);
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS roles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
description TEXT
);
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS role_permissions (
role_id INTEGER NOT NULL,
permission_id INTEGER NOT NULL,
FOREIGN KEY (role_id) REFERENCES roles (id),
FOREIGN KEY (permission_id) REFERENCES permissions (id),
PRIMARY KEY (role_id, permission_id)
);
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE,
email TEXT NOT NULL
);
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS user_roles (
user_id INTEGER NOT NULL,
role_id INTEGER NOT NULL,
FOREIGN KEY (user_id) REFERENCES users (id),
FOREIGN KEY (role_id) REFERENCES roles (id),
PRIMARY KEY (user_id, role_id)
);
""")
await db.commit()
async def fetch_all(query: str, *args):
async with aiosqlite.connect(DATABASE_URL) as db:
async with db.execute(query, args) as cursor:
return await cursor.fetchall()
async def fetch_one(query: str, *args):
async with aiosqlite.connect(DATABASE_URL) as db:
async with db.execute(query, args) as cursor:
return await cursor.fetchone()
async def execute_query(query: str, *args):
async with aiosqlite.connect(DATABASE_URL) as db:
await db.execute(query, args)
await db.commit()
This code sets up the database connection and provides helper functions for executing queries. We also create the database tables if they don't exist.
Implementing API Endpoints
Now comes the fun part: implementing the API endpoints! We'll create routes for managing roles, permissions, and user assignments, as well as the access-check endpoint. For each endpoint, we'll need to:
- Define the route and HTTP method (GET, POST, PUT, DELETE).
- Parse the request body (if any) using Pydantic models.
- Interact with the database to perform the requested operation.
- Return the appropriate response.
Here's an example of how to implement the GET /roles endpoint:
@app.get("/roles", response_model=List[Role])
async def get_roles():
roles = await fetch_all("SELECT id, name, description FROM roles")
return [Role(id=role[0], name=role[1], description=role[2]) for role in roles]
This code retrieves all roles from the database and returns them as a list of Role objects. We'll implement the other endpoints in a similar fashion, using the database helper functions we defined earlier.
Optimizing the access-check Route
The access-check route is the most performance-critical part of our system. It will be called frequently by the Marketing API, so we need to make it as efficient as possible. One way to optimize this route is to cache permission data in KV storage. KV storage offers extremely fast read operations, which can significantly reduce the latency of access checks.
The idea is to store a mapping of user_id:permission to allowed in KV storage. When an access check request comes in, we first check if the permission is cached in KV storage. If it is, we return the cached result immediately. If not, we query the D1 database, store the result in KV storage, and then return it.
Here's a simplified example of how we might implement this:
import json
# Assuming you have a KV binding named 'RBAC_KV'
async def check_permission_in_kv(user_id: int, permission: str) -> Optional[bool]:
key = f"{user_id}:{permission}"
cached_result = await RBAC_KV.get(key)
if cached_result:
return json.loads(cached_result)
return None
async def store_permission_in_kv(user_id: int, permission: str, allowed: bool):
key = f"{user_id}:{permission}"
await RBAC_KV.put(key, json.dumps(allowed))
@app.post("/access/check", response_model=AccessCheckResponse)
async def access_check(request: AccessCheckRequest):
cached_result = await check_permission_in_kv(request.user_id, request.permission)
if cached_result is not None:
return AccessCheckResponse(allowed=cached_result)
# Query the database to check permission
query = """
SELECT COUNT(*) FROM users u
INNER JOIN user_roles ur ON u.id = ur.user_id
INNER JOIN roles r ON ur.role_id = r.id
INNER JOIN role_permissions rp ON r.id = rp.role_id
INNER JOIN permissions p ON rp.permission_id = p.id
WHERE u.id = ? AND p.name = ?
"""
result = await fetch_one(query, request.user_id, request.permission)
allowed = result[0] > 0
await store_permission_in_kv(request.user_id, request.permission, allowed)
return AccessCheckResponse(allowed=allowed)
This code first checks if the permission is cached in KV storage. If it is, it returns the cached result. If not, it queries the database, stores the result in KV storage, and then returns it. This caching mechanism can significantly improve the performance of the access-check route.
Deploying to Cloudflare Workers
Once we've implemented and optimized our RBAC system, we can deploy it to Cloudflare Workers. This involves creating a Cloudflare Workers project, configuring the necessary bindings (such as the D1 database and KV storage), and deploying the code. Cloudflare Workers allows us to run our API on Cloudflare's global network, ensuring low latency and high availability.
The deployment process typically involves using the Cloudflare Workers CLI (wrangler) to build and deploy the project. We'll need to configure wrangler.toml to specify the project settings, including the bindings to our D1 database and KV storage.
Conclusion
Building a robust RBAC system is crucial for securing your applications. In this article, we've walked through the process of designing and implementing an RBAC system using FastAPI and Cloudflare Workers. We've covered the importance of RBAC, the key architectural decisions, the API endpoints, the implementation details with FastAPI, and the optimization strategies using KV storage. By following these steps, you can create a secure and efficient RBAC system for your applications. Remember, security is an ongoing process, and it's essential to continuously review and update your RBAC system as your application evolves. Keep your system secure, your users safe, and your applications robust. Cheers, guys!