FastAPI Security Checklist

Security should never be an afterthought when developing web applications. FastAPI provides several built-in features to help secure your APIs, but it's essential to understand and implement them correctly. This guide will walk you through the key security considerations for your FastAPI applications.

Introduction

When developing APIs with FastAPI, implementing proper security measures protects your application from common vulnerabilities and ensures that your users' data remains confidential and intact. This checklist covers essential security aspects from authentication and authorization to data validation and dependency management.

1. Authentication & Authorization

JWT Authentication

JSON Web Tokens (JWT) provide a compact and secure way to represent claims between parties.

python

from datetime import datetime, timedelta
from typing import Optional

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

# Secret key for JWT signing (store in environment variables in production)
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

# OAuth2 setup
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

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

class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None

# Verify password function
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

# Create access token function
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
        
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# Get current user function
async def get_current_user(token: str = Depends(oauth2_scheme)):
    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
    except JWTError:
        raise credentials_exception
        
    # In a real app, fetch the user from the database here
    user = get_user_from_database(username)
    if user is None:
        raise credentials_exception
    return user

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    # Authenticate user (check against database)
    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"},
        )
    
    # Create and return access token
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, 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: User = Depends(get_current_user)):
    return current_user

OAuth2 with Third-Party Providers

For more complex authentication needs, integrate with OAuth2 providers like Google, GitHub, or Auth0:

python

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2AuthorizationCodeBearer
from starlette.config import Config
from starlette.requests import Request
from authlib.integrations.starlette_client import OAuth

# Load configuration
config = Config(".env")
oauth = OAuth(config)

# Configure OAuth provider (e.g., GitHub)
oauth.register(
    name="github",
    client_id=config("GITHUB_CLIENT_ID"),
    client_secret=config("GITHUB_CLIENT_SECRET"),
    authorize_url="https://github.com/login/oauth/authorize",
    access_token_url="https://github.com/login/oauth/access_token",
    api_base_url="https://api.github.com/",
)

app = FastAPI()

@app.get("/login/github")
async def login_github(request: Request):
    redirect_uri = request.url_for("auth_github")
    return await oauth.github.authorize_redirect(request, redirect_uri)

@app.get("/auth/github")
async def auth_github(request: Request):
    token = await oauth.github.authorize_access_token(request)
    user = await oauth.github.get("user", token=token)
    # Process user data and create session/JWT
    return {"user": user.json()}

2. HTTPS and TLS Configuration

Always use HTTPS in production to encrypt data in transit. FastAPI can be deployed with HTTPS through various methods:

Using Uvicorn with HTTPS

python

# Generate self-signed certificates for development:
# openssl req -x509 -newkey rsa:4096 -nodes -out cert.pem -keyout key.pem -days 365

# Run with HTTPS
import uvicorn

if __name__ == "__main__":
    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8443,
        ssl_keyfile="./key.pem",
        ssl_certfile="./cert.pem"
    )

Behind a Reverse Proxy (Recommended for Production)

In production, it's usually better to handle HTTPS at the reverse proxy level (Nginx, Apache, etc.):

# Example Nginx configuration
server {
    listen 443 ssl;
    server_name example.com;
    
    ssl_certificate /etc/letsencrypt/live/example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/example.com/privkey.pem;
    
    location / {
        proxy_pass http://localhost:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

3. CORS (Cross-Origin Resource Sharing)

Configure CORS to control which domains can interact with your API:

python

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# Configure CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://trusted-frontend.com"],  # List of allowed origins
    allow_credentials=True,  # Allow cookies
    allow_methods=["*"],  # Allow all methods
    allow_headers=["*"],  # Allow all headers
)

For more restricted CORS settings in production:

python

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://www.yourfrontend.com", "https://app.yourfrontend.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
    max_age=3600,  # Cache preflight requests for 1 hour
)

4. Input Validation and Sanitization

FastAPI uses Pydantic for automatic request validation, but consider these additional practices:

