FastAPI Authentication & Authorization | JWT + RSA Encryption + Admin CRUD

Build a powerful authentication and authorization system with FastAPI using JWT tokens, RSA encryption, and a full Admin CRUD interface.
This step-by-step tutorial covers securing your FastAPI backend with JWT signed by RSA keys, protecting routes based on user roles, and implementing admin features to manage users.

This tutorial walks you through creating a production-ready REST API using FastAPI with:

• RSA-encrypted login passwords

• JWT authentication with roles (admin, user)

• Admin-only user management (CRUD)

• MySQL database integration

• Token blacklist for logout support

• Full Postman testing

Tech Stack

• Python 3.10+

• FastAPI

• MySQL 8+

• JWT (python-jose)

• RSA Encryption (OpenSSL)

• bcrypt (password hashing)

• Postman (API testing)

Authentication & Authorization Flow Overview

1. Client encrypts login password with RSA Public Key.

2. Server decrypts password with RSA Private Key.

3. Password is verified using bcrypt.

4. JWT token issued containing user role.

5. Middleware validates token and user role.

6. Admin-only APIs protected by role.

7. Logout adds tokens to a blacklist to revoke access.

Step 1: Create Project and Setup Virtual Environment

mkdir python-fastapi-rsa-auth
cd python-fastapi-rsa-auth
python3 -m venv venv
source venv/bin/activate

Step 2: Install Dependencies

pip install fastapi uvicorn python-jose bcrypt mysql-connector-python python-dotenv cryptography

Step 3: Project Structure

app/
 ├── api/
 │   ├── auth/
 │   │   ├── register.py
 │   │   ├── login.py
 │   │   ├── profile.py
 │   │   └── logout.py
 │   └── users/
 │       └── user_crud.py
 ├── middleware/
 │   └── auth.py
 ├── utils/
 │   ├── jwt.py
 │   └── rsa.py
 ├── db/
 │   └── mysql.py
 └── main.py

storage/
 └── rsa/
     ├── private.pem
     └── public.pem

.env
.gitignore

Step 4: Configure .env file

Create .env in your project root:

DB_HOST=localhost
DB_USER=root
DB_PASSWORD=
DB_NAME=python_rsa_auth

JWT_SECRET=super_secret_key
JWT_EXPIRES=900  # 15 minutes

Step 5: MySQL Database Setup

Run these SQL commands:

CREATE DATABASE python_rsa_auth;
USE python_rsa_auth;

