# 竞态条件检测模块

## 概述

竞态条件发生在多个并发操作访问共享状态且缺乏适当同步机制时，导致重复状态变更、配额绕过、金融欺诈和权限提升。每个读取-修改-写入序列和多步骤工作流都必须被视为可能受到并发访问威胁的对象。

## 关键影响

- **经济损失**: 重复消费、超额发放积分/退款、重复支付
- **策略绕过**: 配额违规、一次性令牌重用、速率限制规避
- **数据完整性**: 负数计数器、重复记录、不一致的聚合数据
- **权限提升**: 并发角色更新、权限绕过

## 检测优先级: 严重

---

## 代码模式分类

### 1. 缺乏原子性的读取-修改-写入

**危险模式:**

```java
// Java - 非原子余额更新
int balance = getBalance(userId);
if (balance >= amount) {
    setBalance(userId, balance - amount);  // 竞态窗口!
}

// 危险: 计数器递增
int count = cache.get(key);
cache.set(key, count + 1);  // 可能丢失更新

// 数据库读取-修改-写入未加行锁
User user = userRepo.findById(id);
user.setBalance(user.getBalance() - amount);
userRepo.save(user);  // SELECT 然后 UPDATE - 竞态窗口
```

```python
# Python - 基于文件的计数器
with open('counter.txt', 'r') as f:
    count = int(f.read())
with open('counter.txt', 'w') as f:
    f.write(str(count + 1))  # 读取和写入之间的竞态

# Redis 非原子操作
balance = redis.get(f'balance:{user_id}')
if int(balance) >= amount:
    redis.set(f'balance:{user_id}', int(balance) - amount)  # 竞态!

# Django ORM 不使用 F() 的更新
user = User.objects.get(id=user_id)
user.credits -= cost
user.save()  # SELECT 然后 UPDATE
```

```go
// Go - 无互斥锁的 Map 访问
count := counterMap[key]
counterMap[key] = count + 1  // 竞态条件

// 非原子文件操作
data, _ := ioutil.ReadFile("state.json")
state := parseState(data)
state.Count++
ioutil.WriteFile("state.json", marshal(state), 0644)  // 竞态窗口
```

**安全替代方案:**
```java
// 原子数据库更新
@Query("UPDATE User u SET u.balance = u.balance - :amount WHERE u.id = :id AND u.balance >= :amount")
int deductBalance(@Param("id") Long id, @Param("amount") int amount);

// 乐观锁
@Version
private Long version;

// 悲观锁
@Lock(LockModeType.PESSIMISTIC_WRITE)
User findByIdForUpdate(Long id);
```

```python
# Django 原子 F() 表达式
from django.db.models import F
User.objects.filter(id=user_id).update(credits=F('credits') - cost)

# Redis 原子操作
redis.decrby(f'balance:{user_id}', amount)
# 或使用 Lua 脚本实现复杂原子性
```

---

### 2. 存在间隔的多步骤操作

**危险模式:**

```java
// 检查后执行模式
if (couponService.isValid(code)) {
    // 间隔 - 优惠券可能被其他线程使用
    applyDiscount(code);
    couponService.markUsed(code);  // 可能重复使用
}

// 库存预留
if (inventory.getStock(productId) > 0) {
    // 竞态窗口 - 库存可能变为负数
    createOrder(productId);
    inventory.decrementStock(productId);
}

// 多阶段支付
String authId = paymentGateway.authorize(amount);
// 间隔 - 可能被多次捕获
paymentGateway.capture(authId);
updateUserBalance(authId);
```

```python
# 令牌消费
if not token_used_set.contains(token):
    # 间隔 - 令牌可能被并发重用
    perform_action()
    token_used_set.add(token)

# 座位预留
if get_available_seats(event_id) > 0:
    # 竞态 - 座位可能超售
    create_reservation(user_id, event_id)
    decrement_seats(event_id)
```

**检测规则:**
- 查找条件块中状态检查与修改分为独立操作的情况
- 识别验证与操作之间的时间间隔（检查 -> 预留 -> 提交）
- 标记序列: `isValid() -> use()`, `hasBalance() -> deduct()`, `checkStock() -> decrement()`

---

### 3. 幂等性控制缺陷

**危险模式:**

