name: API请求验证 description: 通过模式验证、类型安全性和清理，对API请求进行数据完整性、安全性和业务逻辑执行的验证。

API请求验证

概览

API请求验证是在处理请求正文、查询参数、路径参数和头部之前，对传入的API请求进行验证的过程，以确保数据完整性、安全性和业务逻辑执行。

为什么这很重要

数据一致性：通过高达90%的方式减少无效数据
安全性：输入验证有助于防止SQL注入和XSS攻击
用户体验：清晰的错误消息减少支持工单，提高用户满意度
开发效率：可重用的验证模式减少样板代码

核心概念

1. 模式验证

定义：为请求/响应数据定义验证模式

// 使用Zod (TypeScript)
import { z } from 'zod';

// 请求模式
const createUserSchema = z.object({
  email: z.string().email(),
  password: z.string().min(8).regex(/^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)/),
  age: z.number().min(18).max(120),
  name: z.string().min(2).max(100),
});

// 响应模式
const userResponseSchema = z.object({
  id: z.string().uuid(),
  email: z.string().email(),
  name: z.string(),
  createdAt: z.date(),
});

# 使用Pydantic (Python)
from pydantic import BaseModel, EmailStr, Field, validator

class CreateUserSchema(BaseModel):
    email: EmailStr
    password: str = Field(..., min_length=8)
    age: int = Field(..., ge=18, le=120)
    name: str = Field(..., min_length=2, max_length=100)

    @validator('password')
    def validate_password(cls, v):
        if not any(c.isupper() for c in v):
            raise ValueError('密码必须包含大写字母')
        if not any(c.islower() for c in v):
            raise ValueError('密码必须包含小写字母')
        if not any(c.isdigit() for c in v):
            raise ValueError('密码必须包含数字')
        return v

2. 类型安全性

定义：使用TypeScript/Python类型提示进行类型安全验证

// NestJS与class-validator
import { IsEmail, IsString, Min, Max } from 'class-validator';
import { Type } from 'class-transformer';

class CreateUserDto {
  @IsEmail()
  email: string;

  @IsString()
  @Min(8)
  password: string;

  @Type(() => Number)
  @Min(18)
  @Max(120)
  age: number;
}

3. 验证中间件

定义：中间件用于验证传入请求

// Express中间件
import { Request, Response, NextFunction } from 'express';
import { AnyZodObject, ZodError } from 'zod';

export const validate =
  (schema: AnyZodObject) =>
  async (req: Request, res: Response, next: NextFunction) => {
    try {
      await schema.parseAsync({
        body: req.body,
        query: req.query,
        params: req.params,
      });
      next();
    } catch (error) {
      if (error instanceof ZodError) {
        return res.status(400).json({
          error: '验证失败',
          details: error.errors,
        });
      }
      next(error);
    }
  };

# FastAPI与依赖注入
from fastapi import FastAPI, HTTPException, Depends
from pydantic import ValidationError

app = FastAPI()

def validate_request(data, schema_class):
    try:
        return schema_class(**data)
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=e.errors())

@app.post('/users')
async def create_user(user_data: dict = Body(...)):
    validated_user = validate_request(user_data, CreateUserSchema)
    # 处理验证过的用户
    return validated_user

4. 错误处理

定义：对无效数据返回适当的错误响应

// 自定义错误格式化器
export const formatZodError = (error: ZodError) => {
  return {
    errors: error.errors.map((err) => ({
      field: err.path.join('.'),
      message: err.message,
      code: err.code,
    })),
  };
};

// 在中间件中的使用
res.status(400).json(formatZodError(error));

# FastAPI自动错误处理
@app.exception_handler(ValidationError)
async def validation_exception_handler(request: Request, exc: ValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "errors": [
                {
                    "field": ".".join(str(loc) for loc in err["loc"]),
                    "message": err["msg"],
                    "type": err["type"],
                }
                for err in exc.errors()
            ]
        },
    )

5. 输入清理

定义：清理和清理用户输入以防止攻击

// 使用DOMPurify (浏览器)或sanitize-html (Node.js)
import DOMPurify from 'dompurify';

const sanitizeHtml = (html: string): string => {
  return DOMPurify.sanitize(html, {
    ALLOWED_TAGS: ['b', 'i', 'em', 'strong', 'a'],
    ALLOWED_ATTR: ['href', 'target'],
  });
};

