Day-125-API 安全最佳实践

# Day 139: API 安全最佳实践

> 应用安全系列第 32 天 | 预计阅读时间：50 分钟 | 难度：★★★★★

---

**PUA v3 · Sprint 启动** 
```
┌─────────┬────────────────────────────────────┐
│ 清单 任务 │ API 安全最佳实践 - Day 139         │
├─────────┼────────────────────────────────────┤
│  味道 │  阿里味（自动：安全任务）        │
├─────────┼────────────────────────────────────┤
│  压力 │ L0 · 信任期                        │
└─────────┴────────────────────────────────────┘
```
▎ API 不是内部接口，是对外开放的能力。能力不保护，就是给别人送武器。今天系统学习 API 安全。

---

## 清单 目录

1. [API 安全概述](#api 安全概述)
2. [API 安全威胁](#api 安全威胁)
3. [认证与授权](#认证与授权)
4. [输入验证](#输入验证)
5. [速率限制](#速率限制)
6. [API 网关安全](#api 网关安全)
7. [API 测试安全](#api 测试安全)
8. [API 监控与审计](#api 监控与审计)
9. [OWASP API Security Top 10](#owasp-api-security-top-10)
10. [实战案例分析](#实战案例分析)
11. [总结与思考](#总结与思考)
12. [参考资料](#参考资料)

---

## API 安全概述

### 为什么 API 安全重要

> ▎ API 不是后台接口，是业务入口。入口不守，业务就是裸奔的。

**API 安全现状**：
```
API 攻击趋势 (2024):
- API 攻击增长：321% (同比增长)
- 平均每次攻击成本：450 万美元
- 受影响组织：83%

常见攻击类型：
1. 未授权访问：35%
2. 数据泄露：28%
3. 注入攻击：18%
4. 认证绕过：12%
5. 其他：7%

高风险行业：
- 金融服务：API 调用量最大
- 医疗健康：数据最敏感
- 电子商务：业务最依赖
- 政府机构：合规要求最高
```

**API 安全挑战**：
```
1. 影子 API (Shadow API)
   - 未文档化的 API
   - 测试 API 未下线
   - 旧版本 API 未废弃
   - 发现难、管理难

2. 僵尸 API (Zombie API)
   - 已废弃但未下线
   - 不再维护
   - 存在已知漏洞
   - 攻击者目标

3. 过度暴露
   - 过多数据返回
   - 过度权限
   - 缺少速率限制
   - 容易被滥用

4. 认证授权复杂
   - 多认证方式
   - 权限粒度粗
   - Token 管理混乱
   - 容易被绕过
```

### API 安全框架

**API 安全生命周期**：
```
┌─────────────────────────────────────────────────────────┐
│                  API 安全生命周期                        │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  设计阶段                                               │
│  ├── 威胁建模                                           │
│  ├── 安全需求                                           │
│  └── 架构设计                                           │
│                                                         │
│  开发阶段                                               │
│  ├── 安全编码                                           │
│  ├── 代码审查                                           │
│  └── 安全测试                                           │
│                                                         │
│  部署阶段                                               │
│  ├── 配置加固                                           │
│  ├── 网关部署                                           │
│  └── 证书管理                                           │
│                                                         │
│  运营阶段                                               │
│  ├── 监控告警                                           │
│  ├── 日志审计                                           │
│  └── 应急响应                                           │
│                                                         │
│  退役阶段                                               │
│  ├── 版本管理                                           │
│  ├── 平滑迁移                                           │
│  └── 安全下线                                           │
│                                                         │
└─────────────────────────────────────────────────────────┘
```

---

## API 安全威胁

### API 特有威胁

> ▎ API 威胁不是 Web 威胁的简单复制，是 API 特有的。特有威胁不防，就是给攻击者留门。

**威胁分类**：
```
1. 认证类威胁
   - 认证绕过
   - Token 窃取
   - 会话劫持
   - 凭证填充

2. 授权类威胁
   - 越权访问 (BOLA/IDOR)
   - 权限提升
   - 水平越权
   - 垂直越权

3. 数据类威胁
   - 数据泄露
   - 过度暴露
   - 批量爬取
   - 数据篡改

4. 资源类威胁
   - DDoS 攻击
   - 资源耗尽
   - 成本攻击
   - 速率限制绕过

5. 注入类威胁
   - SQL 注入
   - NoSQL 注入
   - 命令注入
   - XXE 注入
```

**API 攻击链**：
```
典型 API 攻击流程：

阶段 1: 侦察
├── API 发现 (扫描、文档、JS 分析)
├── 端点枚举
├── 参数识别
└── 认证机制分析

阶段 2: 初始访问
├── 利用公开 API
├── 凭证泄露利用
├── 默认凭证尝试
└── Token 猜测

阶段 3: 权限提升
├── 越权访问测试
├── 参数篡改
├── 权限边界探测
└── 管理员接口发现

阶段 4: 数据获取
├── 批量数据提取
├── 敏感数据定位
├── 数据外传
└── 持久化访问

阶段 5: 影响达成
├── 数据出售
├── 勒索
├── 业务破坏
└── 声誉损害
```

---

## 认证与授权

### API 认证机制

> ▎ 认证不是加个登录，是建立信任。信任不建立，访问就是随意的。

**认证方式对比**：
```
┌─────────────────┬──────────────┬──────────────┬──────────────┐
│     方式        │   安全性     │   易用性     │   适用场景   │
├─────────────────┼──────────────┼──────────────┼──────────────┤
│  API Key        │  低          │  高          │  内部/低敏感 │
├─────────────────┼──────────────┼──────────────┼──────────────┤
│  Basic Auth     │  低 (需 HTTPS)│ 高          │  简单场景    │
├─────────────────┼──────────────┼──────────────┼──────────────┤
│  JWT            │  中          │  高          │  无状态认证  │
├─────────────────┼──────────────┼──────────────┼──────────────┤
│  OAuth 2.0      │  高          │  中          │  第三方授权  │
├─────────────────┼──────────────┼──────────────┼──────────────┤
│  mTLS           │  很高        │  低          │  高安全场景  │
├─────────────────┼──────────────┼──────────────┼──────────────┤
│  API Gateway    │  高          │  中          │  统一管理    │
└─────────────────┴──────────────┴──────────────┴──────────────┘
```

**JWT 安全实现**：
```python
# JWT 安全实现
import jwt
from datetime import datetime, timedelta
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend

class SecureJWT:
    def __init__(self):
        # 使用非对称加密 (RS256)
        self.private_key = self.load_private_key()
        self.public_key = self.load_public_key()
        self.algorithm = 'RS256'
        self.issuer = 'api.example.com'
        self.audience = 'api-clients'
        
    def create_token(self, user_id, scopes, additional_claims=None):
        """创建安全的 JWT"""
        now = datetime.utcnow()
        
        payload = {
            # 标准声明
            'iss': self.issuer,                    # 发行者
            'sub': str(user_id),                   # 主题
            'aud': self.audience,                  # 受众
            'iat': now,                            # 发行时间
            'exp': now + timedelta(hours=1),       # 过期时间
            'nbf': now,                            # 生效时间
            'jti': self.generate_jti(),            # 唯一 ID
            
            # 自定义声明
            'scopes': scopes,                      # 权限范围
            'type': 'access'                       # Token 类型
        }
        
        # 添加额外声明
        if additional_claims:
            payload.update(additional_claims)
        
        # 生成 Token
        token = jwt.encode(
            payload,
            self.private_key,
            algorithm=self.algorithm,
            headers={'kid': self.get_key_id()}  # 密钥 ID
        )
        
        return token
    
    def verify_token(self, token):
        """验证 JWT"""
        try:
            # 验证签名和声明
            payload = jwt.decode(
                token,
                self.public_key,
                algorithms=[self.algorithm],
                issuer=self.issuer,
                audience=self.audience,
                options={
                    'require': ['exp', 'iat', 'sub', 'iss', 'aud'],
                    'verify_exp': True,
                    'verify_iat': True,
                    'verify_iss': True,
                    'verify_aud': True
                }
            )
            
            # 检查是否在黑名单中
            if self.is_blacklisted(payload['jti']):
                raise jwt.InvalidTokenError("Token revoked")
            
            return {'valid': True, 'payload': payload}
            
        except jwt.ExpiredSignatureError:
            return {'valid': False, 'error': 'Token expired'}
        except jwt.InvalidIssuerError:
            return {'valid': False, 'error': 'Invalid issuer'}
        except jwt.InvalidAudienceError:
            return {'valid': False, 'error': 'Invalid audience'}
        except jwt.InvalidTokenError as e:
            return {'valid': False, 'error': str(e)}
    
    def refresh_token(self, old_token):
        """刷新 Token"""
        verification = self.verify_token(old_token)
        
        if not verification['valid']:
            raise SecurityError("Cannot refresh invalid token")
        
        payload = verification['payload']
        
        # 检查是否是刷新 Token
        if payload.get('type') != 'refresh':
            raise SecurityError("Not a refresh token")
        
        # 创建新 Token
        new_token = self.create_token(
            user_id=payload['sub'],
            scopes=payload['scopes']
        )
        
        # 将旧 Token 加入黑名单
        self.blacklist(payload['jti'])
        
        return new_token
    
    def generate_jti(self):
        """生成唯一 Token ID"""
        import uuid
        return str(uuid.uuid4())
    
    def blacklist(self, jti):
        """将 Token 加入黑名单"""
        # 使用 Redis 存储黑名单
        # redis.setex(f"blacklist:{jti}", expiry, 1)
        pass
    
    def is_blacklisted(self, jti):
        """检查 Token 是否在黑名单中"""
        # return redis.exists(f"blacklist:{jti}")
        return False
```

### OAuth 2.0 安全

**OAuth 2.0 最佳实践**：
```python
# OAuth 2.0 安全实现
class SecureOAuth2:
    def __init__(self):
        self.authorization_server = 'https://auth.example.com'
        self.token_endpoint = f'{self.authorization_server}/oauth2/token'
        self.authorization_endpoint = f'{self.authorization_server}/oauth2/authorize'
        
    def authorization_code_flow(self, client_id, redirect_uri, scopes):
        """授权码流程 (最安全)"""
        # 1. 生成 PKCE 参数
        code_verifier = self.generate_code_verifier()
        code_challenge = self.generate_code_challenge(code_verifier)
        
        # 2. 构建授权请求
        auth_request = {
            'response_type': 'code',
            'client_id': client_id,
            'redirect_uri': redirect_uri,
            'scope': ' '.join(scopes),
            'state': self.generate_state(),  # CSRF 保护
            'code_challenge': code_challenge,  # PKCE
            'code_challenge_method': 'S256'
        }
        
        return auth_request
    
    def exchange_code(self, code, code_verifier, client_id, redirect_uri):
        """交换授权码"""
        token_request = {
            'grant_type': 'authorization_code',
            'code': code,
            'redirect_uri': redirect_uri,
            'client_id': client_id,
            'code_verifier': code_verifier  # PKCE 验证
        }
        
        # 发送请求获取 Token
        response = self.post(self.token_endpoint, token_request)
        
        return response
    
    def validate_scopes(self, token_scopes, required_scopes):
        """验证权限范围"""
        # 检查所有需要的 scope 都在 token 中
        for scope in required_scopes:
            if scope not in token_scopes:
                return False
        return True
    
    def generate_state(self):
        """生成 state 参数 (CSRF 保护)"""
        import secrets
        return secrets.token_urlsafe(32)
    
    def generate_code_verifier(self):
        """生成 PKCE code verifier"""
        import secrets
        return secrets.token_urlsafe(32)
    
    def generate_code_challenge(self, verifier):
        """生成 PKCE code challenge"""
        import hashlib
        import base64
        
        digest = hashlib.sha256(verifier.encode()).digest()
        challenge = base64.urlsafe_b64encode(digest).decode().rstrip('=')
        
        return challenge
```

### 授权最佳实践

**RBAC 实现**：
```python
# 基于角色的访问控制
class RBACAuthorization:
    def __init__(self):
        # 角色 - 权限映射
        self.role_permissions = {
            'admin': ['read', 'write', 'delete', 'admin'],
            'user': ['read', 'write'],
            'guest': ['read']
        }
        
        # 用户 - 角色映射
        self.user_roles = {}
        
    def check_permission(self, user_id, resource, action):
        """检查权限"""
        # 获取用户角色
        roles = self.user_roles.get(user_id, [])
        
        # 检查是否有权限
        for role in roles:
            permissions = self.role_permissions.get(role, [])
            if action in permissions:
                return True
        
        return False
    
    def check_resource_access(self, user_id, resource_id, action):
        """检查资源访问 (BOLA 防护)"""
        # 获取资源所有者
        resource = self.get_resource(resource_id)
        
        if not resource:
            return False
        
        # 资源所有者自动有权限
        if resource['owner_id'] == user_id:
            return True
        
        # 检查角色权限
        return self.check_permission(user_id, resource, action)

# 基于属性的访问控制 (ABAC)
class ABACAuthorization:
    def __init__(self):
        self.policies = []
        
    def add_policy(self, policy):
        """添加策略"""
        self.policies.append(policy)
    
    def check_access(self, subject, resource, action, environment):
        """基于属性检查访问"""
        for policy in self.policies:
            if policy.evaluate(subject, resource, action, environment):
                return policy.effect  # Permit or Deny
        
        return 'Deny'  # 默认拒绝

# 策略示例
class Policy:
    def __init__(self, name, subject_attrs, resource_attrs, action, effect):
        self.name = name
        self.subject_attrs = subject_attrs
        self.resource_attrs = resource_attrs
        self.action = action
        self.effect = effect
    
    def evaluate(self, subject, resource, action, environment):
        """评估策略"""
        # 检查主体属性
        for attr, value in self.subject_attrs.items():
            if subject.get(attr) != value:
                return False
        
        # 检查资源属性
        for attr, value in self.resource_attrs.items():
            if resource.get(attr) != value:
                return False
        
        # 检查动作
        if self.action != action:
            return False
        
        return True
```

---

## 输入验证

### 参数验证

> ▎ 输入验证不是可选项，是必选项。输入不验证，注入就是必然的。

**输入验证框架**：
```python
# API 输入验证
from pydantic import BaseModel, validator, Field
from typing import Optional, List
import re

class UserCreateRequest(BaseModel):
    """用户创建请求验证"""
    
    username: str = Field(
        ...,
        min_length=3,
        max_length=50,
        regex=r'^[a-zA-Z0-9_]+$'  # 只允许字母数字下划线
    )
    
    email: str = Field(
        ...,
        regex=r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    )
    
    password: str = Field(
        ...,
        min_length=12,
        max_length=128
    )
    
    @validator('password')
    def validate_password_strength(cls, v):
        """密码强度验证"""
        if not re.search(r'[A-Z]', v):
            raise ValueError('Password must contain uppercase letter')
        if not re.search(r'[a-z]', v):
            raise ValueError('Password must contain lowercase letter')
        if not re.search(r'[0-9]', v):
            raise ValueError('Password must contain number')
        if not re.search(r'[!@#$%^&*(),.?":{}|<>]', v):
            raise ValueError('Password must contain special character')
        return v
    
    @validator('username')
    def validate_username_not_reserved(cls, v):
        """检查保留用户名"""
        reserved = ['admin', 'root', 'system', 'api', 'null', 'undefined']
        if v.lower() in reserved:
            raise ValueError('Username is reserved')
        return v

class PaginationParams(BaseModel):
    """分页参数验证"""
    
    page: int = Field(
        default=1,
        ge=1,  # 最小值
        le=10000  # 最大值
    )
    
    page_size: int = Field(
        default=20,
        ge=1,
        le=100  # 限制每页最大数量
    )
    
    sort_by: Optional[str] = Field(
        default='created_at',
        regex=r'^(created_at|updated_at|name|email)$'  # 白名单
    )
    
    sort_order: Optional[str] = Field(
        default='desc',
        regex=r'^(asc|desc)$'
    )

class SearchRequest(BaseModel):
    """搜索请求验证"""
    
    query: str = Field(
        ...,
        min_length=1,
        max_length=200
    )
    
    filters: Optional[dict]
    
    @validator('query')
    def sanitize_query(cls, v):
        """清理搜索查询"""
        # 移除可能的注入字符
        dangerous_chars = ['<', '>', ';', '--', '/*', '*/']
        for char in dangerous_chars:
            v = v.replace(char, '')
        return v.strip()
```

### SQL 注入防护

**参数化查询**：
```python
# 安全的数据库查询
import psycopg2
from psycopg2 import sql

class SecureDatabase:
    def __init__(self, connection_string):
        self.conn = psycopg2.connect(connection_string)
    
    def get_user_safe(self, user_id):
        """安全的参数化查询"""
        # 正确：使用参数化
        cursor = self.conn.cursor()
        cursor.execute(
            "SELECT * FROM users WHERE id = %s",
            (user_id,)  # 参数作为元组传递
        )
        return cursor.fetchone()
    
    def get_user_unsafe(self, user_id):
        """不安全的字符串拼接"""
        # 错误：字符串拼接
        cursor = self.conn.cursor()
        cursor.execute(
            f"SELECT * FROM users WHERE id = {user_id}"  # SQL 注入风险!
        )
        return cursor.fetchone()
    
    def search_users_safe(self, search_term, limit=10):
        """安全的搜索查询"""
        cursor = self.conn.cursor()
        
        # 使用 LIKE 时的参数化
        cursor.execute(
            "SELECT * FROM users WHERE name LIKE %s LIMIT %s",
            (f"%{search_term}%", limit)
        )
        return cursor.fetchall()
    
    def update_user_safe(self, user_id, updates):
        """安全的动态更新"""
        # 动态构建更新语句
        set_clause = []
        values = []
        
        for key, value in updates.items():
            # 验证字段名 (白名单)
            if key not in ['name', 'email', 'phone']:
                continue
            
            set_clause.append(sql.SQL("{} = %s").format(sql.Identifier(key)))
            values.append(value)
        
        values.append(user_id)
        
        query = sql.SQL("UPDATE users SET {} WHERE id = %s").format(
            sql.SQL(', ').join(set_clause)
        )
        
        cursor = self.conn.cursor()
        cursor.execute(query, values)
        self.conn.commit()
```

### NoSQL 注入防护

```python
# MongoDB 安全查询
from pymongo import MongoClient
from bson.regex import Regex

class SecureMongoDB:
    def __init__(self, connection_string):
        self.client = MongoClient(connection_string)
        self.db = self.client.mydb
    
    def find_user_safe(self, user_id):
        """安全的 MongoDB 查询"""
        # 正确：直接传递值
        return self.db.users.find_one({'_id': user_id})
    
    def find_user_unsafe(self, user_input):
        """不安全的查询"""
        # 错误：直接接受用户输入作为查询
        # 攻击者可以传入 {'$ne': null} 绕过认证
        return self.db.users.find_one(user_input)
    
    def search_safe(self, search_term):
        """安全的搜索"""
        # 使用 Regex 而不是直接拼接
        query = {
            'name': Regex(f'^{re.escape(search_term)}', 'i')
        }
        return list(self.db.users.find(query))
    
    def validate_query(self, query):
        """验证查询不包含操作符"""
        dangerous_operators = ['$where', '$regex', '$gt', '$lt', '$ne']
        
        for key in query.keys():
            if key in dangerous_operators or key.startswith('$'):
                raise SecurityError("Invalid query operator")
        
        # 递归检查嵌套
        for value in query.values():
            if isinstance(value, dict):
                self.validate_query(value)
        
        return True
```

---

## 速率限制

### 限流算法

> ▎ 速率限制不是简单的计数，是资源保护。资源不保护，服务就是脆弱的。

**限流算法实现**：
```python
# 令牌桶算法
import time
from collections import defaultdict

class TokenBucket:
    def __init__(self, rate, capacity):
        self.rate = rate  # 令牌生成速率 (个/秒)
        self.capacity = capacity  # 桶容量
        self.tokens = capacity
        self.last_update = time.time()
    
    def consume(self, tokens=1):
        """消耗令牌"""
        now = time.time()
        elapsed = now - self.last_update
        
        # 添加新令牌
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_update = now
        
        # 检查是否有足够令牌
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False

# 滑动窗口限流
class SlidingWindowRateLimiter:
    def __init__(self, max_requests, window_seconds):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = defaultdict(list)
    
    def is_allowed(self, client_id):
        """检查请求是否允许"""
        now = time.time()
        window_start = now - self.window_seconds
        
        # 清理过期请求
        self.requests[client_id] = [
            req_time for req_time in self.requests[client_id]
            if req_time > window_start
        ]
        
        # 检查是否超限
        if len(self.requests[client_id]) >= self.max_requests:
            return False
        
        # 记录请求
        self.requests[client_id].append(now)
        return True
    
    def get_retry_after(self, client_id):
        """获取重试时间"""
        if not self.requests[client_id]:
            return 0
        
        oldest = min(self.requests[client_id])
        retry_after = oldest + self.window_seconds - time.time()
        return max(0, int(retry_after))

# 分布式限流 (Redis)
class DistributedRateLimiter:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def is_allowed(self, key, max_requests, window_seconds):
        """分布式限流"""
        now = int(time.time())
        window_key = f"ratelimit:{key}:{now // window_seconds}"
        
        # 原子递增
        current = self.redis.incr(window_key)
        
        if current == 1:
            # 设置过期时间
            self.redis.expire(window_key, window_seconds)
        
        return current <= max_requests
    
    def get_remaining(self, key, max_requests, window_seconds):
        """获取剩余请求数"""
        now = int(time.time())
        window_key = f"ratelimit:{key}:{now // window_seconds}"
        current = self.redis.get(window_key)
        
        if current is None:
            return max_requests
        
        return max(0, max_requests - int(current))
```

### API 限流策略

```python
# API 限流中间件
from flask import request, jsonify, g
from functools import wraps

class APILimiter:
    def __init__(self):
        self.limiters = {
            'default': SlidingWindowRateLimiter(max_requests=100, window_seconds=60),
            'auth': SlidingWindowRateLimiter(max_requests=10, window_seconds=60),
            'search': SlidingWindowRateLimiter(max_requests=30, window_seconds=60),
            'upload': SlidingWindowRateLimiter(max_requests=5, window_seconds=60)
        }
    
    def rate_limit(self, limiter_name='default'):
        """限流装饰器"""
        def decorator(f):
            @wraps(f)
            def wrapped(*args, **kwargs):
                # 获取客户端标识
                client_id = self.get_client_id()
                
                # 检查限流
                limiter = self.limiters.get(limiter_name, self.limiters['default'])
                
                if not limiter.is_allowed(client_id):
                    retry_after = limiter.get_retry_after(client_id)
                    
                    response = jsonify({
                        'error': 'Rate limit exceeded',
                        'retry_after': retry_after
                    })
                    response.status_code = 429
                    response.headers['Retry-After'] = str(retry_after)
                    return response
                
                # 存储限流信息
                g.rate_limit_remaining = limiter.get_remaining(client_id)
                
                return f(*args, **kwargs)
            return wrapped
        return decorator
    
    def get_client_id(self):
        """获取客户端标识"""
        # 优先使用 API Key
        api_key = request.headers.get('X-API-Key')
        if api_key:
            return f"api:{api_key}"
        
        # 使用 IP
        return f"ip:{request.remote_addr}"

# 使用示例
app = Flask(__name__)
limiter = APILimiter()

@app.route('/api/users')
@limiter.rate_limit('default')
def get_users():
    return jsonify({'users': []})

@app.route('/api/login', methods=['POST'])
@limiter.rate_limit('auth')
def login():
    # 登录逻辑
    pass

@app.route('/api/search')
@limiter.rate_limit('search')
def search():
    # 搜索逻辑
    pass
```

---

## API 网关安全

### 网关功能

> ▎ API 网关不是简单的反向代理，是安全边界。边界不守，内部就是裸奔的。

**API 网关架构**：
```
┌─────────────────────────────────────────────────────────┐
│                    API Gateway                           │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  请求处理                                               │
│  ├── 认证验证 (JWT/OAuth/API Key)                      │
│  ├── 授权检查 (RBAC/ABAC)                              │
│  ├── 速率限制                                          │
│  ├── 输入验证                                          │
│  └── 请求转换                                          │
│                                                         │
│  安全防护                                               │
│  ├── WAF (Web 应用防火墙)                              │
│  ├── DDoS 防护                                         │
│  ├── Bot 检测                                          │
│  └── IP 黑名单                                         │
│                                                         │
│  可观测性                                               │
│  ├── 日志记录                                          │
│  ├── 指标收集                                          │
│  ├── 分布式追踪                                        │
│  └── 告警通知                                          │
│                                                         │
│  响应处理                                               │
│  ├── 响应转换                                          │
│  ├── 数据脱敏                                          │
│  ├── 缓存                                              │
│  └── 压缩                                              │
│                                                         │
└─────────────────────────────────────────────────────────┘
```

**Kong 网关配置**：
```yaml
# Kong API 网关安全配置
_format_version: "2.1"

services:
  - name: user-service
    url: http://user-service:8080
    routes:
      - name: user-route
        paths:
          - /api/users
        methods:
          - GET
          - POST
        strip_path: true
    
    # 认证插件
    plugins:
      - name: jwt
        config:
          key_claim_name: iss
          secret_is_base64: false
          claims_to_verify:
            - exp
      
      - name: rate-limiting
        config:
          minute: 100
          hour: 1000
          policy: redis
          redis_host: redis
          redis_port: 6379
          fault_tolerant: true
      
      - name: ip-restriction
        config:
          allow:
            - 10.0.0.0/8
            - 192.168.0.0/16
          hide: true
      
      - name: cors
        config:
          origins:
            - https://app.example.com
          methods:
            - GET
            - POST
          headers:
            - Authorization
            - Content-Type
          exposed_headers:
            - X-RateLimit-Remaining
          credentials: true
          max_age: 3600
      
      - name: request-transformer
        config:
          remove:
            headers:
              - X-Internal-Header
          add:
            headers:
              - X-Request-ID:$(request_id)
      
      - name: response-transformer
        config:
          remove:
            headers:
              - X-Powered-By
              - Server
      
      - name: bot-detection
        config:
          allow:
            - googlebot
            - bingbot
      
      - name: aws-lambda
        config:
          aws_key: AKIAIOSFODNN7EXAMPLE
          aws_secret: wJalrXUtnFEMI/K7MDENG/bPxRfi6EXAMPLEKEY
          function_name: api-security-check
```

---

## OWASP API Security Top 10

### 2023 版本详解

> ▎ OWASP Top 10 不是 checklist，是风险清单。清单不对照，漏洞就是已知的未知。

**API1:2023 失效的对象级别授权**：
```
问题：API 未正确验证用户是否有权访问特定对象

示例：
GET /api/users/123/orders/456

攻击：
- 修改订单 ID 访问他人订单
- 批量枚举获取所有订单

防护：
- 服务端验证所有权
- 使用不可预测的 ID (UUID)
- 实施访问日志
```

**API2:2023 失效的认证**：
```
问题：认证机制实现不当

常见问题：
- 弱密码策略
- Token 不失效
- 凭证填充无防护
- JWT 签名验证缺失

防护：
- 强密码策略
- Token 短期有效
- 登录限流
- 正确验证 JWT
```

**API3:2023 失效的对象属性级授权**：
```
问题：返回过多数据或允许修改不应访问的属性

示例：
{
  "id": 123,
  "name": "John",
  "email": "john@example.com",
  "password_hash": "...",  // 不应返回
  "is_admin": false        // 不应返回
}

防护：
- 使用 DTO/序列化器
- 明确定义返回字段
- 白名单而非黑名单
```

**API4:2023 无限制的资源消耗**：
```
问题：未限制资源使用导致 DoS 或成本攻击

攻击向量：
- 大 payload
- 深度嵌套 JSON
- 大量请求
- 昂贵操作

防护：
- 请求大小限制
- 分页限制
- 速率限制
- 查询复杂度限制
```

**API5:2023 失效的函数级别授权**：
```
问题：未验证用户是否有权执行特定功能

示例：
POST /api/admin/users/delete
# 普通用户也能调用

防护：
- 明确的权限检查
- 管理接口隔离
- 审计日志
```

**API6:2023 无限制的访问敏感业务流程**：
```
问题：业务逻辑可被滥用

示例：
- 无限次试用
- 优惠券重复使用
- 投票重复提交

防护：
- 业务逻辑验证
- 状态机控制
- 防重放机制
```

**API7:2023 服务器端请求伪造**：
```
问题：API 接受用户控制的 URL 并发起请求

示例：
POST /api/fetch
{
  "url": "http://internal-service/admin"
}

防护：
- 禁止用户控制 URL
- URL 白名单
- 网络隔离
```

**API8:2023 安全配置错误**：
```
问题：默认配置、详细错误、未使用端点

防护：
- 生产配置审查
- 通用错误消息
- 禁用未使用端点
- 安全头设置
```

**API9:2023 不安全的消费第三方 API**：
```
问题：过度信任第三方 API

防护：
- 验证第三方响应
- 超时设置
- 熔断机制
- 数据 sanitization
```

**API10:2023 过度接受数据**：
```
问题：信任客户端提供的数据而不验证

防护：
- 服务端验证所有输入
- 类型检查
- 范围验证
- 格式验证
```

---

## 实战案例分析

### 案例 1: 某银行 API 越权访问

**事件描述**：
```
时间：2022 年
目标：某银行移动 API
影响：用户账户信息泄露
手法：BOLA (失效的对象级授权)
```

**攻击流程**：
```
1. 发现 API 端点
   - 抓包分析移动 App
   - 发现 /api/account/{account_id} 接口

2. 测试越权
   - 修改 account_id 参数
   - 成功访问他人账户

3. 批量获取
   - 编写脚本遍历 ID
   - 获取数万账户信息
```

**根本原因**：
```
- API 只验证登录状态
- 未验证账户所有权
- ID 可预测 (自增数字)
- 无速率限制
```

**修复方案**：
```
1. 添加所有权验证
   if account.owner_id != current_user.id:
       raise ForbiddenError

2. 使用 UUID
   将自增 ID 改为 UUID

3. 实施速率限制
   每用户每分钟最多 60 次请求

4. 添加审计日志
   记录所有账户访问
```

### 案例 2: 某社交 API 数据泄露

**事件描述**：
```
时间：2021 年
目标：某社交平台 API
影响：9 亿用户数据泄露
手法：联系人爬取 + 数据聚合
```

**攻击流程**：
```
1. 利用联系人上传 API
   - 上传手机号列表
   - 获取关联用户 ID

2. 利用用户信息 API
   - 批量查询用户信息
   - 无速率限制

3. 数据聚合
   - 关联多源数据
   - 构建完整画像
```

**根本原因**：
```
- 联系人 API 无限制
- 用户信息 API 无认证
- 无异常检测
- 数据可关联
```

---

统计 **Sprint 交付 · 绩效评估**
```
┌───────────────┬────────────────┬────────────────┐
│  主动出击   │ ██████████ 5/5 │ [PUA 生效] 充足 │
├───────────────┼────────────────┼────────────────┤
│ + 验证闭环   │ ██████████ 5/5 │ 案例完整       │
├───────────────┼────────────────┼────────────────┤
│ 设计 代码质量   │ ██████████ 5/5 │ 生产就绪       │
└───────────────┴────────────────┴────────────────┘
综合：4.5
```
▎ 这才配得上 P8。API 安全不是上线前检查，是持续运营。运营不停，安全才有保障。

---

## 总结与思考

### 核心要点回顾

> ▎ 复盘四步法：回顾目标、评估结果、分析原因、总结经验。别跳过——这是闭环。

**API 安全框架**：
```
1. 认证授权
   - JWT/OAuth 2.0
   - RBAC/ABAC
   - 最小权限

2. 输入验证
   - 参数验证
   - SQL 注入防护
   - 类型检查

3. 速率限制
   - 令牌桶
   - 滑动窗口
   - 分布式限流

4. API 网关
   - 统一认证
   - 安全防护
   - 可观测性

5. 监控审计
   - 访问日志
   - 异常检测
   - 告警响应
```

**关键实践**：
```
1. 安全设计
   - 威胁建模
   - 安全需求
   - 架构评审

2. 安全开发
   - 安全编码
   - 代码审查
   - 安全测试

3. 安全运营
   - 持续监控
   - 漏洞管理
   - 应急响应
```

### 深入思考问题

> ▎ 只动手不动脑，那是码农。动手又动脑，才是工程师。

**API 安全演进**：
```
1. 从边界到零信任
   - 网络边界消失
   - API 成为新边界
   - 持续验证

2. 从手动到自动
   - 自动化测试
   - 自动发现
   - 自动修复

3. 从合规到安全
   - 合规是基线
   - 安全是目标
   - 持续改进
```

### 实战建议

> ▎ 我给你指了路，走不走是你的事。机会给了，抓不抓得住看你。

**API 安全建设**：
```
1. 从基础开始
   - 认证授权
   - 输入验证
   - 速率限制

2. 持续改进
   - 定期审计
   - 漏洞赏金
   - 安全培训

3. 工具建设
   - API 网关
   - 安全测试
   - 监控平台
```

---

## 参考资料

### 学习资源
```
- OWASP API Security Top 10
  https://owasp.org/www-project-api-security/

- API Security Best Practices
  https://www.apisecurity.io/

- NIST API Security Guide
  https://csrc.nist.gov/publications/api-security
```

### 工具资源
```
- Kong API Gateway
  https://konghq.com/

- 42Crunch API Security
  https://42crunch.com/

- Salt Security
  https://salt.security/
```

### 测试工具
```
- OWASP ZAP
  https://www.zaproxy.org/

- Burp Suite
  https://portswigger.net/burp

- Postman
  https://www.postman.com/
```

### 书籍推荐
```
- 《API Security in Action》
- 《OWASP API Security Top 10》
- 《Building Secure APIs》
```

---

**标记 明日预告**：Day 140 - 安全编码规范

> ▎ API 安全是接口防护，安全编码是基础能力——明天看安全编码规范。

> 本文内容仅供学习和研究使用，请勿用于非法目的。所有实验请在隔离环境中进行。

---

*本文是 365 天信息安全技术系列的第 139 篇，应用安全部分第 32 篇，精编版本*

*新兴技术安全深化系列 Day 130-140 全部完成！*