FastAPI JWT Authentication with OAuth2

Problem Statement

Modern web applications and APIs require stateless, scalable, and secure authentication mechanisms to verify client identities across microservices, mobile apps, and Single Page Applications (SPAs). Traditional session-based authentication requires server-side state (such as Redis or database session tables), creating scaling bottlenecks and cross-domain policy complexities.

Without a standardized OAuth2 implementation, systems are prone to insecure token generation, weak password hashing, and loose authorization checks. This script provides a production-ready template that implements OAuth2 with Password Flow and JSON Web Tokens (JWT). It solves the problem of secure user registration, password hashing using bcrypt, JWT token generation with expiration, and dependency-injected endpoint protection.

How It Works

The architecture follows the OAuth2 Password Flow specification using HTTP Bearer authentication. When a user logs in, they send credentials (username and password) to the /token endpoint using standard form data (application/x-www-form-urlencoded). The application verifies the password hash using Passlib's bcrypt context. If valid, the server generates a signed JWT containing claims such as subject (sub), issued time (iat), and expiration time (exp).

For protected routes, clients include the token in the Authorization: Bearer <token> HTTP header. FastAPI's dependency injection system intercepts requests using OAuth2PasswordBearer. It decodes the token with python-jose using HMAC-SHA256 and a server secret key, verifies its expiration and integrity, and yields the authenticated user object to the route handler.

Full Script

#!/usr/bin/env python3
"""
FastAPI JWT Authentication with OAuth2
Production-ready authentication API featuring user registration, bcrypt password
hashing, JWT token issuance, protected endpoints, and role-based access control.
"""

import os
import sys
from datetime import datetime, timedelta, timezone
from typing import Annotated, Optional

import uvicorn
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, EmailStr, Field

# ==============================================================================
# Configuration
# ==============================================================================

# In production, ALWAYS load these from secure environment variables or secret managers
SECRET_KEY: str = os.environ.get(
    "JWT_SECRET_KEY", 
    "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"  # Dev fallback key
)
ALGORITHM: str = os.environ.get("JWT_ALGORITHM", "HS256")
ACCESS_TOKEN_EXPIRE_MINUTES: int = int(os.environ.get("ACCESS_TOKEN_EXPIRE_MINUTES", "30"))

# Password hashing context setup using bcrypt
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2 scheme specifying the token endpoint URL for OpenAPI integration
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# ==============================================================================
# Pydantic Schemas
# ==============================================================================

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: Optional[str] = None

class UserBase(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    full_name: Optional[str] = None
    role: str = Field(default="user")

class UserCreate(UserBase):
    password: str = Field(..., min_length=8, max_length=100)

class User(UserBase):
    disabled: bool = False

    class Config:
        from_attributes = True

class UserInDB(User):
    hashed_password: str

# ==============================================================================
# In-Memory Database (Replace with SQLAlchemy / Motor in production)
# ==============================================================================

fake_users_db: dict[str, UserInDB] = {}

# Seed initial admin user for testing
initial_admin_hash = pwd_context.hash("AdminPass123!")
fake_users_db["admin"] = UserInDB(
    username="admin",
    email="admin@example.com",
    full_name="System Administrator",
    role="admin",
    disabled=False,
    hashed_password=initial_admin_hash,
)

# ==============================================================================
# Utility & Security Functions
# ==============================================================================

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

def get_password_hash(password: str) -> str:
    """Hash a password for storing."""
    return pwd_context.hash(password)

def get_user(db: dict, username: str) -> Optional[UserInDB]:
    """Retrieve user record from database by username."""
    return db.get(username)

def authenticate_user(db: dict, username: str, password: str) -> UserInDB | bool:
    """Authenticate a user by username and password."""
    user = get_user(db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    """Create a signed JWT access token with expiration claims."""
    to_encode = data.copy()
    now = datetime.now(timezone.utc)
    if expires_delta:
        expire = now + expires_delta
    else:
        expire = now + timedelta(minutes=15)
    
    to_encode.update({
        "exp": expire,
        "iat": now,
    })
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# ==============================================================================
# Dependency Injection Functions
# ==============================================================================

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> UserInDB:
    """Dependency to extract and validate user from JWT token header."""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception

    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(
    current_user: Annotated[UserInDB, Depends(get_current_user)]
) -> UserInDB:
    """Dependency to verify user account is active."""
    if current_user.disabled:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, 
            detail="Inactive user account"
        )
    return current_user

async def require_admin_role(
    current_user: Annotated[UserInDB, Depends(get_current_active_user)]
) -> UserInDB:
    """Dependency to enforce admin role permissions."""
    if current_user.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Operation not permitted: Admin access required"
        )
    return current_user

# ==============================================================================
# FastAPI Application & Routes
# ==============================================================================

app = FastAPI(
    title="OAuth2 JWT Authentication API",
    description="Production-ready FastAPI setup with JWT authentication and RBAC.",
    version="1.0.0",
)