Advanced Validation with Pydantic

python

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, EmailStr, validator
from typing import Optional
import re

app = FastAPI()

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    password: str = Field(..., min_length=8)
    
    @validator('password')
    def password_strength(cls, v):
        if not re.search(r'[A-Z]', v):
            raise ValueError('Password must contain at least one uppercase letter')
        if not re.search(r'[a-z]', v):
            raise ValueError('Password must contain at least one lowercase letter')
        if not re.search(r'[0-9]', v):
            raise ValueError('Password must contain at least one digit')
        if not re.search(r'[^a-zA-Z0-9]', v):
            raise ValueError('Password must contain at least one special character')
        return v
    
    @validator('username')
    def username_alphanumeric(cls, v):
        if not v.isalnum():
            raise ValueError('Username must be alphanumeric')
        return v

@app.post("/users/")
async def create_user(user: UserCreate):
    # Process validated user data
    return {"username": user.username, "email": user.email}

5. Rate Limiting

Implement rate limiting to prevent abuse and DoS attacks:

python

import time
from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.security import APIKeyHeader
from typing import Dict, List, Tuple

app = FastAPI()

# Simple in-memory rate limiter
class RateLimiter:
    def __init__(self, requests_limit: int, time_window: int):
        self.requests_limit = requests_limit  # Number of allowed requests
        self.time_window = time_window  # Time window in seconds
        self.clients: Dict[str, List[float]] = {}  # Store client request timestamps
    
    def is_allowed(self, client_id: str) -> bool:
        now = time.time()
        
        # Initialize client if not exists
        if client_id not in self.clients:
            self.clients[client_id] = []
        
        # Remove old timestamps
        self.clients[client_id] = [ts for ts in self.clients[client_id] 
                                  if ts > now - self.time_window]
        
        # Check if client exceeded limit
        if len(self.clients[client_id]) >= self.requests_limit:
            return False
        
        # Add current timestamp
        self.clients[client_id].append(now)
        return True

# Create a rate limiter instance (10 requests per minute)
limiter = RateLimiter(requests_limit=10, time_window=60)

# Dependency to check rate limit
async def check_rate_limit(request: Request):
    client_id = request.client.host  # Use IP address as client ID
    if not limiter.is_allowed(client_id):
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded. Try again later."
        )

@app.get("/protected-endpoint/", dependencies=[Depends(check_rate_limit)])
async def protected_endpoint():
    return {"message": "This endpoint is rate limited"}

For production use, consider using Redis or another distributed cache for rate limiting across multiple instances.

6. Environment Variables and Secret Management

Never hardcode secrets in your code. Use environment variables instead:

python

import os
from fastapi import FastAPI
from dotenv import load_dotenv

# Load environment variables from .env file in development
load_dotenv()

# Get secrets from environment variables
DATABASE_URL = os.getenv("DATABASE_URL")
API_SECRET_KEY = os.getenv("API_SECRET_KEY")
JWT_SECRET = os.getenv("JWT_SECRET")

# Validate required secrets are present
if not JWT_SECRET:
    raise ValueError("JWT_SECRET environment variable must be set")

app = FastAPI()

7. SQL Injection Prevention

FastAPI with SQLAlchemy or other ORMs helps prevent SQL injection, but be careful with raw queries:

python

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

app = FastAPI()

# Database setup
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Dependency to get database session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/{username}")
def get_user(username: str, db: Session = Depends(get_db)):
    # GOOD: Parameterized query using ORM
    user = db.query(User).filter(User.username == username).first()
    
    # BAD: Do NOT do this - vulnerable to SQL injection
    # query = f"SELECT * FROM users WHERE username = '{username}'"
    # user = db.execute(text(query)).first()
    
    # If you need raw SQL, use parameterized queries
    # user = db.execute(text("SELECT * FROM users WHERE username = :username"),
    #                   {"username": username}).first()
    
    if not user:
        return {"error": "User not found"}
    return {"id": user.id, "username": user.username}

