How concurrent asynchronous execution creates Time-of-Check to Time-of-Use (TOCTOU) windows, allowing attackers to double-spend account balances and exploit promo codes.
Imagine an ATM checks if you have $100 before giving you cash. If you send two withdrawal requests that arrive at the exact same microsecond, both checks might see '$100 available' before either withdrawal subtracts the money. The ATM dispenses $200 because the check happened before the balance was updated. That gap is a Race Condition.
An attacker observes that an endpoint checks coupon validity: if not coupon.is_redeemed: redeem().
Using HTTP/2 multiplexing or asynchronous scripts, the attacker dispatches 30 identical redemption requests arriving simultaneously.
Multiple server threads query the database simultaneously. All 30 threads see is_redeemed == False before any thread marks it True.
The server applies the $50 discount 30 times, generating $1,500 of unauthorized credits.
# VULNERABLE: Check Followed by Unlocked Update (TOCTOU)
from flask import Flask, request, jsonify
from db import get_db
app = Flask(__name__)
@app.route("/api/coupons/redeem", methods=["POST"])
def redeem_coupon():
code = request.json.get("code")
db = get_db()
# 1. TIME OF CHECK: Queries whether coupon is used
coupon = db.execute("SELECT * FROM coupons WHERE code = ?", (code,)).fetchone()
if not coupon or coupon["used"]:
return jsonify({"error": "Coupon already used"}), 400
# Simulate processing delay (network, payment API) - EXPLOIT WINDOW!
apply_discount(coupon["discount_amount"])
# 2. TIME OF USE: Updates database state too late!
db.execute("UPDATE coupons SET used = TRUE WHERE code = ?", (code,))
db.commit()
return jsonify({"success": True})
# SECURE: Atomic Database Update with Row Locking
from flask import Flask, request, jsonify
from db import get_db
app = Flask(__name__)
@app.route("/api/coupons/redeem", methods=["POST"])
def redeem_coupon():
code = request.json.get("code")
db = get_db()
# Execute state change ATOMICALLY: update only if used is currently FALSE
# The database row lock guarantees only one concurrent query succeeds!
with db.transaction():
cursor = db.execute(
"""
UPDATE coupons
SET used = TRUE, redeemed_at = NOW()
WHERE code = ? AND used = FALSE
RETURNING discount_amount;
""",
(code,)
)
row = cursor.fetchone()
# If rowcount == 0, another concurrent request already claimed the coupon
if not row:
return jsonify({"error": "Coupon invalid or already claimed"}), 400
discount = row["discount_amount"]
apply_discount(discount)
return jsonify({"success": True, "discount": discount})
UPDATE ... WHERE status = 'available') rather than separate SELECT then UPDATE statements.SELECT ... FOR UPDATE) when complex multi-step validations must precede mutations.UNIQUE(user_id, coupon_id)) as an immutable safety backstop.