Express 高级模式与生产实践 / Express Advanced Patterns and Production Practices

📅 创建时间：2026-07-28 🏷️ 标签：#Express #Advanced #Authentication #FileUpload #API #Security #Performance 📚 前置知识：[[03-express-deep-dive]]

---

📋 本章目标

• 掌握 JWT（access + refresh token）、Session、OAuth 2.0、API Key 四种认证模式的实现与选型
• 理解文件上传从 multer 到生产级分片直传的完整链路
• 实现 API 版本化（URL / Header / Query）并建立废弃与迁移策略
• 用 Zod schema 构建类型安全的请求验证管道
• 设计统一的成功 / 错误 / 分页 / HATEOAS 响应格式
• 利用 zod-to-openapi 自动生成 Swagger 文档，实现代码到文档的单向数据流
• 配置 rate limiting、helmet、CORS、CSRF 构建纵深防御
• 调优 Express 性能：compression、缓存、连接池、cluster

---

第1部分：认证中间件设计

1.1 四种认证模式全景

┌─────────────────────────────────────────────────────────────┐
│                    认证模式决策矩阵                           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  场景                       推荐方案                         │
│  ─────────────────────     ──────────────────────────────   │
│  SPA + REST API             JWT (access + refresh)          │
│  传统服务端渲染              Session (cookie-based)          │
│  第三方登录 / SSO            OAuth 2.0 (Passport.js)         │
│  服务间调用 / Webhook        API Key                        │
│                                                             │
└─────────────────────────────────────────────────────────────┘

1.2 JWT 双 Token 模式

核心矛盾：单一 token 有效期短则频繁登录，有效期长则泄露危害大。双 token 化解此矛盾——access token 短寿命（15分钟）鉴权，refresh token 长寿命（7天）仅用于续期。

┌─────────────────────────────────────────────────────────────┐
│                    双 Token 认证流程                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  登录:    Client → POST /auth/login → {access, refresh}     │
│  请求:    Client → Bearer <accessToken> → 资源              │
│  过期:    Server 返回 401                                    │
│  刷新:    Client → POST /auth/refresh {refreshToken}        │
│          Server → {newAccess, newRefresh}                   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

import jwt from "jsonwebtoken";
import crypto from "crypto";

interface TokenPayload { userId: string; role: "user" | "admin"; }

const ACCESS_SECRET = process.env.JWT_ACCESS_SECRET!;
const REFRESH_SECRET = process.env.JWT_REFRESH_SECRET!;

function generateTokenPair(payload: TokenPayload) {
  return {
    accessToken: jwt.sign(payload, ACCESS_SECRET, {
      expiresIn: "15m",
      jwtid: crypto.randomUUID(),
    }),
    refreshToken: jwt.sign(
      { userId: payload.userId, tokenFamily: crypto.randomUUID() },
      REFRESH_SECRET,
      { expiresIn: "7d" }
    ),
  };
}

function verifyAccessToken(token: string): TokenPayload {
  return jwt.verify(token, ACCESS_SECRET) as TokenPayload;
}

可组合认证中间件

import { Request, Response, NextFunction } from "express";

declare global {
  namespace Express {
    interface Request {
      user?: TokenPayload;
      apiClient?: { clientId: string; tier: string };
    }
  }
}

// JWT 认证中间件工厂
function authenticateJwt(req: Request, res: Response, next: NextFunction) {
  const header = req.headers.authorization;
  if (!header?.startsWith("Bearer ")) {
    res.status(401).json({ error: "Missing or malformed authorization header" });
    return;
  }
  try {
    req.user = verifyAccessToken(header.slice(7));
    next();
  } catch (err) {
    if (err instanceof jwt.TokenExpiredError) {
      res.status(401).json({ error: "Token expired", code: "TOKEN_EXPIRED" });
      return;
    }
    res.status(401).json({ error: "Invalid token" });
  }
}

// 可选认证：有 token 则解析，无则放行
function authenticateOptional(req: Request, _res: Response, next: NextFunction) {
  const header = req.headers.authorization;
  if (header?.startsWith("Bearer ")) {
    try { req.user = verifyAccessToken(header.slice(7)); } catch { /* 静默失败 */ }
  }
  next();
}

// 角色守卫中间件工厂
function requireRole(...roles: string[]) {
  return (req: Request, res: Response, next: NextFunction) => {
    if (!req.user || !roles.includes(req.user.role)) {
      res.status(403).json({ error: "Insufficient permissions" });
      return;
    }
    next();
  };
}