CREATE TABLE users (
  id INT AUTO_INCREMENT PRIMARY KEY,
  name VARCHAR(100),
  email VARCHAR(100) UNIQUE,
  password VARCHAR(255),
  role VARCHAR(20) DEFAULT 'user',
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE token_blacklist (
  id INT AUTO_INCREMENT PRIMARY KEY,
  token TEXT,
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Step 6: Generate RSA Keys

Generate RSA key pair:

openssl genrsa -out storage/rsa/private.pem 2048
openssl rsa -in storage/rsa/private.pem -pubout -out storage/rsa/public.pem

• Use public.pem on frontend to encrypt passwords.

• Use private.pem on backend to decrypt.

Warning: Never commit private.pem to source control!

Step 7: Setup MySQL Connection

Create app/db/mysql.py:

import mysql.connector
import os
from dotenv import load_dotenv

load_dotenv()

db = mysql.connector.connect(
    host=os.getenv("DB_HOST"),
    user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD"),
    database=os.getenv("DB_NAME")
)

Step 8: JWT Helper

Create app/utils/jwt.py:

from jose import jwt
import os, time

def sign_token(user):
    payload = {
        "id": user["id"],
        "email": user["email"],
        "role": user.get("role", "user"),
        "exp": time.time() + int(os.getenv("JWT_EXPIRES"))
    }
    return jwt.encode(payload, os.getenv("JWT_SECRET"), algorithm="HS256")

def verify_token(token):
    return jwt.decode(token, os.getenv("JWT_SECRET"), algorithms=["HS256"])

Step 9: RSA Helper

Create app/utils/rsa.py:

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import padding
import base64

with open("storage/rsa/private.pem", "rb") as f:
    private_key = serialization.load_pem_private_key(f.read(), password=None)

def decrypt_password(encrypted):
    decrypted = private_key.decrypt(
        base64.b64decode(encrypted),
        padding.PKCS1v15()
    )
    return decrypted.decode()

Step 🔟 Middleware for Authentication and Role-based Access

Create app/middleware/auth.py:

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer
from app.utils.jwt import verify_token
from app.db.mysql import db

security = HTTPBearer()

def auth_required(credentials=Depends(security)):
    token = credentials.credentials
    cursor = db.cursor()
    cursor.execute("SELECT id FROM token_blacklist WHERE token=%s", (token,))
    if cursor.fetchone():
        raise HTTPException(status_code=401, detail="Token has been revoked")
    return verify_token(token)

def admin_required(credentials=Depends(security)):
    user = auth_required(credentials)
    if user.get("role") != "admin":
        raise HTTPException(status_code=403, detail="Admin access required")
    return user

Step 1️⃣1️⃣ Register API

Create app/api/auth/register.py:

from fastapi import APIRouter
import bcrypt
from app.db.mysql import db

router = APIRouter()

@router.post("/register")
def register(data: dict):
    cursor = db.cursor()
    hashed = bcrypt.hashpw(data["password"].encode(), bcrypt.gensalt())
    cursor.execute(
        "INSERT INTO users (name, email, password, role) VALUES (%s, %s, %s, %s)",
        (data["name"], data["email"], hashed, data.get("role", "user"))
    )
    db.commit()
    return {"message": "User registered successfully"}

Step 1️⃣2️⃣ Login API

Create app/api/auth/login.py:

from fastapi import APIRouter, HTTPException
import bcrypt
from app.utils.rsa import decrypt_password
from app.utils.jwt import sign_token
from app.db.mysql import db

router = APIRouter()

@router.post("/login")
def login(data: dict):
    cursor = db.cursor(dictionary=True)
    cursor.execute("SELECT * FROM users WHERE email=%s", (data["email"],))
    user = cursor.fetchone()
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    decrypted = decrypt_password(data["password"])
    if not bcrypt.checkpw(decrypted.encode(), user["password"].encode()):
        raise HTTPException(status_code=401, detail="Invalid credentials")

    return {
        "user": {
            "id": user["id"],
            "email": user["email"],
            "role": user["role"]
        },
        "access_token": sign_token(user),
        "token_type": "Bearer"
    }

Step 1️⃣3️⃣ Profile API

Create app/api/auth/profile.py:

from fastapi import APIRouter, Depends
from app.middleware.auth import auth_required

router = APIRouter()

@router.get("/profile")
def profile(user=Depends(auth_required)):
    return user

Step 1️⃣4️⃣ Logout API with Token Blacklist

Create app/api/auth/logout.py:

from fastapi import APIRouter, Depends
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from app.db.mysql import db

router = APIRouter()
security = HTTPBearer()

@router.post("/logout")
def logout(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    cursor = db.cursor()
    cursor.execute("INSERT INTO token_blacklist (token) VALUES (%s)", (token,))
    db.commit()
    return {"message": "Successfully logged out"}

Step 1️⃣5️⃣ Admin User Management CRUD

Create app/api/users/user_crud.py:

from fastapi import APIRouter, HTTPException, Depends, Request
from app.middleware.auth import admin_required
from app.db.mysql import db

router = APIRouter()

@router.get("/users")
def list_users(user=Depends(admin_required)):
    cursor = db.cursor(dictionary=True)
    cursor.execute("SELECT id, name, email, role, created_at FROM users")
    users = cursor.fetchall()
    return {"users": users}

@router.get("/users/{user_id}")
def get_user(user_id: int, user=Depends(admin_required)):
    cursor = db.cursor(dictionary=True)
    cursor.execute("SELECT id, name, email, role, created_at FROM users WHERE id=%s", (user_id,))
    user_data = cursor.fetchone()
    if not user_data:
        raise HTTPException(status_code=404, detail="User not found")
    return user_data

@router.put("/users/{user_id}")
async def update_user(user_id: int, request: Request, user=Depends(admin_required)):
    data = await request.json()
    name = data.get("name")
    email = data.get("email")
    role = data.get("role")

    if not name or not email or not role:
        raise HTTPException(status_code=400, detail="Name, Email, and Role are required")

    cursor = db.cursor()
    cursor.execute("UPDATE users SET name=%s, email=%s, role=%s WHERE id=%s", (name, email, role, user_id))
    db.commit()

    return {"message": "User updated successfully"}

@router.delete("/users/{user_id}")
def delete_user(user_id: int, user=Depends(admin_required)):
    cursor = db.cursor()
    cursor.execute("DELETE FROM users WHERE id=%s", (user_id,))
    db.commit()
    return {"message": "User deleted successfully"}

Step 1️⃣6️⃣ Main Application

Create app/main.py:

from fastapi import FastAPI
from app.api.auth import register, login, profile, logout
from app.api.users import user_crud

app = FastAPI()

app.include_router(register.router, prefix="/api/auth")
app.include_router(login.router, prefix="/api/auth")
app.include_router(profile.router, prefix="/api/auth")
app.include_router(logout.router, prefix="/api/auth")

app.include_router(user_crud.router, prefix="/api")

Step 1️⃣7️⃣ Run Application

Start the server:

uvicorn app.main:app --reload

API Base URL:

http://localhost:8000/api

Step 1️⃣8️⃣ Postman Testing (Step-by-Step)

1. Register Admin

POST /api/auth/register

Body (raw JSON):

{
  "name": "Admin User",
  "email": "admin@example.com",
  "password": "<RSA encrypted password>",
  "role": "admin"
}

2. Login Admin

POST /api/auth/login

Body (raw JSON):

{
  "email": "admin@example.com",
  "password": "<RSA encrypted password>"
}

Copy access_token from response.

3. List Users (Admin only)

GET /api/users

Headers:

Authorization: Bearer <ADMIN_TOKEN>

4. Test Normal User Access

Try GET /api/users with a non-admin user token → should return 403 Forbidden

5. Get User Profile

GET /api/auth/profile

Headers:

Authorization: Bearer <TOKEN>

Accessible by all logged-in users.

6. Logout

POST /api/auth/logout

Headers:

Authorization: Bearer <TOKEN>

The token will be blacklisted and cannot be used further.

Production Best Practices

• Use HTTPS

• Never commit private RSA keys

• Use short-lived JWT tokens

• Rotate RSA keys regularly

• Secure your .env file

• Rate-limit login endpoints

Final Result

You now have a secure, role-based FastAPI REST API with:

• RSA-encrypted login

• JWT authentication with role control

• Admin-only user management

• Token blacklisting (logout)

• MySQL backend

• Fully tested with Postman

🔗 Source Code

Want the full source code?
👉 Download the complete Python JWT example from my GitHub repository