implementing-api-rate-limiting-and-throttling

name	implementing-api-rate-limiting-and-throttling
description	使用令牌桶、滑动窗口和固定窗口算法实施API速率限制和节流控制，防止暴力破解、凭据填充、 资源耗尽和API滥用。使用Redis支持的计数器、API网关插件或应用中间件配置按用户、按IP和按端点的速率限制， 并实施带Retry-After头的正确HTTP 429响应。
domain	cybersecurity
subdomain	api-security
tags	["api-security","rate-limiting","throttling","redis","token-bucket","abuse-prevention"]
version	1.0.0
author	mahipal
license	Apache-2.0

实施API速率限制和节流

适用场景

• 保护认证端点免受暴力破解和凭据填充攻击
• 防止来自自动化脚本和机器人的API滥用和资源耗尽
• 为不同的API消费者层级（免费、高级、企业）实施公平使用配额
• 在应用层防御拒绝服务攻击
• 满足要求实施API滥用防护控制的合规要求

不适用将速率限制作为抵御攻击的唯一防御。需与认证、授权和WAF规则结合使用。

前置条件

• Redis 6.0+用于分布式速率限制计数器（或单实例部署的内存存储）
• API框架（Express.js、FastAPI、Spring Boot或Django REST Framework）
• 速率限制指标监控系统（Prometheus、CloudWatch、Datadog）
• 了解API的正常流量模式和峰值使用量
• 负载测试工具（k6、Gatling或Locust）用于验证速率限制行为

工作流程

步骤1：速率限制策略设计

按端点类别和用户层级定义速率限制：

# 速率限制配置
RATE_LIMITS = {
    # 认证端点（最严格）
    "auth": {
        "login": {"requests": 5, "window_seconds": 60, "by": "ip"},
        "register": {"requests": 3, "window_seconds": 300, "by": "ip"},
        "forgot_password": {"requests": 3, "window_seconds": 3600, "by": "ip"},
        "verify_mfa": {"requests": 5, "window_seconds": 300, "by": "user"},
    },
    # 标准API端点
    "api": {
        "free": {"requests": 60, "window_seconds": 60, "by": "user"},
        "premium": {"requests": 300, "window_seconds": 60, "by": "user"},
        "enterprise": {"requests": 1000, "window_seconds": 60, "by": "user"},
    },
    
    : {
        : {: , : , : },
        : {: , : , : },
        : {: , : , : },
    },
    
    : {
        : {: , : , : },
        : {: , : , : },
    },
}

步骤2：滑动窗口速率限制器（Redis）

import redis
import time
import hashlib
from functools import wraps
from flask import Flask, request, jsonify, g

app = Flask(__name__)
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

class SlidingWindowRateLimiter:
    """使用Redis有序集合的滑动窗口速率限制器。"""

    def __init__(self, redis_conn):
        self.redis = redis_conn

    def is_allowed(self, key, max_requests, window_seconds):
        """检查请求是否被允许并记录它。"""
        now = time.time()
        window_start = now - window_seconds
        pipe = self.redis.pipeline()

        # 删除过期条目
        pipe.zremrangebyscore(key, 0, window_start)
        # 计算当前窗口内的请求数
        pipe.zcard(key)
        # 添加当前请求
        pipe.zadd(key, {f"{now}:{hashlib.md5(str(now).encode()).hexdigest()[:8]}": now})
        # 设置键的TTL
        pipe.expire(key, window_seconds + 1)

        results = pipe.execute()
        current_count = results[1]

        if current_count >= max_requests:
            # 计算重试等待时间
            oldest = self.redis.zrange(key, 0, 0, withscores=True)
            if oldest:
                retry_after = int(oldest[][] + window_seconds - now) + 
            :
                retry_after = window_seconds
             , current_count, max_requests, retry_after

         , current_count + , max_requests, 

rate_limiter = SlidingWindowRateLimiter(redis_client)

 ():
    
     ():

         ():
            
             key_func:
                identifier = key_func()
             (g, ):
                identifier = 
            :
                identifier = 

            key = 
            allowed, current, limit, retry_after = rate_limiter.is_allowed(
                key, max_requests, window_seconds)

            
            headers = {
                : (limit),
                : ((, limit - current)),
                : ((time.time()) + window_seconds),
            }

              allowed:
                headers[] = (retry_after)
                response = jsonify({
                    : ,
                    : ,
                    : retry_after
                })
                response.status_code = 
                 h, v  headers.items():
                    response.headers[h] = v
                 response

            response = f(*args, **kwargs)
             h, v  headers.items():
                response.headers[h] = v
             response
         wrapped
     decorator




 ():
    
     jsonify({: })



 ():
    
     jsonify({: })



 ():
    
     jsonify({: []})

