条件竞争漏洞¶

竞态条件是一种常见的漏洞类型，通常与业务逻辑缺陷密切相关。当网站同时处理多个请求而没有适当的防护措施时，就会引发竞态条件。这种情况下，多个线程可能会同时访问相同的数据，从而导致“冲突”，进而使应用程序出现意外的行为。攻击者可以利用这种竞态条件，通过精心安排时间的请求来引发预期中的冲突，从而利用这种意外行为达到恶意目的。

发生碰撞的可能时间区间被称为“竞争窗口”。例如，这可以是两次与数据库交互之间所间隔的极短时间。

与其他逻辑缺陷一样，竞争条件的影响也很大程度上取决于应用程序以及该缺陷出现的特定功能。

Note

更多参考：https://portswigger.net/research/smashing-the-state-machine

超出限制的条件竞争¶

最著名的竞争条件类型是指那种能够突破应用程序业务逻辑所设定的某种限制的情况。

例如，想象一个在线商店，用户在结账时可以输入促销代码以获得一次性的折扣。为了享受这个折扣，应用程序可能需要执行以下高级操作：

请检查您是否已经使用了这个代码
将折扣金额应用于订单的总金额中
在数据库中更新记录，以表明您现在已经使用了这段代码

如果你之后试图再次使用这段代码，那么在流程开始时所进行的初步检查应该能够阻止你这样做。

现在想象一下，如果有一个从未使用过此优惠码的用户，在几乎同一时间尝试两次使用该优惠码，会发生什么情况：

该应用程序会经历一个临时的子状态；也就是说，在请求处理完成之前，它会进入这个子状态然后再次退出。在这种情况下，子状态从服务器开始处理第一个请求时开始，直到更新数据库以表明您已经使用了这段代码为止。这样就在一定的时间内，您可以多次申请该折扣。

这种攻击方式有很多种变体，包括：

多次兑换礼品卡
多次评价某个产品
提取或转账超过账户余额的现金
重复使用同一个验证码解决方案
绕过反暴力破解率的限制机制

利用Turbo Intruder工具检测条件竞争¶

“Turbo Intruder”这个工具需要一定的 Python 编程能力，但非常适合执行更复杂的攻击，比如那些需要多次尝试、错开请求时间或执行大量请求的操作。

在 Turbo Intruder 中使用单包攻击：

请确保目标支持 HTTP/2 协议。单包攻击与 HTTP/1 协议不兼容。
请设置请求引擎的 engine=Engine.BURP2 和 concurrentConnections=1 配置选项。
在排队处理请求时，请将这些请求分组，并使用 engine.queue() 方法的 gate 参数将它们分配给某个名为“gate”的接口。
要发送给定组中的所有请求，可以使用 engine.openGate() 方法打开相应的界面。

Python

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

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

    # send all requests in gate '1' in parallel
    engine.openGate('1')

更多详细信息，请参考 Turbo Intruder 默认示例目录中提供的 race-single-packet-attack.py 模板。

多步骤业务中的条件竞争¶

在实践中，一个单独的请求可能会触发一系列多步骤的操作序列。在请求处理完成之前，应用程序会进入多个隐藏状态，然后再次退出这些状态。我们将这些状态称为“子状态”。

如果你能够识别出一个或多个会导致与相同数据交互的 HTTP 请求，那么你就有可能利用这些子状态来利用那些常见于多步骤工作流程中的逻辑缺陷。这样就能实现一种称为“竞态条件”的攻击方式，这种攻击方式远远超出了单纯的时间超时的范畴。

例如，你可能熟悉有缺陷的多因素认证工作流程。在这种工作流程中，用户可以使用已知的凭据完成登录的第一步，然后直接通过强制浏览行为进入应用程序，从而完全绕过多因素认证。

以下伪代码展示了如何使一个网站容易受到这种攻击的竞态变化的影响：

Text Only

session['userid'] = user.userid
if user.mfa_enabled:
session['enforce_mfa'] = True
# generate and send MFA code to user
# redirect browser to MFA code entry form

如您所见，这实际上是一个在单个请求过程中分多步进行的序列。最重要的是，这个过程会经过一个子状态，在該状态下，用户暂时拥有有效的登录会话，但多因素认证尚未生效。攻击者可以利用这一机会，通过发送包含登录请求的请求，同时附上对敏感资源的访问请求，从而发起攻击。

Note

更多参考：https://portswigger.net/web-security/race-conditions

漏洞示例¶

支付类条件竞争¶

初始化数据库

SQL

