request-validation

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Request Validation

请求验证

Never trust user input. Validate everything.
永远不要信任用户输入,所有内容都要验证。

When to Use This Skill

何时使用该方案

  • All API endpoints accepting user input
  • Form submissions
  • File uploads
  • Query parameters
  • Webhook payloads
  • 所有接收用户输入的API端点
  • 表单提交
  • 文件上传
  • 查询参数
  • Webhook负载

TypeScript Implementation (Zod)

TypeScript实现方案(基于Zod)

Schema Definition

模式定义

typescript
// schemas/user.ts
import { z } from 'zod';

export const createUserSchema = z.object({
  email: z.string().email('Invalid email address'),
  password: z
    .string()
    .min(8, 'Password must be at least 8 characters')
    .regex(/[A-Z]/, 'Password must contain uppercase letter')
    .regex(/[0-9]/, 'Password must contain a number'),
  name: z.string().min(1).max(100).trim(),
  role: z.enum(['user', 'admin']).default('user'),
  metadata: z.record(z.string()).optional(),
});

export const updateUserSchema = createUserSchema.partial().omit({ password: true });

export const userIdSchema = z.object({
  id: z.string().uuid('Invalid user ID'),
});

export const listUsersSchema = z.object({
  page: z.coerce.number().int().positive().default(1),
  limit: z.coerce.number().int().min(1).max(100).default(20),
  search: z.string().max(100).optional(),
  role: z.enum(['user', 'admin']).optional(),
  sortBy: z.enum(['name', 'createdAt', 'email']).default('createdAt'),
  sortOrder: z.enum(['asc', 'desc']).default('desc'),
});

// Infer types from schemas
export type CreateUserInput = z.infer<typeof createUserSchema>;
export type UpdateUserInput = z.infer<typeof updateUserSchema>;
export type ListUsersQuery = z.infer<typeof listUsersSchema>;
typescript
// schemas/user.ts
import { z } from 'zod';

export const createUserSchema = z.object({
  email: z.string().email('Invalid email address'),
  password: z
    .string()
    .min(8, 'Password must be at least 8 characters')
    .regex(/[A-Z]/, 'Password must contain uppercase letter')
    .regex(/[0-9]/, 'Password must contain a number'),
  name: z.string().min(1).max(100).trim(),
  role: z.enum(['user', 'admin']).default('user'),
  metadata: z.record(z.string()).optional(),
});

export const updateUserSchema = createUserSchema.partial().omit({ password: true });

export const userIdSchema = z.object({
  id: z.string().uuid('Invalid user ID'),
});

export const listUsersSchema = z.object({
  page: z.coerce.number().int().positive().default(1),
  limit: z.coerce.number().int().min(1).max(100).default(20),
  search: z.string().max(100).optional(),
  role: z.enum(['user', 'admin']).optional(),
  sortBy: z.enum(['name', 'createdAt', 'email']).default('createdAt'),
  sortOrder: z.enum(['asc', 'desc']).default('desc'),
});

// Infer types from schemas
export type CreateUserInput = z.infer<typeof createUserSchema>;
export type UpdateUserInput = z.infer<typeof updateUserSchema>;
export type ListUsersQuery = z.infer<typeof listUsersSchema>;

Validation Middleware

验证中间件

typescript
// middleware/validate.ts
import { Request, Response, NextFunction } from 'express';
import { ZodSchema, ZodError } from 'zod';

interface ValidationSchemas {
  body?: ZodSchema;
  query?: ZodSchema;
  params?: ZodSchema;
}

function validate(schemas: ValidationSchemas) {
  return async (req: Request, res: Response, next: NextFunction) => {
    const errors: Array<{ field: string; message: string }> = [];

    try {
      if (schemas.body) {
        req.body = schemas.body.parse(req.body);
      }
      if (schemas.query) {
        req.query = schemas.query.parse(req.query);
      }
      if (schemas.params) {
        req.params = schemas.params.parse(req.params);
      }
      next();
    } catch (error) {
      if (error instanceof ZodError) {
        const formattedErrors = error.errors.map(err => ({
          field: err.path.join('.'),
          message: err.message,
        }));

        return res.status(400).json({
          error: 'Validation failed',
          details: formattedErrors,
        });
      }
      next(error);
    }
  };
}

export { validate };
typescript
// middleware/validate.ts
import { Request, Response, NextFunction } from 'express';
import { ZodSchema, ZodError } from 'zod';

interface ValidationSchemas {
  body?: ZodSchema;
  query?: ZodSchema;
  params?: ZodSchema;
}

