Implement FastAPI authentication with JWT and OAuth2 for secure API development

Update system packages

Start by updating your package manager to ensure you have the latest packages available.

sudo apt update && sudo apt upgrade -y

sudo dnf update -y

Install Python 3.12 and development tools

Install Python 3.12 along with pip and virtual environment tools for FastAPI development.

sudo apt install -y python3.12 python3.12-venv python3.12-dev python3-pip build-essential

sudo dnf install -y python3.12 python3.12-venv python3.12-devel python3-pip gcc gcc-c++ make

Create project directory and virtual environment

Set up a dedicated directory for your FastAPI authentication project with an isolated Python environment.

mkdir fastapi-auth && cd fastapi-auth
python3.12 -m venv fastapi-env
source fastapi-env/bin/activate

Install FastAPI and authentication dependencies

Install FastAPI, Uvicorn ASGI server, and the required packages for JWT authentication and password hashing.

pip install fastapi[all] uvicorn[standard] python-jose[cryptography] passlib[bcrypt] python-multipart

Create the main application structure

Set up the basic directory structure for your FastAPI authentication application.

mkdir -p app/{auth,models,routers,utils}
touch app/__init__.py app/main.py app/auth/__init__.py app/models/__init__.py app/routers/__init__.py app/utils/__init__.py

Configure environment variables

Create a secure configuration file for JWT secrets and application settings.

SECRET_KEY=your-256-bit-secret-key-here-change-in-production
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30
REFRESH_TOKEN_EXPIRE_DAYS=7

Create user models and database schemas

Define Pydantic models for user authentication, token handling, and database schemas.

from pydantic import BaseModel, EmailStr
from typing import Optional, List
from enum import Enum

class UserRole(str, Enum):
    ADMIN = "admin"
    USER = "user"
    MODERATOR = "moderator"

class UserBase(BaseModel):
    email: EmailStr
    username: str
    full_name: Optional[str] = None
    roles: List[UserRole] = [UserRole.USER]
    is_active: bool = True

class UserCreate(UserBase):
    password: str

class UserInDB(UserBase):
    id: int
    hashed_password: str

class User(UserBase):
    id: int
    
    class Config:
        from_attributes = True

Create token models

Define Pydantic models for JWT tokens and token data validation.

from pydantic import BaseModel
from typing import Optional, List
from app.models.user import UserRole

