Fastapi

FastAPI Development Skill

File Organization

• SKILL.md: Core principles, patterns, essential security (this file)
• references/security-examples.md: CVE details and OWASP implementations
• references/advanced-patterns.md: Advanced FastAPI patterns
• references/threat-model.md: Attack scenarios and STRIDE analysis

Validation Gates

Gate 0.2: Vulnerability Research (BLOCKING for HIGH-RISK)

• Status: PASSED (5+ CVEs documented)
• Research Date: 2025-11-20
• CVEs: CVE-2024-47874, CVE-2024-12868, CVE-2023-30798, Starlette DoS variants

---

1. Overview

Risk Level: HIGH

Justification: FastAPI applications handle authentication, database access, file uploads, and external API communication. DoS vulnerabilities in Starlette, injection risks, and improper validation can compromise availability and security.

You are an expert FastAPI developer creating secure, performant REST APIs and WebSocket services. You configure proper validation, authentication, and security headers.

Core Expertise Areas

• Pydantic validation and dependency injection
• Authentication: OAuth2, JWT, API keys
• Security headers and CORS configuration
• Rate limiting and DoS protection
• Database integration with async ORMs
• WebSocket security

---

2. Core Responsibilities

Fundamental Principles

1. TDD First: Write tests before implementation code 2. Performance Aware: Connection pooling, caching, async patterns 3. Validate Everything: Use Pydantic models for all inputs 4. Secure by Default: HTTPS, security headers, strict CORS 5. Rate Limit: Protect all endpoints from abuse 6. Authenticate & Authorize: Verify identity and permissions 7. Handle Errors Safely: Never leak internal details

---

3. Technical Foundation

Version Recommendations

Security Dependencies

[project]
dependencies = [
    "fastapi>=0.115.3",
    "starlette>=0.40.0",
    "pydantic>=2.5",
    "python-jose[cryptography]>=3.3",
    "passlib[argon2]>=1.7",
    "python-multipart>=0.0.6",
    "slowapi>=0.1.9",
    "secure>=0.3",
]

---

4. Implementation Patterns

Pattern 1: Secure Application Setup

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

app = FastAPI(
    title="Secure API",
    docs_url=None if PRODUCTION else "/docs",  # Disable in prod
    redoc_url=None,
)

# Security headers
secure_headers = SecureHeaders()

@app.middleware("http")
async def add_security_headers(request, call_next):
    response = await call_next(request)
    secure_headers.framework.fastapi(response)
    return response

# Restrictive CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],  # Never ["*"]
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

Pattern 2: Input Validation

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, field_validator, EmailStr

class UserCreate(BaseModel):
    username: str = Field(min_length=3, max_length=50, pattern=r'^[a-zA-Z0-9_-]+$')
    email: EmailStr
    password: str = Field(min_length=12)

    @field_validator('password')
    @classmethod
    def validate_password(cls, v):
        if not any(c.isupper() for c in v):
            raise ValueError('Must contain uppercase')
        if not any(c.isdigit() for c in v):
            raise ValueError('Must contain digit')
        return v

@app.post("/users")
async def create_user(user: UserCreate):
    # Input already validated by Pydantic
    return await user_service.create(user)

Pattern 3: JWT Authentication

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from datetime import datetime, timedelta

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

SECRET_KEY = os.environ["JWT_SECRET"]
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict) -> str:
    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)) -> User:
    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])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = await user_service.get(user_id)
    if user is None:
        raise credentials_exception
    return user

Pattern 4: Rate Limiting

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")  # Strict for auth endpoints
async def login(request: Request, credentials: LoginRequest):
    return await auth_service.login(credentials)

@app.get("/data")
@limiter.limit("100/minute")
async def get_data(request: Request):
    return await data_service.get_all()

Pattern 5: Safe File Upload

from fastapi import UploadFile, File, HTTPException
import magic

ALLOWED_TYPES = {"image/jpeg", "image/png", "application/pdf"}
MAX_SIZE = 10 * 1024 * 1024  # 10MB

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    # Check size
    content = await file.read()
    if len(content) > MAX_SIZE:
        raise HTTPException(400, "File too large")

    # Check magic bytes, not just extension
    mime_type = magic.from_buffer(content, mime=True)
    if mime_type not in ALLOWED_TYPES:
        raise HTTPException(400, f"File type not allowed: {mime_type}")

    # Generate safe filename
    safe_name = f"{uuid4()}{Path(file.filename).suffix}"

    # Store outside webroot
    file_path = UPLOAD_DIR / safe_name
    file_path.write_bytes(content)

    return {"filename": safe_name}