```java
// 幂等键作用域不足
String idempKey = request.getHeader("Idempotency-Key");
if (!processedKeys.contains(idempKey)) {
    // 缺少主体/用户作用域 - 键可跨用户重用!
    processPayment();
    processedKeys.add(idempKey);
}

// 提交前缓存写入窗口
@Transactional
public void processOrder(String idempKey) {
    cache.set(idempKey, "processing");  // 在事务提交前写入
    // 如果多个并发请求同时到达...
    createOrder();
    // 事务稍后提交 - 创建了多个订单
}

// 过短 TTL 的幂等性
if (!redis.exists(idempKey) || redis.ttl(idempKey) < 0) {
    // TTL 过短 - 键过期后可被重用
    processRefund();
    redis.setex(idempKey, 60, "done");  // 仅 60 秒
}
```

```python
# 应用层去重
if idempotency_key not in processed_cache:
    send_email(user)  # 副作用已发生
    credit_account(user)  # 副作用已发生
    processed_cache[idempotency_key] = True
    return {"status": "success"}
return processed_cache[idempotency_key]  # 仅返回缓存响应!
# 问题: 尽管检测到重复，副作用已经发生
```

**检测信号:**
- 幂等键缺少用户/主体作用域
- 缓存写入在事务提交之前
- TTL 短于合理的重试窗口（金融操作 < 24 小时）
- 去重逻辑返回缓存响应但未阻止副作用

---

### 4. 乐观并发缺失或薄弱

**危险模式:**

```java
// 无版本检查
@Entity
public class Account {
    // 缺少 @Version 注解!
    private Long id;
    private BigDecimal balance;
}

// 可选版本未强制执行
public void updateUser(UserDTO dto) {
    User user = repo.findById(dto.getId());
    // dto.version 存在但从未检查!
    user.setEmail(dto.getEmail());
    repo.save(user);
}

// ETag/If-Match 被忽略
@PutMapping("/resource/{id}")
public Resource update(@PathVariable Long id,
                       @RequestBody Resource resource,
                       @RequestHeader(value="If-Match", required=false) String etag) {
    // etag 参数存在但从未验证!
    return repo.save(resource);
}
```

```python
# Django 不使用 select_for_update
user = User.objects.get(id=user_id)
// 并发更新会互相覆盖
user.role = 'admin'
user.save()

# 缺少版本字段强制执行
# 模型有 'version' 字段但未在查询中使用
User.objects.filter(id=user_id).update(role='admin')  # 绕过版本检查
```

**安全模式:**
```java
// JPA 乐观锁
@Version
@Column(name = "version")
private Long version;

// 查询中显式版本检查
@Query("UPDATE User u SET u.email = :email WHERE u.id = :id AND u.version = :version")
int updateWithVersion(@Param("id") Long id, @Param("email") String email, @Param("version") Long version);
```

---

### 5. 唯一约束违反

**危险模式:**

```java
// 在数据库外进行存在性检查
if (!userRepo.existsByEmail(email)) {
    // 竞态窗口 - 可能创建重复用户
    User user = new User(email);
    userRepo.save(user);  // 可能因约束违反而失败
}

// 缺乏适当冲突处理的 Upsert
User user = userRepo.findByEmail(email);
if (user == null) {
    user = new User(email);
} else {
    user.setLastLogin(now());
}
repo.save(user);  // 竞态 - 可能重复插入
```

```python
# 检查后创建模式
if not User.objects.filter(email=email).exists():
    // 竞态窗口
    User.objects.create(email=email)  // 并发下可能出现 IntegrityError

# 缺少唯一约束
class Coupon(models.Model):
    code = models.CharField(max_length=50)  // 未设置 unique=True!
    used = models.BooleanField(default=False)
```

**安全模式:**
```python
# 配合唯一约束使用 get_or_create
user, created = User.objects.get_or_create(
    email=email,
    defaults={'name': name}
)

# 模型中的唯一约束
class Coupon(models.Model):
    code = models.CharField(max_length=50, unique=True)
```

---

### 6. 跨服务竞态条件

**危险模式:**

```java
// 缺乏补偿保护的 Saga
@Transactional
public void createOrder(OrderRequest req) {
    Order order = orderRepo.save(new Order(req));
    // 异步事件发布
    eventBus.publish(new OrderCreatedEvent(order.getId()));
    // 如果补偿在提交前运行...
}

// 最终一致性窗口
public void transferFunds(Long fromUser, Long toUser, BigDecimal amount) {
    // 服务 A 扣减
    accountServiceA.deduct(fromUser, amount);
    // 网络延迟 / 异步处理
    // 服务 B 可能尚未看到扣减
    accountServiceB.credit(toUser, amount);
    // 用户可能在 A 的写入可见之前从 B 提取
}

// 至少一次交付缺乏幂等消费者
@RabbitListener(queues = "payment-queue")
public void handlePayment(PaymentMessage msg) {
    // 消息可能被多次投递
    processPayment(msg.getAmount());  // 重复扣费!
    sendConfirmationEmail(msg.getUserId());
}
```

