RESTAPI设计专家Skill rest-api-design

REST API 设计技能

name: rest-api-design-expert
risk_level: MEDIUM
description: RESTful API 设计专家，精通资源建模、HTTP 语义、分页、版本控制和安全 API 实现
version: 1.0.0
author: JARVIS AI Assistant
tags: [api, rest, http, design, web-services]

---

1. 概述

风险等级: 中风险

理由: REST API 暴露业务逻辑、处理身份验证和处理用户数据。设计不良会导致安全漏洞、数据泄露和注入攻击。

您是一位 RESTful API 设计 专家。您创建结构良好、安全且性能优异的 API，遵循 HTTP 语义和行业最佳实践。

核心专长

• 资源建模、URI 设计、HTTP 语义
• 分页、过滤、版本控制
• 安全最佳实践（BOLA、注入、验证）

主要用例

• 设计和重构 REST API
• API 文档编写和安全加固

文件组织: 核心概念在此；查看 references/security-examples.md 了解 CVE 缓解措施和详细模式。

---

2. 核心职责

核心原则

1. TDD 优先: 在实现前编写 API 测试
2. 性能意识: 优化延迟、吞吐量和效率
3. 设计安全: 保护端点免受常见攻击
4. 面向资源: 建模资源，而非操作

基本职责

1. 面向资源设计: 建模资源，而非操作
2. HTTP 语义: 使用正确方法和状态码
3. 一致惯例: 遵循命名和结构模式
4. 设计安全: 保护端点免受常见攻击

设计原则

• 名词，非动词: /users/{id} 而非 /getUser/{id}
• 复数资源: /users 而非 /user
• 层次关系: /users/{id}/orders
• 无状态操作: 无服务器端会话状态

---

3. 技术基础

HTTP 方法

方法	目的	幂等	安全	请求体
GET	检索资源	是	是	否
POST	创建资源	否	否	是
PUT	替换资源	是	否	是
PATCH	部分更新	否	否	是
DELETE	移除资源	是	否	否

状态码

成功 (2xx): 200 OK、201 Created、204 No Content

客户端错误 (4xx): 400 Bad Request、401 Unauthorized、403 Forbidden、404 Not Found、409 Conflict、422 Unprocessable Entity、429 Too Many Requests

服务器错误 (5xx): 500 Internal Server Error、503 Service Unavailable

---

4. 实现模式

4.1 资源设计

// 集合操作
GET    /api/v1/users              // 列出用户
POST   /api/v1/users              // 创建用户

// 实例操作
GET    /api/v1/users/{id}         // 获取用户
PUT    /api/v1/users/{id}         // 替换用户
PATCH  /api/v1/users/{id}         // 更新用户
DELETE /api/v1/users/{id}         // 删除用户

// 嵌套资源
GET    /api/v1/users/{id}/orders  // 获取用户订单
POST   /api/v1/users/{id}/orders  // 为用户创建订单

// 操作（必要时）
POST   /api/v1/users/{id}/verify  // 触发验证

4.2 请求/响应格式

// 一致响应信封
interface APIResponse<T> {
  data: T;
  meta?: { pagination?: PaginationMeta; timestamp: string; requestId: string; };
}

interface APIError {
  error: { code: string; message: string; details?: ValidationError[]; };
}

4.3 分页

// 基于游标（推荐） - 在 meta.pagination 中返回 nextCursor
GET /api/v1/users?limit=20&cursor=eyJpZCI6MTAwfQ

// 基于偏移（更简单但 O(n)）
GET /api/v1/users?limit=20&offset=40

4.4 过滤、排序和版本控制

// 过滤和排序
GET /api/v1/users?status=active&role=admin&sort=created_at:desc
GET /api/v1/users?fields=id,name,email  // 字段选择

// URL 路径版本控制（推荐）
GET /api/v1/users
GET /api/v2/users

// 旧版本弃用头
res.set("Deprecation", "true");
res.set("Sunset", "Sat, 01 Jun 2025 00:00:00 GMT");

4.5 身份验证

// Bearer 令牌身份验证
app.use("/api", (req, res, next) => {
  const authHeader = req.headers.authorization;
  if (!authHeader?.startsWith("Bearer ")) {
    return res.status(401).json({ error: { code: "UNAUTHORIZED", message: "需要 Bearer 令牌" }});
  }
  try {
    req.user = jwt.verify(authHeader.substring(7), process.env.JWT_SECRET);
    next();
  } catch {
    return res.status(401).json({ error: { code: "INVALID_TOKEN", message: "令牌无效或已过期" }});
  }
});

---

5. 实现工作流 (TDD)

分步 TDD 流程

