FastAPI安全防护指南：构建坚不可摧的参数处理体系-牛翰网

扫描二维码关注或者微信搜一搜：编程智域前端至全栈交流与成长

探索数千个预构建的 AI 应用，开启你的下一个伟大创意

第一章：输入验证体系

1.1 类型安全革命

from pydantic import BaseModel, PaymentCardNumber
from pydantic.types import SecretStr


class UserRequest(BaseModel):
    username: str = Field(min_length=4, regex="^[a-zA-Z0-9_]+$")
    credit_card: PaymentCardNumber
    password: SecretStr
    ip_address: IPv4Address

# 自动完成：
# 1. 信用卡格式验证
# 2. 密码内存加密
# 3. IP地址合法性检测

1.2 深度校验策略

from pydantic import validator, root_validator


class OrderRequest(BaseModel):
    items: list[int]
    total_price: float

    @validator('items', each_item=True)
    def check_item_ids(cls, v):
        if v <= 0:
            raise ValueError("非法商品ID")
        return v

    @root_validator
    def check_price_match(cls, values):
        items = values.get('items')
        price = values.get('total_price')
        # 查询数据库验证价格一致性
        real_price = calc_real_price(items)
        if abs(price - real_price) > 1e-6:
            raise ValueError("价格不匹配")
        return values

第二章：注入攻击防护

2.1 SQL注入防护矩阵

# 危险示例（绝对禁止）
@app.get("/items")
async def get_items(name: str):
    # 直接拼接SQL语句
    query = f"SELECT * FROM items WHERE name = '{name}'"
    return await database.fetch_all(query)


# 安全方案
from sqlalchemy import text


@app.get("/items")
async def safe_get_items(name: str):
    # 参数化查询
    query = text("SELECT * FROM items WHERE name = :name")
    return await database.fetch_all(query, {"name": name})

2.2 NoSQL注入防护

from bson import json_util
from fastapi.encoders import jsonable_encoder


class QuerySanitizer:
    @classmethod
    def sanitize(cls, query: dict):
        safe_query = {}
        for k, v in jsonable_encoder(query).items():
            if isinstance(v, str):
                safe_query[k] = {"$eq": v}
            else:
                safe_query[k] = v
        return json_util.dumps(safe_query)


# 使用示例
raw_query = {"name": {"$ne": "admin"}}
safe_query = QuerySanitizer.sanitize(raw_query)  # 转换为安全查询

第三章：敏感数据处理

3.1 数据遮蔽中间件

from fastapi import Request
from fastapi.middleware import Middleware


class DataMaskingMiddleware:
    def __init__(self, app):
        self.app = app
        self.sensitive_keys = {'password', 'token', 'credit_card'}

    async def __call__(self, request: Request, call_next):
        response = await call_next(request)
        body = await response.body()

        # 对敏感字段进行遮蔽
        masked_body = self.mask_sensitive_data(json.loads(body))
        return JSONResponse(
            content=masked_body,
            status_code=response.status_code,
            headers=dict(response.headers)
        )

    def mask_sensitive_data(self, data):
        if isinstance(data, dict):
            return {k: self._mask_value(k, v) for k, v in data.items()}
        return data

    def _mask_value(self, key, value):
        if key in self.sensitive_keys:
            return "***MASKED***"
        return value

3.2 密码学存储方案

from cryptography.fernet import Fernet
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
fernet = Fernet(config.SECRET_KEY)


class PasswordManager:
    @staticmethod
    def hash_password(plain: str) -> str:
        return pwd_context.hash(plain)

    @staticmethod
    def encrypt_data(data: str) -> bytes:
        return fernet.encrypt(data.encode())

    @staticmethod
    def decrypt_data(cipher: bytes) -> str:
        return fernet.decrypt(cipher).decode()


# 使用示例
hashed_pwd = PasswordManager.hash_password("user123")
encrypted_data = PasswordManager.encrypt_data("sensitive_info")

第四章：高级安全策略

4.1 请求签名验证

import hmac
from hashlib import sha256


class SignatureValidator:
    @classmethod
    def generate_signature(cls, data: dict, secret: str) -> str:
        sorted_str = "&".join(f"{k}={v}" for k, v in sorted(data.items()))
        return hmac.new(secret.encode(), sorted_str.encode(), sha256).hexdigest()

    @classmethod
    def validate_signature(cls, data: dict, signature: str, secret: str) -> bool:
        actual = cls.generate_signature(data, secret)
        return hmac.compare_digest(actual, signature)


# 在依赖项中进行验证
async def verify_request(
        request: Request,
        body: dict = Body(...),
        signature: str = Header(...)
):
    secret = config.API_SECRET
    if not SignatureValidator.validate_signature(body, signature, secret):
        raise HTTPException(403, "非法请求")
    return body

4.2 速率限制防御

from fastapi import Depends
from fastapi_limiter import FastAPILimiter
from fastapi_limiter.depends import RateLimiter


@app.on_event("startup")
async def startup():
    await FastAPILimiter.init(config.REDIS_URL)


@app.get("/sensitive", dependencies=[Depends(RateLimiter(times=5, seconds=60))])
async def sensitive_operation():
    return {"detail": "敏感操作成功"}

第五章：错误处理与日志

5.1 安全错误标准化

from fastapi import HTTPException


class SecurityException(HTTPException):
    def __init__(self, detail: str):
        super().__init__(
            status_code=403,
            detail=detail,
            headers={"WWW-Authenticate": "Bearer"},
        )


@app.exception_handler(SecurityException)
async def security_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
        headers=exc.headers
    )

5.2 安全日志审计

import logging
from logging.handlers import SysLogHandler

security_logger = logging.getLogger("api.security")
security_logger.setLevel(logging.INFO)
handler = SysLogHandler(address=('logs.papertrailapp.com', 12345))
security_logger.addHandler(handler)


class SecurityLogger:
    @staticmethod
    def log_suspicious(request: Request):
        log_data = {
            "ip": request.client.host,
            "path": request.url.path,
            "method": request.method,
            "user_agent": request.headers.get("user-agent")
        }
        security_logger.warning("可疑请求: %s", json.dumps(log_data))

课后Quiz

Q1：哪种方式能有效防止SQL注入？
A) 使用ORM的参数化查询
B) 拼接用户输入到SQL语句
C) 用正则过滤特殊字符
D) 限制数据库权限

Q2：敏感信息遮蔽的正确时机是？

数据库存储时
日志记录时
API响应时
全部正确

Q3：请求签名验证的主要作用是？

提升性能
防止请求篡改
压缩数据体积
验证请求来源合法性

错误代码速查表

错误码	场景	解决方案
422	参数校验失败	检查字段类型与格式约束
403	签名验证失败	检查请求签名生成算法
429	请求频率超限	降低操作频率或联系管理员
500	密钥配置错误	检查加密密钥加载逻辑