レッスン 11: セキュリティのベスト プラクティス

Python FastAPI: 基本から高度まで

パート 3: 認証とセキュリティ

xdev.asia

1. CORS の構成

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.config import settings

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.cors_origins,  # ["https://example.com"]
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
    expose_headers=["X-Total-Count", "X-Request-ID"],
    max_age=600,  # Cache preflight requests for 10 minutes
)

2. レート制限

uv add slowapi

# app/core/rate_limit.py
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address

limiter = Limiter(
    key_func=get_remote_address,
    default_limits=["100/minute"],
    storage_uri="redis://localhost:6379/1",
)

# app/main.py
from app.core.rate_limit import limiter

app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# Sử dụng trong routes
from slowapi import Limiter
from fastapi import Request

@app.post("/api/v1/auth/login")
@limiter.limit("5/minute")  # Giới hạn login attempts
async def login(request: Request):
    ...

@app.get("/api/v1/items")
@limiter.limit("60/minute")
async def list_items(request: Request):
    ...

3. セキュリティヘッダーミドルウェア

# app/middleware/security.py
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response


class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        response = await call_next(request)

        # Security headers
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"
        response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        response.headers["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()"
        response.headers["Content-Security-Policy"] = (
            "default-src 'self'; "
            "script-src 'self'; "
            "style-src 'self' 'unsafe-inline'; "
            "img-src 'self' data: https:; "
            "font-src 'self'"
        )

        # Remove server header
        response.headers.pop("server", None)

        return response


# app/main.py
app.add_middleware(SecurityHeadersMiddleware)

4. 入力のサニタイズ

# app/core/sanitize.py
import re
import html
from pydantic import BeforeValidator
from typing import Annotated


def sanitize_string(value: str) -> str:
    """Sanitize string input - loại bỏ HTML/script injection."""
    if not isinstance(value, str):
        return value
    # Escape HTML entities
    value = html.escape(value)
    # Remove potential script tags
    value = re.sub(r'<script.*?>.*?</script>', '', value, flags=re.IGNORECASE | re.DOTALL)
    # Remove null bytes
    value = value.replace('\x00', '')
    return value.strip()


# Custom type sử dụng trong Pydantic
SafeString = Annotated[str, BeforeValidator(sanitize_string)]


# Sử dụng trong schemas
from pydantic import BaseModel

class CommentCreate(BaseModel):
    content: SafeString  # Tự động sanitize
    title: SafeString

5. SQL インジェクションの防止

# ❌ TUYỆT ĐỐI KHÔNG LÀM
# Truyền trực tiếp user input vào raw SQL
@app.get("/users/search")
async def bad_search(q: str, session: AsyncSession = Depends(get_db)):
    result = await session.execute(
        text(f"SELECT * FROM users WHERE name = '{q}'")  # SQL INJECTION!
    )

# ✅ ĐÚNG CÁCH 1: Sử dụng parameterized queries
from sqlalchemy import text

@app.get("/users/search")
async def safe_search(q: str, session: AsyncSession = Depends(get_db)):
    result = await session.execute(
        text("SELECT * FROM users WHERE name = :name"),
        {"name": q}  # Parameterized - safe
    )

# ✅ ĐÚNG CÁCH 2: Sử dụng SQLAlchemy ORM (recommended)
@app.get("/users/search")
async def safe_search_orm(q: str, session: AsyncSession = Depends(get_db)):
    result = await session.execute(
        select(User).where(User.name.ilike(f"%{q}%"))  # SQLAlchemy handles escaping
    )
    return result.scalars().all()

6. リクエストID追跡

# app/middleware/request_id.py
import uuid

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response


class RequestIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        # Lấy request ID từ header hoặc generate mới
        request_id = request.headers.get("X-Request-ID") or str(uuid.uuid4())

        # Lưu vào request state
        request.state.request_id = request_id

        # Process request
        response = await call_next(request)

        # Thêm request ID vào response header
        response.headers["X-Request-ID"] = request_id

        return response

7. 信頼できるホストとHTTPS

# app/main.py
from starlette.middleware.trustedhost import TrustedHostMiddleware
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware

# Chỉ cho phép request từ trusted hosts
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "*.example.com"],
)

# Redirect HTTP → HTTPS (production only)
if settings.environment == "production":
    app.add_middleware(HTTPSRedirectMiddleware)

8. OWASP FastAPI のトップ 10 チェックリスト

#	脆弱性	FastAPI でのソリューション
A01	壊れたアクセス制御	RBAC の依存関係、所有権のチェック
A02	暗号化の失敗	Bcrypt パスワード、強力な秘密を持つ JWT
A03	注射	Pydantic 検証、SQLAlchemy ORM
A04	安全でない設計	レート制限、入力検証
A05	セキュリティの構成ミス	セキュリティヘッダー、CORS、環境変数
A06	脆弱なコンポーネント	UV監査、依存関係の更新
A07	認証の失敗	OAuth2、JWT ローテーション、トークン ブラックリスト
A08	データ整合性の失敗	Pydantic バリデーター、CSRF 保護
A09	ログの失敗	構造化ロギング、リクエストID追跡
A10	SSRF	URL検証、許可リスト

概要

FastAPI のセキュリティのベスト プラクティス:

• コルス: 厳密な構成。必要なオリジンのみを許可します。
• レート制限: API を悪用から保護する
• セキュリティヘッダー: すべての応答にセキュリティ ヘッダーを追加します
• 入力のサニタイズ: 処理前にユーザー入力をサニタイズします
• SQLインジェクション: ORM またはパラメータ化されたクエリを使用する
• リクエスト追跡: システム全体でリクエストをトレースします。

次の記事では、ソーシャル ログインと OAuth2 プロバイダーを実装します。