步骤3：令牌桶速率限制器

import redis
import time

class TokenBucketRateLimiter:
    """允许在限制内突发流量的令牌桶速率限制器。"""

    def __init__(self, redis_conn):
        self.redis = redis_conn

    def is_allowed(self, key, max_tokens, refill_rate, refill_interval=1):
        """
        令牌桶算法：
        - max_tokens: 最大突发容量
        - refill_rate: 每refill_interval添加的令牌数
        - refill_interval: 补充之间的秒数
        """
        now = time.time()
        bucket_key = f"tb:{key}"

        # 用于原子令牌桶操作的Lua脚本
        lua_script = """
        local key = KEYS[1]
        local max_tokens = tonumber(ARGV[1])
        local refill_rate = tonumber(ARGV[2])
        local refill_interval = tonumber(ARGV[3])
        local now = tonumber(ARGV[4])

        local bucket = redis.call('hmget', key, 'tokens', 'last_refill')
        local tokens = tonumber(bucket[1])
        local last_refill = tonumber(bucket[2])

        if tokens == nil then
            tokens = max_tokens
            last_refill = now
        end

        -- 补充令牌
        local elapsed = now - last_refill
        local refills = math.floor(elapsed / refill_interval)
        if refills > 0 then
            tokens = math.min(max_tokens, tokens + (refills * refill_rate))
            last_refill = last_refill + (refills * refill_interval)
        end

        local allowed = 0
        if tokens >= 1 then
            tokens = tokens - 1
            allowed = 1
        end

        redis.call('hmset', key, 'tokens', tokens, 'last_refill', last_refill)
        redis.call('expire', key, math.ceil(max_tokens / refill_rate * refill_interval) + 10)

        return {allowed, tokens, max_tokens}
        """

        result = self.redis.eval(lua_script, 1, bucket_key,
                                  max_tokens, refill_rate, refill_interval, now)
        allowed = bool(result[0])
        remaining = int(result[1])
        limit = int(result[2])

         allowed, remaining, limit

步骤4：基于用户计划的分层速率限制

from enum import Enum

class UserTier(Enum):
    FREE = "free"
    PREMIUM = "premium"
    ENTERPRISE = "enterprise"

TIER_LIMITS = {
    UserTier.FREE: {
        "default": (60, 60),          # 60 req/min
        "search": (10, 60),           # 10 req/min
        "export": (5, 3600),          # 5 req/hour
        "daily_total": (1000, 86400), # 1000 req/day
    },
    UserTier.PREMIUM: {
        "default": (300, 60),
        "search": (50, 60),
        "export": (20, 3600),
        "daily_total": (10000, 86400),
    },
    UserTier.ENTERPRISE: {
        "default": (1000, 60),
        "search": (200, 60),
        "export": (100, 3600),
        "daily_total": (100000, 86400),
    },
}

def get_rate_limit_for_request(user_tier, endpoint_category=):
    
    tier_config = TIER_LIMITS.get(user_tier, TIER_LIMITS[UserTier.FREE])
    limit_config = tier_config.get(endpoint_category, tier_config[])
     limit_config  

 :
    

     ():
        .app = app
        .limiter = SlidingWindowRateLimiter(redis_conn)

     ():
        
        user_id = environ.get()
        user_tier = UserTier(environ.get(, ))
        endpoint = environ.get(, )

        
        category = 
           endpoint:
            category = 
           endpoint:
            category = 

        max_requests, window = get_rate_limit_for_request(user_tier, category)
        key = 

        allowed, current, limit, retry_after = .limiter.is_allowed(
            key, max_requests, window)

          allowed:
            status = 
            headers = [
                (, ),
                (, (retry_after)),
                (, (limit)),
                (, ),
            ]
            start_response(status, headers)
            body = 
             [body.encode()]

         .app(environ, start_response)

步骤5：微服务的分布式速率限制

# 使用Redis Cluster的集中式速率限制服务
import redis
from redis.cluster import RedisCluster

