📖 第 05 章 API 安全¶

图源：项目内既有威胁建模示意图（人工复核：未见 AI 生成痕迹）。该图适合帮助读者从接口、数据流、外部系统调用与缓解措施的角度理解 API 安全。

📚 章节概述¶

本章将深入讲解 API 安全，包括 API 网关、速率限制、认证与防御性测试等。内容以防御、检测和设计基线为主，术语尽量靠近 OWASP API Security Top 10 的稳定口径。

🎯 学习目标¶

完成本章后，你将能够：

理解 API 安全的核心概念
掌握 API 网关的使用
了解速率限制技术
掌握 API 认证方法
能够保护 API 安全

5.1 API 安全概述¶

5.1.1 什么是 API 安全¶

API 安全是保护 API 免受未授权访问、滥用和攻击的实践。

API 安全威胁¶

未授权访问
缺少认证
弱认证机制
会话管理不当
数据泄露
过度暴露数据
敏感信息泄露
错误信息泄露
滥用与资源消耗
DDoS 攻击
暴力破解
速率限制设计不当

5.1.2 API 安全常见稳妥做法¶

认证
强认证机制
OAuth2/JWT
多因素认证
授权
RBAC/ABAC
最小权限
权限审查
速率限制
防止滥用
保护资源
公平使用

5.2 API 网关¶

5.2.1 API 网关概述¶

API 网关是管理、保护和路由 API 请求的中间层。

API 网关功能¶

路由
请求路由
负载均衡
服务发现
安全
认证授权
速率限制
WAF 防护

5.2.2 API 网关实现¶

Python

from flask import Flask, request, jsonify
from functools import wraps
import time

app = Flask(__name__)

class APIGateway:
    """API网关类"""

    def __init__(self):
        self.rate_limits = {}
        self.blocked_ips = set()

    def rate_limit(self, max_requests=100, window=60):
        """速率限制装饰器"""
        def decorator(f):
            @wraps(f)
            def wrapper(*args, **kwargs):
                ip = request.remote_addr

                if ip in self.blocked_ips:
                    return jsonify({'error': 'IP blocked'}), 403

                current_time = time.time()

                if ip not in self.rate_limits:
                    self.rate_limits[ip] = []

                # 清理过期请求
                self.rate_limits[ip] = [
                    req_time for req_time in self.rate_limits[ip]
                    if current_time - req_time < window
                ]

                # 检查速率限制
                if len(self.rate_limits[ip]) >= max_requests:
                    return jsonify({'error': 'Rate limit exceeded'}), 429

                # 记录请求
                self.rate_limits[ip].append(current_time)

                return f(*args, **kwargs)

            return wrapper
        return decorator

    def block_ip(self, ip):
        """阻止IP"""
        self.blocked_ips.add(ip)

    def unblock_ip(self, ip):
        """解除阻止IP"""
        if ip in self.blocked_ips:
            self.blocked_ips.remove(ip)

# 使用API网关
gateway = APIGateway()

@app.route('/api/users', methods=['GET'])
@gateway.rate_limit(max_requests=10, window=60)
def get_users():
    """获取用户列表"""
    # 实际的业务逻辑
    return jsonify({'users': []})

@app.route('/api/users', methods=['POST'])
@gateway.rate_limit(max_requests=5, window=60)
def create_user():
    """创建用户"""
    # 实际的业务逻辑
    return jsonify({'message': 'User created'})

注：该示例使用进程内内存实现速率限制，只适合单进程教学演示。生产环境通常需要 Redis、网关产品或服务网格等集中式限流能力，并配合身份维度、租户维度和代理链真实来源识别。

5.3 速率限制¶

5.3.1 速率限制概述¶

速率限制是控制 API 请求频率的技术。

速率限制策略¶

固定窗口
固定时间窗口
简单实现
可能突发
滑动窗口
滑动时间窗口
更平滑
更复杂
令牌桶
令牌桶算法
平滑限流
广泛使用

5.3.2 速率限制实现¶

Python

import time
from collections import deque

