FastAPI RBAC - Full Implementation Tutorial

Implementing RBAC (Role-Based Access Control) in a FastAPI application ensures that each user has the appropriate level of access based on predefined roles.

In this FastAPI RBAC guide, we’ll create a secure contact management app, complete with a fine-grained authorization layer that can be easily integrated into your existing FastAPI workflow.

The application we’re implementing will allow administrators and regular users to interact with a contact management system based on their assigned roles. Here’s what each role can do:

• Admin Capabilities:
  • View, add, update, and delete any contact in the system.
  • Manage user roles and permissions.
  • Access system-wide audit logs.
• Regular User Capabilities:
  • View, add, update, and delete only their own contacts.
  • No access to administrative functions like role management or audit logs.

There are a lot of authentication solutions out there, and it's quite uncommon to build your own authentication solution from scratch. In this tutorial, we’ll extend this capability and integrate more fine-grained access control using Permit.io - an authorization-as-a-service provider.

Prerequisites and Tech Stack

For you to be able to follow along with the tutorial, you should have:

• Some Python Experience
• A Permit.io account

You can access the code in this GitHub repo.

For the tech stack, we have:

• FastAPI: A modern, fast (high-performance) web framework for building APIs with Python 3.7+.
• Uvicorn: A lightning-fast ASGI server, used to run FastAPI application.
• PostgreSQL: A powerful, open-source relational database management system.
• SQLAlchemy: A SQL toolkit and Object-Relational Mapping (ORM) library for Python.
• Alembic: A database migration tool for SQLAlchemy.
• Jinja: A template engine

Project Setup

To get started, let’s clone the project:

git clone <https://github.com/uma-victor1/FastAPI-Permit-RBAC.git>

After installing run the commands:

mkdir FastAPI-Permit-RBAC
cd FastAPI-Permit-RBAC
python -m venv venv
source venv/bin/activate  # macOS/Linux
# or venv\\\\Scripts\\\\activate on Windows

Now we have our project setup, let’s install the necessary dependencies.

pip install -r requirements.txt

Here is what our project structure looks like:

    contact_app/
    ├── app
    │   ├── controllers
    │   │   ├── auth_controller.py     # Authentication routes (login, register)
    │   │   └── contact_controller.py  # Contact CRUD routes
    │   ├── models
    │   │   └── db_models.py           # SQLAlchemy models
    │   ├── schemas
    │   │   └── pydantic_models.py     # Pydantic models (for validation)
    │   ├── services
    │   │   ├── auth_service.py        # Login, registration logic
    │   │   └── contact_service.py     # Business logic for handling contacts
    │   ├── main.py                    # FastAPI entry point
    │   └── ...
    ├── requirements.txt
    └── README.md

With our starter template, we have authentication set up, meaning we can get the currently logged-in user using the user_dependency function in the dependencies module. We use PostgreSQL as our database and go with SQLAlchemy for our ORM, giving us a solid data layer that makes accessing data easy and safe.

To proceed with the tutorial, we need to understand the requirements for our contact management app, set up our database layout for the users and contacts tables, and use Permit.io in our app to manage who can do what. This way, only people with the right access can perform important actions, like adding, changing, or deleting contacts.

Feature Requirements for Our FastAPI RBAC Application

The main requirement for this application is that an admin user has the ability to manage all contacts in the application. The admin has the ability to:

• Add a contact to any user's contact list
• View all contacts in the application
• Update any contact
• Delete any contact by any user

This is what our admin page looks like:

Here is a representation in table form of the access an admin and a normal user has:

Database Schema for our App

For our app, we need just two tables for User and Contacts. These two tables are enough to demonstrate RBAC in our app. Here is what our SQLAlchemy Models look like:

    from sqlalchemy import Column, Integer, String, ForeignKey, DateTime
    from sqlalchemy.orm import relationship, declarative_base
    from datetime import datetime

    Base = declarative_base()

    class User(Base):
        __tablename__ = 'users'

        id = Column(Integer, primary_key=True, index=True)
        username = Column(String, unique=True, nullable=False)
        email = Column(String, unique=True, nullable=False)
        password = Column(String, nullable=False)
        role = Column(String, nullable=False, default="user")
        contacts = relationship("Contact", back_populates="owner")

    class Contact(Base):
        __tablename__ = 'contacts'

        id = Column(Integer, primary_key=True, index=True)
        user_id = Column(Integer, ForeignKey('users.id'))
        name = Column(String, nullable=False)
        phone = Column(String)
        email = Column(String)
        notes = Column(String)
        created_at = Column(DateTime, default=datetime.utcnow)
        updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

        owner = relationship("User", back_populates="contacts")

Implementing RBAC

To manage the admin and user roles in this application, we need to use the Permit.io platform to map out resources and actions. In the previous section, we listed what roles we need and the permissions we need for them. All we have to do is create those roles in our Permit dashboard, create the resources, and manage the permissions for resources in the policy editor.