class DistributedRateLimiter:
    """使用Redis Cluster的微服务架构速率限制器。"""

    def __init__(self):
        self.redis = RedisCluster(
            startup_nodes=[
                {"host": "redis-node-1", "port": 6379},
                {"host": "redis-node-2", "port": 6379},
                {"host": "redis-node-3", "port": 6379},
            ],
            decode_responses=True
        )

    def check_and_increment(self, service_name, user_id, endpoint,
                             max_requests, window_seconds):
        """使用Redis Lua脚本进行原子检查和递增。"""
        key = f"rl:{{{service_name}}}:{user_id}:{endpoint}"

        # Lua脚本确保检查和递增的原子性
        lua_script = """
        local key = KEYS[1]
        local max_requests = tonumber(ARGV[1])
        local window = tonumber(ARGV[2])
        local now = tonumber(ARGV[3])
        local window_start = now - window

        -- 删除旧条目
        redis.call('zremrangebyscore', key, '-inf', window_start)

        -- 计算当前条目数
        local count = redis.call('zcard', key)

        if count >= max_requests then
            -- 获取最旧条目用于重试等待时间计算
            local oldest = redis.call('zrange', key, 0, 0, 'WITHSCORES')
            local retry_after = 0
            if #oldest > 0 then
                retry_after = math.ceil(tonumber(oldest[2]) + window - now)
            end
            return {0, count, retry_after}
        end

        -- 添加新条目
        redis.call('zadd', key, now, now .. ':' .. math.random(100000))
        redis.call('expire', key, window + 1)

        return {1, count + 1, 0}
        """

        result = .redis.(lua_script, , key,
                                  max_requests, window_seconds, time.time())
         {
            : (result[]),
            : (result[]),
            : (result[]),
        }

核心概念

术语	定义
滑动窗口（Sliding Window）	在滚动时间窗口内跟踪请求的速率限制算法，比固定窗口提供更平滑的速率强制
令牌桶（Token Bucket）	令牌以固定速率添加并按请求消耗的算法，允许在桶容量范围内的受控突发
固定窗口（Fixed Window）	最简单的速率限制，按固定时间窗口（如每分钟）计数请求，在窗口边界处容易发生突发
429 Too Many Requests	表示客户端已超过速率限制的HTTP状态码，伴随Retry-After头
Retry-After头（Retry-After Header）	告知客户端重试前需等待多少秒的HTTP响应头，对行为良好的API客户端至关重要
分布式速率限制（Distributed Rate Limiting）	使用共享状态（Redis、Memcached）跨多个服务器实例进行速率限制，维护准确的全局计数器

工具和系统

• Redis：用于分布式速率限制计数器的内存数据存储，通过Lua脚本支持原子操作
• Kong Rate Limiting Plugin：支持Redis后端固定窗口和滑动窗口速率限制的API网关插件
• express-rate-limit：支持Redis、Memcached或内存存储的Express.js速率限制中间件
• Flask-Limiter：支持多种后端和按端点可配置限制的Flask速率限制扩展
• Envoy Rate Limit Service：基于Envoy的服务网格架构的集中式速率限制服务

常见场景

场景：为公共API实施速率限制

背景：公司推出包含免费、高级和企业层级的公共API，需要在防止滥用的同时为付费客户提供公平访问。API在AWS ALB后面运行在6个实例上。

方法：

1. 部署Redis Cluster（3个节点）用于分布式速率限制状态
2. 使用Redis有序集合和Lua脚本实现原子性的滑动窗口速率限制器
3. 配置按层级限制：免费（60 req/min）、高级（300 req/min）、企业（1000 req/min）
4. 无论层级如何，对认证端点添加更严格的限制（每IP每分钟5个请求）
5. 实施资源密集型端点限制（搜索：免费版每分钟10个，导出：每小时5个）
6. 在每个响应上设置速率限制响应头（X-RateLimit-Limit、X-RateLimit-Remaining、X-RateLimit-Reset）
7. 超出限制时返回带Retry-After头和JSON错误体的429响应
8. 设置Prometheus指标用于速率限制命中，设置CloudWatch告警用于异常模式

注意事项：

• 使用内存速率限制而没有跨实例共享状态，允许通过访问不同服务器绕过限制
• 未将认证端点的速率限制与通用API限制分开实施
• 使用固定窗口导致在窗口边界出现突发（短时间内达到限制的2倍）
• 在成功响应上不包含速率限制头，使客户端看不到其配额
• 信任X-Forwarded-For用于IP识别而不验证来自负载均衡器的有效性

输出格式

## 速率限制实施报告

**API**: Public API v2
**算法**: 滑动窗口（Redis有序集合）
**后端**: Redis Cluster（3个节点）
**部署**: AWS ALB后面的6个API实例

### 速率限制配置

| 层级 | 默认 | 搜索 | 导出 | 认证（每IP） |
|------|---------|--------|--------|---------------|
| 免费 | 60/min | 10/min | 5/hour | 5/min |
| 高级 | 300/min | 50/min | 20/hour | 5/min |
| 企业 | 1000/min | 200/min | 100/hour | 10/min |

### 验证结果（k6负载测试）

- 免费层：第61个请求时速率限制（正确）
- 高级层：第301个请求时速率限制（正确）
- 跨实例：所有6个实例速率限制一致
- Redis故障转移：Redis不可用时速率限制优雅降级（允许流量通过）
- Retry-After头：实际重置时间1秒内准确
- 响应开销：每个速率限制检查增加<2ms延迟