---

5. Implementation Workflow (TDD)

Step 1: Write Failing Test First

Always start with tests that define expected behavior:

import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.mark.asyncio
async def test_create_item_success():
    """Test successful item creation with valid data."""
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.post(
            "/items",
            json={"name": "Test Item", "price": 29.99},
            headers={"Authorization": "Bearer valid_token"}
        )
        assert response.status_code == 201
        data = response.json()
        assert data["name"] == "Test Item"
        assert "id" in data

@pytest.mark.asyncio
async def test_create_item_validation_error():
    """Test validation rejects invalid price."""
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.post(
            "/items",
            json={"name": "Test", "price": -10},
            headers={"Authorization": "Bearer valid_token"}
        )
        assert response.status_code == 422

@pytest.mark.asyncio
async def test_create_item_unauthorized():
    """Test endpoint requires authentication."""
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.post("/items", json={"name": "Test", "price": 10})
        assert response.status_code == 401

Step 2: Implement Minimum to Pass

Write only the code needed to make tests pass:

@app.post("/items", status_code=201)
async def create_item(
    item: ItemCreate,
    user: User = Depends(get_current_user)
) -> ItemResponse:
    created = await item_service.create(item, user.id)
    return ItemResponse.from_orm(created)

Step 3: Refactor if Needed

Improve code quality while keeping tests green. Extract common patterns, improve naming, optimize queries.

Step 4: Run Full Verification

# Run all tests with coverage
pytest --cov=app --cov-report=term-missing

# Type checking
mypy app --strict

# Security scan
bandit -r app -ll

# All must pass before committing

---

6. Performance Patterns

Pattern 1: Connection Pooling for Database

# BAD - Creates new connection per request
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    conn = await asyncpg.connect(DATABASE_URL)
    try:
        return await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
    finally:
        await conn.close()

# GOOD - Uses connection pool
from contextlib import asynccontextmanager

pool: asyncpg.Pool = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global pool
    pool = await asyncpg.create_pool(
        DATABASE_URL,
        min_size=5,
        max_size=20,
        command_timeout=60
    )
    yield
    await pool.close()

app = FastAPI(lifespan=lifespan)

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    async with pool.acquire() as conn:
        return await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)

Pattern 2: Concurrent Requests with asyncio.gather

# BAD - Sequential external API calls
@app.get("/dashboard")
async def get_dashboard(user_id: int):
    profile = await fetch_profile(user_id)      # 100ms
    orders = await fetch_orders(user_id)        # 150ms
    notifications = await fetch_notifications(user_id)  # 80ms
    return {"profile": profile, "orders": orders, "notifications": notifications}
    # Total: ~330ms

# GOOD - Concurrent calls
@app.get("/dashboard")
async def get_dashboard(user_id: int):
    profile, orders, notifications = await asyncio.gather(
        fetch_profile(user_id),
        fetch_orders(user_id),
        fetch_notifications(user_id)
    )
    return {"profile": profile, "orders": orders, "notifications": notifications}
    # Total: ~150ms (slowest call)

Pattern 3: Response Caching

# BAD - Recomputes expensive data every request
@app.get("/stats")
async def get_stats():
    return await compute_expensive_stats()  # 500ms each time

# GOOD - Cache with Redis
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache

@asynccontextmanager
async def lifespan(app: FastAPI):
    redis = aioredis.from_url("redis://localhost")
    FastAPICache.init(RedisBackend(redis), prefix="api-cache")
    yield

@app.get("/stats")
@cache(expire=300)  # Cache for 5 minutes
async def get_stats():
    return await compute_expensive_stats()

# GOOD - In-memory cache for simpler cases
from functools import lru_cache
from datetime import datetime, timedelta

_cache = {}
_cache_time = {}

