Configure FastAPI rate limiting and security middleware with Redis and authentication

Update system packages and install Redis

Start by updating your system and installing Redis server which will store rate limiting data and session information.

sudo apt update && sudo apt upgrade -y
sudo apt install -y redis-server python3-pip python3-venv

sudo dnf update -y
sudo dnf install -y redis python3-pip python3-venv

Configure Redis for production use

Secure Redis by requiring authentication and binding to localhost only for security.

# Bind to localhost only
bind 127.0.0.1

# Set a strong password
requirepass your_strong_redis_password_here

# Enable persistence
save 900 1
save 300 10
save 60 10000

# Set max memory policy
maxmemory 256mb
maxmemory-policy allkeys-lru

# Security settings
protected-mode yes
port 6379

Start and enable Redis service

Enable Redis to start automatically on boot and verify it's running correctly.

sudo systemctl enable --now redis
sudo systemctl status redis
redis-cli ping

Create Python virtual environment and install dependencies

Set up an isolated Python environment for your FastAPI application with all required security and rate limiting packages.

mkdir -p /opt/fastapi-secure
cd /opt/fastapi-secure
python3 -m venv venv
source venv/bin/activate
pip install fastapi[all] redis slowapi python-jose[cryptography] passlib[bcrypt] python-multipart

Create rate limiting middleware

Implement Redis-backed rate limiting using SlowAPI to prevent abuse and control request frequency per client.

import redis
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from fastapi import Request, HTTPException
import os

# Redis connection with authentication
redis_client = redis.Redis(
    host='127.0.0.1',
    port=6379,
    password='your_strong_redis_password_here',
    decode_responses=True,
    socket_connect_timeout=5,
    socket_timeout=5,
    retry_on_timeout=True
)

# Custom key function that combines IP and user ID if authenticated
def get_rate_limit_key(request: Request):
    # Get client IP
    forwarded = request.headers.get('X-Forwarded-For')
    if forwarded:
        client_ip = forwarded.split(',')[0].strip()
    else:
        client_ip = get_remote_address(request)
    
    # If user is authenticated, include user ID in key
    user_id = getattr(request.state, 'user_id', None)
    if user_id:
        return f"rate_limit:{client_ip}:{user_id}"
    return f"rate_limit:{client_ip}"

# Initialize limiter with Redis backend
limiter = Limiter(
    key_func=get_rate_limit_key,
    storage_uri="redis://127.0.0.1:6379",
    storage_options={"password": "your_strong_redis_password_here"}
)

# Custom rate limit exceeded handler
def custom_rate_limit_handler(request: Request, exc: RateLimitExceeded):
    response = JSONResponse(
        status_code=429,
        content={
            "error": "Rate limit exceeded",
            "message": f"Rate limit exceeded: {exc.detail}",
            "retry_after": exc.retry_after
        }
    )
    response.headers["Retry-After"] = str(exc.retry_after)
    return response

Create JWT authentication system

Implement secure JWT token handling with proper password hashing and token validation.

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import HTTPException, status, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import redis
import json

# JWT Configuration
SECRET_KEY = "your-secret-key-change-this-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 7

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

# Security scheme
security = HTTPBearer()

# Redis client for token blacklisting
redis_client = redis.Redis(
    host='127.0.0.1',
    port=6379,
    password='your_strong_redis_password_here',
    decode_responses=True
)

class AuthManager:
    @staticmethod
    def verify_password(plain_password: str, hashed_password: str) -> bool:
        return pwd_context.verify(plain_password, hashed_password)
    
    @staticmethod
    def get_password_hash(password: str) -> str:
        return pwd_context.hash(password)
    
    @staticmethod
    def create_access_token(data: dict, expires_delta: timedelta = None):
        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
    
    @staticmethod
    def create_refresh_token(data: dict):
        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
    
    @staticmethod
    def verify_token(token: str, token_type: str = "access"):
        try:
            # Check if token is blacklisted
            if redis_client.get(f"blacklist:{token}"):
                raise HTTPException(
                    status_code=status.HTTP_401_UNAUTHORIZED,
                    detail="Token has been revoked"
                )
            
            payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
            if payload.get("type") != token_type:
                raise HTTPException(
                    status_code=status.HTTP_401_UNAUTHORIZED,
                    detail="Invalid token type"
                )
            
            username: str = payload.get("sub")
            user_id: str = payload.get("user_id")
            
            if username is None or user_id is None:
                raise HTTPException(
                    status_code=status.HTTP_401_UNAUTHORIZED,
                    detail="Could not validate credentials"
                )
            
            return {"username": username, "user_id": user_id}
        
        except JWTError:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Could not validate credentials"
            )
    
    @staticmethod
    def blacklist_token(token: str, expires_in: int = None):
        """Add token to Redis blacklist"""
        if expires_in is None:
            expires_in = ACCESS_TOKEN_EXPIRE_MINUTES * 60
        redis_client.setex(f"blacklist:{token}", expires_in, "true")

