fastapi

What this skill does

# 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

| 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_un