第 11 課：安全最佳實踐

Python FastAPI：從基礎到進階

第 3 部分：身份驗證和安全性

亞洲開發網

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 十大清單

#	漏洞	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 提供者。