API Security Patterns
Authentication
# JWT verification — pin algorithm, check expiry
import jwt
from fastapi import HTTPException, Security
from fastapi.security import HTTPBearer
security = HTTPBearer()
def verify_token(token: str) -> dict:
try:
payload = jwt.decode(
token,
settings.JWT_PUBLIC_KEY,
algorithms=["RS256"], # Pin algorithm — never accept "none"
options={"require": ["exp", "iat", "sub"]},
audience="api.example.com",
)
return payload
except jwt.ExpiredSignatureError:
raise HTTPException(401, "Token expired")
except jwt.InvalidTokenError:
raise HTTPException(401, "Invalid token")
# API Key — hash stored, compared in constant time
import hashlib, hmac, secrets
def create_api_key() -> tuple[str, str]:
raw = secrets.token_urlsafe(32)
hashed = hashlib.sha256(raw.encode()).hexdigest()
return raw, hashed # return raw once, store hashed
def verify_api_key(provided: str, stored_hash: str) -> bool:
provided_hash = hashlib.sha256(provided.encode()).hexdigest()
return hmac.compare_digest(provided_hash, stored_hash)
Authorization — BOLA Prevention
# Broken Object Level Authorization — always check ownership
from fastapi import Depends
async def get_invoice(
invoice_id: int,
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
invoice = await db.get(Invoice, invoice_id)
if not invoice:
raise HTTPException(404)
# CRITICAL: verify ownership — never skip this
if invoice.owner_id != current_user.id:
raise HTTPException(403, "Access denied") # Not 404 — don't leak existence
return invoice
# For admin endpoints — verify role server-side (not just in UI)
def require_role(role: str):
def dependency(user: User = Depends(get_current_user)):
if role not in user.roles:
raise HTTPException(403, f"Role {role} required")
return user
return dependency
Rate Limiting
# FastAPI with slowapi
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
@app.post("/api/login")
@limiter.limit("5/minute") # 5 attempts per minute per IP
async def login(request: Request, credentials: LoginRequest):
...
@app.post("/api/send-otp")
@limiter.limit("3/hour") # Prevent OTP flooding
async def send_otp(request: Request):
...
# Nginx rate limiting
limit_req_zone $binary_remote_addr zone=api:10m rate=100r/m;
limit_req_zone $binary_remote_addr zone=auth:10m rate=5r/m;
location /api/ {
limit_req zone=api burst=20 nodelay;
limit_req_status 429;
}
location /api/auth/ {
limit_req zone=auth burst=3 nodelay;
limit_req_status 429;
}
Input Validation
from pydantic import BaseModel, validator, constr
import re
class CreateUserRequest(BaseModel):
username: constr(min_length=3, max_length=30, pattern=r'^[a-zA-Z0-9_]+$')
email: str
age: int
@validator('email')
def validate_email(cls, v):
if not re.match(r'^[^@]+@[^@]+\.[^@]+$', v):
raise ValueError('Invalid email format')
return v.lower().strip()
@validator('age')
def validate_age(cls, v):
if not 0 < v < 150:
raise ValueError('Invalid age')
return v
GraphQL Security
# Depth limiting — prevent deeply nested queries
from graphql import build_schema
from graphql_depth_limit import depth_limit_validator
schema = build_schema(type_defs)
# Reject queries deeper than 5 levels
validation_rules = [depth_limit_validator(max_depth=5)]
# Query complexity limiting
from graphql_query_complexity import QualityLimitRule
class ComplexityRule(QualityLimitRule):
def __init__(self, context):
super().__init__(context, max_complexity=100)
# Disable introspection in production
GRAPHQL_INTROSPECTION = settings.DEBUG # False in prod
# Field-level authorization
@strawberry.type
class User:
name: str
@strawberry.field
def email(self, info: Info) -> str:
if info.context.user.id != self.id and not info.context.user.is_admin:
raise PermissionError("Cannot view other users' email")
return self._email
Security Headers
# FastAPI middleware
from fastapi.middleware.cors import CORSMiddleware
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["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
response.headers["Content-Security-Policy"] = "default-src 'self'"
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
response.headers.pop("Server", None) # Don't expose server info
return response
app.add_middleware(SecurityHeadersMiddleware)
app.add_middleware(
CORSMiddleware,
allow_origins=["https://app.example.com"], # Explicit, never "*"
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["Authorization", "Content-Type"],
)
API Security Checklist
Authentication: JWT pins alg, API keys hashed, OAuth PKCE for SPAs
Authorization: BOLA checked server-side on every resource access
Rate Limiting: login ≤5/min, OTP ≤3/hr, general API ≤100/min
Input: schema validation on all inputs, content-type enforced
Output: no stack traces, no internal paths, minimal error details
Transport: HTTPS only, HSTS header, TLS 1.2+ minimum
Headers: security headers set, Server header removed
GraphQL: depth limit, complexity limit, introspection off in prod
Logging: log auth failures, 4xx/5xx, but NOT request bodies with PII
Mattakushi432