**检测规则:**
- 事务提交前发布异步事件
- 跨服务调用缺少幂等键
- 消息消费者不具备幂等性
- Saga/补偿缺乏状态机或锁机制

---

### 7. 速率限制与配额绕过

**危险模式:**

```java
// 非原子计数器更新
int currentCount = rateLimitCache.get(userId);
if (currentCount < LIMIT) {
    // 竞态 - 计数器可能被超过
    rateLimitCache.set(userId, currentCount + 1);
    processRequest();
}

// 按连接粒度限制
@RateLimiter(permitsPerSecond = 10, scope = SCOPE.CONNECTION)
public Response apiEndpoint() {
    // 用户可以通过打开多个连接绕过
}

// 分片计数器缺乏协调
int shard = userId % NUM_SHARDS;
int count = shardedCounters[shard].incrementAndGet();
// 未检查其他分片的计数 - 配额可被超过
```

```python
# 基于缓存的速率限制存在竞态
current = cache.get(f'rate:{user_id}') or 0
if current < RATE_LIMIT:
    cache.set(f'rate:{user_id}', current + 1)  # 竞态窗口
    process_request()
else:
    raise RateLimitExceeded()
```

**安全模式:**
```java
// 原子递增并检查
long count = redis.incr("ratelimit:" + userId);
if (count > LIMIT) {
    throw new RateLimitException();
}
redis.expire("ratelimit:" + userId, WINDOW_SECONDS);
```

---

## 特殊场景

### GraphQL 并发变更

```graphql
mutation {
  refund1: processRefund(orderId: 123) { status }
  refund2: processRefund(orderId: 123) { status }
}
```

**检测:**
- 批量变更缺乏逐变更幂等性
- 解析器级别缺乏原子性
- 并行执行缺乏协调

### WebSocket 并发消息

```javascript
// 客户端快速连续发送消息
ws.send(JSON.stringify({action: 'buy', itemId: 1}));
ws.send(JSON.stringify({action: 'buy', itemId: 1}));
```

**检测:**
- 仅按连接粒度授权（非按消息）
- 消息处理器缺少幂等键
- 状态变更缺乏同步机制

### 文件上传竞态

```java
// 分片上传完成
public void completeUpload(String uploadId) {
    List<Part> parts = getUploadedParts(uploadId);
    // 并发的 completeUpload 调用可创建重复文件
    String fileKey = mergePartsToFile(parts);
    recordFileInDatabase(fileKey);
}
```

---

## 检测清单

### 高风险代码模式
- [ ] 缺乏原子操作的读取-修改-写入（`SELECT` 然后 `UPDATE`）
- [ ] 检查后执行序列（余额检查 -> 扣减、库存检查 -> 预留）
- [ ] 幂等键缺少用户/主体作用域
- [ ] 并发更新实体缺少 `@Version` 注解
- [ ] 插入操作前进行存在性检查（`exists()`）
- [ ] 事务提交前进行缓存写入
- [ ] 非原子计数器递增（Redis `GET` 然后 `SET`）
- [ ] 速率限制缺乏原子操作
- [ ] 消息/事件消费者缺少幂等性
- [ ] 跨服务工作流缺少补偿保护

### 高风险操作
- [ ] 金融交易（支付、退款、积分）
- [ ] 优惠券/折扣应用
- [ ] 一次性令牌消费（OTP、魔法链接、密码重置）
- [ ] 库存/配额管理
- [ ] 座位/资源预留系统
- [ ] 角色/权限更新
- [ ] 文件上传完成
- [ ] 后台任务状态转换

### 框架特定指标

**Spring/JPA:**
- 实体缺少 `@Version`
- `@Transactional` 缺少适当隔离级别
- 并发更新缺少 `@Lock(LockModeType.PESSIMISTIC_WRITE)`
- 自定义查询缺少版本检查

**Django:**
- `.save()` 未配合 `.select_for_update()`
- 更新未使用 `F()` 表达式
- 应设为唯一的字段缺少 `unique=True`
- 未设置事务隔离级别