First, go to your Permit dashboard and create a new project.

After you’ve created a new project, your screen should look similar to the image above. Now, you can create roles and manage resources, which are all done in the policy editor screen. Let’s create a resource.

In Permit, resources refer to the objects or entities within your application that require permission management. In the case of our contact management app, that would be the contacts.

To create the contact resource, Click on the policy tab on the left sidebar, and create the contact resource.

After creating the contact resource, we need to create different user roles and add permissions to each role so that each user assigned a role has permission to carry out the actions determined for that role.

Creating User Roles on Permit

In our application, we only need two roles.

• Admin
• Viewer

To create a role, navigate to the Policy section in the Permit dashboard, click the Roles tab, and add the required roles.

We need to define the specific permissions associated with each role we create. For example, an Admin has permission to delete, create, or update any contact in our application.

Navigate to the Policy Editor tab and adjust permission so it looks like this:

We now have everything set up in our permit dashboard. Let's continue with some code. In the next section, we'll look at how to enforce the permissions we set up using the Permit SDK.

Enforcing the Permissions in Our App

In the previous sections, we have seen how to set roles and permissions in our permit dashboard. This section will show how we can use the Permit API to test and enforce this permission in our FastAPI app.

In your .env file add these credentials:

PERMIT_API_KEY="your_permit_api_key"
PERMIT_PDP_URL="your_permit_pdp_url"

You can find your API key by going to the projects page in the permit dashboard and copying it.

In the projects app folder, create a lib directory and add a file called permit.py. In that file paste the following code to initialize the SDK and connect your Python app to the Permit.io PDP container you've set up

from permit import Permit
from constants import PERMIT_API_KEY, PERMIT_PDP_URL

permit = Permit(
   pdp=PERMIT_PDP_URL,
   token=PERMIT_API_KEY,
)

The pdp is a policy engine needed for evaluating authorization queries based on defined policies. It’s very important for checking permission for roles and although permit provides a pdp URL for testing, permit advises us to deploy our own. For now, we are using the provided pdp URL https://cloudpdp.api.permit.io.

In our utils folder, create a file authorization.py and define the check_permission utility function.

    from fastapi import HTTPException, Depends, Request, Response
    from lib.permit import permit
    from utils.dependencies import get_user
    from models.db import User

    async def check_permission(action: str, resource: str, user: User):
        """
        Checks if a user is authorized to perform a specific action on a resource.
        """
        print("checking permission")
        try:
            allowed = await permit.check(
                str(user.id),
                action,
                {
                    "type": resource,
                },
            )
            if not allowed:
                raise HTTPException(status_code=403, detail="Forbidden: Not allowed")
            return True
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Authorization error: {str(e)}")

This code defines and configures a permission-checking utility check_permission using the Permit SDK in our application. With this initialization, we can manage RBAC anywhere in the FastAPI app.

Securing Privileged Actions in our App

Since we have our configuration set, we can start enforcing some roles and permissions in our app. But something is missing! In our permissions dashboard, we didn’t add any users, so adding roles and permissions is useless. We need a way to sync the users in our app with the users on Permit.

To achieve this, we need a unique way to identify our users. It doesn’t matter what method of authentication we are using; we just need a unique ID for each user. For this project, we are using JWT, so we can decode our JWT and use the user ID or email to sync users to permit.

The perfect place to do this is during the signup process. First, we need to add utility functions for creating a user on Permit and assigning them a role.

In our utils/authorization.py file, add the following code:

    async def assign_role(user_id: str, role: str):
        try:
            await permit.api.users.assign_role(
                {
                    # the user key
                    "user": user_id,
                    # the role key
                    "role": role,
                    # the tenant key
                    "tenant": "default",
                }
            )
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Authorization error: {str(e)}")

    async def create_role(user):
        try:
            await permit.api.users.sync(
                {
                    "key": user["id"],
                    "email": user["email"],
                    "first_name": user["surname"],
                    "last_name": user["surname"],
                }
            )
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Authorization error: {str(e)}")

In the code above, we use Permit’s API to create a user with permit.api.users.sync and assign that new user the role of viewer with permit.api.users.assign_role