# Dependency to get current user
async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    return AuthManager.verify_token(token)

# Dependency for optional authentication
async def get_current_user_optional(credentials: HTTPAuthorizationCredentials = Depends(security)):
    if not credentials:
        return None
    try:
        token = credentials.credentials
        return AuthManager.verify_token(token)
    except HTTPException:
        return None

Create security middleware

Implement comprehensive security headers and CORS configuration to protect against common web vulnerabilities.

from fastapi import Request, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
import time
import uuid

class SecurityHeadersMiddleware:
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            request = Request(scope, receive)
            
            # Add request ID for tracking
            request.state.request_id = str(uuid.uuid4())
            
            # Process request
            response = await self.app(scope, receive, send)
            
            # Add security headers
            if hasattr(response, 'headers'):
                self.add_security_headers(response)
            
            return response
        
        return await self.app(scope, receive, send)
    
    def add_security_headers(self, response: Response):
        # Prevent XSS attacks
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"
        
        # HTTPS enforcement
        response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
        
        # Content Security Policy
        response.headers["Content-Security-Policy"] = (
            "default-src 'self'; "
            "script-src 'self'; "
            "style-src 'self' 'unsafe-inline'; "
            "img-src 'self' data: https:; "
            "font-src 'self'; "
            "connect-src 'self'; "
            "frame-ancestors 'none';"
        )
        
        # Referrer Policy
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        
        # Feature Policy
        response.headers["Permissions-Policy"] = (
            "geolocation=(), "
            "microphone=(), "
            "camera=()"
        )
        
        # Remove server information
        response.headers["Server"] = "FastAPI"

def configure_cors(app):
    """Configure CORS with secure defaults"""
    app.add_middleware(
        CORSMiddleware,
        allow_origins=[
            "https://example.com",  # Replace with your domain
            "https://app.example.com",
        ],
        allow_credentials=True,
        allow_methods=["GET", "POST", "PUT", "DELETE"],
        allow_headers=[
            "Accept",
            "Accept-Language",
            "Content-Language",
            "Content-Type",
            "Authorization"
        ],
        expose_headers=["X-Request-ID"]
    )

def configure_trusted_hosts(app):
    """Configure trusted hosts middleware"""
    app.add_middleware(
        TrustedHostMiddleware,
        allowed_hosts=[
            "example.com",  # Replace with your domain
            "*.example.com",
            "127.0.0.1",
            "localhost"
        ]
    )

Create the main FastAPI application

Integrate all security components into a complete FastAPI application with protected and public endpoints.

from fastapi import FastAPI, HTTPException, Depends, Request, status
from fastapi.responses import JSONResponse
from fastapi.security import HTTPAuthorizationCredentials
from pydantic import BaseModel, EmailStr
from slowapi.errors import RateLimitExceeded
import uvicorn
import logging
from datetime import timedelta

# Import our modules
from rate_limiter import limiter, custom_rate_limit_handler
from auth import AuthManager, get_current_user, get_current_user_optional
from security_middleware import SecurityHeadersMiddleware, configure_cors, configure_trusted_hosts

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Create FastAPI app
app = FastAPI(
    title="Secure FastAPI Application",
    description="FastAPI with rate limiting, security middleware, and authentication",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

# Add rate limiter
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, custom_rate_limit_handler)

# Add security middleware
app.add_middleware(SecurityHeadersMiddleware)
configure_cors(app)
configure_trusted_hosts(app)