class Token(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"

class TokenData(BaseModel):
    username: Optional[str] = None
    user_id: Optional[int] = None
    roles: List[UserRole] = []
    expires_at: Optional[float] = None

Implement password hashing utilities

Create secure password hashing and verification functions using bcrypt.

from passlib.context import CryptContext
from datetime import datetime, timedelta
from typing import Union, Any
from jose import jwt, JWTError
import os
from dotenv import load_dotenv

load_dotenv()

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

SECRET_KEY = os.getenv("SECRET_KEY")
ALGORITHM = os.getenv("ALGORITHM", "HS256")
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", 30))
REFRESH_TOKEN_EXPIRE_DAYS = int(os.getenv("REFRESH_TOKEN_EXPIRE_DAYS", 7))

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Verify a plain password against its hash."""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """Generate password hash."""
    return pwd_context.hash(password)

def create_access_token(
    data: dict, expires_delta: Union[timedelta, None] = None
) -> str:
    """Create JWT access token."""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    
    to_encode.update({"exp": expire, "type": "access"})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def create_refresh_token(data: dict) -> str:
    """Create JWT refresh token."""
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    to_encode.update({"exp": expire, "type": "refresh"})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def verify_token(token: str, token_type: str = "access") -> Union[dict, None]:
    """Verify and decode JWT token."""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != token_type:
            return None
        return payload
    except JWTError:
        return None

Create user database and authentication functions

Implement a simple in-memory user database and authentication functions for demonstration purposes.

from typing import Dict, Optional, List
from app.models.user import UserInDB, UserCreate, UserRole
from app.utils.security import get_password_hash

# In-memory user database (replace with real database in production)
users_db: Dict[str, UserInDB] = {
    "admin@example.com": UserInDB(
        id=1,
        email="admin@example.com",
        username="admin",
        full_name="Admin User",
        roles=[UserRole.ADMIN],
        hashed_password=get_password_hash("admin123"),
        is_active=True
    ),
    "user@example.com": UserInDB(
        id=2,
        email="user@example.com",
        username="testuser",
        full_name="Test User",
        roles=[UserRole.USER],
        hashed_password=get_password_hash("user123"),
        is_active=True
    )
}

def get_user_by_email(email: str) -> Optional[UserInDB]:
    """Get user by email address."""
    return users_db.get(email)

def get_user_by_username(username: str) -> Optional[UserInDB]:
    """Get user by username."""
    for user in users_db.values():
        if user.username == username:
            return user
    return None

def create_user(user: UserCreate) -> UserInDB:
    """Create new user in database."""
    hashed_password = get_password_hash(user.password)
    user_id = len(users_db) + 1
    
    db_user = UserInDB(
        id=user_id,
        email=user.email,
        username=user.username,
        full_name=user.full_name,
        roles=user.roles,
        hashed_password=hashed_password,
        is_active=user.is_active
    )
    
    users_db[user.email] = db_user
    return db_user

Implement OAuth2 authentication scheme

Create OAuth2 password bearer authentication with JWT token validation and user dependency injection.

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from app.utils.security import verify_password, verify_token
from app.auth.database import get_user_by_email, get_user_by_username
from app.models.user import User, UserInDB, UserRole
from app.models.token import TokenData
from typing import Optional, List

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")

def authenticate_user(username: str, password: str) -> Optional[UserInDB]:
    """Authenticate user with username/email and password."""
    user = get_user_by_username(username) or get_user_by_email(username)
    if not user:
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    """Get current user from JWT token."""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    payload = verify_token(token, "access")
    if payload is None:
        raise credentials_exception
    
    username: str = payload.get("sub")
    if username is None:
        raise credentials_exception
    
    user = get_user_by_username(username) or get_user_by_email(username)
    if user is None:
        raise credentials_exception
    
    if not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Inactive user"
        )
    
    return User(
        id=user.id,
        email=user.email,
        username=user.username,
        full_name=user.full_name,
        roles=user.roles,
        is_active=user.is_active
    )

def require_roles(allowed_roles: List[UserRole]):
    """Dependency to check if user has required roles."""
    def role_checker(current_user: User = Depends(get_current_user)) -> User:
        if not any(role in current_user.roles for role in allowed_roles):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient permissions"
            )
        return current_user
    return role_checker

Create authentication routes

Implement login, registration, token refresh, and user management endpoints.

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from app.models.user import User, UserCreate, UserRole
from app.models.token import Token
from app.auth.oauth2 import authenticate_user, get_current_user, require_roles
from app.auth.database import create_user, get_user_by_email, get_user_by_username
from app.utils.security import create_access_token, create_refresh_token, verify_token
from datetime import timedelta

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

@router.post("/login", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    """Login endpoint that returns JWT tokens."""
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    access_token = create_access_token(
        data={"sub": user.username, "user_id": user.id, "roles": [role.value for role in user.roles]}
    )
    refresh_token = create_refresh_token(
        data={"sub": user.username, "user_id": user.id}
    )
    
    return Token(
        access_token=access_token,
        refresh_token=refresh_token,
        token_type="bearer"
    )

@router.post("/register", response_model=User)
async def register_user(user: UserCreate):
    """Register new user endpoint."""
    if get_user_by_email(user.email):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Email already registered"
        )
    
    if get_user_by_username(user.username):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already taken"
        )
    
    db_user = create_user(user)
    return User(
        id=db_user.id,
        email=db_user.email,
        username=db_user.username,
        full_name=db_user.full_name,
        roles=db_user.roles,
        is_active=db_user.is_active
    )

@router.post("/refresh", response_model=Token)
async def refresh_token(refresh_token: str):
    """Refresh access token using refresh token."""
    payload = verify_token(refresh_token, "refresh")
    if payload is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid refresh token"
        )
    
    username = payload.get("sub")
    user = get_user_by_username(username) or get_user_by_email(username)
    
    if not user or not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not found or inactive"
        )
    
    access_token = create_access_token(
        data={"sub": user.username, "user_id": user.id, "roles": [role.value for role in user.roles]}
    )
    new_refresh_token = create_refresh_token(
        data={"sub": user.username, "user_id": user.id}
    )
    
    return Token(
        access_token=access_token,
        refresh_token=new_refresh_token,
        token_type="bearer"
    )

@router.get("/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
    """Get current user information."""
    return current_user

Create protected API routes with role-based access

Implement example API endpoints with different access levels based on user roles.

from fastapi import APIRouter, Depends
from app.models.user import User, UserRole
from app.auth.oauth2 import get_current_user, require_roles
from typing import List

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

@router.get("/public")
async def public_endpoint():
    """Public endpoint that doesn't require authentication."""
    return {"message": "This is a public endpoint"}

@router.get("/private")
async def private_endpoint(current_user: User = Depends(get_current_user)):
    """Private endpoint that requires authentication."""
    return {
        "message": f"Hello {current_user.full_name or current_user.username}!",
        "user_id": current_user.id,
        "roles": current_user.roles
    }

@router.get("/admin")
async def admin_endpoint(current_user: User = Depends(require_roles([UserRole.ADMIN]))):
    """Admin-only endpoint that requires admin role."""
    return {
        "message": "Admin access granted",
        "admin_user": current_user.username,
        "sensitive_data": "This is admin-only information"
    }

@router.get("/moderator")
async def moderator_endpoint(
    current_user: User = Depends(require_roles([UserRole.ADMIN, UserRole.MODERATOR]))
):
    """Moderator endpoint that requires admin or moderator role."""
    return {
        "message": "Moderator access granted",
        "user": current_user.username,
        "roles": current_user.roles,
        "moderation_tools": ["ban_user", "delete_post", "edit_content"]
    }

@router.get("/users")
async def list_users(current_user: User = Depends(require_roles([UserRole.ADMIN]))):
    """Admin endpoint to list all users."""
    from app.auth.database import users_db
    
    users = [
        {
            "id": user.id,
            "username": user.username,
            "email": user.email,
            "full_name": user.full_name,
            "roles": user.roles,
            "is_active": user.is_active
        }
        for user in users_db.values()
    ]
    
    return {"users": users, "total": len(users)}

Create the main FastAPI application

Set up the main application with CORS, security headers, and route registration.

from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.responses import JSONResponse
from app.routers import auth, protected
import time

app = FastAPI(
    title="FastAPI Authentication Demo",
    description="JWT and OAuth2 authentication with role-based access control",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

# Security middleware
app.add_middleware(
    TrustedHostMiddleware, 
    allowed_hosts=["localhost", "127.0.0.1", "*.example.com"]
)

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

@app.middleware("http")
async def add_security_headers(request: Request, call_next):
    """Add security headers to all responses."""
    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"
    response.headers["