// 按需组合使用
router.get("/me", authenticateJwt, meHandler);
router.delete("/users/:id", authenticateJwt, requireRole("admin"), deleteHandler);
router.get("/articles/:id", authenticateOptional, articleHandler);

1.3 Session 认证与 OAuth 2.0

┌─────────────────────────────────────────────────────────────┐
│              JWT vs Session 选型                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  JWT: 无状态、水平扩展友好，但吊销困难（需黑名单）            │
│  Session: 有状态、吊销即时，但每次请求需查存储               │
│                                                             │
│  规则：REST API / 微服务 → JWT；传统 SSR → Session。         │
│                                                             │
└─────────────────────────────────────────────────────────────┘

import session from "express-session";
import connectRedis from "connect-redis";
import Redis from "ioredis";
import passport from "passport";
import { Strategy as GoogleStrategy } from "passport-google-oauth20";

// Session 配置（防 session fixation：登录后 regenerate）
const RedisStore = connectRedis(session);
app.use(session({
  store: new RedisStore({ client: new Redis(process.env.REDIS_URL) }),
  secret: process.env.SESSION_SECRET!,
  resave: false,
  saveUninitialized: false,
  name: "sid",
  cookie: { httpOnly: true, secure: true, sameSite: "lax", maxAge: 86400000 },
}));

// OAuth 2.0 — Authorization Code Flow
passport.use(new GoogleStrategy({
  clientID: process.env.GOOGLE_CLIENT_ID!,
  clientSecret: process.env.GOOGLE_CLIENT_SECRET!,
  callbackURL: "/auth/google/callback",
}, async (_at, _rt, profile, done) => {
  let user = await User.findOne({ googleId: profile.id });
  if (!user) user = await User.create({ googleId: profile.id, email: profile.emails?.[0]?.value });
  done(null, user);
}));

app.use(passport.initialize());
router.get("/auth/google", passport.authenticate("google", { scope: ["profile", "email"] }));
router.get("/auth/google/callback",
  passport.authenticate("google", { failureRedirect: "/login", session: false }),
  (req, res) => {
    const tokens = generateTokenPair({ userId: (req.user as any).id, role: "user" });
    res.redirect(`/oauth-success?access=${tokens.accessToken}&refresh=${tokens.refreshToken}`);
  }
);

1.4 API Key 认证 + 可组合架构

import { createHash } from "crypto";

// 数据库只存 hash，不存明文 key
async function createApiKey(clientId: string, permissions: string[]) {
  const rawKey = `sk_${crypto.randomUUID().replace(/-/g, "")}`;
  const hash = createHash("sha256").update(rawKey).digest("hex");
  await db.apiKeys.insert({ clientId, keyHash: hash, keyPrefix: rawKey.slice(0, 8), permissions });
  return rawKey; // 只在创建时返回一次
}

function authenticateApiKey(req: Request, res: Response, next: NextFunction) {
  const rawKey = req.headers["x-api-key"] as string | undefined;
  if (!rawKey) { res.status(401).json({ error: "API key required" }); return; }

  const hash = createHash("sha256").update(rawKey).digest("hex");
  db.apiKeys.findOne({ keyHash: hash }).then(record => {
    if (!record) { res.status(401).json({ error: "Invalid API key" }); return; }
    req.apiClient = { clientId: record.clientId, tier: record.tier };
    next();
  }).catch(next);
}

// 统一认证入口——按策略类型分发
type AuthStrategy = "jwt" | "apikey" | "session";

function authenticate(strategies: AuthStrategy[]) {
  return async (req: Request, res: Response, next: NextFunction) => {
    for (const strategy of strategies) {
      try {
        if (strategy === "jwt") await runJwtAuth(req);
        if (strategy === "apikey") await runApiKeyAuth(req);
        if (req.user || req.apiClient) break;
      } catch { /* 当前策略失败，尝试下一个 */ }
    }
    if (!req.user && !req.apiClient) {
      res.status(401).json({ error: "Authentication required", accepted: strategies });
      return;
    }
    next();
  };
}

// 使用：接受 JWT 或 API Key
app.use("/api/v2", authenticate(["jwt", "apikey"]));

---

第2部分：文件上传策略

2.1 上传策略选择指南

┌─────────────────────────────────────────────────────────────┐
│  文件大小    推荐方案                                        │
│  ─────────  ──────────────────────────────────────────────  │
│  < 1MB      内存存储 (multer memoryStorage)                  │
│  1-100MB    磁盘临时 + 异步推送到云存储                      │
│  > 100MB    分片上传 (chunked upload, 断点续传)              │
│  任意大小    预签名 URL 前端直传 (bypass 服务器)              │
└─────────────────────────────────────────────────────────────┘