@app.post("/register", response_model=User, status_code=status.HTTP_201_CREATED)
async def register_user(user_in: UserCreate):
    """Register a new user account."""
    if user_in.username in fake_users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already registered",
        )
    
    hashed_pwd = get_password_hash(user_in.password)
    user_db = UserInDB(
        **user_in.model_dump(exclude={"password"}),
        hashed_password=hashed_pwd,
        disabled=False,
    )
    fake_users_db[user_in.username] = user_db
    return user_db

@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
):
    """Authenticate user and return JWT access token."""
    user = authenticate_user(fake_users_db, 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_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username, "role": user.role},
        expires_delta=access_token_expires,
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me", response_model=User)
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_active_user)]
):
    """Fetch current authenticated user details."""
    return current_user

@app.get("/admin/dashboard")
async def read_admin_dashboard(
    admin_user: Annotated[User, Depends(require_admin_role)]
):
    """Protected endpoint accessible only by administrators."""
    return {
        "message": f"Welcome to the admin panel, {admin_user.full_name}!",
        "system_status": "All systems operational",
        "total_users": len(fake_users_db),
    }

@app.get("/health")
async def health_check():
    """Unprotected health check endpoint."""
    return {"status": "ok", "timestamp": datetime.now(timezone.utc).isoformat()}

# ==============================================================================
# Script Execution Entrypoint
# ==============================================================================

if __name__ == "__main__":
    host = os.environ.get("HOST", "127.0.0.1")
    port = int(os.environ.get("PORT", "8000"))
    
    print(f"Starting server on http://{host}:{port}")
    uvicorn.run("auth_api:app", host=host, port=port, reload=True)

Configuration

The application reads configuration settings from environment variables with safe default fallbacks for local development:

Generating a Secure Production Secret Key

Run this command in your terminal to generate a 256-bit secret key:

python3 -c "import secrets; print(secrets.token_hex(32))"

Usage Examples

1. Launch the Server

Save the script as auth_api.py and run it:

export JWT_SECRET_KEY="c8f1e3a9..."
python3 auth_api.py

2. Register a New User

curl -X POST "http://127.0.0.1:8000/register" \
  -H "Content-Type: application/json" \
  -d '{
    "username": "johndoe",
    "email": "john@example.com",
    "full_name": "John Doe",
    "password": "SuperSecretPassword123!",
    "role": "user"
  }'

Expected Output:

{
  "username": "johndoe",
  "email": "john@example.com",
  "full_name": "John Doe",
  "role": "user",
  "disabled": false
}

3. Obtain Access Token (Login)

OAuth2 standard uses form data for token requests (application/x-www-form-urlencoded):

curl -X POST "http://127.0.0.1:8000/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "username=johndoe&password=SuperSecretPassword123!"

Expected Output:

{
  "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "token_type": "bearer"
}

4. Access Protected Route (/users/me)

Set your token in a variable and pass it in the Authorization header:

TOKEN="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

curl -X GET "http://127.0.0.1:8000/users/me" \
  -H "Authorization: Bearer $TOKEN"

Expected Output:

{
  "username": "johndoe",
  "email": "john@example.com",
  "full_name": "John Doe",
  "role": "user",
  "disabled": false
}

5. Access Admin Endpoint with Seeded Credentials

# Get Admin Token
ADMIN_TOKEN=$(curl -s -X POST "http://127.0.0.1:8000/token" \
  -d "username=admin&password=AdminPass123!" | jq -r .access_token)

# Access Admin Endpoint
curl -X GET "http://127.0.0.1:8000/admin/dashboard" \
  -H "Authorization: Bearer $ADMIN_TOKEN"

Expected Output:

{
  "message": "Welcome to the admin panel, System Administrator!",
  "system_status": "All systems operational",
  "total_users": 2
}

Error Handling

The application handles common authentication failure modes by raising explicit HTTPException instances:

• HTTP 401 Unauthorized:
  • Triggered when invalid credentials are provided to /token.
  • Triggered when an HTTP request lacks an Authorization header, uses an invalid JWT signature, or presents an expired token.
  • Headers include WWW-Authenticate: Bearer per RFC 6750 standards.
• HTTP 400 Bad Request:
  • Triggered when trying to register a username that already exists.
  • Triggered when an authenticated user account has disabled: true.
• HTTP 403 Forbidden:
  • Triggered when a non-admin user attempts to access endpoints guarded by require_admin_role.
• HTTP 422 Unprocessable Entity:
  • Automatically raised by Pydantic when registration payloads fail email formatting, field length limits, or missing required fields.

Extending Ideas

1. Refresh Tokens: Implement a dual-token architecture using short-lived access tokens (15 mins) and long-lived refresh tokens (7 days) stored securely in HttpOnly cookies.
2. Database Integration: Replace the in-memory fake_users_db dictionary with an asynchronous ORM such as SQLAlchemy with asyncpg or Motor for MongoDB.
3. Public Key Cryptography (RS256): Upgrade signing from symmetric HS256 key to asymmetric RS256 (RSA private/public keypair). This enables backend microservices to verify tokens using only the public key.
4. OAuth2 Scopes: Enhance role handling by using OAuth2 scopes (security.OAuth2PasswordBearer(..., scopes=...)) for fine-grained permission management across API endpoints.