fastapi

FastAPI 開発スキル

ファイル構成

• SKILL.md：基本原則、パターン、必須のセキュリティ対策（このファイル）
• references/security-examples.md：CVE の詳細と OWASP の実装例
• references/advanced-patterns.md：FastAPI の高度なパターン
• references/threat-model.md：攻撃シナリオと STRIDE 分析

検証ゲート

Gate 0.2：脆弱性調査（高リスクの場合はブロッキング）

• ステータス：合格（5 件以上の CVE を記録）
• 調査日：2025-11-20
• CVE：CVE-2024-47874、CVE-2024-12868、CVE-2023-30798、Starlette の DoS 関連バリアント

---

1. 概要

リスクレベル：高

理由：FastAPI アプリケーションは、認証、データベースアクセス、ファイルアップロード、外部 API との通信を処理します。Starlette の DoS 脆弱性、インジェクションのリスク、不適切な検証により、可用性とセキュリティが損なわれる可能性があります。

あなたは、安全で高性能な REST API と WebSocket サービスを開発する、FastAPI のエキスパートです。適切な検証、認証、セキュリティヘッダーを設定します。

主な専門分野

• Pydantic による検証と依存性注入
• 認証：OAuth2、JWT、API キー
• セキュリティヘッダーと CORS の設定
• レート制限と DoS 対策
• 非同期 ORM によるデータベース連携
• WebSocket のセキュリティ

---

2. 主な責務

基本原則

1. まず TDD：実装コードを書く前にテストを作成します
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 のマルチパート 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 アクセス制御	HIGH	認証に依存性注入を使用し、権限デコレーターを適用
A02 暗号化の失敗	HIGH	適切なアルゴリズムを使用した JWT、Argon2 パスワード
A03 インジェクション	HIGH	Pydantic による検証、パラメーター化クエリ
A04 安全でない設計	MEDIUM	型安全性、検証レイヤー
A05 セキュリティ設定の不備	HIGH	セキュリティヘッダー、本番環境でドキュメントを無効化
A06 脆弱なコンポーネント	CRITICAL	Starlette／FastAPI を最新の状態に保つ
A07 認証の失敗	HIGH	認証のレート制限、安全な 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 ドキュメントを無効にします