2.2 multer 基础 + 魔数检测

import multer from "multer";
import path from "path";

const memoryUpload = multer({
  storage: multer.memoryStorage(),
  limits: { fileSize: 5 * 1024 * 1024, files: 1 },
});

// 魔数表——文件头部固定字节序列，无法通过改名伪造
const MAGIC_BYTES: Record<string, number[]> = {
  "image/jpeg": [0xff, 0xd8, 0xff],
  "image/png":  [0x89, 0x50, 0x4e, 0x47],
  "image/gif":  [0x47, 0x49, 0x46, 0x38],
  "application/pdf": [0x25, 0x50, 0x44, 0x46],
};

function validateMagicBytes(buffer: Buffer, allowed: string[]): boolean {
  return allowed.some(mime => {
    const sig = MAGIC_BYTES[mime];
    return sig && sig.every((b, i) => buffer[i] === b);
  });
}

// 集成到路由
router.post("/upload", memoryUpload.single("file"), async (req, res) => {
  if (!req.file) { res.status(400).json({ error: "No file" }); return; }
  if (!validateMagicBytes(req.file.buffer, ["image/jpeg", "image/png"])) {
    res.status(400).json({ error: "Invalid file type (magic bytes check failed)" });
    return;
  }
  const url = await uploadToCloud(req.file.buffer, req.file.originalname);
  res.json({ url });
});

2.3 分片上传 + 预签名 URL 直传

┌─────────────────────────────────────────────────────────────┐
│                    分片上传流程                               │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  POST /upload/init       → { uploadId, chunkSize }          │
│  POST /upload/{id}/chunk/{n}  (可并行上传多个分片)           │
│  GET  /upload/{id}/status → { uploadedParts: [1,2,4] }      │
│  POST /upload/{id}/complete → { url, key }                  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

import { S3Client, CreateMultipartUploadCommand, UploadPartCommand,
         CompleteMultipartUploadCommand } from "@aws-sdk/client-s3";
import { getSignedUrl } from "@aws-sdk/s3-request-presigner";
import { PutObjectCommand } from "@aws-sdk/client-s3";
import { PassThrough } from "stream";

const s3 = new S3Client({
  region: "auto",
  endpoint: process.env.R2_ENDPOINT,
  credentials: { accessKeyId: process.env.R2_ACCESS_KEY!, secretAccessKey: process.env.R2_SECRET_KEY! },
});

// 分片初始化
router.post("/upload/multipart/init", authenticateJwt, async (req, res) => {
  const { filename, totalSize, contentType } = req.body;
  const key = `uploads/${req.user!.userId}/${Date.now()}-${filename}`;
  const { UploadId } = await s3.send(new CreateMultipartUploadCommand({
    Bucket: process.env.R2_BUCKET!, Key: key, ContentType: contentType,
  }));
  await redis.hset(`upload:${UploadId}`, { key, totalSize, userId: req.user!.userId });
  res.json({ uploadId: UploadId, key, chunkSize: 5 * 1024 * 1024,
    totalParts: Math.ceil(totalSize / (5 * 1024 * 1024)) });
});

// 上传分片
router.post("/upload/multipart/:uploadId/chunk/:partNumber",
  memoryUpload.single("chunk"), async (req, res) => {
    const partNumber = parseInt(req.params.partNumber);
    const { ETag } = await s3.send(new UploadPartCommand({
      Bucket: process.env.R2_BUCKET!,
      Key: await getKeyFromUploadId(req.params.uploadId),
      UploadId: req.params.uploadId,
      PartNumber: partNumber,
      Body: req.file!.buffer,
    }));
    await redis.hset(`upload:${req.params.uploadId}:parts`, String(partNumber), ETag!);
    res.json({ partNumber, etag: ETag });
  });

// 预签名 URL 直传——服务器零带宽消耗
router.post("/upload/presign", authenticateJwt, async (req, res) => {
  const { filename, contentType } = req.body;
  const key = `uploads/${req.user!.userId}/${Date.now()}-${filename}`;
  const presignedUrl = await getSignedUrl(s3, new PutObjectCommand({
    Bucket: process.env.R2_BUCKET!, Key: key, ContentType: contentType,
  }), { expiresIn: 300 });
  await redis.setex(`pending_upload:${key}`, 600, req.user!.userId);
  res.json({ presignedUrl, key, method: "PUT" });
});

---

第3部分：API 版本化策略

3.1 三种方式对比