class RateLimiter:
    """速率限制器"""

    def __init__(self, max_requests=100, window=60):
        self.max_requests = max_requests
        self.window = window
        self.requests = {}

    def is_allowed(self, identifier):
        """检查是否允许请求"""
        current_time = time.time()

        if identifier not in self.requests:
            self.requests[identifier] = deque()

        # 清理过期请求
        while self.requests[identifier] and current_time - self.requests[identifier][0] > self.window:
            self.requests[identifier].popleft()

        # 检查是否超过限制
        if len(self.requests[identifier]) >= self.max_requests:
            return False

        # 记录请求
        self.requests[identifier].append(current_time)
        return True

    def get_remaining(self, identifier):
        """获取剩余请求数"""
        if identifier not in self.requests:
            return self.max_requests

        return self.max_requests - len(self.requests[identifier])

    def get_retry_after(self, identifier):
        """获取重试时间"""
        if identifier not in self.requests or not self.requests[identifier]:
            return 0

        oldest_request = self.requests[identifier][0]
        current_time = time.time()
        return int(self.window - (current_time - oldest_request))

5.4 API 认证¶

5.4.1 API 认证方法¶

API Key
简单易用
适合内部 API
需要安全管理
OAuth 2.0 / OIDC
标准协议
适合第三方集成
复杂实现
JWT
无状态
适合分布式系统
需要密钥管理

5.4.2 API 认证实现¶

Python

from flask import Flask, request, jsonify, g
from datetime import datetime, timedelta, timezone
import jwt
from functools import wraps

app = Flask(__name__)
# ⚠️ 仅用于本地学习演示！生产环境必须从环境变量获取
import os
app.secret_key = os.environ.get('FLASK_SECRET_KEY', 'your-secret-key-FOR-DEV-ONLY')

class APIAuth:
    """API认证类"""

    def __init__(self, secret_key):
        self.secret_key = secret_key

    def generate_token(self, user_id):
        """生成JWT令牌"""
        payload = {
            'user_id': user_id,
            'iat': datetime.now(timezone.utc),
            'iss': 'demo-api',
            'aud': 'demo-client',
            'exp': datetime.now(timezone.utc) + timedelta(hours=24)
        }

        token = jwt.encode(payload, self.secret_key, algorithm='HS256')
        return token

    def verify_token(self, token):
        """验证JWT令牌"""
        try:  # try/except捕获异常
            payload = jwt.decode(
                token,
                self.secret_key,
                algorithms=['HS256'],
                audience='demo-client',
                issuer='demo-api'
            )
            return payload
        except jwt.ExpiredSignatureError:
            return None
        except jwt.InvalidTokenError:
            return None

    def require_auth(self, f):
        """认证装饰器"""
        @wraps(f)
        def wrapper(*args, **kwargs):  # *args接收任意位置参数；**kwargs接收任意关键字参数
            token = request.headers.get('Authorization')

            if not token:
                return jsonify({'error': 'Missing token'}), 401

            if not token.startswith('Bearer '):
                return jsonify({'error': 'Invalid token format'}), 401

            token = token.split(' ')[1]
            payload = self.verify_token(token)

            if not payload:
                return jsonify({'error': 'Invalid token'}), 401

            g.user = payload
            return f(*args, **kwargs)

        return wrapper

# 使用API认证
auth = APIAuth(app.secret_key)

@app.route('/api/login', methods=['POST'])
def login():
    """用户登录"""
    data = request.json
    # 实际的认证逻辑
    user_id = 1  # 假设认证成功

    token = auth.generate_token(user_id)

    return jsonify({'token': token})

@app.route('/api/protected', methods=['GET'])
@auth.require_auth
def protected():
    """受保护的API"""
    return jsonify({'message': 'Access granted', 'user_id': g.user['user_id']})

说明： JWT 不是“默认更安全”的同义词。是否使用 JWT、Opaque Token、Session Cookie 或 OAuth 2.0/OIDC，要结合调用方类型、撤销需求、密钥轮换、网关能力和合规要求综合评估。

5.5 API 安全测试¶

5.5.1 API 安全测试工具¶

Postman
API 测试
自动化测试
集合管理
OWASP ZAP
安全扫描
漏洞检测
报告生成

5.5.2 API 安全测试实践¶

Python

import requests
import json