Then in our auth controller file app/controllers/auth_controller.py, import the permit utility functions and sync users to Permit.

    from datetime import datetime
    from datetime import timezone
    from fastapi import APIRouter
    from fastapi import HTTPException
    from fastapi import status
    from fastapi import Response, Form
    from fastapi.responses import JSONResponse, RedirectResponse
    from fastapi.requests import Request
    from utils import formating
    from models import db
    from models import dto
    from services import user_service
    from services import jwt_service
    from utils.bcrypt_hashing import HashLib
    from utils import dependencies
    from constants import COOKIES_KEY_NAME
    from constants import SESSION_TIME
    from utils.authorization import assign_role, create_role

    router = APIRouter(prefix="/auth", tags=["Auth"])

    @router.post(
        "/register", status_code=status.HTTP_201_CREATED, response_model=dto.GetUser
    )
    async def register(
        res: Response,
        email: str = Form(...),
        password: str = Form(...),
        name: str = Form(...),
        surname: str = Form(...),
    ):
        user = dto.CreateUser(name=name, surname=surname, email=email, password=password)
        email = formating.format_string(user.email)
        NOW = datetime.now(timezone.utc)
        if not email:
            raise HTTPException(
                detail="Email can not be empty",
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            )
        if not user.password:
            raise HTTPException(
                detail="Password can not be empty",
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            )
        exist_user = user_service.get_by_email(email)
        if exist_user:
            raise HTTPException(
                detail=f"User '{email}' exist",
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            )
        created_user = user_service.create(
            user.name, user.surname, db.User.Role.USER, email, user.password
        )
        created_user_dict = created_user.to_dict()
        exp_date = NOW + SESSION_TIME
        token = jwt_service.encode(
            created_user_dict["id"], str(created_user_dict["role"]), exp_date
        )
        await create_role(created_user_dict)
        await assign_role(created_user_dict["id"], "viewer")
        res.set_cookie(
            key=COOKIES_KEY_NAME,
            value=token,
            expires=exp_date,
            httponly=True,
            secure=False,
            samesite="Lax",
        )
        # redirect to home page
        return RedirectResponse(
            url="/", status_code=status.HTTP_303_SEE_OTHER, headers=res.headers
        )

From the code above, any time a new user signs up, a user is created and synced with Permit, and they are assigned a viewer user role.

Securing Admin dashboard and actions

With all we have done in the previous section, syncing users and roles to Permit, the next step is to secure the Admin Dashboard and associated privileged actions. This involves ensuring that only users with the Admin role can access certain routes, view sensitive data, and perform administrative actions such as managing other users' contacts.

In our project, we have a dashboard route that returns a dashboard template we created. The route is located in the page_controller, and the template is located in templates/dashboard.jinja.

To protect the dashboard route, we’ll use the check_permission utility function we created earlier in the /dashboard route:

    @router.get("/dashboard")
    async def dashboard(
        request: Request,
        user: db.User = Depends(get_user),
    ):
        """
        Render the dashboard page showing all contacts.
        """
        if user is None:
            return RedirectResponse(url="/login")
        await check_permission(action="readany", resource="contact", user=user)
        try:
            contacts = contact_service.get_all_contacts()
            return templates.TemplateResponse(
                "dashboard.jinja",
                {
                    "request": request,
                    "contacts": contacts,
                    "user": user,
                },
            )
        except Exception as e:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))

Here, all we are doing is checking if the signed-in user has permission to view the dashboard page with the read any action we set up in our Permit dashboard.

Admin Privileged Actions
Admins have broader permissions than regular users. To secure these actions, enforce role-based permissions with the check_permission function in our dashboard controller file.

    from fastapi import APIRouter, HTTPException, Depends, status, Body, Request, Query
    from sqlalchemy.orm import Session
    from typing import List
    from utils.authorization import check_permission
    from services import contact_service
    from utils.dependencies import get_user
    from db.context import get_db
    from models.dto import CreateContact, UpdateContact, GetContact
    from models.db import User

    router = APIRouter(prefix="/dashboard", tags=["dashboard"])

    @router.post("/{user_id}", status_code=status.HTTP_201_CREATED)
    async def admin_add_contact(
        user_id: int,
        contact_data: CreateContact = Body(...),
        user: User = Depends(get_user),
    ):
        """
        Admin adds a contact to a specific user's contact list.
        """
        try:
            await check_permission(action="createany", resource="contact", user=user)
            contact = contact_service.create_contact(user.id, contact_data)
            return contact
        except ValueError as e:
            raise HTTPException(status_code=400, detail=str(e))

    @router.put("/{id}")
    async def update_any_contact(
        id: int,
        contact_data: UpdateContact,
        user: User = Depends(get_user),
    ):
        try:
            await check_permission(action="updateany", resource="contact", user=user)
            contact_service.update_any_contact(id, contact_data)
            return {"message": "Contact updated successfully"}
        except ValueError as e:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))

    @router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
    async def delete_any_contact(
        id: int,
        user: User = Depends(get_user),
    ):
        try:
            await check_permission(action="deleteany", resource="contact", user=user)
            contact_service.delete_any_contact(id)
        except ValueError as e:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))

For each controller method, we use the check_permission utility function to check if the signed-in user has permission to carry out actions on the contacts resource.
We can also use the check_permission utility function as a FastAPI dependency in our routes.

Wrapping Up

With these implementations, your contact management app is now secure, scalable, and ready for real-world usage. Whether you’re building for a small team or a large enterprise, you can confidently manage user roles and permissions with Permit.io.

This has been a long read. Hopefully, you’ve grasped how Permit.io can be used to implement authorization in your FastAPI application. You can learn more by visiting the Permit.io Docs or reaching out to me on X @umavictor.