┌─────────────────────────────────────────────────────────────┐
│  方式     示例                    缓存友好  推荐度           │
│  ──────── ──────────────────────  ────────  ──────          │
│  URL前缀   /v2/users             是        ⭐⭐⭐⭐⭐          │
│  Header    Accept: app+v2+json   否(Vary)  ⭐⭐⭐             │
│  Query     ?version=2            是        ⭐⭐               │
│                                                             │
│  推荐：URL 前缀作为主策略。                                   │
└─────────────────────────────────────────────────────────────┘

3.2 URL 前缀版本化 + 废弃策略

import { Router, Request, Response, NextFunction } from "express";

// 版本化路由结构
const v1UsersRouter = Router();
v1UsersRouter.get("/", (_req, res) => res.json({ users: [{ id: 1, name: "Alice" }] }));

const v2UsersRouter = Router();
v2UsersRouter.get("/", (_req, res) => {
  res.json({ data: [{ id: 1, name: "Alice", email: "alice@e.com" }], meta: { page: 1, total: 1 } });
});

app.use("/api/v1/users", deprecationWarning({
  sunset: "2026-06-01T00:00:00Z",
  alternative: "/api/v2/users",
}), v1UsersRouter);
app.use("/api/v2/users", v2UsersRouter);

// 废弃中间件：在响应头告知客户端
function deprecationWarning(opts: { sunset: string; alternative: string }) {
  return (_req: Request, res: Response, next: NextFunction) => {
    res.set({
      "Deprecation": "true",
      "Sunset": opts.sunset,
      "Link": `<${opts.alternative}>; rel="successor-version"`,
    });
    next();
  };
}

// Header 版本化方案（辅助）
function apiVersioning(req: Request, _res: Response, next: NextFunction) {
  const match = (req.headers.accept ?? "").match(/vnd\.myapp\.v(\d+)/);
  req.apiVersion = match ? match[1] : "1";
  next();
}

┌─────────────────────────────────────────────────────────────┐
│                    版本生命周期                               │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  CURRENT ──▶ DEPRECATED (警告+文档) ──▶ REMOVED (410 Gone) │
│                                                             │
│  至少 6 个月废弃期，最多同时维护 2 个版本。                   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

---

第4部分：请求验证

4.1 Zod + Express 集成

核心理念：定义一次 Zod schema，同时获得运行时验证和 TypeScript 类型。

import { z } from "zod";

const CreateUserSchema = z.object({
  body: z.object({
    username: z.string().min(3).max(30).regex(/^[a-zA-Z0-9_]+$/, "只能包含字母数字下划线"),
    email: z.string().email("邮箱格式不正确"),
    password: z.string().min(8).regex(/[A-Z]/, "必须包含大写字母")
      .regex(/[a-z]/, "必须包含小写字母").regex(/[0-9]/, "必须包含数字"),
    age: z.number().int().min(0).max(150).optional(),
  }),
});

const GetUsersQuerySchema = z.object({
  query: z.object({
    page: z.coerce.number().int().min(1).default(1),
    pageSize: z.coerce.number().int().min(1).max(100).default(20),
    sort: z.enum(["createdAt", "username"]).default("createdAt"),
    order: z.enum(["asc", "desc"]).default("desc"),
  }),
});

// 通用验证中间件工厂
function validate(schema: z.ZodSchema) {
  return (req: Request, res: Response, next: NextFunction) => {
    const result = schema.safeParse({ body: req.body, query: req.query, params: req.params });
    if (!result.success) {
      const errors = result.error.errors.map(e => ({ field: e.path.join("."), message: e.message, code: e.code }));
      res.status(422).json({ error: "Validation failed", code: "VALIDATION_ERROR", details: errors });
      return;
    }
    if (result.data.body) req.body = result.data.body;
    if (result.data.query) (req.query as any) = result.data.query;
    next();
  };
}

// 使用——req.body 和 req.query 已被验证且类型正确
router.post("/users", validate(CreateUserSchema), createHandler);
router.get("/users", validate(GetUsersQuerySchema), listHandler);

4.2 条件验证与 Schema 复用

// discriminatedUnion：根据 method 切换 schema
const PaymentSchema = z.discriminatedUnion("method", [
  z.object({ method: z.literal("credit_card"), cardNumber: z.string().regex(/^\d{16}$/), cvv: z.string() }),
  z.object({ method: z.literal("wechat"), wechatCode: z.string() }),
]);

// 字段复用
const EmailField = z.string().email();
const PasswordField = z.string().min(8).regex(/[A-Z]/);

const RegisterSchema = z.object({ body: z.object({ username: z.string(), email: EmailField, password: PasswordField }) });
const LoginSchema = z.object({ body: z.object({ email: EmailField, password: z.string().min(1) }) });

