Rate limiting appears simple at first glance — allow N requests per time window, reject the rest. In practice, effective rate limiting requires careful decisions about: which algorithm fits your traffic pattern, what you’re counting against (IP, user, endpoint, or all three), how to handle distributed deployments, and how to communicate limits to legitimate callers.
OWASP API Security Top 10 2023 lists “Unrestricted Resource Consumption” (API4:2023) as a top API security risk. APIs without rate limits are trivially abusable for credential stuffing, scraping, enumeration attacks, and denial of service through resource exhaustion.
The Four Core Algorithms
1. Fixed Window Counter
Divide time into fixed windows (e.g. per minute). Count requests per window. If the count exceeds the limit, reject.
import redis
import time
def fixed_window_check(r: redis.Redis, key: str, limit: int, window: int) -> bool:
"""Returns True if request is allowed."""
current_window = int(time.time()) // window
redis_key = f"ratelimit:{key}:{current_window}"
count = r.incr(redis_key)
if count == 1:
r.expire(redis_key, window)
return count <= limitProblem: boundary bursting. A client can make limit requests at the end of one window and limit more at the start of the next, for 2 * limit requests in a short window. Not suitable for protection against burst abuse.
2. Sliding Window Log
Store a sorted set of request timestamps per client. On each request, remove entries older than the window, count remaining entries, and reject if over limit.
def sliding_window_log_check(r: redis.Redis, key: str, limit: int, window: int) -> bool:
redis_key = f"ratelimit:swl:{key}"
now = time.time()
window_start = now - window
pipe = r.pipeline()
pipe.zremrangebyscore(redis_key, 0, window_start)
pipe.zadd(redis_key, {str(now): now})
pipe.zcard(redis_key)
pipe.expire(redis_key, window)
_, _, count, _ = pipe.execute()
return count <= limitAccurate but memory-intensive: every request timestamp is stored. Fine for moderate traffic, problematic for high-throughput APIs.
3. Sliding Window Counter
A memory-efficient approximation of the sliding window log. Uses two fixed window counters and a weighted calculation:
def sliding_window_counter_check(r: redis.Redis, key: str, limit: int, window: int) -> bool:
now = time.time()
current_window = int(now) // window
previous_window = current_window - 1
current_key = f"ratelimit:swc:{key}:{current_window}"
previous_key = f"ratelimit:swc:{key}:{previous_window}"
pipe = r.pipeline()
pipe.incr(current_key)
pipe.expire(current_key, window * 2)
pipe.get(previous_key)
results = pipe.execute()
current_count = results[0]
previous_count = int(results[2] or 0)
# Weight previous window by how much of it falls within the sliding window
elapsed_in_window = (now % window) / window
estimated_count = previous_count * (1 - elapsed_in_window) + current_count
return estimated_count <= limitThis is the algorithm used by Cloudflare’s rate limiting. Low memory overhead, good accuracy, handles bursts better than fixed window.
4. Token Bucket
Conceptually, a bucket holds tokens (capacity = burst limit). Tokens accumulate at a refill rate. Each request consumes one token. Requests are rejected when the bucket is empty.
def token_bucket_check(r: redis.Redis, key: str, capacity: int,
refill_rate: float) -> bool:
"""
capacity: max tokens (burst limit)
refill_rate: tokens added per second
"""
redis_key = f"ratelimit:tb:{key}"
now = time.time()
# Lua script for atomic token bucket update
lua_script = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local data = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(data[1]) or capacity
local last_refill = tonumber(data[2]) or now
-- Refill tokens based on elapsed time
local elapsed = now - last_refill
tokens = math.min(capacity, tokens + elapsed * refill_rate)
if tokens >= 1 then
tokens = tokens - 1
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
redis.call('EXPIRE', key, math.ceil(capacity / refill_rate) + 1)
return 1
else
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
return 0
end
"""
result = r.eval(lua_script, 1, redis_key, capacity, refill_rate, now)
return bool(result)Token bucket is the best choice for most APIs: it allows short bursts up to capacity while enforcing a sustained rate of refill_rate requests/second. Preferred for user-facing endpoints where occasional burst is legitimate.
Multi-Level Rate Limiting
Effective rate limiting applies at multiple granularities simultaneously:
from functools import wraps
import redis
from flask import request, jsonify
r = redis.Redis(host='redis', port=6379, decode_responses=True)
def rate_limit(per_ip_limit=60, per_user_limit=300, per_endpoint_limit=1000):
def decorator(f):
@wraps(f)
def decorated(*args, **kwargs):
endpoint = request.endpoint
ip = request.remote_addr
user_id = getattr(request, 'user_id', None)
# Layer 1: per-IP limit (catches unauthenticated scraping/crawling)
if not sliding_window_counter_check(r, f"ip:{ip}", per_ip_limit, 60):
return rate_limit_response(retry_after=60)
# Layer 2: per-user limit (catches authenticated abuse)
if user_id and not token_bucket_check(r, f"user:{user_id}", per_user_limit, 5.0):
return rate_limit_response(retry_after=10)
# Layer 3: per-endpoint global limit (protects expensive operations)
if not sliding_window_counter_check(r, f"endpoint:{endpoint}", per_endpoint_limit, 60):
return rate_limit_response(retry_after=5)
return f(*args, **kwargs)
return decorated
return decorator
def rate_limit_response(retry_after: int):
response = jsonify({
"error": "rate_limit_exceeded",
"message": "Too many requests. Please retry after the specified delay.",
"retry_after": retry_after
})
response.status_code = 429
response.headers["Retry-After"] = str(retry_after)
response.headers["X-RateLimit-Reset"] = str(int(time.time()) + retry_after)
return responseCommunicating Limits to Clients
Include rate limit headers in every response, not just 429s:
// Express.js middleware — adds standard rate limit headers
function addRateLimitHeaders(res, limit, remaining, resetTime) {
res.set({
'X-RateLimit-Limit': limit,
'X-RateLimit-Remaining': Math.max(0, remaining),
'X-RateLimit-Reset': resetTime,
'X-RateLimit-Policy': `${limit};w=60` // IETF draft format
});
}
// On 429 responses, add Retry-After per RFC 6585
function send429(res, retryAfterSeconds) {
res.set('Retry-After', retryAfterSeconds);
return res.status(429).json({
error: 'rate_limit_exceeded',
retry_after: retryAfterSeconds
});
}Authentication Endpoint Hardening
Login and password reset endpoints require tighter limits and additional considerations:
- Exponential backoff for repeated failures: After N failed login attempts, increase the delay before the next attempt is allowed
- Account-level lockout (separate from rate limiting): lock specific accounts after M failed attempts regardless of source IP
- CAPTCHA gating after threshold: require CAPTCHA solving after a configurable number of failed attempts
- Credential stuffing detection: monitor for low-rate distributed attacks that stay under per-IP limits by rotating through many source IPs
# Redis-based login attempt tracker with progressive delay
def check_login_allowed(r, username: str) -> tuple[bool, int]:
key = f"login_attempts:{username}"
attempts = int(r.get(key) or 0)
if attempts >= 10:
# Account locked
ttl = r.ttl(key)
return False, max(ttl, 300)
# Progressive delay: 0, 0, 0, 1s, 2s, 4s, 8s, 16s, 32s, 64s
delay = 0 if attempts < 3 else 2 ** (attempts - 3)
return True, delay
def record_failed_login(r, username: str):
key = f"login_attempts:{username}"
attempts = r.incr(key)
# Lock for 15 minutes after 10 attempts
if attempts >= 10:
r.expire(key, 900)
else:
r.expire(key, 3600)Distributed Deployments
In multi-instance deployments, rate limit state must live in a shared store (Redis, Memcached). Without a shared backend, per-instance counters allow N * instance_count requests.
If Redis availability is a concern, implement a fallback:
def rate_limit_with_fallback(redis_client, key, limit, window):
try:
return sliding_window_counter_check(redis_client, key, limit, window)
except redis.RedisError:
# On Redis failure, fail open (allow request) and log the error
# Alternatively, fail closed (deny request) for high-security endpoints
logger.error("Rate limit Redis unavailable — failing open")
return True # or False for high-security pathsConsider Redis Cluster or Sentinel for high-availability rate limiting on critical endpoints. A down rate limiter is a business risk, not just a security risk, if it blocks legitimate traffic.