async def get_cached_config(key: str, ttl: int = 60):
    now = datetime.utcnow()
    if key in _cache and _cache_time[key] > now:
        return _cache[key]

    value = await fetch_config(key)
    _cache[key] = value
    _cache_time[key] = now + timedelta(seconds=ttl)
    return value

Pattern 4: Pagination for Large Datasets

# BAD - Returns all records
@app.get("/items")
async def list_items():
    return await db.fetch("SELECT * FROM items")  # Could be millions

# GOOD - Cursor-based pagination
from pydantic import BaseModel

class PaginatedResponse(BaseModel):
    items: list
    next_cursor: str | None
    has_more: bool

@app.get("/items")
async def list_items(
    cursor: str | None = None,
    limit: int = Query(default=20, le=100)
) -> PaginatedResponse:
    query = "SELECT * FROM items"
    params = []

    if cursor:
        query += " WHERE id > $1"
        params.append(decode_cursor(cursor))

    query += f" ORDER BY id LIMIT {limit + 1}"

    rows = await db.fetch(query, *params)
    has_more = len(rows) > limit
    items = rows[:limit]

    return PaginatedResponse(
        items=items,
        next_cursor=encode_cursor(items[-1]["id"]) if items else None,
        has_more=has_more
    )

Pattern 5: Background Tasks for Heavy Operations

# BAD - Blocks response for slow operations
@app.post("/reports")
async def create_report(request: ReportRequest):
    report = await generate_report(request)  # Takes 30 seconds
    await send_email(request.email, report)
    return {"status": "completed"}

# GOOD - Return immediately, process in background
from fastapi import BackgroundTasks

@app.post("/reports", status_code=202)
async def create_report(
    request: ReportRequest,
    background_tasks: BackgroundTasks
):
    report_id = str(uuid4())
    background_tasks.add_task(process_report, report_id, request)
    return {"report_id": report_id, "status": "processing"}

async def process_report(report_id: str, request: ReportRequest):
    report = await generate_report(request)
    await save_report(report_id, report)
    await send_email(request.email, report)

@app.get("/reports/{report_id}")
async def get_report_status(report_id: str):
    return await get_report(report_id)

---

7. Security Standards

7.1 Domain Vulnerability Landscape

7.2 OWASP Top 10 Mapping

7.3 Error Handling

from fastapi import HTTPException
from fastapi.responses import JSONResponse
import logging

logger = logging.getLogger(__name__)

@app.exception_handler(Exception)
async def global_exception_handler(request, exc):
    # Log full details
    logger.error(f"Unhandled error: {exc}", exc_info=True)

    # Return safe message
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"}
    )

@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail}
    )

---

6. Testing & Validation

Security Tests

import pytest
from fastapi.testclient import TestClient

def test_rate_limiting():
    client = TestClient(app)
    # Exceed rate limit
    for _ in range(10):
        response = client.post("/login", json={"username": "test", "password": "test"})
    assert response.status_code == 429

def test_invalid_jwt_rejected():
    client = TestClient(app)
    response = client.get(
        "/protected",
        headers={"Authorization": "Bearer invalid.token.here"}
    )
    assert response.status_code == 401

def test_sql_injection_prevented():
    client = TestClient(app)
    response = client.get("/users", params={"search": "'; DROP TABLE users; --"})
    assert response.status_code in [200, 400]
    # Should not cause 500 (SQL error)

def test_file_upload_type_validation():
    client = TestClient(app)
    # Try uploading executable disguised as image
    response = client.post(
        "/upload",
        files={"file": ("test.jpg", b"MZ\x90\x00", "image/jpeg")}  # EXE magic bytes
    )
    assert response.status_code == 400

---

8. Common Mistakes & Anti-Patterns

Anti-Pattern 1: Permissive CORS

# NEVER
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True)

# ALWAYS
app.add_middleware(CORSMiddleware, allow_origins=["https://app.example.com"])

Anti-Pattern 2: No Rate Limiting

# NEVER - allows brute force
@app.post("/login")
async def login(creds): ...

# ALWAYS
@app.post("/login")
@limiter.limit("5/minute")
async def login(request, creds): ...

Anti-Pattern 3: Exposing Docs in Production

# NEVER
app = FastAPI()

# ALWAYS
app = FastAPI(
    docs_url=None if os.environ.get("ENV") == "production" else "/docs",
    redoc_url=None
)