// 分页字段复用
const PaginationFields = z.object({ page: z.coerce.number().int().min(1).default(1), pageSize: z.coerce.number().min(1).max(100).default(20) });
const ListUsersSchema = z.object({ query: PaginationFields.extend({ role: z.enum(["user", "admin"]).optional() }) });

---

第5部分：响应格式标准化

import crypto from "crypto";

// 统一成功响应
function sendSuccess<T>(res: Response, data: T, opts?: { status?: number; links?: Record<string, string> }) {
  res.status(opts?.status ?? 200).json({
    success: true, data,
    meta: { timestamp: new Date().toISOString(), requestId: (res.req as any).requestId ?? crypto.randomUUID(), version: "2" },
    links: opts?.links,
  });
}

// 统一错误响应
const ErrorCodes = { VALIDATION_ERROR: "VALIDATION_ERROR", NOT_FOUND: "NOT_FOUND",
  UNAUTHORIZED: "UNAUTHORIZED", FORBIDDEN: "FORBIDDEN", RATE_LIMITED: "RATE_LIMITED", INTERNAL_ERROR: "INTERNAL_ERROR" };

function sendError(res: Response, status: number, message: string, opts?: { code?: string; details?: unknown }) {
  res.status(status).json({
    success: false,
    error: { code: opts?.code ?? ErrorCodes.INTERNAL_ERROR, message, details: opts?.details },
    meta: { timestamp: new Date().toISOString(), requestId: (res.req as any).requestId ?? "unknown" },
  });
}

// 分页响应（含 HATEOAS links）
function sendPaginated<T>(req: Request, res: Response, data: T[], total: number, page: number, pageSize: number) {
  const totalPages = Math.ceil(total / pageSize);
  const baseUrl = `${req.protocol}://${req.get("host")}${req.baseUrl}${req.path}`;
  const buildUrl = (p: number) => `${baseUrl}?${new URLSearchParams({ ...req.query as any, page: String(p), pageSize: String(pageSize) })}`;

  res.json({
    success: true, data,
    pagination: { page, pageSize, total, totalPages, hasNext: page < totalPages, hasPrev: page > 1 },
    links: { self: buildUrl(page), first: buildUrl(1), prev: page > 1 ? buildUrl(page - 1) : null, next: page < totalPages ? buildUrl(page + 1) : null, last: buildUrl(totalPages) },
  });
}

// 全局错误处理（放所有路由之后）
function globalErrorHandler(err: Error, _req: Request, res: Response, _next: NextFunction) {
  if (err instanceof z.ZodError) {
    sendError(res, 422, "Validation failed", { code: ErrorCodes.VALIDATION_ERROR,
      details: err.errors.map(e => ({ field: e.path.join("."), message: e.message })) });
    return;
  }
  sendError(res, 500, process.env.NODE_ENV === "production" ? "Internal server error" : err.message);
}
app.use(globalErrorHandler);

┌─────────────────────────────────────────────────────────────┐
│              HATEOAS 响应示例                                 │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  GET /api/v2/users/42                                       │
│  {                                                          │
│    "success": true, "data": { "id": 42, "name": "Alice" },  │
│    "links": {                                               │
│      "self": "/api/v2/users/42",                             │
│      "update": "/api/v2/users/42",                           │
│      "delete": "/api/v2/users/42",                           │
│      "articles": "/api/v2/users/42/articles"                 │
│    }                                                        │
│  }                                                          │
│                                                             │
│  客户端通过 links 发现可用操作，无需硬编码 URL。              │
└─────────────────────────────────────────────────────────────┘

---

第6部分：OpenAPI/Swagger 文档

┌─────────────────────────────────────────────────────────────┐
│                代码到文档的单向数据流                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  Zod Schema ──▶ zod-to-openapi ──▶ OpenAPI Spec             │
│       │                                │                    │
│       ▼                                ▼                    │
│  运行时验证                       Swagger UI                 │
│                                                             │
│  优势：修改 Schema = 同时更新验证 + 文档，永不漂移。          │
└─────────────────────────────────────────────────────────────┘

import { OpenAPIRegistry, OpenApiGeneratorV3 } from "@asteasolutions/zod-to-openapi";
import swaggerUi from "swagger-ui-express";

const registry = new OpenAPIRegistry();

registry.registerComponent("securitySchemes", "bearerAuth", { type: "http", scheme: "bearer", bearerFormat: "JWT" });