// SQL注入防护
const sanitizeQuery = (query: string): string => {
  return query.replace(/['";\\]/g, '');
};

# 使用bleach进行HTML清理
import bleach

def sanitize_html(html: str) -> str:
    return bleach.clean(
        html,
        tags=['b', 'i', 'em', 'strong', 'a'],
        attributes={'a': ['href']},
        strip=True
    )

# SQL注入防护与参数化查询
# 始终使用参数化查询而不是字符串格式化

6. 自定义验证器

定义：可重用的自定义验证逻辑

// 自定义Zod验证器
const strongPassword = z
  .string()
  .min(8)
  .regex(/^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)/, {
    message: '密码必须包含大写字母、小写字母和数字',
  });

const uuidSchema = z.string().uuid();

const urlSchema = z.string().url();

// 带异步检查的自定义验证器
const uniqueEmail = z.string().email().refine(
  async (email) => {
    const exists = await checkEmailExists(email);
    return !exists;
  },
  { message: '电子邮件已注册' }
);

# 自定义Pydantic验证器
from pydantic import validator

class UserSchema(BaseModel):
    email: EmailStr
    username: str

    @validator('username')
    def username_alphanumeric(cls, v):
        if not v.isalnum():
            raise ValueError('用户名必须是字母数字')
        return v

    @validator('email')
    def email_unique(cls, v):
        # 与数据库核对
        if check_email_exists(v):
            raise ValueError('电子邮件已注册')
        return v

7. 异步验证

定义：支持异步验证（数据库检查、外部API）

// 异步Zod验证
const uniqueUsername = z
  .string()
  .min(3)
  .max(20)
  .refine(
    async (username) => {
      const user = await prisma.user.findUnique({
        where: { username },
      });
      return !user;
    },
    { message: '用户名已被占用' }
  );

# 异步Pydantic验证器
import asyncio
from pydantic import validator

class UserSchema(BaseModel):
    email: EmailStr

    @validator('email')
    async def email_must_be_unique(cls, v):
        is_taken = await check_email_in_database(v)
        if is_taken:
            raise ValueError('电子邮件已注册')
        return v

架构图

┌─────────────────────────────────────────────────┐
│              API请求验证架构               │
├─────────────────────────────────────────────────┤
│                                                  │
│  ┌───────────────────────────────────────────┐  │
│  │              客户端请求                │  │
│  └───────────────────────────────────────────┘  │
│                       │                          │
│                       ▼                          │
│  ┌───────────────────────────────────────────┐  │
│  │         中间件层                     │  │
│  │  ┌─────────┐  ┌─────────┐  ┌───────────┐ │  │
│  │  │ 正文    │  │ 查询    │  │ 参数      │ │  │
│  │  │ 解析器   │  │ 解析器   │  │ 解析器     │ │  │
│  │  └─────────┘  └─────────┘  └───────────┘ │  │
│  │       │              │              │        │  │
│  │       └──────────────┼──────────────┘        │  │
│  │                      ▼                       │  │
│  │         ┌─────────────────────┐              │  │
│  │         │   模式验证器         │              │  │
│  │         └─────────────────────┘              │  │
│  │                      │                       │  │
│  │              ┌───────┴───────┐               │  │
│  │              ▼               ▼               │  │
│  │        ┌──────────┐    ┌──────────┐          │  │
│  │        │  有效     │    │  无效     │          │  │
│  │        └────┬─────┘    └────┬─────┘          │  │
│  └─────────────┼─────────────┼─────────────────┘  │
│                │             │                     │
│                ▼             ▼                     │
│  ┌──────────────────┐  ┌──────────────────┐       │
│  │ 控制器         │  │ 错误处理器       │       │
│  └──────────────────┘  └──────────────────┘       │
└─────────────────────────────────────────────────────┘

快速开始

TypeScript与Zod

npm install zod express express-async-errors

import express from 'express';
import { z } from 'zod';

const app = express();
app.use(express.json());

// 定义模式
const userSchema = z.object({
  name: z.string().min(2),
  email: z.string().email(),
});

app.post('/users', validate(userSchema), (req, res) => {
  const user = req.body; // 验证过的数据
  // 处理用户
  res.json(user);
});

app.listen(3000);

Python与Pydantic

pip install fastapi uvicorn pydantic

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr, Field

app = FastAPI()

class UserCreate(BaseModel):
    name: str = Field(..., min_length=2)
    email: EmailStr

@app.post('/users')
async def create_user(user: UserCreate):
    # 处理用户
    return user

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)

生产清单

[ ] 所有端点都有验证模式
[ ] 实施输入清理
[ ] SQL注入防护（参数化查询）
[ ] XSS防护HTML输入
[ ] 清晰、可操作的错误消息
[ ] 验证端点的速率限制
[ ] 验证性能<10ms
[ ] 测试自定义验证器
[ ] 正确处理异步验证器
[ ] 错误日志记录以便于调试

反模式

仅客户端验证：永远不要单独信任客户端验证
静默失败：始终返回清晰的错误消息
过于宽松的正则表达式：避免接受无效数据的正则表达式
忽视安全性：始终清理输入以防止注入攻击
跳过验证：即使“安全”的输入也要验证

集成点

NestJS：内置的class-validator和class-transformer
Express：使用Zod/Joi的中间件
FastAPI：自动Pydantic验证
Django：表单和DRF序列化器
TypeORM：实体验证装饰器

API请求验证 APIRequestValidation