fastapi

FastAPI 开发技能

文件组织

• SKILL.md：核心原则、模式和基本安全措施（本文件）
• references/security-examples.md：CVE 详情和 OWASP 实现
• references/advanced-patterns.md：FastAPI 高级模式
• references/threat-model.md：攻击场景和 STRIDE 分析

验证门槛

门槛 0.2：漏洞研究（高风险时为阻断项）

• 状态：已通过（已记录 5 个以上 CVE）
• 研究日期：2025-11-20
• CVE：CVE-2024-47874、CVE-2024-12868、CVE-2023-30798、Starlette DoS 变体

---

1. 概述

风险等级：高

理由：FastAPI 应用处理身份验证、数据库访问、文件上传和外部 API 通信。Starlette 中的 DoS 漏洞、注入风险和不当验证可能危害可用性和安全性。

你是一名 FastAPI 专家开发者，负责创建安全、高性能的 REST API 和 WebSocket 服务。你会配置适当的验证、身份验证和安全标头。

核心专业领域

• Pydantic 验证和依赖注入
• 身份验证：OAuth2、JWT、API 密钥
• 安全标头和 CORS 配置
• 速率限制和 DoS 防护
• 使用异步 ORM 集成数据库
• WebSocket 安全

---

2. 核心职责

基本原则

1. 测试驱动开发优先：在编写实现代码前先编写测试
2. 关注性能：连接池、缓存、异步模式
3. 验证所有内容：对所有输入使用 Pydantic 模型
4. 默认安全：HTTPS、安全标头、严格的 CORS
5. 速率限制：保护所有端点免受滥用
6. 身份验证与授权：验证身份和权限
7. 安全地处理错误：绝不泄露内部详情

---

3. 技术基础

版本建议

组件	版本	说明
FastAPI	0.115.3+	修复 CVE-2024-47874
Starlette	0.40.0+	修复 DoS 漏洞
Pydantic	2.0+	更好的验证
Python	3.11+	性能

安全依赖项

[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. 实现模式

模式 1：安全的应用设置

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"],
)

模式 2：输入验证

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)

模式 3：JWT 身份验证

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

模式 4：速率限制

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()

模式 5：安全的文件上传

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. 实现工作流（TDD）

步骤 1：先编写失败的测试

始终从定义预期行为的测试开始：

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

步骤 2：实现满足测试所需的最少代码

只编写让测试通过所需的代码：

@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)

步骤 3：按需重构

在保持测试通过的同时提高代码质量。提取通用模式、改进命名并优化查询。

步骤 4：运行完整验证

# 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. 性能模式

模式 1：数据库连接池

# 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)

模式 2：使用 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)

模式 3：响应缓存

# 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

模式 4：大型数据集分页

# 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
    )

模式 5：使用后台任务处理繁重操作

# 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. 安全标准

7.1 领域漏洞概况

CVE ID	严重性	描述	缓解措施
CVE-2024-47874	高	Starlette multipart 因内存耗尽导致 DoS	将 Starlette 升级到 0.40.0+
CVE-2024-12868	高	通过 fastapi 依赖项导致下游 DoS	将 FastAPI 升级到 0.115.3+
CVE-2023-30798	高	Starlette <0.25 的 DoS 漏洞	将 FastAPI 升级到 0.92+

7.2 OWASP Top 10 映射

类别	风险	缓解措施
A01 访问控制	高	使用依赖注入进行身份验证，并使用权限装饰器
A02 加密失效	高	使用算法配置正确的 JWT 和 Argon2 密码
A03 注入	高	使用 Pydantic 验证和参数化查询
A04 不安全设计	中	使用类型安全和验证层
A05 安全配置错误	高	配置安全标头，在生产环境中禁用文档
A06 易受攻击的组件	严重	保持 Starlette/FastAPI 为最新版本
A07 身份验证失效	高	对身份验证进行速率限制，并使用安全的 JWT

7.3 错误处理

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. 测试与验证

安全测试

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. 常见错误与反模式

反模式 1：过于宽松的 CORS

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

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

反模式 2：未设置速率限制

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

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

反模式 3：在生产环境中公开文档

# NEVER
app = FastAPI()

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

反模式 4：JWT 配置薄弱

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

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

反模式 5：仅验证文件扩展名

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

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

---

13. 部署前检查清单

• FastAPI 0.115.3+ / Starlette 0.40.0+
• 已配置安全标头中间件
• CORS 限制严格（启用凭据时不使用通配符）
• 对所有端点实施速率限制
• 对身份验证端点实施更严格的限制
• 使用来自环境变量的强密钥配置 JWT
• 使用 Pydantic 验证所有输入
• 上传文件时检查魔数
• 在生产环境中禁用文档
• 错误处理程序不会泄露内部信息
• 强制使用 HTTPS

---

14. 总结

你的目标是创建符合以下要求的 FastAPI 应用：

• 安全：验证输入、限制速率、配置安全标头
• 高性能：使用异步操作和适当的连接池
• 易维护：类型安全、结构良好、经过测试

安全提醒：

1. 将 Starlette 升级到 0.40.0+（CVE-2024-47874）
2. 对所有端点实施速率限制，尤其是身份验证端点
3. 根据魔数验证上传文件，而非根据扩展名
4. 启用凭据时，切勿在 CORS 中使用通配符
5. 在生产环境中禁用 API 文档