// 复用已验证好的 Zod schema 注册路径
registry.registerPath({
  method: "post", path: "/api/v2/users", description: "创建新用户", tags: ["Users"],
  security: [{ bearerAuth: [] }],
  request: { body: { content: { "application/json": { schema: CreateUserSchema.shape.body } } } },
  responses: {
    201: { description: "用户创建成功", content: { "application/json": { schema: z.object({ success: z.literal(true), data: z.object({ id: z.string(), username: z.string() }) }) } } },
    422: { description: "验证失败", content: { "application/json": { schema: z.object({ success: z.literal(false), error: z.object({ code: z.string(), message: z.string(), details: z.array(z.object({ field: z.string(), message: z.string() })) }) }) } } },
  },
});

registry.registerPath({
  method: "get", path: "/api/v2/users", description: "用户列表", tags: ["Users"],
  security: [{ bearerAuth: [] }],
  request: { query: GetUsersQuerySchema.shape.query },
  responses: {
    200: { description: "用户列表", content: { "application/json": { schema: z.object({ success: z.literal(true), data: z.array(z.object({ id: z.string(), username: z.string() })), pagination: z.object({ page: z.number(), pageSize: z.number(), total: z.number(), totalPages: z.number() }) }) } } },
  },
});

const generator = new OpenApiGeneratorV3(registry.definitions);
const openApiDoc = generator.generateDocument({
  openapi: "3.0.3",
  info: { title: "My API", version: "2.0.0", description: "Auto-generated OpenAPI docs" },
  servers: [{ url: "/api/v2", description: "current" }, { url: "/api/v1", description: "deprecated" }],
});

app.use("/api/docs", swaggerUi.serve, swaggerUi.setup(openApiDoc));
app.get("/api/openapi.json", (_req, res) => res.json(openApiDoc));

---

第7部分：Rate Limiting 与安全加固

7.1 纵深防御

┌─────────────────────────────────────────────────────────────┐
│  Layer 1: Rate Limiting    ← 防滥用/暴力破解                 │
│  Layer 2: Helmet           ← 安全 HTTP 头                   │
│  Layer 3: CORS             ← 跨域控制                       │
│  Layer 4: CSRF             ← 防跨站伪造                      │
│  Layer 5: Input Validation ← Zod schema                     │
│  Layer 6: AuthN / AuthZ    ← 认证 + 授权                    │
│  每层独立运作，任一失守不影响其他层。                         │
└─────────────────────────────────────────────────────────────┘

7.2 Rate Limiting

import rateLimit from "express-rate-limit";
import RedisStore from "rate-limit-redis";

// 全局限流
app.use(rateLimit({
  windowMs: 15 * 60 * 1000,
  max: 200,
  standardHeaders: true,
  legacyHeaders: false,
  store: process.env.NODE_ENV === "production" ? new RedisStore({ sendCommand: (...args: string[]) => redisClient.call(...args) as any }) : undefined,
}));

// 登录接口——严格防暴力破解
const authLimiter = rateLimit({
  windowMs: 15 * 60 * 1000, max: 10,
  skipSuccessfulRequests: true,
  keyGenerator: (req) => `${req.ip}-${req.body.username ?? "unknown"}`,
  handler: (_req, res) => res.status(429).json({ success: false, error: { code: "RATE_LIMITED", message: "登录尝试过于频繁" } }),
});
router.post("/auth/login", authLimiter, loginHandler);

// 按用户限流（需放在 authenticateJwt 之后）
const apiLimiter = rateLimit({
  windowMs: 60 * 1000, max: 60,
  keyGenerator: (req) => (req as any).user?.userId ?? req.ip,
});
app.use("/api", authenticateJwt, apiLimiter);

7.3 Helmet + CORS + CSRF

import helmet from "helmet";
import cors from "cors";
import { doubleCsrf } from "csrf-csrf";

// Helmet——14 个安全 header
app.use(helmet({
  contentSecurityPolicy: {
    directives: {
      defaultSrc: ["'self'"],
      scriptSrc: ["'self'"],
      imgSrc: ["'self'", "data:", "https://cdn.example.com"],
    },
  },
  hsts: { maxAge: 31536000, includeSubDomains: true },
  frameguard: { action: "deny" },
}));

// CORS——动态白名单
const allowedOrigins = (process.env.CORS_ORIGINS ?? "").split(",").map(s => s.trim());
app.use(cors({
  origin: (origin, cb) => {
    if (!origin || allowedOrigins.includes(origin) || allowedOrigins.includes("*")) cb(null, true);
    else cb(new Error(`Origin ${origin} not allowed`));
  },
  methods: ["GET", "POST", "PUT", "PATCH", "DELETE"],
  allowedHeaders: ["Content-Type", "Authorization", "X-API-Key"],
  exposedHeaders: ["X-Request-Id", "X-RateLimit-Remaining"],
  credentials: true,
  maxAge: 86400,
}));

