Side Channel Attack in Fastapi

FastAPI provides several native mechanisms to mitigate side channel vulnerabilities. For timing attacks, use constant-time comparison functions and uniform response patterns:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import hmac
import time
from typing import Optional

app = FastAPI()

class LoginRequest(BaseModel):
    username: str
    password: str

# Constant-time comparison function
def timing_safe_compare(val1: str, val2: str) -> bool:
    return hmac.compare_digest(val1, val2)

@app.post("/login")
def login(request: LoginRequest):
    start_time = time.time()
    
    # Uniform response time regardless of username existence
    valid_user = request.username == "valid_user"
    
    # Constant-time password comparison
    password_valid = timing_safe_compare(request.password, "correct_password")
    
    # Uniform error response
    if not (valid_user and password_valid):
        # Add delay to mask timing differences
        elapsed = time.time() - start_time
        if elapsed < 0.1:  # Minimum response time
            time.sleep(0.1 - elapsed)
        raise HTTPException(status_code=401, detail="Invalid credentials")
    
    return {"token": "valid_token"}

This implementation ensures uniform response times regardless of authentication success, preventing timing-based information leakage.

For database queries, use uniform query patterns and add artificial delays:

from fastapi import FastAPI
from databases import Database
import asyncio
import time

app = FastAPI()
DATABASE_URL = "sqlite:///./test.db"
db = Database(DATABASE_URL)

@app.on_event("startup")
async def startup():
    await db.connect()

@app.on_event("shutdown")
async def shutdown():
    await db.disconnect()

@app.get("/user/{user_id}")
async def read_user(user_id: int):
    start_time = time.time()
    
    # Always execute the same query pattern
    query = "SELECT * FROM users WHERE id = $1"
    user = await db.fetch_one(query, [user_id])
    
    # Add uniform delay to mask database timing differences
    elapsed = time.time() - start_time
    if elapsed < 0.2:  # Minimum response time for user endpoints
        await asyncio.sleep(0.2 - elapsed)
    
    if user:
        return {"user": user}
    else:
        return {"error": "User not found"}

This approach ensures that database query timing doesn't leak information about user existence.

For JWT token validation, use uniform error handling and constant-time operations:

from fastapi import FastAPI, HTTPException
from jose import jwt
import time
from typing import Optional

app = FastAPI()
SECRET_KEY = "supersecretkey"

def constant_time_jwt_decode(token: str, secret: str) -> Optional[dict]:
    try:
        return jwt.decode(token, secret, algorithms=["HS256"])
    except:
        # Simulate decoding time for invalid tokens
        time.sleep(0.05)
        return None

@app.get("/protected")
async def protected_endpoint(token: str = Depends(oauth2_scheme)):
    start_time = time.time()
    
    payload = constant_time_jwt_decode(token, SECRET_KEY)
    if not payload:
        # Uniform error response
        elapsed = time.time() - start_time
        if elapsed < 0.1:
            time.sleep(0.1 - elapsed)
        raise HTTPException(status_code=401, detail="Invalid token")
    
    if payload["exp"] < time.time():
        # Uniform error response for expired tokens
        elapsed = time.time() - start_time
        if elapsed < 0.1:
            time.sleep(0.1 - elapsed)
        raise HTTPException(status_code=401, detail="Invalid token")
    
    return {"data": "protected resource"}

This implementation ensures uniform response times for all authentication failure scenarios.