Api security

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