// CSRF——double submit cookie 模式
const { generateToken, doubleCsrfProtection } = doubleCsrf({
  getSecret: () => process.env.CSRF_SECRET!,
  cookieName: "__Host-psifi.x-csrf-token",
  cookieOptions: { httpOnly: true, sameSite: "strict", secure: true },
  getTokenFromRequest: (req) => req.headers["x-csrf-token"] as string,
});

router.get("/csrf-token", (req, res) => res.json({ token: generateToken(req, res) }));
app.use("/api", doubleCsrfProtection);

---

第8部分：Express 性能调优

8.1 Compression + 静态资源缓存

import compression from "compression";
import express from "express";

// 只压缩 > 1KB 的响应，跳过已压缩的图片/视频
app.use(compression({
  threshold: 1024,
  level: 6,
  filter: (req, res) => {
    if (req.headers["x-no-compression"]) return false;
    const ct = res.getHeader("Content-Type") as string;
    if (/^(image|video|audio)\//.test(ct) || ct === "application/pdf") return false;
    return compression.filter(req, res);
  },
}));

// 带 hash 的资源永久缓存，HTML 不缓存
app.use("/static", express.static("public", {
  maxAge: "30d",
  immutable: true,
  setHeaders: (res, fp) => {
    if (fp.endsWith(".html")) res.setHeader("Cache-Control", "no-cache");
    if (/\.(js|css)$/.test(fp)) res.setHeader("Cache-Control", "public, max-age=31536000, immutable");
  },
}));

8.2 连接池 + 避免阻塞事件循环

import mysql from "mysql2/promise";
import { createClient } from "redis";
import { Agent as HttpsAgent } from "https";

// 连接池黄金法则：poolSize * 实例数 < 数据库 max_connections
const dbPool = mysql.createPool({
  host: process.env.DB_HOST, user: process.env.DB_USER, database: process.env.DB_NAME,
  connectionLimit: 10, waitForConnections: true, enableKeepAlive: true,
});

const redisClient = createClient({ url: process.env.REDIS_URL });
await redisClient.connect();

const httpsAgent = new HttpsAgent({ keepAlive: true, maxSockets: 50, maxFreeSockets: 10, timeout: 60000 });

┌─────────────────────────────────────────────────────────────┐
│                    事件循环红线                               │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ❌ fs.readFileSync、crypto.pbkdf2Sync、JSON.parse(超大)     │
│  ❌ 正则 ReDoS（catastrophic backtracking）                  │
│  ❌ 请求 handler 中大循环（1M+ 迭代）                        │
│                                                             │
│  ✅ 始终使用异步版本（fs.promises, crypto.pbkdf2）           │
│  ✅ 大计算任务 → Worker Threads                              │
│  ✅ 用户输入正则 → re2 库或设置超时                          │
│                                                             │
└─────────────────────────────────────────────────────────────┘

8.3 Cluster 多核 + 优雅关闭

// cluster.ts
import cluster from "cluster";
import os from "os";
import { app } from "./app";

const PORT = process.env.PORT ?? 3000;

if (cluster.isPrimary) {
  console.log(`[Master] PID ${process.pid} starting ${os.cpus().length} workers`);
  for (let i = 0; i < os.cpus().length; i++) cluster.fork();

  cluster.on("exit", (worker, code, signal) => {
    console.error(`[Master] Worker ${worker.process.pid} died. Restarting...`);
    setTimeout(() => cluster.fork(), 1000);
  });
} else {
  const server = app.listen(PORT, () =>
    console.log(`[Worker] PID ${process.pid} listening on ${PORT}`));

  // 优雅关闭
  process.on("SIGTERM", async () => {
    console.log(`[Worker] ${process.pid} shutting down...`);
    server.close(async () => { await dbPool.end(); await redisClient.quit(); process.exit(0); });
    setTimeout(() => process.exit(1), 30000); // 30s 强制退出
  });
}

┌─────────────────────────────────────────────────────────────┐
│              Cluster vs PM2                                   │
├─────────────────────────────────────────────────────────────┤
│  Cluster: 零依赖内置，简单场景                               │
│  PM2: 进程管理 + 日志 + 零停机重载（pm2 reload），生产首选    │
│  Docker: 由编排平台管理多实例                                │
└─────────────────────────────────────────────────────────────┘

---

核心总结

┌─────────────────────────────────────────────────────────────┐
│                    速查表                                     │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  认证                                                       │
│  ├─ JWT: access(15min) + refresh(7d), tokenFamily 支持撤销   │
│  ├─ Session: Redis 存储, regenerate 防 fixation              │
│  ├─ OAuth: Passport + Authorization Code Flow               │
│  ├─ API Key: SHA256 hash 存储, 前缀识别                      │
│  └─ 可组合: authenticate(["jwt","apikey"])                  │
│                                                             │
│  文件上传                                                    │
│  ├─ <1MB: memoryStorage                                     │
│  ├─ 1-100MB: diskStorage + 异步 S3                          │
│  ├─ >100MB: 分片上传 + 断点续传                              │
│  └─ 最佳: 预签名 URL 直传（零服务器带宽）                     │
│                                                             │
│  验证 & 文档                                                 │
│  ├─ Zod Schema → 验证 + TS 类型 + OpenAPI 文档               │
│  └─ 代码 → 文档单向数据流，永不漂移                          │
│                                                             │
│  安全                                                       │
│  ├─ Rate Limit: 全局 + 登录 + 按用户，三层叠加               │
│  ├─ Helmet: 14 个安全 header                                │
│  ├─ CORS: 动态白名单                                        │
│  └─ CSRF: doubleCsrf + SameSite cookie                     │
│                                                             │
│  性能                                                       │
│  ├─ compression: threshold 1KB, level 6                    │
│  ├─ 缓存: hash 资源 immutable, HTML no-cache               │
│  ├─ 连接池: 异步 + keepAlive                                │
│  ├─ 不阻塞: 异步一切，大计算 Worker Thread                  │
│  └─ 多核: cluster/PM2                                      │
│                                                             │
└─────────────────────────────────────────────────────────────┘

---

章节测试

1. 双 Token 模式：为什么 access token 设为 15 分钟而不是 1 小时？refresh token 泄露后的止损手段是什么？

2. 魔数检测：为什么只检查扩展名不够？说出一种绕过方式，并解释魔数检测如何防御。

3. API 版本化：某团队用 Accept header 做版本化，发现 CDN 缓存命中率极低。分析原因并给出改进方案。

4. Zod 条件验证：编写 schema：method 为 "credit_card" 时必须提供 cardNumber + cvv；为 "wire" 时必须提供 bankCode + accountNumber。

5. 多层限流：如何同时实现"全局 100 req/min"和"每用户 1000 req/hour"？描述 keyGenerator 选择。

6. 阻塞事件循环：fs.readFileSync、crypto.pbkdf2Sync、含嵌套量词的正则匹配长字符串——哪些会阻塞？如何安全处理？

7. 预签名 URL：相比服务器转发上传，预签名 URL 直传的三个主要优势是什么？有什么安全隐患需要防范？

---

参考答案

1. 短时效缩小泄露影响范围（15 分钟后自动失效）。refresh token 泄露后立即将 tokenFamily 加入 Redis 黑名单（撤销家族），使攻击者无法续期。

2. 攻击者改扩展名 .exe → .jpg。魔数检测读取文件头部字节序列（JPEG 为 FF D8 FF），这些字节由格式本身决定，改名无法改变。

3. CDN 按 Accept 缓存，不同版本号视为不同缓存 key，变体爆炸。改为 URL 前缀版本化（/v1/、/v2/），CDN 正常按路径缓存。

4. 使用 z.discriminatedUnion("method", [z.object({method: z.literal("credit_card"), cardNumber: ..., cvv: ...}), z.object({method: z.literal("wire"), bankCode: ..., accountNumber: ...})])。

5. 两个独立 rateLimit 中间件。全局用 keyGenerator: req => req.ip，按用户用 keyGenerator: req => req.user.userId（放认证之后）。任一触发即 429。

6. 全部可能阻塞（正则 ReDoS）。处理：改用 fs.promises、crypto.pbkdf2（异步）、正则用 re2 库或 runWithTimeout。

7. 优势：(1) 零服务器带宽 (2) 零内存占用 (3) 上传速度优化（用户直连 S3）。隐患：需限制 contentType 白名单、限制文件大小（S3 侧 policy）、设置短有效期（5 分钟）、确认回调验证所有权。

---

相关笔记

• [[03-express-deep-dive]] - Express 中间件洋葱模型
• [[02-async-patterns-and-error-handling]] - 异步模式与错误处理
• [[../03-security/01-web-security-basics]] - Web 安全深入

---

下一步学习

1. NestJS 架构对比——装饰器 + 依赖注入如何消除 Express 样板代码
2. API Gateway 模式——将限流、认证、版本化上移到 Kong / Traefik 网关层
3. gRPC——服务间通信中性能、类型安全和流式传输的终极方案
4. 文件处理流水线——上传后异步生成缩略图、病毒扫描的队列设计

---

学习状态：🟡 开始学习