function validate(schemas: ValidationSchemas) {
  return async (req: Request, res: Response, next: NextFunction) => {
    const errors: Array<{ field: string; message: string }> = [];

    try {
      if (schemas.body) {
        req.body = schemas.body.parse(req.body);
      }
      if (schemas.query) {
        req.query = schemas.query.parse(req.query);
      }
      if (schemas.params) {
        req.params = schemas.params.parse(req.params);
      }
      next();
    } catch (error) {
      if (error instanceof ZodError) {
        const formattedErrors = error.errors.map(err => ({
          field: err.path.join('.'),
          message: err.message,
        }));

        return res.status(400).json({
          error: 'Validation failed',
          details: formattedErrors,
        });
      }
      next(error);
    }
  };
}

export { validate };

Route Usage

路由使用

typescript
// routes/users.ts
import { validate } from '../middleware/validate';
import { createUserSchema, userIdSchema, listUsersSchema } from '../schemas/user';

router.post(
  '/users',
  validate({ body: createUserSchema }),
  async (req, res) => {
    // req.body is typed and validated
    const user = await userService.create(req.body);
    res.status(201).json(user);
  }
);

router.get(
  '/users/:id',
  validate({ params: userIdSchema }),
  async (req, res) => {
    const user = await userService.findById(req.params.id);
    res.json(user);
  }
);

router.get(
  '/users',
  validate({ query: listUsersSchema }),
  async (req, res) => {
    // req.query is typed with defaults applied
    const users = await userService.list(req.query);
    res.json(users);
  }
);
typescript
// routes/users.ts
import { validate } from '../middleware/validate';
import { createUserSchema, userIdSchema, listUsersSchema } from '../schemas/user';

router.post(
  '/users',
  validate({ body: createUserSchema }),
  async (req, res) => {
    // req.body is typed and validated
    const user = await userService.create(req.body);
    res.status(201).json(user);
  }
);

router.get(
  '/users/:id',
  validate({ params: userIdSchema }),
  async (req, res) => {
    const user = await userService.findById(req.params.id);
    res.json(user);
  }
);

router.get(
  '/users',
  validate({ query: listUsersSchema }),
  async (req, res) => {
    // req.query is typed with defaults applied
    const users = await userService.list(req.query);
    res.json(users);
  }
);

Custom Validators

自定义验证器

typescript
// schemas/custom.ts
import { z } from 'zod';

// Phone number with formatting
export const phoneSchema = z
  .string()
  .transform(val => val.replace(/\D/g, ''))
  .refine(val => val.length >= 10 && val.length <= 15, 'Invalid phone number');

// Slug (URL-safe string)
export const slugSchema = z
  .string()
  .min(1)
  .max(100)
  .regex(/^[a-z0-9]+(?:-[a-z0-9]+)*$/, 'Invalid slug format');

// Date string to Date object
export const dateSchema = z
  .string()
  .datetime()
  .transform(val => new Date(val));

// Sanitized HTML (strip dangerous tags)
export const safeHtmlSchema = z
  .string()
  .transform(val => sanitizeHtml(val, { allowedTags: ['b', 'i', 'em', 'strong', 'a', 'p'] }));

// File upload validation
export const fileSchema = z.object({
  mimetype: z.enum(['image/jpeg', 'image/png', 'application/pdf']),
  size: z.number().max(10 * 1024 * 1024, 'File too large (max 10MB)'),
  originalname: z.string(),
});
typescript
// schemas/custom.ts
import { z } from 'zod';

// Phone number with formatting
export const phoneSchema = z
  .string()
  .transform(val => val.replace(/\D/g, ''))
  .refine(val => val.length >= 10 && val.length <= 15, 'Invalid phone number');

// Slug (URL-safe string)
export const slugSchema = z
  .string()
  .min(1)
  .max(100)
  .regex(/^[a-z0-9]+(?:-[a-z0-9]+)*$/, 'Invalid slug format');

// Date string to Date object
export const dateSchema = z
  .string()
  .datetime()
  .transform(val => new Date(val));

// Sanitized HTML (strip dangerous tags)
export const safeHtmlSchema = z
  .string()
  .transform(val => sanitizeHtml(val, { allowedTags: ['b', 'i', 'em', 'strong', 'a', 'p'] }));

// File upload validation
export const fileSchema = z.object({
  mimetype: z.enum(['image/jpeg', 'image/png', 'application/pdf']),
  size: z.number().max(10 * 1024 * 1024, 'File too large (max 10MB)'),
  originalname: z.string(),
});

Python Implementation (Pydantic)

Python实现方案(基于Pydantic)

python
undefined
python
undefined

schemas/user.py

schemas/user.py

from pydantic import BaseModel, EmailStr, Field, field_validator from typing import Optional from enum import Enum
class UserRole(str, Enum): user = "user" admin = "admin"
class CreateUserInput(BaseModel): email: EmailStr password: str = Field(min_length=8) name: str = Field(min_length=1, max_length=100) role: UserRole = UserRole.user metadata: Optional[dict[str, str]] = None
@field_validator('password')
@classmethod
def password_strength(cls, v: str) -> str:
    if not any(c.isupper() for c in v):
        raise ValueError('Password must contain uppercase letter')
    if not any(c.isdigit() for c in v):
        raise ValueError('Password must contain a number')
    return v