Anti-Pattern 4: Weak JWT Configuration

# NEVER
jwt.encode(data, "secret", algorithm="HS256")  # Hardcoded weak secret

# ALWAYS
jwt.encode(data, os.environ["JWT_SECRET"], algorithm="RS256")  # Env var, strong algo

Anti-Pattern 5: File Extension Validation Only

# NEVER
if file.filename.endswith('.jpg'): ...

# ALWAYS
mime = magic.from_buffer(content, mime=True)
if mime not in ALLOWED_TYPES: ...

---

13. Pre-Deployment Checklist

• [ ] FastAPI 0.115.3+ / Starlette 0.40.0+
• [ ] Security headers middleware configured
• [ ] CORS restrictive (no wildcard with credentials)
• [ ] Rate limiting on all endpoints
• [ ] Stricter limits on auth endpoints
• [ ] JWT with strong secret from environment
• [ ] Pydantic validation on all inputs
• [ ] File uploads check magic bytes
• [ ] Docs disabled in production
• [ ] Error handlers don't leak internals
• [ ] HTTPS enforced

---

14. Summary

Your goal is to create FastAPI applications that are:

• Secure: Validated inputs, rate limits, security headers
• Performant: Async operations, proper connection pooling
• Maintainable: Type-safe, well-structured, tested

Security Reminder: 1. Upgrade Starlette to 0.40.0+ (CVE-2024-47874) 2. Rate limit all endpoints, especially authentication 3. Validate file uploads by magic bytes, not extension 4. Never use wildcard CORS with credentials 5. Disable API docs in production

FastAPI Advanced Patterns Reference

Dependency Injection Patterns

Layered Dependencies

from fastapi import Depends

async def get_db() -> AsyncSession:
    async with async_session() as session:
        yield session

async def get_user_repo(db: AsyncSession = Depends(get_db)) -> UserRepository:
    return UserRepository(db)

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    repo: UserRepository = Depends(get_user_repo)
) -> User:
    user_id = decode_token(token)
    return await repo.get(user_id)

@app.get("/profile")
async def get_profile(user: User = Depends(get_current_user)):
    return user

Scoped Dependencies

from fastapi import Depends
from uuid import uuid4

def get_request_id():
    return str(uuid4())

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request.state.request_id = get_request_id()
    response = await call_next(request)
    response.headers["X-Request-ID"] = request.state.request_id
    return response

---

Background Tasks

Async Background Processing

from fastapi import BackgroundTasks

async def send_email(email: str, content: str):
    # Async email sending
    await email_service.send(email, content)

@app.post("/register")
async def register(user: UserCreate, background_tasks: BackgroundTasks):
    new_user = await user_service.create(user)

    # Queue email for later
    background_tasks.add_task(
        send_email,
        user.email,
        "Welcome to our service!"
    )

    return new_user

---

Response Optimization

Streaming Responses

from fastapi.responses import StreamingResponse

async def generate_large_file():
    for chunk in range(100):
        yield f"data chunk {chunk}\n".encode()
        await asyncio.sleep(0.1)

@app.get("/stream")
async def stream_data():
    return StreamingResponse(
        generate_large_file(),
        media_type="text/plain"
    )

Response Caching

from fastapi_cache import FastAPICache
from fastapi_cache.decorator import cache

@app.get("/data")
@cache(expire=300)  # 5 minutes
async def get_cached_data():
    return await expensive_operation()

---

WebSocket Integration

from fastapi import WebSocket, WebSocketDisconnect

class ConnectionManager:
    def __init__(self):
        self.connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Message: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)

---

Testing Patterns

Test Client Setup

import pytest
from httpx import AsyncClient
from fastapi.testclient import TestClient

@pytest.fixture
def client():
    return TestClient(app)

@pytest.fixture
async def async_client():
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

@pytest.fixture
async def authenticated_client(async_client):
    # Login and get token
    response = await async_client.post("/token", data={
        "username": "test",
        "password": "testpass"
    })
    token = response.json()["access_token"]

    async_client.headers["Authorization"] = f"Bearer {token}"
    yield async_client

Mocking Dependencies

from unittest.mock import AsyncMock

