api-review

API 安全审查技能

摘要

用于开发 API 端点的全面安全检查清单。确保在部署前实现适当的身份验证、授权、输入验证、输出安全和安全日志记录。

使用时机

• 合并包含 API 更改的任何 PR 之前
• 创建新的 API 端点时
• 修改身份验证／授权逻辑时
• 进行安全审计时
• 审查 API 路由代码时

快速检查清单

部署前安全审计

• 身份验证：路由要求有效的用户身份
• 授权：已实现所有权／权限检查
• 输入验证：所有输入均通过架构进行验证（Zod/Joi 等）
• 输出安全：响应中不会暴露敏感数据
• 日志记录：已适当记录安全事件
• 速率限制：已配置防止滥用的保护措施
• 错误处理：错误中不会泄露系统信息

---

身份验证

要求

每个 API 端点都必须先验证用户身份，再处理请求。

使用 Clerk 的 Next.js（App Router）

import { auth } from '@clerk/nextjs';
import { NextResponse } from 'next/server';

export async function GET(request: Request) {
  // 1. Authenticate request
  const { userId } = await auth();

  if (!userId) {
    return NextResponse.json(
      { error: "Unauthorized" },
      { status: 401 }
    );
  }

  // Continue with authenticated request...
}

使用 JWT 的 Express.js

import jwt from 'jsonwebtoken';
import { Request, Response, NextFunction } from 'express';

function authenticateToken(req: Request, res: Response, next: NextFunction) {
  const authHeader = req.headers['authorization'];
  const token = authHeader && authHeader.split(' ')[1];

  if (!token) {
    return res.sendStatus(401);
  }

  jwt.verify(token, process.env.JWT_SECRET!, (err, user) => {
    if (err) return res.sendStatus(403);
    req.user = user;
    next();
  });
}

app.get('/api/protected', authenticateToken, (req, res) => {
  // Request is authenticated
});

使用 OAuth2 的 FastAPI

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = await verify_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

@app.get("/api/protected")
async def protected_route(current_user: User = Depends(get_current_user)):
    return {"user": current_user.email}

Django REST Framework

from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response

@api_view(['GET'])
@permission_classes([IsAuthenticated])
def protected_view(request):
    # request.user is authenticated
    return Response({'user': request.user.email})

---

授权

资源所有权验证

身份验证用于证明用户是谁。授权用于证明用户有权访问该资源。

Next.js 示例

import { auth } from '@clerk/nextjs';
import { NextResponse } from 'next/server';
import { db } from '@/lib/db';
import { eq } from 'drizzle-orm';

export async function GET(
  request: Request,
  { params }: { params: { id: string } }
) {
  // 1. Authenticate
  const { userId } = await auth();
  if (!userId) {
    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
  }

  // 2. Fetch resource
  const resource = await db.query.resources.findFirst({
    where: eq(resources.id, params.id)
  });

  if (!resource) {
    return NextResponse.json({ error: "Not found" }, { status: 404 });
  }

  // 3. Authorize - Check ownership
  if (resource.ownerId !== userId) {
    return NextResponse.json({ error: "Forbidden" }, { status: 403 });
  }

  // 4. Return authorized data
  return NextResponse.json(resource);
}

基于角色的访问控制（RBAC）

enum Role {
  USER = 'user',
  ADMIN = 'admin',
  MODERATOR = 'moderator'
}

function requireRole(allowedRoles: Role[]) {
  return async (request: Request) => {
    const { userId } = await auth();
    if (!userId) {
      return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
    }

    const user = await db.query.users.findFirst({
      where: eq(users.clerkId, userId)
    });

    if (!user || !allowedRoles.includes(user.role)) {
      return NextResponse.json({ error: "Forbidden" }, { status: 403 });
    }

    return null; // Authorized
  };
}

export async function DELETE(request: Request) {
  const authError = await requireRole([Role.ADMIN, Role.MODERATOR])(request);
  if (authError) return authError;

  // User is authorized as admin or moderator
}

多租户数据隔离

// CRITICAL: Prevent cross-tenant data leaks

// ❌ WRONG - No tenant check
const orders = await db.query.orders.findMany({
  where: eq(orders.userId, userId)
});

// ✅ CORRECT - Tenant isolation
const user = await db.query.users.findFirst({
  where: eq(users.clerkId, userId)
});

const orders = await db.query.orders.findMany({
  where: and(
    eq(orders.userId, userId),
    eq(orders.tenantId, user.tenantId) // CRITICAL: tenant boundary
  )
});

---

输入验证

Zod 架构验证（TypeScript）

import { z } from 'zod';
import { NextResponse } from 'next/server';

