1. CORS Configuration
# 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. Rate Limiting
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. Security Headers Middleware
# 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. Input Sanitization
# 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 Injection Prevention
# ❌ 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. Request ID Tracking
# 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. Trusted Host & 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 Top 10 Checklist cho FastAPI
| # | Vulnerability | Giải pháp trong FastAPI |
|---|---|---|
| A01 | Broken Access Control | RBAC dependencies, ownership checks |
| A02 | Cryptographic Failures | Bcrypt passwords, JWT với strong secret |
| A03 | Injection | Pydantic validation, SQLAlchemy ORM |
| A04 | Insecure Design | Rate limiting, input validation |
| A05 | Security Misconfiguration | Security headers, CORS, env variables |
| A06 | Vulnerable Components | uv audit, dependency updates |
| A07 | Auth Failures | OAuth2, JWT rotation, token blacklist |
| A08 | Data Integrity Failures | Pydantic validators, CSRF protection |
| A09 | Logging Failures | Structured logging, request ID tracking |
| A10 | SSRF | URL validation, allowlists |
Tổng kết
Security best practices cho FastAPI:
- CORS: Cấu hình strict, chỉ allow origins cần thiết
- Rate Limiting: Bảo vệ API khỏi abuse
- Security Headers: Thêm headers bảo mật vào mọi response
- Input Sanitization: Sanitize user input trước khi xử lý
- SQL Injection: Sử dụng ORM hoặc parameterized queries
- Request Tracking: Trace requests qua toàn bộ system
Bài tiếp theo sẽ triển khai Social Login và OAuth2 Providers.