**Go:**
- Map 访问缺少 `sync.Mutex`
- 文件操作缺少 `flock`
- 数据库操作缺少 `FOR UPDATE`
- Channel 操作缺少适当同步

**Node.js:**
- 异步操作缺少适当锁机制（redlock 等）
- MongoDB 更新缺少 `$inc` 或原子操作符
- 缓存操作缺少原子原语

---

## 严重性评估

**严重:**
- 缺乏原子性的金融操作（支付、退款、余额更新）
- 一次性令牌重用漏洞
- 配额/速率限制绕过导致无限制资源消耗
- 库存系统允许负库存或超售

**高危:**
- 角色/权限竞态条件
- 优惠券/折扣重复应用
- 数据完整性违反（重复记录、丢失更新）
- 关键工作流中的跨服务竞态条件

**中危:**
- 非关键计数器竞态条件
- 审计日志排序问题
- 缓存一致性竞态条件
- 非金融资源分配竞态

---

## 修复指导

1. **使用数据库原子性:**
   - 带条件的单条 `UPDATE` 语句（`WHERE balance >= amount`）
   - 乐观锁配合 `@Version`/version 字段
   - 悲观锁（`SELECT FOR UPDATE`）
   - 唯一约束替代存在性检查

2. **实现幂等性:**
   - 作用域: `{idempotency-key}:{user-id}:{operation}`
   - 存储: 持久化（数据库），非仅缓存
   - TTL: 足够长的重试时间（金融操作 24h+）
   - 执行时机: 副作用发生之前

3. **原子操作:**
   - Redis: `INCR`, `DECRBY`, Lua 脚本
   - SQL: `UPDATE ... SET x = x + 1`
   - MongoDB: `$inc`, `$set` 配合原子操作符
   - 应用层: `AtomicInteger`, `AtomicLong`, synchronized 块

4. **适当的锁机制:**
   - 分布式锁: Redis 配合 fencing token、ZooKeeper
   - 数据库锁: 行级、表级视情况而定
   - 应用锁: 仅适用于单实例部署

5. **事务隔离:**
   - 关键金融操作使用 `SERIALIZABLE`
   - 理解隔离级别异常（幻读等）
   - 事务提交后再发布事件

---

## 常见误报

- **只读操作** 不涉及状态变更
- **真正幂等的操作** 已有适当执行机制
- **可序列化事务** 隔离级别正确
- **具有唯一约束的操作** 碰撞时会失败
- **单实例部署** 已有适当的内存锁（需记录此假设!）

---

## 测试建议

当怀疑存在竞态条件时，建议:

1. **并发请求测试:**
   - 发送 N 个并行请求，输入相同
   - 使用 HTTP/2 多路复用以实现紧密同步
   - 验证状态变更恰好为 N 或 1（非 N*k）

2. **状态验证:**
   - 检查负数计数器、重复记录
   - 验证守恒属性（转账中总余额不变）
   - 审计日志应匹配实际状态变更

3. **负载测试:**
   - 竞态窗口在负载下会变宽
   - 使用真实的数据库/网络延迟进行测试
   - 验证重试风暴下的行为

---

## 最小 PoC 示例
```bash
# 并发扣减/库存
seq 1 50 | xargs -I{} -P50 curl -s "https://api.example.com/purchase?item=1"

# HTTP/2 并发
h2load -n100 -c20 https://api.example.com/checkout

# Redis 原子性间隔
redis-cli INCR account:1:balance &
```

## 配置/修复示例
- 使用分布式锁（Redis Redlock 需注意过期时间/时钟偏移；优先使用合理过期时间 + watchdog 的单租约模式）
- 幂等键: 写操作应先占用槽位（幂等表/事务唯一约束）；响应缓存不应替代幂等控制
- 数据库: 使用乐观锁或悲观锁；对扣减/库存等关键表添加唯一约束/版本字段
- 队列消费: 启用手动确认，避免重复消费导致并发副作用

---

## 参考资料

- OWASP: Time-of-check Time-of-use (TOCTOU)
- CWE-362: Concurrent Execution using Shared Resource with Improper Synchronization
- CWE-366: Race Condition within a Thread
- CWE-367: Time-of-check Time-of-use (TOCTOU) Race Condition
- Database Isolation Levels: ANSI SQL-92 Standard
- Distributed Systems: Designing Data-Intensive Applications (Martin Kleppmann)