为每个 API 端点遵循此工作流：

步骤 1: 先编写失败测试

# tests/test_users_api.py
import pytest
from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_create_user_returns_201():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post("/api/v1/users", json={"name": "John", "email": "john@example.com"})
    assert response.status_code == 201
    assert "id" in response.json()["data"]

@pytest.mark.asyncio
async def test_create_user_validates_email():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post("/api/v1/users", json={"name": "John", "email": "invalid"})
    assert response.status_code == 422

@pytest.mark.asyncio
async def test_get_user_requires_auth():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/api/v1/users/123")
    assert response.status_code == 401

步骤 2: 实现最小代码以通过测试

# app/routers/users.py
from fastapi import APIRouter, HTTPException, Depends
from pydantic import BaseModel, EmailStr

router = APIRouter(prefix="/api/v1/users", tags=["users"])

class CreateUserRequest(BaseModel):
    name: str
    email: EmailStr

@router.post("", status_code=201)
async def create_user(request: CreateUserRequest):
    user = await db.users.create(request.model_dump())
    return {"data": {"id": user.id, "name": user.name, "email": user.email}}

步骤 3: 重构和添加边缘案例

@pytest.mark.asyncio
async def test_get_user_prevents_bola():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/api/v1/users/other-id", headers={"Authorization": f"Bearer {user_a_token}"})
    assert response.status_code == 403

@pytest.mark.asyncio
async def test_list_users_pagination():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/api/v1/users?limit=10", headers={"Authorization": f"Bearer {admin_token}"})
    assert len(response.json()["data"]) <= 10

步骤 4: 运行完整验证

# 运行所有测试
pytest tests/test_users_api.py -v

# 运行覆盖率
pytest --cov=app --cov-report=term-missing

# 运行安全重点测试
pytest -m security -v

---

6. 性能模式

6.1 分页（基于游标）

# 坏：偏移分页 - O(n) 扫描
@router.get("/users")
async def list_users(offset: int = 0, limit: int = 20):
    return await db.execute(f"SELECT * FROM users LIMIT {limit} OFFSET {offset}")

# 好：基于游标的分页 - O(1) 查找
@router.get("/users")
async def list_users(cursor: str | None = None, limit: int = 20):
    query = "SELECT * FROM users"
    if cursor:
        query += f" WHERE id > '{base64.b64decode(cursor).decode()}'"
    query += f" ORDER BY id LIMIT {limit + 1}"

    results = await db.execute(query)
    has_more = len(results) > limit
    return {
        "data": results[:limit],
        "meta": {"pagination": {"limit": limit, "hasMore": has_more,
            "nextCursor": base64.b64encode(results[-1]["id"].encode()).decode() if has_more else None}}
    }

6.2 缓存头

# 坏：无缓存策略
@router.get("/products/{id}")
async def get_product(id: str):
    return await db.products.find_by_id(id)

# 好：ETag 和 Cache-Control 头
@router.get("/products/{id}")
async def get_product(id: str, request: Request, response: Response):
    product = await db.products.find_by_id(id)
    etag = f'"{hashlib.md5(json.dumps(product).encode()).hexdigest()}"'

    if request.headers.get("If-None-Match") == etag:
        return Response(status_code=304)  # 未修改

    response.headers["ETag"] = etag
    response.headers["Cache-Control"] = "public, max-age=300, must-revalidate"
    return {"data": product}

6.3 压缩

# 坏：无压缩
app = FastAPI()

# 好：启用 gzip 中间件
from fastapi.middleware.gzip import GZipMiddleware
app = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=1000)  # 压缩大于 1KB 的响应

6.4 速率限制

# 坏：无速率限制
@router.post("/api/auth/login")
async def login(credentials: LoginRequest):
    return await authenticate(credentials)

# 好：分层速率限制与 slowapi
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@router.post("/api/auth/login")
@limiter.limit("5/分钟")  # 对身份验证严格
async def login(request: Request, credentials: LoginRequest):
    return await authenticate(credentials)

@router.get("/api/v1/users")
@limiter.limit("100/分钟")  # API 标准
async def list_users(request: Request):
    return await get_users()

6.5 连接保持活动

# 坏：每次请求创建新连接
async def call_external_api():
    async with httpx.AsyncClient() as client:  # 每次新连接
        return await client.get("https://api.example.com/data")

# 好：应用级客户端与连接池
http_client: httpx.AsyncClient | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global http_client
    http_client = httpx.AsyncClient(
        limits=httpx.Limits(max_keepalive_connections=20, max_connections=100)
    )
    yield
    await http_client.aclose()

app = FastAPI(lifespan=lifespan)

---

7. 安全标准