-- init_db.sql
CREATE DATABASE IF NOT EXISTS race_demo CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

CREATE USER IF NOT EXISTS 'demo'@'localhost' IDENTIFIED BY 'demo123';

GRANT ALL PRIVILEGES ON race_demo.* TO 'demo'@'localhost';

FLUSH PRIVILEGES;

USE race_demo;

DROP TABLE IF EXISTS users;

CREATE TABLE users (
    username VARCHAR(50) PRIMARY KEY,
    balance DECIMAL(10,2) NOT NULL
);

INSERT INTO users (username, balance) VALUES ('alice', 100.00);

server

Python

# app.py
from flask import Flask, request, jsonify
from flask_cors import CORS  # type: ignore
import pymysql  # type: ignore
import time
import threading

app = Flask(__name__)
CORS(app)

# 数据库配置
DB_CONFIG = {
    'host': 'localhost',
    'user': 'yifan',
    'password': 'yifan',
    'database': 'payment_demo',
    'charset': 'utf8mb4',
    'autocommit': False  # 手动控制事务
}


def get_conn():
    conn = pymysql.connect(**DB_CONFIG)
    # 设置事务隔离级别为 READ COMMITTED，减少锁
    with conn.cursor() as cur:
        cur.execute("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;")
    return conn


# === ❌ 不安全接口：最大竞争窗口（可被 Turbo Intruder 触发）===
@app.route('/pay/unsafe', methods=['POST'])
def pay_unsafe():
    if not request.is_json or request.json is None:
        return jsonify({'error': '请求体必须为JSON'}), 400
    amount = request.json.get('amount', 0)
    if amount <= 0:
        return jsonify({'error': '无效金额'}), 400

    conn = get_conn()
    try:
        with conn.cursor() as cur:
            # 1. 查询余额（无锁）
            cur.execute("SELECT balance FROM users WHERE username = 'alice'")
            row = cur.fetchone()
            if not row:
                return jsonify({'error': '用户不存在'}), 400
            current_balance = float(row[0])

            print(f"[{threading.current_thread().name}] 读取余额: {current_balance}")

            # 🔥 关键：放大竞争窗口（2秒）
            time.sleep(2)

            if current_balance >= amount:
                # 再次延迟
                time.sleep(2)
                # 执行扣款
                cur.execute("UPDATE users SET balance = balance - %s WHERE username = 'alice'",
                            (amount,))
                conn.commit()
                new_balance = current_balance - amount
                print(f"[{threading.current_thread().name}] 扣款成功")
                return jsonify({'success': True, 'balance': new_balance})
            else:
                return jsonify({'error': '余额不足'}), 400
    except Exception as e:
        conn.rollback()
        return jsonify({'error': f'系统错误[{e}]'}), 500
    finally:
        conn.close()


# === ✅ 安全接口：使用 SELECT ... FOR UPDATE 修复竞争 ===
@app.route('/pay/safe', methods=['POST'])
def pay_safe():
    if not request.is_json or request.json is None:
        return jsonify({'error': '请求体必须为JSON'}), 400
    amount = request.json.get('amount', 0)
    if amount <= 0:
        return jsonify({'error': '无效金额'}), 400

    conn = get_conn()
    try:
        with conn.cursor() as cur:
            # 开始事务
            conn.begin()
            # 🔐 使用 FOR UPDATE 加行锁
            cur.execute("SELECT balance FROM users WHERE username = 'alice' FOR UPDATE")
            row = cur.fetchone()
            if not row:
                conn.rollback()
                return jsonify({'error': '用户不存在'}), 400
            current_balance = float(row[0])

            if current_balance >= amount:
                new_balance = current_balance - amount
                cur.execute("UPDATE users SET balance = %s WHERE username = 'alice'",
                            (new_balance,))
                conn.commit()
                return jsonify({'success': True, 'balance': new_balance})
            else:
                conn.rollback()
                return jsonify({'error': '余额不足'}), 400
    except Exception as e:
        conn.rollback()
        return jsonify({'error': f'系统错误[{e}]'}), 500
    finally:
        conn.close()


# === 查询余额 ===
@app.route('/balance', methods=['GET'])
def get_balance():
    conn = get_conn()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT balance FROM users WHERE username = 'alice'")
            row = cur.fetchone()
            return jsonify({'balance': float(row[0]) if row else 0})
    finally:
        conn.close()


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8000, threaded=True)

client

HTML