@field_validator('name')
@classmethod
def strip_name(cls, v: str) -> str:
    return v.strip()
class UpdateUserInput(BaseModel): email: Optional[EmailStr] = None name: Optional[str] = Field(None, min_length=1, max_length=100) role: Optional[UserRole] = None
class ListUsersQuery(BaseModel): page: int = Field(default=1, ge=1) limit: int = Field(default=20, ge=1, le=100) search: Optional[str] = Field(None, max_length=100) role: Optional[UserRole] = None
undefined
from pydantic import BaseModel, EmailStr, Field, field_validator from typing import Optional from enum import Enum
class UserRole(str, Enum): user = "user" admin = "admin"
class CreateUserInput(BaseModel): email: EmailStr password: str = Field(min_length=8) name: str = Field(min_length=1, max_length=100) role: UserRole = UserRole.user metadata: Optional[dict[str, str]] = None
@field_validator('password')
@classmethod
def password_strength(cls, v: str) -> str:
    if not any(c.isupper() for c in v):
        raise ValueError('Password must contain uppercase letter')
    if not any(c.isdigit() for c in v):
        raise ValueError('Password must contain a number')
    return v

@field_validator('name')
@classmethod
def strip_name(cls, v: str) -> str:
    return v.strip()
class UpdateUserInput(BaseModel): email: Optional[EmailStr] = None name: Optional[str] = Field(None, min_length=1, max_length=100) role: Optional[UserRole] = None
class ListUsersQuery(BaseModel): page: int = Field(default=1, ge=1) limit: int = Field(default=20, ge=1, le=100) search: Optional[str] = Field(None, max_length=100) role: Optional[UserRole] = None
undefined

FastAPI Usage

FastAPI使用

python
undefined
python
undefined

routes/users.py

routes/users.py

from fastapi import APIRouter, Query, Path, HTTPException from schemas.user import CreateUserInput, UpdateUserInput, ListUsersQuery
router = APIRouter()
@router.post("/users", status_code=201) async def create_user(data: CreateUserInput): # data is validated and typed user = await user_service.create(data.model_dump()) return user
@router.get("/users/{user_id}") async def get_user(user_id: str = Path(pattern=r'^[0-9a-f-]{36}$')): user = await user_service.find_by_id(user_id) if not user: raise HTTPException(404, "User not found") return user
@router.get("/users") async def list_users( page: int = Query(default=1, ge=1), limit: int = Query(default=20, ge=1, le=100), search: str = Query(default=None, max_length=100), ): return await user_service.list(page=page, limit=limit, search=search)
undefined
from fastapi import APIRouter, Query, Path, HTTPException from schemas.user import CreateUserInput, UpdateUserInput, ListUsersQuery
router = APIRouter()
@router.post("/users", status_code=201) async def create_user(data: CreateUserInput): # data is validated and typed user = await user_service.create(data.model_dump()) return user
@router.get("/users/{user_id}") async def get_user(user_id: str = Path(pattern=r'^[0-9a-f-]{36}$')): user = await user_service.find_by_id(user_id) if not user: raise HTTPException(404, "User not found") return user
@router.get("/users") async def list_users( page: int = Query(default=1, ge=1), limit: int = Query(default=20, ge=1, le=100), search: str = Query(default=None, max_length=100), ): return await user_service.list(page=page, limit=limit, search=search)
undefined

Error Response Format

错误响应格式

json
{
  "error": "Validation failed",
  "details": [
    { "field": "email", "message": "Invalid email address" },
    { "field": "password", "message": "Password must be at least 8 characters" }
  ]
}
json
{
  "error": "Validation failed",
  "details": [
    { "field": "email", "message": "Invalid email address" },
    { "field": "password", "message": "Password must be at least 8 characters" }
  ]
}

Best Practices

最佳实践

  1. Validate at the edge - Before any business logic
  2. Use schema inference - Don't duplicate types
  3. Provide helpful messages - Tell users how to fix errors
  4. Sanitize strings - Trim whitespace, escape HTML
  5. Set sensible defaults - Reduce required fields
  1. 在边缘层验证 - 在执行业务逻辑之前完成验证
  2. 使用模式推导 - 不要重复定义类型
  3. 提供友好的提示消息 - 告诉用户如何修复错误
  4. 清理字符串 - 去除空格、转义HTML内容
  5. 设置合理的默认值 - 减少必填字段数量

Common Mistakes

常见错误

  • Validating after database operations
  • Generic error messages ("Invalid input")
  • Not validating query parameters
  • Missing max length on strings (DoS vector)
  • Not sanitizing HTML content
  • 在数据库操作之后才进行验证
  • 使用通用的错误消息(如“输入无效”)
  • 未验证查询参数
  • 未设置字符串的最大长度(可能导致DoS攻击)
  • 未清理HTML内容