# Request/Response models
class LoginRequest(BaseModel):
    username: str
    password: str

class RegisterRequest(BaseModel):
    username: str
    email: EmailStr
    password: str

class TokenResponse(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str

class RefreshRequest(BaseModel):
    refresh_token: str

# Mock user database (replace with real database)
fake_users_db = {
    "testuser": {
        "user_id": "1",
        "username": "testuser",
        "email": "test@example.com",
        "hashed_password": AuthManager.get_password_hash("testpassword")
    }
}

# Public endpoints with rate limiting
@app.get("/")
@limiter.limit("100/minute")
async def root(request: Request):
    return {"message": "Welcome to Secure FastAPI", "status": "healthy"}

@app.get("/health")
@limiter.limit("200/minute")
async def health_check(request: Request):
    return {"status": "healthy", "timestamp": "2024-01-01T00:00:00Z"}

# Authentication endpoints
@app.post("/auth/register", response_model=dict)
@limiter.limit("5/minute")  # Strict limit for registration
async def register(request: Request, user_data: RegisterRequest):
    # Check if user exists
    if user_data.username in fake_users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already registered"
        )
    
    # Create new user
    user_id = str(len(fake_users_db) + 1)
    fake_users_db[user_data.username] = {
        "user_id": user_id,
        "username": user_data.username,
        "email": user_data.email,
        "hashed_password": AuthManager.get_password_hash(user_data.password)
    }
    
    logger.info(f"New user registered: {user_data.username}")
    return {"message": "User registered successfully", "username": user_data.username}

@app.post("/auth/login", response_model=TokenResponse)
@limiter.limit("10/minute")  # Strict limit for login attempts
async def login(request: Request, credentials: LoginRequest):
    # Verify user
    user = fake_users_db.get(credentials.username)
    if not user or not AuthManager.verify_password(credentials.password, user["hashed_password"]):
        logger.warning(f"Failed login attempt for user: {credentials.username}")
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password"
        )
    
    # Create tokens
    access_token = AuthManager.create_access_token(
        data={"sub": user["username"], "user_id": user["user_id"]}
    )
    refresh_token = AuthManager.create_refresh_token(
        data={"sub": user["username"], "user_id": user["user_id"]}
    )
    
    logger.info(f"User logged in: {credentials.username}")
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer"
    }

@app.post("/auth/refresh", response_model=dict)
@limiter.limit("20/minute")
async def refresh_token(request: Request, refresh_data: RefreshRequest):
    # Verify refresh token
    user_data = AuthManager.verify_token(refresh_data.refresh_token, "refresh")
    
    # Create new access token
    new_access_token = AuthManager.create_access_token(
        data={"sub": user_data["username"], "user_id": user_data["user_id"]}
    )
    
    return {
        "access_token": new_access_token,
        "token_type": "bearer"
    }

@app.post("/auth/logout")
@limiter.limit("30/minute")
async def logout(request: Request, current_user: dict = Depends(get_current_user)):
    # Get token from request
    auth_header = request.headers.get("authorization")
    if auth_header and auth_header.startswith("Bearer "):
        token = auth_header.split(" ")[1]
        AuthManager.blacklist_token(token)
    
    logger.info(f"User logged out: {current_user['username']}")
    return {"message": "Successfully logged out"}

# Protected endpoints
@app.get("/protected")
@limiter.limit("50/minute")
async def protected_route(request: Request, current_user: dict = Depends(get_current_user)):
    # Store user_id in request state for rate limiting
    request.state.user_id = current_user["user_id"]
    
    return {
        "message": "This is a protected route",
        "user": current_user["username"],
        "user_id": current_user["user_id"]
    }

@app.get("/user/profile")
@limiter.limit("30/minute")
async def get_profile(request: Request, current_user: dict = Depends(get_current_user)):
    request.state.user_id = current_user["user_id"]
    
    user = fake_users_db.get(current_user["username"])
    return {
        "username": user["username"],
        "email": user["email"],
        "user_id": user["user_id"]
    }

# Public API with optional authentication
@app.get("/api/data")
@limiter.limit("100/minute")
async def get_data(request: Request, current_user: dict = Depends(get_current_user_optional)):
    if