<!-- static/payment.html -->
<!DOCTYPE html>
<html lang="zh">
<head>
  <meta charset="UTF-8" />
  <title>支付测试页面</title>
  <style>
    body { font-family: Arial, sans-serif; padding: 20px; }
    .form-group { margin: 10px 0; }
    label { display: inline-block; width: 100px; }
    input[type="text"], input[type="number"] { padding: 5px; width: 300px; }
    button { padding: 10px 15px; background: #007cba; color: white; border: none; cursor: pointer; }
    button:hover { background: #005a87; }
    #result { margin-top: 20px; padding: 10px; border: 1px solid #ccc; display: none; }
  </style>
</head>
<body>

<h2>支付系统 - 条件竞争测试</h2>

<form id="pay-form">
  <div class="form-group">
    <label>服务端地址：</label>
    <input type="text" id="server-url" placeholder="例如：http://192.168.1.100:5000" value="" />
  </div>
  <div class="form-group">
    <label>支付模式：</label>
    <select id="mode">
      <option value="unsafe">不安全（有漏洞）</option>
      <option value="safe">安全（已修复）</option>
    </select>
  </div>
  <div class="form-group">
    <label>金额：</label>
    <input type="number" id="amount" value="60" min="1" max="1000" />
  </div>
  <button type="submit">发起支付</button>
  <button type="button" onclick="checkBalance()">查询余额</button>
</form>

<div id="result"></div>

<script>
  // 从 URL 参数读取 server
  const urlParams = new URLSearchParams(window.location.search);
  const serverFromUrl = urlParams.get('server');
  if (serverFromUrl) {
    document.getElementById('server-url').value = serverFromUrl;
  }

  function showResult(msg, isError = false) {
    const resultDiv = document.getElementById('result');
    resultDiv.style.display = 'block';
    resultDiv.style.color = isError ? 'red' : 'green';
    resultDiv.innerHTML = msg;
  }

  document.getElementById('pay-form').onsubmit = function (e) {
    e.preventDefault();

    const server = document.getElementById('server-url').value.trim();
    const mode = document.getElementById('mode').value;
    const amount = parseFloat(document.getElementById('amount').value);

    if (!server) {
      alert("请填写服务端地址，例如：http://192.168.1.100:5000");
      return;
    }
    if (isNaN(amount) || amount <= 0) {
      alert("请输入有效金额");
      return;
    }

    const url = `${server}/pay/${mode}`;
    const data = { amount };

    fetch(url, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(data)
    })
    .then(res => res.json())
    .then(data => {
      if (data.success) {
        showResult(`✅ 支付成功！余额：${data.balance}`);
      } else {
        showResult(`❌ 支付失败：${data.error}`, true);
      }
    })
    .catch(err => {
      showResult(`🚨 请求失败：${err.message}<br>请检查服务端地址是否正确或网络是否可达。`, true);
    });
  };

  function checkBalance() {
    const server = document.getElementById('server-url').value.trim();
    if (!server) {
      alert("请填写服务端地址");
      return;
    }

    fetch(`${server}/balance`)
      .then(res => res.json())
      .then(data => {
        showResult(`📊 当前余额：${data.balance}`);
      })
      .catch(() => {
        showResult("❌ 无法获取余额，请检查服务端是否运行", true);
      });
  }
</script>

</body>
</html>

unsafe场景¶

关键代码：

Python

@app.route('/pay/unsafe', methods=['POST'])
def pay_unsafe():
    if not request.is_json or request.json is None:
        return jsonify({'error': '请求体必须为JSON'}), 400
    amount = request.json.get('amount', 0)
    if amount <= 0:
        return jsonify({'error': '无效金额'}), 400

    conn = get_conn()
    try:
        with conn.cursor() as cur:
            # 1. 查询余额（无锁）
            cur.execute("SELECT balance FROM users WHERE username = 'alice'")
            row = cur.fetchone()
            if not row:
                return jsonify({'error': '用户不存在'}), 400
            current_balance = float(row[0])

            print(f"[{threading.current_thread().name}] 读取余额: {current_balance}")

            # 🔥 关键：放大竞争窗口（2秒）
            time.sleep(2)

            if current_balance >= amount:
                # 再次延迟
                time.sleep(2)
                # 执行扣款
                cur.execute("UPDATE users SET balance = balance - %s WHERE username = 'alice'",
                            (amount,))
                conn.commit()
                new_balance = current_balance - amount
                print(f"[{threading.current_thread().name}] 扣款成功")
                return jsonify({'success': True, 'balance': new_balance})
            else:
                return jsonify({'error': '余额不足'}), 400
    except Exception as e:
        conn.rollback()
        return jsonify({'error': f'系统错误[{e}]'}), 500
    finally:
        conn.close()

此时余额为100，一次转60，按照后端逻辑分析第二次转账会提示余额不足，但这仅限于单用户单次操作时，而在高并发时，就会出现问题，我们使用burp turbo intruder插件来验证

拦截转账的请求

右键发送到插件

需要注意，该插件设计出来是用于参数化爆破的，所以即使我们只需要高并发重放，不需要参数化，也要在请求头添加%s标识参数化标识，请求头名任意，选择的python脚本为basic，可以自行调节连接数、每个连接的并发数等

Python

def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                           concurrentConnections=30,
                           requestsPerConnection=1,
                           pipeline=False
                           )

    # the 'gate' argument blocks the final byte of each request until openGate is invoked
    for i in range(30):
        engine.queue(target.req, target.baseInput, gate='race1')

    # wait until every 'race1' tagged request is ready
    # then send the final byte of each request
    # (this method is non-blocking, just like queue)
    engine.openGate('race1')

    engine.complete(timeout=60)


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