参见 references/security-examples.md 了解完整 CVE 细节和缓解模式。

顶级 API 漏洞

• BOLA: 未经授权访问其他用户资源
• 大规模赋值: 通过请求体更新受保护字段
• 注入: 通过参数的 SQL/NoSQL 注入
• 过度数据暴露: 返回敏感字段

输入验证与授权

import { z } from "zod";

const CreateUserSchema = z.object({
  name: z.string().min(1).max(100),
  email: z.string().email(),
  password: z.string().min(12).max(100)
});

app.post("/api/v1/users", async (req, res) => {
  const validation = CreateUserSchema.safeParse(req.body);
  if (!validation.success) {
    return res.status(422).json({ error: { code: "VALIDATION_ERROR", details: validation.error.errors }});
  }
  res.status(201).json({ data: await createUser(validation.data) });
});

// BOLA 预防 - 始终检查对象所有权
app.get("/api/v1/users/:id", async (req, res) => {
  if (req.user.id !== req.params.id && !req.user.isAdmin) {
    return res.status(403).json({ error: { code: "FORBIDDEN", message: "访问被拒绝" }});
  }
  res.json({ data: await getUser(req.params.id) });
});

速率限制与安全头

import rateLimit from "express-rate-limit";

app.use("/api", rateLimit({ windowMs: 60000, max: 100 }));
app.use("/api/v1/auth", rateLimit({ windowMs: 60000, max: 5 }));  // 对身份验证更严格

// 安全头
app.use((req, res, next) => {
  res.set({ "Content-Type": "application/json", "X-Content-Type-Options": "nosniff", "X-Frame-Options": "DENY" });
  next();
});

---

8. 测试

describe("API 安全", () => {
  it("需要身份验证", async () => {
    expect((await request(app).get("/api/v1/users")).status).toBe(401);
  });
  it("防止 BOLA", async () => {
    const res = await request(app).get("/api/v1/users/other-id").set("Authorization", `Bearer ${userAToken}`);
    expect(res.status).toBe(403);
  });
  it("验证输入", async () => {
    expect((await request(app).post("/api/v1/users").send({ email: "bad" })).status).toBe(422);
  });
});

---

9. 常见错误

// 坏：返回未过滤数据（暴露密码哈希！）
res.json({ data: await db.users.findById(id) });
// 好：选择特定字段
const user = await db.users.findById(id, { select: ["id", "name", "email"] });

// 坏：无授权检查
app.delete("/api/v1/users/:id", async (req, res) => {
  await db.users.delete(req.params.id);  // 任何人都可以删除！
});
// 好：检查所有权
if (req.user.id !== req.params.id && !req.user.isAdmin) {
  return res.status(403).json({ error: { message: "禁止访问" } });
}

// 坏：大规模赋值漏洞
await db.users.update(id, req.body);  // 用户可以设置 isAdmin！
// 好：白名单允许字段
const ALLOWED = ["name", "email", "avatar"];
const updates = Object.fromEntries(ALLOWED.filter(f => req.body[f]).map(f => [f, req.body[f]]));

---

10. 预实现检查清单

阶段 1: 编码前

• [ ] 为所有端点编写失败测试（TDD 优先）
• [ ] 定义请求/响应模式的 API 合约
• [ ] 规划遵循 REST 惯例的资源 URI
• [ ] 确定身份验证和授权要求
• [ ] 审查性能需求（分页、缓存需求）

阶段 2: 实现期间

• [ ] 实现最小代码以通过每个测试
• [ ] 资源是名词，正确使用 HTTP 方法
• [ ] 适当状态码和一致响应格式
• [ ] 所有受保护端点身份验证
• [ ] 授权检查（BOLA 预防）
• [ ] 使用 Pydantic/Zod 模式进行输入验证
• [ ] 输出过滤仅必要字段
• [ ] 配置每个端点层级的速率限制
• [ ] 适当设置缓存头

阶段 3: 提交前

• [ ] 所有测试通过：pytest -v
• [ ] 覆盖率达到阈值：pytest --cov=app
• [ ] 安全测试通过：pytest -m security
• [ ] OpenAPI/Swagger 规范完整并包含示例
• [ ] 身份验证和错误代码已文档化
• [ ] CORS 配置限制性，强制 HTTPS
• [ ] 使用预期负载进行性能测试

---

11. 总结

设计 直观（REST 惯例、HTTP 语义）、安全（验证输入、授权访问、过滤输出）和 一致（统一响应、错误、分页）的 REST API。

安全要点: 检查对象级授权、使用模式验证输入、过滤输出字段、使用参数化查询、实施速率限制。

构建默认安全且易于正确使用的 API。