class APISecurityTester:
    """API安全测试器"""

    def __init__(self, base_url):
        self.base_url = base_url
        self.vulnerabilities = []

    def test_authentication(self):
        """在授权测试环境中验证认证边界"""
        print("Testing authentication controls...")

        # 测试无认证
        response = requests.get(f"{self.base_url}/api/users")
        if response.status_code == 200:
            self.vulnerabilities.append({
                'type': 'Missing Authentication',
                'endpoint': '/api/users',
                'severity': 'High'
            })

        # 测试弱认证
        response = requests.get(
            f"{self.base_url}/api/users",
            headers={'Authorization': 'Bearer weak-token'}
        )
        if response.status_code == 200:
            self.vulnerabilities.append({
                'type': 'Weak Authentication',
                'endpoint': '/api/users',
                'severity': 'High'
            })

    def test_rate_limiting(self):
        """验证速率限制是否生效"""
        print("Testing rate limiting...")

        # 快速发送多个请求
        responses = []
        for i in range(20):
            response = requests.get(f"{self.base_url}/api/users")
            responses.append(response.status_code)

        # 检查是否被限制
        if 429 not in responses:
            self.vulnerabilities.append({
                'type': 'Missing Rate Limiting',
                'endpoint': '/api/users',
                'severity': 'Medium'
            })

    def test_input_validation(self):
        """使用脱敏样例验证输入处理是否稳健"""
        print("Testing input validation...")

        test_inputs = [
            "",
            "A" * 2048,
            "<test-marker>",
            "control-'\\\"-marker"
        ]

        for candidate in test_inputs:
            response = requests.get(
                f"{self.base_url}/api/search",
                params={'q': candidate}
            )

            if response.status_code == 200 and candidate in response.text:
                self.vulnerabilities.append({
                    'type': 'Insufficient Output Handling',
                    'endpoint': '/api/search',
                    'sample': candidate[:40],
                    'severity': 'Medium'
                })

    def generate_report(self):
        """生成报告"""
        return {
            'base_url': self.base_url,
            'total_vulnerabilities': len(self.vulnerabilities),
            'vulnerabilities': self.vulnerabilities
        }

注：安全测试应限定在授权的预发、实验室或专门靶场环境中。这里保留的是防御性验证思路，而不是攻击链教学；真实项目中还应补充越权、对象级授权、敏感字段暴露、日志脱敏与 OpenAPI 契约校验。

5.6 练习题¶

基础题¶

选择题
API 安全威胁不包括什么？
- A. 未授权访问
- B. 数据泄露
- C. 滥用攻击
- D. 性能优化
简答题
解释 API 安全的核心概念。
说明 API 网关的功能。

进阶题¶

实践题
搭建 API 网关。
实施速率限制。
实施 API 认证。
设计题
设计 API 安全架构。
设计 API 安全测试流程。

答案¶

1. 选择题答案¶

D （ API 安全威胁不包括性能优化）

2. 简答题答案¶

API 安全的核心概念： - 未授权访问、数据泄露、滥用攻击

API 网关的功能： - 路由、安全、监控

3. 实践题答案¶

参见 5.2-5.4 节的示例。

4. 设计题答案¶

参见 5.1-5.5 节的设计方案。

5.7 面试准备¶

关键复盘问题¶

字节跳动¶

解释 API 安全的核心概念。
API 网关的功能是什么？
如何实施速率限制？
如何设计 API 认证？

腾讯¶

API 安全有哪些常见稳妥做法？
如何设计 API 网关？
如何设计 API 授权？
如何设计 API 监控？

阿里云¶

API 安全的挑战是什么？
如何设计 API 安全架构？
如何处理 API 滥用？
如何设计 API 文档？

📚 参考资料¶

OWASP API Security ： https://owasp.org/www-project-api-security/
API Gateway 文档： https://docs.aws.amazon.com/apigateway/
《 API 安全实战》

🎯 本章小结¶

本章深入讲解了 API 安全，包括：

API 安全的核心概念
API 网关的使用
速率限制技术
API 认证方法

通过本章学习，你可以建立 API 安全的基本设计与验证框架。是否达到生产可落地水平，还需要结合网关产品、身份系统、密钥管理、日志体系与威胁建模继续完善。

⚠️ 核验说明（2026-04-03）：本页已于 2026-04-03 按网络安全专题完成复核。由于安全标准、项目版本、术语口径和威胁态势会持续变化，文中涉及 OWASP / NIST / CWE / 法规 / 工具的内容请以官方文档、授权范围和实际环境为准；高风险内容应以防御、检测、加固和合规学习为边界。

最后更新日期：2026-04-03