@pytest.fixture
def mock_user_service():
    mock = AsyncMock()
    mock.get.return_value = User(id=1, username="test")
    return mock

def test_with_mock(client, mock_user_service):
    app.dependency_overrides[get_user_service] = lambda: mock_user_service

    response = client.get("/users/1")
    assert response.status_code == 200

    # Cleanup
    app.dependency_overrides.clear()

---

OpenAPI Customization

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema

    openapi_schema = get_openapi(
        title="Secure API",
        version="1.0.0",
        routes=app.routes,
    )

    # Add security schemes
    openapi_schema["components"]["securitySchemes"] = {
        "bearerAuth": {
            "type": "http",
            "scheme": "bearer",
            "bearerFormat": "JWT"
        }
    }

    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

FastAPI Security Examples Reference

CVE Details and Mitigations

CVE-2024-47874: Starlette Multipart DoS

Severity: HIGH (CVSS 8.7) Affected: Starlette < 0.40.0 CWE: CWE-400 (Resource Exhaustion)

Description: Multipart form fields without filename are buffered entirely in memory without high-water mark limits, allowing memory exhaustion.

Mitigation:

pip install 'starlette>=0.40.0' 'fastapi>=0.115.3'

# Additional protection: request size limits
@app.middleware("http")
async def limit_body_size(request, call_next):
    MAX_BODY_SIZE = 10 * 1024 * 1024  # 10MB

    if request.headers.get("content-length"):
        if int(request.headers["content-length"]) > MAX_BODY_SIZE:
            return JSONResponse(413, {"detail": "Request too large"})

    return await call_next(request)

---

Complete OWASP Examples

A01: Broken Access Control

from fastapi import Depends, HTTPException

def require_role(role: str):
    async def checker(user: User = Depends(get_current_user)):
        if role not in user.roles:
            raise HTTPException(403, "Insufficient permissions")
        return user
    return checker

@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    current_user: User = Depends(require_role("admin"))
):
    # Only admins can delete
    await user_service.delete(user_id)

A02: Cryptographic Failures

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

A03: Injection

# SQL Injection Prevention
from sqlalchemy import select

async def get_user(db, username: str):
    stmt = select(User).where(User.username == username)
    return (await db.execute(stmt)).scalar_one_or_none()

# Command Injection Prevention
import subprocess

def safe_command(filename: str):
    if not filename.isalnum():
        raise ValueError("Invalid filename")
    subprocess.run(["cat", filename], check=True)

A05: Security Misconfiguration

from fastapi import FastAPI
from secure import SecureHeaders

secure_headers = SecureHeaders(
    hsts=True,
    xfo="DENY",
    xxp=True,
    content="nosniff",
    csp="default-src 'self'"
)

@app.middleware("http")
async def add_security_headers(request, call_next):
    response = await call_next(request)
    secure_headers.framework.fastapi(response)
    return response

A07: Authentication Failures

from slowapi import Limiter

limiter = Limiter(key_func=get_remote_address)

@app.post("/token")
@limiter.limit("5/minute")
async def login(request: Request, form: OAuth2PasswordRequestForm = Depends()):
    user = await authenticate(form.username, form.password)
    if not user:
        raise HTTPException(401, "Incorrect credentials")
    return {"access_token": create_token(user), "token_type": "bearer"}

---

Additional Security Patterns

API Key Authentication

from fastapi import Security
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

async def verify_api_key(api_key: str = Security(api_key_header)):
    if not api_key:
        raise HTTPException(401, "API key required")

    key = await get_api_key(api_key)
    if not key or not key.is_active:
        raise HTTPException(403, "Invalid API key")

    return key

@app.get("/data")
async def get_data(key: APIKey = Depends(verify_api_key)):
    return {"data": "secret"}

Request Validation Middleware

@app.middleware("http")
async def validate_requests(request: Request, call_next):
    # Check content type
    if request.method in ["POST", "PUT", "PATCH"]:
        content_type = request.headers.get("content-type", "")
        if not content_type.startswith(("application/json", "multipart/form-data")):
            return JSONResponse(415, {"detail": "Unsupported media type"})

    return await call_next(request)

Secure Database Operations

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

engine = create_async_engine(
    os.environ["DATABASE_URL"],
    pool_size=20,
    max_overflow=10,
    pool_timeout=30,
    pool_pre_ping=True  # Check connections
)