8. Content Security Policy

Implement Content Security Policy headers to prevent XSS and data injection attacks:

python

from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response

app = FastAPI()

# Add trusted host middleware
app.add_middleware(
    TrustedHostMiddleware, allowed_hosts=["api.example.com", "*.example.com"]
)

# Custom middleware for security headers
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        
        # Content Security Policy
        response.headers["Content-Security-Policy"] = "default-src 'self'; img-src 'self' data:; font-src 'self'; style-src 'self' 'unsafe-inline'"
        
        # Other security headers
        response.headers["X-XSS-Protection"] = "1; mode=block"
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        
        return response

app.add_middleware(SecurityHeadersMiddleware)

9. Dependency Security

Regularly audit and update your dependencies to patch security vulnerabilities:

bash

# Install safety to check for vulnerabilities in dependencies
pip install safety

# Run a security check
safety check

# Generate requirements with pinned versions
pip freeze > requirements.txt

Consider using Dependabot or similar tools to automatically check for vulnerability updates in your repository.

10. Logging and Monitoring

Implement proper logging to detect and respond to security events:

python

import logging
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
import time
import uuid

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("app.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger("fastapi")

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    request_id = str(uuid.uuid4())
    
    # Add request ID to request state
    request.state.request_id = request_id
    
    # Log request details
    logger.info(f"Request {request_id} started: {request.method} {request.url.path}")
    
    start_time = time.time()
    
    try:
        response = await call_next(request)
        
        # Log successful response
        process_time = time.time() - start_time
        logger.info(f"Request {request_id} completed: {response.status_code} in {process_time:.3f}s")
        
        # Add request ID to response headers
        response.headers["X-Request-ID"] = request_id
        return response
        
    except Exception as e:
        # Log exceptions
        process_time = time.time() - start_time
        logger.error(f"Request {request_id} failed: {str(e)} in {process_time:.3f}s")
        
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content={"detail": "Internal server error"}
        )

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    # Log exceptions from request handlers
    request_id = getattr(request.state, "request_id", "unknown")
    logger.error(f"Request {request_id} error: {str(exc)}")
    
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"detail": "Internal server error"}
    )

@app.get("/secure-endpoint")
async def secure_endpoint():
    logger.info("Secure endpoint accessed")
    return {"message": "You accessed the secure endpoint"}

Summary

This security checklist covers essential aspects of securing your FastAPI applications:

1. Authentication & Authorization: Implement JWT, OAuth2, or other auth systems
2. HTTPS and TLS: Always use HTTPS in production
3. CORS: Configure proper Cross-Origin Resource Sharing
4. Input Validation: Use Pydantic's validation capabilities
5. Rate Limiting: Protect against DoS attacks
6. Environment Variables: Secure secret management
7. SQL Injection Prevention: Use ORMs and parameterized queries
8. Content Security Policy: Add security headers
9. Dependency Security: Regularly audit dependencies
10. Logging and Monitoring: Track and alert on security events

Security is an ongoing process, not a one-time implementation. Regularly review your application's security posture and keep up with evolving best practices.

Additional Resources

• FastAPI Security Documentation
• OWASP API Security Top 10
• JWT.io: Learn about JSON Web Tokens
• Python Security Best Practices

Exercises

1. Implement JWT authentication for a simple FastAPI application with user registration and login endpoints.
2. Set up proper CORS configuration for a FastAPI application that needs to work with a frontend on a different domain.
3. Create a custom rate limiter middleware that uses Redis to track request counts across multiple API instances.
4. Implement a comprehensive logging system that captures security events and alerts administrators of suspicious activity.
5. Perform a security audit on one of your existing FastAPI applications using the checklist from this guide.

By implementing these security measures, you'll significantly reduce the risk of security incidents in your FastAPI applications and protect both your system and your users.