通过响应码和响应内容，我们判断存在条件竞争漏洞，此时我们关闭burp拦截并返回前端查询余额验证

此时余额变为负数

safe场景¶

关键代码：

Python

@app.route('/pay/safe', methods=['POST'])
def pay_safe():
    if not request.is_json or request.json is None:
        return jsonify({'error': '请求体必须为JSON'}), 400
    amount = request.json.get('amount', 0)
    if amount <= 0:
        return jsonify({'error': '无效金额'}), 400

    conn = get_conn()
    try:
        with conn.cursor() as cur:
            # 开始事务
            conn.begin()
            # 🔐 使用 FOR UPDATE 加行锁
            cur.execute("SELECT balance FROM users WHERE username = 'alice' FOR UPDATE")
            row = cur.fetchone()
            if not row:
                conn.rollback()
                return jsonify({'error': '用户不存在'}), 400
            current_balance = float(row[0])

            if current_balance >= amount:
                new_balance = current_balance - amount
                cur.execute("UPDATE users SET balance = %s WHERE username = 'alice'",
                            (new_balance,))
                conn.commit()
                return jsonify({'success': True, 'balance': new_balance})
            else:
                conn.rollback()
                return jsonify({'error': '余额不足'}), 400
    except Exception as e:
        conn.rollback()
        return jsonify({'error': f'系统错误[{e}]'}), 500
    finally:
        conn.close()

同上操作，但由于使用了事务和数据库锁，所以不存在条件竞争

此时只有第一次转账成功了，后面的转账都没有成功

漏洞成因总结（基于对比）¶

为什么 `unsafe` 有漏洞？¶

非原子操作：SELECT 和 UPDATE 被业务逻辑和 sleep() 分割。
无锁读取：多个并发请求都能读到 balance = 100。
无事务保护：每个 UPDATE 是独立事务，无法感知其他请求。
竞争窗口大：sleep(2) + sleep(2) = 4 秒，足够 10+ 请求进入。

👉 结果：3 个请求都通过 if balance >= 60，都执行扣款 → 余额 = 100 - 60×3 = -140

为什么 `safe` 无漏洞？¶

加锁读取：SELECT ... FOR UPDATE 为 alice 行加 排他锁（X-Lock）。
事务隔离：其他请求执行 FOR UPDATE 时会 阻塞等待，直到第一个事务 COMMIT。
原子性保障：读、判、改在同一个事务中完成，不可分割。
串行化执行：即使 100 个并发，也只有一个能进入扣款逻辑。

👉 结果：仅第一个请求成功，其余失败或超时 → 余额 = 40

对比纬度	不安全版本(pay/unsafe)	安全版本(pay/safe)	安全影响
事务控制	❌ 无显式事务控制（依赖 autocommit）	✅ 显式`conn.begin()`开启事务	安全版本确保“读-判-改”在同一个事务中
读取方式	`SELECT balance`（普通读）	`SELECT ... FOR UPDATE`（加排他锁）	安全版本阻止其他事务并发修改
锁机制	❌ 无锁，多个请求可同时读到`100`	✅`FOR UPDATE`加行级排他锁	其他请求必须等待锁释放
原子性	❌ “读-判-改”被`sleep(2)`分割	✅ “读-判-改”在事务中连续执行	安全版本是原子操作
竞争窗口	🔥 极大（4秒）：多个请求可同时通过判断	✅ 极小：第一个请求加锁后，其余被阻塞	漏洞版本易被 Turbo Intruder 利用
并发行为	多个请求可同时执行`UPDATE`	仅一个请求能进入`FOR UPDATE`	安全版本防止超扣
最终结果	余额可变为负数（如 -140）	余额最多扣到 40，其余失败	修复有效