const updateUserSchema = z.object({
  id: z.string().uuid(),
  email: z.string().email().optional(),
  age: z.number().int().min(0).max(150).optional(),
  role: z.enum(['user', 'admin', 'moderator']).optional(),
});

export async function PATCH(request: Request) {
  const { userId } = await auth();
  if (!userId) {
    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
  }

  // Parse and validate input
  const body = await request.json();
  const result = updateUserSchema.safeParse(body);

  if (!result.success) {
    return NextResponse.json({
      error: "Validation failed",
      details: result.error.issues
    }, { status: 400 });
  }

  // Safe to use validated data
  const validatedData = result.data;
  // ... update logic
}

Pydantic 验证（Python）

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

class UpdateUser(BaseModel):
    id: str = Field(..., regex=r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$')
    email: EmailStr | None = None
    age: int | None = Field(None, ge=0, le=150)
    role: str | None = Field(None, regex=r'^(user|admin|moderator)$')

    @validator('email')
    def email_must_not_be_disposable(cls, v):
        if v and any(domain in v for domain in ['tempmail.com', '10minutemail.com']):
            raise ValueError('Disposable email addresses not allowed')
        return v

@app.patch("/api/users")
async def update_user(user_data: UpdateUser, current_user: User = Depends(get_current_user)):
    # user_data is validated
    return {"status": "updated"}

SQL 注入防护

// ❌ NEVER: Raw SQL with string interpolation
const userId = request.params.id;
const query = `SELECT * FROM users WHERE id = '${userId}'`; // VULNERABLE!
db.execute(query);

// ✅ ALWAYS: Use ORM or parameterized queries
import { eq } from 'drizzle-orm';
const user = await db.query.users.findFirst({
  where: eq(users.id, userId)
});

// ✅ OR: Parameterized raw query
const [user] = await db.execute(
  'SELECT * FROM users WHERE id = ?',
  [userId]
);

文件上传验证

const MAX_FILE_SIZE = 5 * 1024 * 1024; // 5MB
const ALLOWED_TYPES = ['image/jpeg', 'image/png', 'image/gif'];

export async function POST(request: Request) {
  const formData = await request.formData();
  const file = formData.get('file') as File;

  if (!file) {
    return NextResponse.json({ error: "No file provided" }, { status: 400 });
  }

  // Validate file size
  if (file.size > MAX_FILE_SIZE) {
    return NextResponse.json({
      error: "File too large. Max 5MB"
    }, { status: 400 });
  }

  // Validate file type
  if (!ALLOWED_TYPES.includes(file.type)) {
    return NextResponse.json({
      error: "Invalid file type. Only JPEG, PNG, GIF allowed"
    }, { status: 400 });
  }

  // Process file...
}

---

输出安全

从响应中移除敏感数据

// ❌ WRONG - Exposing sensitive fields
const user = await db.query.users.findFirst({
  where: eq(users.id, userId)
});
return NextResponse.json(user); // Includes password hash, internal IDs, etc.

// ✅ CORRECT - Explicitly select safe fields
const user = await db.query.users.findFirst({
  where: eq(users.id, userId),
  columns: {
    id: true,
    email: true,
    name: true,
    createdAt: true,
    // Exclude: passwordHash, internalNotes, apiKey, etc.
  }
});
return NextResponse.json(user);

// ✅ BETTER - Use DTOs
interface PublicUserDTO {
  id: string;
  email: string;
  name: string;
  createdAt: Date;
}

function toPublicUser(user: User): PublicUserDTO {
  return {
    id: user.id,
    email: user.email,
    name: user.name,
    createdAt: user.createdAt
  };
}

return NextResponse.json(toPublicUser(user));

对日志中的 PII 进行脱敏

function sanitizeForLogging(data: any) {
  const sanitized = { ...data };

  // Mask email
  if (sanitized.email) {
    const [local, domain] = sanitized.email.split('@');
    sanitized.email = `${local.slice(0, 2)}***@${domain}`;
  }

  // Mask SSN
  if (sanitized.ssn) {
    sanitized.ssn = `***-**-${sanitized.ssn.slice(-4)}`;
  }

  // Remove sensitive fields
  delete sanitized.passwordHash;
  delete sanitized.apiKey;

  return sanitized;
}

console.log('User updated:', sanitizeForLogging(user));

安全的错误消息

// ❌ WRONG - Leaking system information
try {
  await db.execute(query);
} catch (error) {
  return NextResponse.json({
    error: error.message, // Might expose SQL, file paths, etc.
    stack: error.stack     // NEVER expose in production
  }, { status: 500 });
}

// ✅ CORRECT - Generic error with logging
try {
  await db.execute(query);
} catch (error) {
  console.error('Database error:', error); // Log full error internally
  return NextResponse.json({
    error: "An error occurred processing your request"
  }, { status: 500 });
}

---

日志记录

安全事件日志记录

enum SecurityEvent {
  AUTH_FAILURE = 'auth_failure',
  UNAUTHORIZED_ACCESS = 'unauthorized_access',
  PERMISSION_DENIED = 'permission_denied',
  RATE_LIMIT_EXCEEDED = 'rate_limit_exceeded',
  INVALID_INPUT = 'invalid_input'
}

function logSecurityEvent(event: SecurityEvent, details: any) {
  console.log(JSON.stringify({
    timestamp: new Date().toISOString(),
    event,
    userId: details.userId || 'anonymous',
    ip: details.ip,
    endpoint: details.endpoint,
    details: sanitizeForLogging(details)
  }));
}

// Usage
export async function GET(request: Request) {
  const { userId } = await auth();

  if (!userId) {
    logSecurityEvent(SecurityEvent.AUTH_FAILURE, {
      ip: request.headers.get('x-forwarded-for'),
      endpoint: request.url
    });
    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
  }
}

请求追踪

import { v4 as uuidv4 } from 'uuid';

export async function middleware(request: Request) {
  const requestId = uuidv4();

  console.log(JSON.stringify({
    requestId,
    method: request.method,
    url: request.url,
    timestamp: new Date().toISOString()
  }));

  // Pass request ID through headers
  const response = await fetch(request.url, {
    headers: {
      ...request.headers,
      'X-Request-ID': requestId
    }
  });

  return response;
}

---

安全端点示例

完整的 Next.js API 路由

import { auth } from '@clerk/nextjs';
import { NextResponse } from 'next/server';
import { db } from '@/lib/db';
import { eq, and } from 'drizzle-orm';
import { z } from 'zod';
import { ratelimit } from '@/lib/ratelimit';

// 1. Input validation schema
const updateResourceSchema = z.object({
  name: z.string().min(1).max(100),
  description: z.string().max(500).optional(),
  isPublic: z.boolean().optional()
});

// 2. DTO for safe output
interface ResourceDTO {
  id: string;
  name: string;
  description: string;
  isPublic: boolean;
  createdAt: Date;
}

function toResourceDTO(resource: any): ResourceDTO {
  return {
    id: resource.id,
    name: resource.name,
    description: resource.description,
    isPublic: resource.isPublic,
    createdAt: resource.createdAt
    // Exclude: ownerId, internalNotes, etc.
  };
}

export async function PATCH(
  request: Request,
  { params }: { params: { id: string } }
) {
  try {
    // 3. Rate limiting
    const { success } = await ratelimit.limit(request.headers.get('x-forwarded-for') || 'anonymous');
    if (!success) {
      return NextResponse.json({ error: "Too many requests" }, { status: 429 });
    }

    // 4. Authentication
    const { userId } = await auth();
    if (!userId) {
      console.log('Auth failure:', { endpoint: request.url });
      return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
    }

    // 5. Input validation
    const body = await request.json();
    const result = updateResourceSchema.safeParse(body);

    if (!result.success) {
      return NextResponse.json({
        error: "Validation failed",
        details: result.error.issues
      }, { status: 400 });
    }

    // 6. Fetch and verify existence
    const resource = await db.query.resources.findFirst({
      where: eq(resources.id, params.id)
    });

    if (!resource) {
      return NextResponse.json({ error: "Not found" }, { status: 404 });
    }

    // 7. Authorization check
    if (resource.ownerId !== userId) {
      console.log('Permission denied:', { userId, resourceId: params.id });
      return NextResponse.json({ error: "Forbidden" }, { status: 403 });
    }

    // 8. Update resource
    const [updatedResource] = await db.update(resources)
      .set({
        ...result.data,
        updatedAt: new Date()
      })
      .where(eq(resources.id, params.id))
      .returning();

    // 9. Return safe response
    return NextResponse.json(toResourceDTO(updatedResource));

  } catch (error) {
    // 10. Safe error handling
    console.error('Error updating resource:', error);
    return NextResponse.json({
      error: "An error occurred"
    }, { status: 500 });
  }
}

---

框架模式

Express.js 中间件模式

import { Request, Response, NextFunction } from 'express';
import { z } from 'zod';

// Validation middleware factory
function validateSchema(schema: z.ZodSchema) {
  return (req: Request, res: Response, next: NextFunction) => {
    const result = schema.safeParse(req.body);
    if (!result.success) {
      return res.status(400).json({
        error: 'Validation failed',
        details: result.error.issues
      });
    }
    req.body = result.data;
    next();
  };
}

// Authorization middleware
async function requireOwnership(req: Request, res: Response, next: NextFunction) {
  const resource = await db.resources.findById(req.params.id);

  if (!resource) {
    return res.status(404).json({ error: 'Not found' });
  }

  if (resource.ownerId !== req.user.id) {
    return res.status(403).json({ error: 'Forbidden' });
  }

  req.resource = resource;
  next();
}

// Usage
const updateSchema = z.object({ name: z.string() });
app.patch('/api/resources/:id',
  authenticate,
  validateSchema(updateSchema),
  requireOwnership,
  async (req, res) => {
    // All checks passed
    const updated = await updateResource(req.resource, req.body);
    res.json(toDTO(updated));
  }
);

FastAPI 依赖注入

from fastapi import Depends, HTTPException
from typing import Annotated

async def verify_ownership(
    resource_id: str,
    current_user: User = Depends(get_current_user)
):
    resource = await db.resources.get(resource_id)
    if not resource:
        raise HTTPException(status_code=404, detail="Not found")
    if resource.owner_id != current_user.id:
        raise HTTPException(status_code=403, detail="Forbidden")
    return resource

@app.patch("/api/resources/{resource_id}")
async def update_resource(
    data: UpdateResourceSchema,
    resource: Resource = Depends(verify_ownership)
):
    # Resource ownership verified
    updated = await resource.update(data.dict())
    return ResourceDTO.from_orm(updated)

---

常见漏洞

OWASP API 安全十大风险

1. 对象级授权失效（BOLA）

// ❌ VULNERABLE
export async function GET(request: Request) {
  const { userId } = await auth();
  const resourceId = new URL(request.url).searchParams.get('id');

  // Missing ownership check!
  const resource = await db.query.resources.findFirst({
    where: eq(resources.id, resourceId)
  });

  return NextResponse.json(resource);
}

// ✅ FIXED
export async function GET(request: Request) {
  const { userId } = await auth();
  const resourceId = new URL(request.url).searchParams.get('id');

  const resource = await db.query.resources.findFirst({
    where: and(
      eq(resources.id, resourceId),
      eq(resources.ownerId, userId) // Ownership check
    )
  });

  if (!resource) {
    return NextResponse.json({ error: "Not found" }, { status: 404 });
  }

  return NextResponse.json(resource);
}

2. 批量赋值

// ❌ VULNERABLE - User can set any field
export async function PATCH(request: Request) {
  const body = await request.json();

  // User could send: { role: 'admin', isVerified: true }
  await db.update(users)
    .set(body) // DANGEROUS!
    .where(eq(users.id, userId));
}

// ✅ FIXED - Explicit allowed fields
const allowedFields = z.object({
  name: z.string().optional(),
  bio: z.string().optional()
  // role, isVerified NOT allowed
});

export async function PATCH(request: Request) {
  const body = await request.json();
  const validated = allowedFields.parse(body);

  await db.update(users)
    .set(validated)
    .where(eq(users.id, userId));
}

3. 数据过度暴露

// ❌ VULNERABLE
return NextResponse.json(user); // All fields exposed

// ✅ FIXED
return NextResponse.json({
  id: user.id,
  name: user.name,
  email: user.email
  // passwordHash, resetToken, etc. excluded
});

4. 速率限制

import { Ratelimit } from '@upstash/ratelimit';
import { Redis } from '@upstash/redis';

const ratelimit = new Ratelimit({
  redis: Redis.fromEnv(),
  limiter: Ratelimit.slidingWindow(10, '10 s'),
});

export async function POST(request: Request) {
  const identifier = request.headers.get('x-forwarded-for') || 'anonymous';
  const { success } = await ratelimit.limit(identifier);

  if (!success) {
    return NextResponse.json({ error: "Too many requests" }, { status: 429 });
  }

  // Process request...
}

---

总结

安全检查清单模板

## Security Review for [Endpoint Name]

### Authentication
- [ ] User identity verified before processing
- [ ] Invalid tokens rejected with 401
- [ ] Token expiration checked

### Authorization
- [ ] Resource ownership verified
- [ ] Role/permission checks implemented
- [ ] Cross-tenant data isolation enforced

### Input Validation
- [ ] All inputs validated with schema (Zod/Pydantic)
- [ ] File uploads size/type limited
- [ ] SQL injection prevented (using ORM)
- [ ] XSS prevention in place

### Output Safety
- [ ] Sensitive fields excluded from responses
- [ ] PII masked in logs
- [ ] Error messages don't leak system info

### Rate Limiting
- [ ] Rate limits configured per endpoint
- [ ] DDoS protection in place

### Logging
- [ ] Failed auth attempts logged
- [ ] Permission denials logged
- [ ] Request IDs for traceability

在代码审查期间以及部署 API 更改之前使用此技能，确保全面覆盖安全要求。