Implements API abuse detection using token bucket, sliding window, and fixed window rate-limiting algorithms backed by Redis, including adaptive limits that tighten during detected attacks and relax during normal traffic. Use when defending APIs against DDoS, brute force login attempts, credential stuffing, or scraping abuse and you need to design or tune rate-limiting logic.
Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Quelldateien prüfen
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
Implements API abuse detection using token bucket, sliding window, and fixed window rate-limiting algorithms backed by Redis, including adaptive limits that tighten during detected attacks and relax during normal traffic. Use when defending APIs against DDoS, brute force login attempts, credential stuffing, or scraping abuse and you need to design or tune rate-limiting logic.
Implementing API Abuse Detection with Rate Limiting
Overview
API rate limiting is a critical security control that restricts the number of requests a client can make within a defined time period. It defends against denial-of-service (DDoS), brute force login attempts, credential stuffing, API scraping, and resource exhaustion attacks. Modern implementations use algorithms like token bucket, sliding window, and fixed window counters, often backed by distributed stores like Redis. Adaptive rate limiting dynamically tightens limits during detected attacks and relaxes during normal operation, achieving a 94% reduction in successful DDoS attempts compared to static IP-based approaches.
When to Use
When deploying or configuring implementing api abuse detection with rate limiting capabilities in your environment
When establishing security controls aligned to compliance requirements
When building or improving security architecture for this domain
When conducting security assessments that require this implementation
Prerequisites
API gateway (Kong, AWS API Gateway, Apigee) or reverse proxy (NGINX, Envoy)
Redis or Memcached for distributed rate limit counters
Monitoring and alerting infrastructure (Prometheus, Grafana, or SIEM)
Understanding of normal API traffic patterns and baselines
Python 3.8+ or Node.js for custom implementation
Rate Limiting Algorithms
Token Bucket Algorithm
The token bucket assigns each client a bucket with a fixed capacity of tokens. Tokens refill at a constant rate. Each request consumes one token. When the bucket is empty, requests are rejected. This allows controlled bursts while maintaining average limits.
"""Token Bucket Rate Limiter with Redis Backend
Implements a distributed token bucket algorithm for API rate limiting
with burst allowance and automatic refill.
"""import time
import redis
import json
from typing importTupleclassTokenBucketRateLimiter:
def__init__(self, redis_client: redis.Redis,
max_tokens: int = 100,
refill_rate: float = ,
key_prefix: = ):
.redis = redis_client
.max_tokens = max_tokens
.refill_rate = refill_rate
.key_prefix = key_prefix
() -> :
() -> [, ]:
key = ._get_key(client_id)
now = time.time()
lua_script =
result = .redis.(
lua_script, , key,
.max_tokens, .refill_rate, now, tokens_required
)
allowed = (result[])
remaining = (result[])
retry_after = (result[])
allowed, {
: remaining,
: .max_tokens,
: retry_after,
: (now + (.max_tokens - remaining) / .refill_rate)
}
10.0
str
"ratelimit:tb"
self
self
self
# tokens per second
self
def
_get_key
self, client_id: str
str
return
f"{self.key_prefix}:{client_id}"
def
allow_request
self, client_id: str, tokens_required: int = 1
Tuple
bool
dict
"""Check if a request should be allowed under the rate limit.
Returns (allowed, info) where info contains remaining tokens
and retry-after seconds.
"""
self
# Atomic token bucket operation using Lua script
"""
local key = KEYS[1]
local max_tokens = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])
local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(bucket[1])
local last_refill = tonumber(bucket[2])
-- Initialize bucket if it doesn't exist
if tokens == nil then
tokens = max_tokens
last_refill = now
end
-- Calculate refilled tokens
local elapsed = now - last_refill
local refilled = elapsed * refill_rate
tokens = math.min(max_tokens, tokens + refilled)
-- Check if enough tokens available
local allowed = 0
if tokens >= requested then
tokens = tokens - requested
allowed = 1
end
-- Update bucket state
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
redis.call('EXPIRE', key, 3600) -- TTL for cleanup
-- Calculate retry-after if denied
local retry_after = 0
if allowed == 0 then
retry_after = math.ceil((requested - tokens) / refill_rate)
end
return {allowed, math.floor(tokens), retry_after}
"""
self
eval
1
self
self
bool
0
int
1
int
2
return
"remaining"
"limit"
self
"retry_after"
"reset"
int
self
self
Sliding Window Rate Limiter
"""Sliding Window Rate Limiter
Tracks requests over a continuously moving time window,
providing smoother rate limiting than fixed windows with
only a 2.3% false positive rate.
"""classSlidingWindowRateLimiter:
def__init__(self, redis_client: redis.Redis,
window_seconds: int = 60,
max_requests: int = 100,
key_prefix: str = "ratelimit:sw"):
self.redis = redis_client
self.window = window_seconds
self.max_requests = max_requests
self.key_prefix = key_prefix
defallow_request(self, client_id: str) -> Tuple[bool, dict]:
key = f"{self.key_prefix}:{client_id}"
now = time.time()
window_start = now - self.window
# Atomic sliding window using sorted set
pipe = self.redis.pipeline()
# Remove expired entries
pipe.zremrangebyscore(key, 0, window_start)
# Add current request
pipe.zadd(key, {f"{now}:{id(now)}": now})
# Count requests in window
pipe.zcard(key)
# Set TTL
pipe.expire(key, self.window + 1)
results = pipe.execute()
current_count = results[2]
allowed = current_count <= self.max_requests
ifnot allowed:
# Remove the request we just added since it's deniedself.redis.zremrangebyscore(key, now, now)
return allowed, {
"remaining": max(0, self.max_requests - current_count),
"limit": self.max_requests,
"window": self.window,
"current_count": current_count
}
Adaptive Rate Limiter
"""Adaptive Rate Limiter
Dynamically adjusts rate limits based on detected attack patterns.
Tightens limits during attacks and relaxes during normal operation.
"""from enum import Enum
from dataclasses import dataclass
classThreatLevel(Enum):
NORMAL = "normal"
ELEVATED = "elevated"
HIGH = "high"
CRITICAL = "critical"@dataclassclassAdaptiveLimits:
requests_per_minute: int
burst_size: int
block_duration_seconds: int
THREAT_LIMITS = {
ThreatLevel.NORMAL: AdaptiveLimits(100, 20, 0),
ThreatLevel.ELEVATED: AdaptiveLimits(50, 10, 60),
ThreatLevel.HIGH: AdaptiveLimits(20, 5, 300),
ThreatLevel.CRITICAL: AdaptiveLimits(5, 2, 3600),
}
classAdaptiveRateLimiter:
def__init__(self, redis_client: redis.Redis):
self.redis = redis_client
self.token_bucket = TokenBucketRateLimiter(redis_client)
self.sliding_window = SlidingWindowRateLimiter(redis_client)
defassess_threat_level(self, client_id: str) -> ThreatLevel:
"""Assess the current threat level for a client based on behavior."""
metrics_key = f"metrics:{client_id}"
metrics = self.redis.hgetall(metrics_key)
ifnot metrics:
return ThreatLevel.NORMAL
error_rate = float(metrics.get(b'error_rate', 0))
auth_failures = int(metrics.get(b'auth_failures_5m', 0))
unique_endpoints = int(metrics.get(b'unique_endpoints_5m', 0))
request_rate = float(metrics.get(b'requests_per_second', 0))
# Scoring-based threat assessment
score = 0if auth_failures > 10:
score += 3elif auth_failures > 5:
score += 2elif auth_failures > 2:
score += 1if error_rate > 0.8:
score += 3elif error_rate > 0.5:
score += 2if request_rate > 50:
score += 2elif request_rate > 20:
score += 1if unique_endpoints > 50:
score += 2# Possible enumerationif score >= 7:
return ThreatLevel.CRITICAL
elif score >= 5:
return ThreatLevel.HIGH
elif score >= 3:
return ThreatLevel.ELEVATED
return ThreatLevel.NORMAL
defallow_request(self, client_id: str, endpoint: str) -> Tuple[bool, dict]:
"""Rate limit with adaptive thresholds based on threat level."""
threat_level = self.assess_threat_level(client_id)
limits = THREAT_LIMITS[threat_level]
# Check if client is currently blocked
block_key = f"blocked:{client_id}"ifself.redis.exists(block_key):
ttl = self.redis.ttl(block_key)
returnFalse, {
"blocked": True,
"threat_level": threat_level.value,
"retry_after": ttl,
"reason": "Temporarily blocked due to suspicious activity"
}
# Apply rate limit with threat-adjusted parametersself.token_bucket.max_tokens = limits.burst_size
self.token_bucket.refill_rate = limits.requests_per_minute / 60.0
allowed, info = self.token_bucket.allow_request(client_id)
ifnot allowed and limits.block_duration_seconds > 0:
# Block the client for the threat-level durationself.redis.setex(block_key, limits.block_duration_seconds, threat_level.value)
info["threat_level"] = threat_level.value
return allowed, info
defrecord_request_outcome(self, client_id: str, status_code: int, endpoint: str):
"""Track request outcomes for threat assessment."""
metrics_key = f"metrics:{client_id}"
pipe = self.redis.pipeline()
pipe.hincrby(metrics_key, 'total_requests', 1)
if status_code in (401, 403):
pipe.hincrby(metrics_key, 'auth_failures_5m', 1)
if status_code >= 400:
pipe.hincrby(metrics_key, 'errors_5m', 1)
# Track unique endpoints for enumeration detection
pipe.sadd(f"endpoints:{client_id}", endpoint)
pipe.expire(metrics_key, 300) # 5-minute window
pipe.expire(f"endpoints:{client_id}", 300)
pipe.execute()