事实上，经测试，竞争窗口的大小对实际漏洞利用影响很小，即使删除unsafe环境下的窗口期，仍然可以利用漏洞，所以漏洞利用的本质原因为：在没有事务等操作时，读取、判断、执行三个操作可以被拆分，而事务会将其看作一个整体来操作，所以事务就不存在可被利用的时间间隔

事务的作用¶

事务是数据库管理系统（DBMS）中用于管理一系列操作的逻辑单元。它的核心作用由 ACID 四个特性定义：


A	Atomicity（原子性）	要么全部成功，要么全部失败，不可分割
C	Consistency（一致性）	操作前后，数据必须满足业务规则（如余额 ≥ 0）
I	Isolation（隔离性）	并发事务之间互不干扰
D	Durability（持久性）	一旦提交，数据永久保存，即使系统崩溃

对于该漏洞，重点关注原子性和隔离性

原子性如何实现？¶

数据库通过 日志（Write-Ahead Log, WAL） 实现：

先写日志（记录“准备修改”）
再修改内存中的数据
最后 COMMIT 时，日志落盘，事务才算成功
如果崩溃，重启后根据日志 重做（Redo）或撤销（Undo）

✅ 所以原子性不是“瞬间完成”，而是“失败时能完全撤销”。

隔离性: 为什么事务能防止条件竞争？¶

这才是防止 条件竞争漏洞 的关键！

无事务时的竞争¶

Text Only

# 请求1：读 balance = 100

# 请求2：读 balance = 100

# 请求1：判断 100 >= 60 → 成功 → 扣款 → 余额 = 40
# 请求2：判断 100 >= 60 → 成功 → 扣款 → 余额 = -20  ← 超扣！

两个请求**同时读到旧值**，因为没有隔离。

事务 + 锁¶

当你在事务中使用 SELECT ... FOR UPDATE：

SQL

BEGIN;
SELECT balance FROM users WHERE username = 'alice' FOR UPDATE;  -- 加排他锁
-- 其他事务无法读取或修改这行
IF balance >= 60:
    UPDATE users SET balance = balance - 60 WHERE username = 'alice';
COMMIT; -- 释放锁

锁的作用¶

第一个事务执行 FOR UPDATE 时，为 alice 这行数据加上 排他锁（X-Lock）
第二个事务执行到 FOR UPDATE 时，被阻塞，必须等待第一个事务 COMMIT
第一个事务完成后，第二个事务才能继续，此时它读到的是 新余额（40）
判断 40 >= 60 失败，扣款被拒绝

✅ 结果：只有一次扣款成功，余额不会为负

事务如何保障“读-判-改”的原子性？¶

虽然“读、判、改”是三条语句，但通过 事务 + 锁，可以把它们变成一个**逻辑上的原子操作**：

Text Only

+---------------------+
| 开始事务 (BEGIN)     |
+----------+----------+
           |
           v
+---------------------+
| 加锁读取 (SELECT FOR UPDATE) ←┐
+----------+----------+         |
           |                    |
           v                    |
+---------------------+         |
| 业务判断 (if...)     |         | 在事务中，其他操作被阻塞
+----------+----------+         |
           |                    |
           v                    |
+---------------------+         |
| 执行修改 (UPDATE)    |         |
+----------+----------+         |
           |                    |
           v                    |
+---------------------+         |
| 提交事务 (COMMIT)    | ————————┘
+---------------------+

综上我们得到结论：单一的事务或加锁机制无法完全防御条件竞争漏洞，只有两者结合，才能做到安全

漏洞利用是否成功的影响因素¶

条件竞争能否被成功利用，不仅取决于代码逻辑，还受数据库类别、数据库配置、版本、开发框架等多重因素影响。

影响维度	是否影响漏洞存在	是否影响漏洞利用	说明
数据库类别	✅ 是	✅ 影响	影响默认隔离级别和锁行为
数据库版本	⚠️ 间接影响	✅ 影响	影响默认隔离级别和锁行为
数据库配置	✅ 是	✅ 是	存储引擎、隔离级别、autocommit 等
是否使用框架	⚠️ 不影响存在	✅ 影响利用难度	框架可能掩盖风险或提供修复工具
代码实现方式	✅ 是	✅ 是	原生 SQL vs ORM，是否加锁

