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
- TDD First: Write tests before implementation code
- Performance Aware: Connection pooling, caching, async patterns
- Validate Everything: Use Pydantic models for all inputs
- Secure by Default: HTTPS, security headers, strict CORS
- Rate Limit: Protect all endpoints from abuse
- Authenticate & Authorize: Verify identity and permissions
- Handle Errors Safely: Never leak internal details
3. Technical Foundation
Version Recommendations
| Component | Version | Notes | |
|---|
| FastAPI | 0.115.3+ | CVE-2024-47874 fix | |
| Starlette | 0.40.0+ | DoS vulnerability fix | |
| Pydantic | 2.0+ | Better validation | |
| Python | 3.11+ | Performance | |
Security Dependencies
toml
[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
python
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
python
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
python
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
python
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
python
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:
python
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:
python
@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
bash
# 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
python
# 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
python
# 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
python
# 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