async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db():
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

FastAPI Threat Model

Threat Model Overview

Domain Risk Level: HIGH

Assets to Protect

1. API Endpoints - Business logic, data access - Sensitivity: HIGH 2. Authentication - User sessions, tokens - Sensitivity: CRITICAL 3. User Data - PII, credentials - Sensitivity: CRITICAL 4. System Resources - Memory, CPU - Sensitivity: MEDIUM (DoS risk)

Attack Surface

• Public API endpoints
• Authentication/token endpoints
• File upload handlers
• WebSocket connections
• Database queries

---

Attack Scenario 1: Multipart Form DoS (CVE-2024-47874)

Threat Category: OWASP A10:2025 / CWE-400

Threat Level: HIGH

Attack Flow:

1. Attacker sends multipart/form-data request
2. Form field has no filename (not file upload)
3. Starlette buffers entire field in memory
4. No high-water mark pause implemented
5. Attacker sends multiple concurrent large requests
6. Server memory exhausted, service crashes

Mitigation:

# Upgrade to Starlette 0.40.0+
# Add request size limits
from starlette.middleware import Middleware
from starlette.middleware.trustedhost import TrustedHostMiddleware

app = FastAPI(middleware=[
    Middleware(TrustedHostMiddleware, allowed_hosts=["example.com"])
])

# Limit request body size
@app.middleware("http")
async def limit_request_size(request, call_next):
    content_length = request.headers.get("content-length")
    if content_length and int(content_length) > 10 * 1024 * 1024:  # 10MB
        return JSONResponse(status_code=413, content={"detail": "Request too large"})
    return await call_next(request)

---

Attack Scenario 2: Authentication Brute Force

Threat Category: OWASP A07:2025 / CWE-307

Threat Level: HIGH

Attack Flow:

1. Attacker targets /login endpoint
2. Sends automated credential stuffing requests
3. Without rate limiting, thousands of attempts/second
4. Eventually guesses valid credentials

Mitigation:

from slowapi import Limiter

limiter = Limiter(key_func=get_remote_address)

@app.post("/login")
@limiter.limit("5/minute")
async def login(request: Request, creds: LoginRequest):
    user = await auth_service.authenticate(creds)
    if not user:
        # Log failed attempt
        logger.warning(f"Failed login: {creds.username} from {request.client.host}")
        # Generic error message
        raise HTTPException(401, "Invalid credentials")
    return {"token": create_token(user)}

---

Attack Scenario 3: JWT Token Manipulation

Threat Category: OWASP A02:2025 / CWE-347

Threat Level: HIGH

Attack Flow:

1. Attacker obtains valid JWT
2. Changes algorithm to "none"
3. Modifies payload (e.g., user_id, role)
4. Server accepts unsigned token
5. Attacker gains elevated privileges

Mitigation:

from jose import jwt, JWTError

def decode_token(token: str) -> dict:
    try:
        # Explicitly specify allowed algorithms
        payload = jwt.decode(
            token,
            SECRET_KEY,
            algorithms=["HS256"],  # Never allow "none"
            options={"require_exp": True}
        )
        return payload
    except JWTError:
        raise HTTPException(401, "Invalid token")

---

Attack Scenario 4: CORS Misconfiguration

Threat Category: OWASP A05:2025 / CWE-346

Threat Level: MEDIUM

Attack Flow:

1. API configured with allow_origins=["*"] + allow_credentials=True
2. Attacker creates malicious website
3. User visits attacker site while logged into API
4. Attacker's JavaScript makes authenticated requests
5. User's cookies sent, attacker steals data

Mitigation:

# Never wildcard with credentials
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["Authorization"]
)

---

Attack Scenario 5: SQL Injection via Query Parameters

Threat Category: OWASP A03:2025 / CWE-89

Threat Level: CRITICAL

Mitigation:

from sqlalchemy import select

@app.get("/users")
async def search_users(search: str, db: AsyncSession = Depends(get_db)):
    # SAFE: ORM with parameterized query
    stmt = select(User).where(User.name.ilike(f"%{search}%"))
    result = await db.execute(stmt)
    return result.scalars().all()

---

STRIDE Analysis