数据库类别的影响¶

数据库	是否支持行级锁	默认隔离级别	存储引擎/架构	漏洞利用难度	说明
MySQL (InnoDB)	✅ 是	`REPEATABLE READ`	行级锁 + MVCC	⭐⭐⭐☆	默认较安全，但可降级为`READ COMMITTED`演示漏洞
MySQL (MyISAM)	❌ 否	无事务	表级锁	⭐⭐⭐⭐⭐	极易触发，无行锁，适合演示
PostgreSQL	✅ 是	`READ COMMITTED`	MVCC + 行锁	⭐⭐⭐	行为清晰，`FOR UPDATE`可靠，适合生产与教学
SQLite	⚠️ 有限	`SERIALIZABLE`	单文件，全局锁	⭐	写操作完全串行，几乎无法复现竞争
SQL Server	✅ 是	`READ COMMITTED`	行版本控制	⭐⭐⭐☆	支持`UPDLOCK, HOLDLOCK`，企业级锁机制
Oracle	✅ 是	`READ COMMITTED`	MVCC + 行锁	⭐⭐⭐	支持`SELECT FOR UPDATE`，行为稳定
MongoDB	❌（原生）	无传统事务（早期）	文档模型	⭐⭐	无行锁，需应用层实现乐观锁或分布式锁

数据库版本的影响¶

版本/引擎	影响说明
MySQL < 5.7 / InnoDB 早期版本	`FOR UPDATE`支持不完整，可能忽略锁，更容易被利用
MySQL 5.7+ / 8.0	默认`REPEATABLE READ`，自动加共享锁，可能掩盖漏洞
MyISAM 引擎	❌ 不支持事务和行锁，极易触发竞争
InnoDB 引擎	✅ 支持行锁，但需主动使用`FOR UPDATE`才能防止竞争
PostgreSQL	支持 MVCC 和`FOR UPDATE`，行为清晰，适合演示和生产

✅ 结论：数据库版本不决定漏洞是否存在，但影响**是否容易被触发**。

数据库配置的影响¶

配置项	配置值	对漏洞的影响
`autocommit`	`ON`	每条语句自动提交，增加竞争窗口
`autocommit`	`OFF`	需手动`commit`，可控制事务范围
`transaction_isolation`	`READ COMMITTED`	允许读取最新提交数据，更容易读到旧值，利于漏洞利用
`transaction_isolation`	`REPEATABLE READ`	默认加共享锁，可能阻塞其他写入，掩盖漏洞
`transaction_isolation`	`SERIALIZABLE`	完全串行化，几乎不可能触发竞争
存储引擎	`MyISAM`	❌ 无事务支持，极易触发
存储引擎	`InnoDB`	✅ 支持行锁，但需主动加锁

✅ 最佳演示配置：

SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;

是否使用框架的影响¶

使用方式	对漏洞的影响
原生 SQL + 手动连接	❗ 漏洞明显，但易出错（如连接泄漏）
SQLAlchemy / Django ORM	✅ 提供`with_for_update()`等工具，便于修复 ❌ 但默认不加锁，可能掩盖开发者对竞争的认知
Flask-SQLAlchemy	封装了 SQLAlchemy，行为一致
使用 ORM 但不加锁	⚠️最危险：开发者误以为“安全”，实际仍存在漏洞
使用 ORM + `FOR UPDATE`	✅ 推荐做法，代码清晰，安全可控

✅ 关键结论：

使用成熟框架不能自动防止条件竞争

框架只是工具，安全仍需开发者主动设计

ORM 可能因“过度封装”而让开发者忽略底层并发问题

条件竞争漏洞¶

超出限制的条件竞争¶

利用Turbo Intruder工具检测条件竞争¶

多步骤业务中的条件竞争¶

漏洞示例¶

支付类条件竞争¶

unsafe场景¶

safe场景¶

漏洞成因总结（基于对比）¶

为什么 unsafe 有漏洞？¶

为什么 safe 无漏洞？¶

事务的作用¶

原子性如何实现？¶

隔离性: 为什么事务能防止条件竞争？¶

无事务时的竞争¶

事务 + 锁¶

锁的作用¶

事务如何保障“读-判-改”的原子性？¶

漏洞利用是否成功的影响因素¶

数据库类别的影响¶

数据库版本的影响¶

数据库配置的影响¶

是否使用框架的影响¶

为什么 `unsafe` 有漏洞？¶

为什么 `safe` 无漏洞？¶