tools-only FastAPI Security Guide
Framework Detection
# Detection patterns
from fastapi import FastAPI
from fastapi import *
app = FastAPI()
Debug Mode and Documentation
Risk Level: MEDIUM
Detection Patterns:
FastAPI\s*\([^)]*docs_url
FastAPI\s*\([^)]*redoc_url
FastAPI\s*\([^)]*openapi_url
app\.debug\s*=\s*True
Vulnerable Code:
# Default docs exposed in production
app = FastAPI() # /docs, /redoc, /openapi.json exposed
# Debug mode
app = FastAPI(debug=True)
Impact:
- API documentation exposes all endpoints
- Attackers can discover all routes and parameters
- Debug mode exposes detailed error information
Remediation:
import os
# Disable docs in production
app = FastAPI(
docs_url="/docs" if os.environ.get("ENV") == "development" else None,
redoc_url="/redoc" if os.environ.get("ENV") == "development" else None,
openapi_url="/openapi.json" if os.environ.get("ENV") == "development" else None,
debug=os.environ.get("DEBUG", "false").lower() == "true"
)
# Or completely disable in production
if os.environ.get("ENV") == "production":
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
else:
app = FastAPI()
SQL Injection
Risk Level: CRITICAL
Detection Patterns:
\.execute\s*\(f['"']
\.execute\s*\([^)]*%
text\s*\(f['"']
Vulnerable Code:
from sqlalchemy import text
from fastapi import FastAPI
@app.get("/users/{user_id}")
async def get_user(user_id: str, db: Session = Depends(get_db)):
# String formatting in query
result = db.execute(f"SELECT * FROM users WHERE id = {user_id}")
return result.fetchone()
@app.get("/search")
async def search(q: str, db: Session = Depends(get_db)):
result = db.execute(text(f"SELECT * FROM users WHERE name LIKE '%{q}%'"))
return result.fetchall()
Remediation:
from sqlalchemy.orm import Session
from sqlalchemy import text
# Use ORM
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
return db.query(User).filter(User.id == user_id).first()
# Parameterized queries
@app.get("/search")
async def search(q: str, db: Session = Depends(get_db)):
result = db.execute(
text("SELECT * FROM users WHERE name LIKE :query"),
{"query": f"%{q}%"}
)
return result.fetchall()
Authentication Vulnerabilities
Risk Level: HIGH
Detection Patterns:
Depends\s*\(\s*\)
@app\.(get|post|put|delete|patch)\s*\([^)]*\)\s*\n\s*(async\s+)?def\s+\w+\s*\([^)]*\):(?![^:]*Depends)
OAuth2PasswordBearer
HTTPBasic
Vulnerable Code:
from fastapi import FastAPI
# No authentication
@app.get("/admin/users")
async def list_users():
return get_all_users()
# Broken JWT validation
@app.get("/protected")
async def protected(token: str):
payload = jwt.decode(token, options={"verify_signature": False}) # No verification!
return payload
# Hardcoded secret
SECRET_KEY = "my-secret-key"
# No token expiry check
def verify_token(token: str):
payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
return payload # No exp check!
Remediation:
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from datetime import datetime, timedelta
import os
SECRET_KEY = os.environ.get("SECRET_KEY")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def create_access_token(data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
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
return username
@app.get("/admin/users")
async def list_users(current_user: str = Depends(get_current_user)):
return get_all_users()
Input Validation
Risk Level: HIGH
Detection Patterns:
def\s+\w+\s*\([^)]*:\s*str[^)]*\)
def\s+\w+\s*\([^)]*:\s*Any[^)]*\)
Query\s*\([^)]*\)(?!.*regex)
Vulnerable Code:
from fastapi import FastAPI, Query
# No input validation
@app.get("/users/{user_id}")
async def get_user(user_id: str): # Should be int
return db.query(f"SELECT * FROM users WHERE id = {user_id}")
# Accepting any input
@app.post("/data")
async def create_data(data: dict): # No schema validation
save_to_db(data)
# Weak validation
@app.get("/search")
async def search(q: str = Query(..., min_length=1)): # No max, no pattern
pass
Remediation:
from fastapi import FastAPI, Query, Path, Body
from pydantic import BaseModel, Field, validator, EmailStr
from typing import Optional
import re
# Use Pydantic models for request bodies
class UserCreate(BaseModel):
username: str = Field(..., min_length=3, max_length=50, regex="^[a-zA-Z0-9_]+$")
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 uppercase letter')
if not re.search(r'[0-9]', v):
raise ValueError('Password must contain digit')
return v
@app.post("/users")
async def create_user(user: UserCreate):
pass
# Use type hints and constraints
@app.get("/users/{user_id}")
async def get_user(
user_id: int = Path(..., gt=0, description="User ID must be positive")
):
pass
@app.get("/search")
async def search(
q: str = Query(..., min_length=1, max_length=100, regex="^[a-zA-Z0-9 ]+$")
):
pass
CORS Misconfiguration
Risk Level: HIGH
Detection Patterns:
CORSMiddleware
allow_origins\s*=\s*\[\s*['"]\*['"]\s*\]
allow_credentials\s*=\s*True.*allow_origins\s*=\s*\[\s*['"]\*['"]\s*\]
Vulnerable Code:
from fastapi.middleware.cors import CORSMiddleware
# Allow all origins with credentials
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True, # Dangerous with allow_origins=["*"]
allow_methods=["*"],
allow_headers=["*"],
)
Impact:
- Any website can make authenticated requests
- Session/cookie theft
- CSRF-like attacks
Remediation:
from fastapi.middleware.cors import CORSMiddleware
# Specific origins
origins = [
"https://example.com",
"https://www.example.com",
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["Authorization", "Content-Type"],
)
# Or dynamic origin validation
from starlette.middleware.cors import CORSMiddleware as StarletteCORS
def is_allowed_origin(origin: str) -> bool:
allowed_patterns = [
r"https://.*\.example\.com$",
r"https://example\.com$",
]
return any(re.match(pattern, origin) for pattern in allowed_patterns)
Path Traversal
Risk Level: HIGH
Detection Patterns:
FileResponse\s*\(
StaticFiles\s*\(
Path\s*\([^)]*\)\s*[^,)]*\.\.
Vulnerable Code:
from fastapi import FastAPI
from fastapi.responses import FileResponse
@app.get("/files/{filename}")
async def get_file(filename: str):
return FileResponse(f"/uploads/{filename}") # Path traversal!
# User-controlled static path
from fastapi.staticfiles import StaticFiles
app.mount("/static", StaticFiles(directory=user_input), name="static")
Remediation:
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
from pathlib import Path
import os
UPLOAD_DIR = Path("/uploads").resolve()
@app.get("/files/{filename}")
async def get_file(filename: str):
# Resolve and validate path
file_path = (UPLOAD_DIR / filename).resolve()
# Ensure path is within upload directory
if not file_path.is_relative_to(UPLOAD_DIR):
raise HTTPException(status_code=400, detail="Invalid filename")
if not file_path.exists():
raise HTTPException(status_code=404, detail="File not found")
return FileResponse(file_path)
Server-Side Request Forgery (SSRF)
Risk Level: HIGH
Detection Patterns:
httpx\.(get|post|put|delete)\s*\([^)]*request
aiohttp\.ClientSession\s*\(\s*\)\.get\s*\([^)]*request
requests\.(get|post)\s*\([^)]*request
Vulnerable Code:
import httpx
from fastapi import FastAPI
@app.get("/fetch")
async def fetch_url(url: str):
async with httpx.AsyncClient() as client:
response = await client.get(url) # SSRF!
return response.text
@app.post("/webhook")
async def webhook(callback_url: str, data: dict):
async with httpx.AsyncClient() as client:
await client.post(callback_url, json=data) # SSRF!
Remediation:
import httpx
from fastapi import FastAPI, HTTPException
from urllib.parse import urlparse
import ipaddress
ALLOWED_HOSTS = ["api.example.com", "cdn.example.com"]
def is_safe_url(url: str) -> bool:
try:
parsed = urlparse(url)
# Only allow HTTPS
if parsed.scheme != "https":
return False
# Check against allowlist
if parsed.hostname not in ALLOWED_HOSTS:
return False
# Block internal IPs
try:
ip = ipaddress.ip_address(parsed.hostname)
if ip.is_private or ip.is_loopback or ip.is_reserved:
return False
except ValueError:
pass # It's a hostname, not IP
return True
except Exception:
return False
@app.get("/fetch")
async def fetch_url(url: str):
if not is_safe_url(url):
raise HTTPException(status_code=400, detail="URL not allowed")
async with httpx.AsyncClient() as client:
response = await client.get(url, follow_redirects=False)
return response.text
Dependency Injection Security
Risk Level: MEDIUM
Detection Patterns:
Depends\s*\([^)]*\)
get_db\s*\(
Vulnerable Code:
from fastapi import Depends
# Database session not properly closed
def get_db():
db = SessionLocal()
return db # Session never closed!
# No exception handling
async def get_current_user(token: str = Depends(oauth2_scheme)):
user = decode_token(token) # Might raise exception
return user
Remediation:
from fastapi import Depends, HTTPException
from contextlib import contextmanager
# Proper session management
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# With exception handling
async def get_current_user(token: str = Depends(oauth2_scheme)):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user = get_user(payload.get("sub"))
if user is None:
raise HTTPException(status_code=401, detail="User not found")
return user
except JWTError:
raise HTTPException(status_code=401, detail="Invalid token")
Response Model Security
Risk Level: HIGH
Detection Patterns:
def\s+\w+\s*\([^)]*\)\s*->.*dict
return\s+\w+\.__dict__
\.dict\s*\(\s*\)
Vulnerable Code:
from fastapi import FastAPI
class User:
def __init__(self):
self.id = 1
self.username = "admin"
self.password_hash = "..." # Sensitive!
self.api_key = "..." # Sensitive!
@app.get("/users/{user_id}")
async def get_user(user_id: int):
user = get_user_from_db(user_id)
return user.__dict__ # Exposes all fields!
@app.get("/me")
async def get_me(user: User = Depends(get_current_user)):
return user.dict() # Includes sensitive fields
Remediation:
from fastapi import FastAPI
from pydantic import BaseModel
# Define response model explicitly
class UserResponse(BaseModel):
id: int
username: str
email: str
class Config:
orm_mode = True
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
return get_user_from_db(user_id)
# Exclude sensitive fields
class UserOut(BaseModel):
id: int
username: str
class Config:
orm_mode = True
# Explicitly exclude fields
fields = {'password_hash': {'exclude': True}}
@app.get("/me", response_model=UserOut)
async def get_me(user: User = Depends(get_current_user)):
return user
Rate Limiting
Risk Level: MEDIUM
Missing Rate Limiting Detection: Check for authentication/sensitive endpoints without rate limiting.
Vulnerable Code:
# No rate limiting on login
@app.post("/login")
async def login(credentials: Credentials):
# Can be brute-forced indefinitely
pass
# No rate limiting on password reset
@app.post("/password-reset")
async def reset_password(email: str):
pass
Remediation:
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.post("/login")
@limiter.limit("5/minute")
async def login(request: Request, credentials: Credentials):
pass
@app.post("/password-reset")
@limiter.limit("3/hour")
async def reset_password(request: Request, email: str):
pass
Security Headers
Risk Level: MEDIUM
Remediation:
from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
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["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
response.headers["Content-Security-Policy"] = "default-src 'self'"
return response
app.add_middleware(SecurityHeadersMiddleware)
Secrets in Configuration
Risk Level: CRITICAL
Detection Patterns:
SECRET_KEY\s*=\s*['"'][^'"']+['"']
API_KEY\s*=\s*['"'][^'"']+['"']
DATABASE_URL\s*=\s*['"'][^'"']+['"']
Vulnerable Code:
SECRET_KEY = "my-secret-key"
DATABASE_URL = "postgresql://user:password@localhost/db"
Remediation:
from pydantic import BaseSettings
class Settings(BaseSettings):
secret_key: str
database_url: str
api_key: str
class Config:
env_file = ".env"
settings = Settings()
# Use settings throughout the app
SECRET_KEY = settings.secret_key