Security Skills Atlas
api security

Implementing API Abuse Detection with Rate Limiting

Implement API abuse detection using token bucket, sliding window, and adaptive rate limiting algorithms to prevent DDoS, brute force, and credential stuffing attacks.

api-abuseapi-gatewayapi-securitybrute-force-preventionddos-protectionrate-limitingsliding-windowtoken-bucket
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

MITRE ATT&CK

T1003 on the official MITRE ATT&CK siteT1059.007 on the official MITRE ATT&CK siteT1110 on the official MITRE ATT&CK siteT1190 on the official MITRE ATT&CK siteT1552.001 on the official MITRE ATT&CK site

NIST CSF 2.0

DE.CM-01 on the official NIST CSF 2.0 siteID.RA-01 on the official NIST CSF 2.0 sitePR.DS-10 on the official NIST CSF 2.0 sitePR.PS-01 on the official NIST CSF 2.0 site

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 import Tuple
 
class TokenBucketRateLimiter:
    def __init__(self, redis_client: redis.Redis,
                 max_tokens: int = 100,
                 refill_rate: float = 10.0,
                 key_prefix: str = "ratelimit:tb"):
        self.redis = redis_client
        self.max_tokens = max_tokens
        self.refill_rate = refill_rate  # tokens per second
        self.key_prefix = key_prefix
 
    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.
        """
        key = self._get_key(client_id)
        now = time.time()
 
        # Atomic token bucket operation using Lua script
        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}
        """
        result = self.redis.eval(
            lua_script, 1, key,
            self.max_tokens, self.refill_rate, now, tokens_required
        )
 
        allowed = bool(result[0])
        remaining = int(result[1])
        retry_after = int(result[2])
 
        return allowed, {
            "remaining": remaining,
            "limit": self.max_tokens,
            "retry_after": retry_after,
            "reset": int(now + (self.max_tokens - remaining) / self.refill_rate)
        }

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.
"""
 
class SlidingWindowRateLimiter:
    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
 
    def allow_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
 
        if not allowed:
            # Remove the request we just added since it's denied
            self.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
 
class ThreatLevel(Enum):
    NORMAL = "normal"
    ELEVATED = "elevated"
    HIGH = "high"
    CRITICAL = "critical"
 
@dataclass
class AdaptiveLimits:
    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),
}
 
class AdaptiveRateLimiter:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.token_bucket = TokenBucketRateLimiter(redis_client)
        self.sliding_window = SlidingWindowRateLimiter(redis_client)
 
    def assess_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)
 
        if not 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 = 0
        if auth_failures > 10:
            score += 3
        elif auth_failures > 5:
            score += 2
        elif auth_failures > 2:
            score += 1
 
        if error_rate > 0.8:
            score += 3
        elif error_rate > 0.5:
            score += 2
 
        if request_rate > 50:
            score += 2
        elif request_rate > 20:
            score += 1
 
        if unique_endpoints > 50:
            score += 2  # Possible enumeration
 
        if score >= 7:
            return ThreatLevel.CRITICAL
        elif score >= 5:
            return ThreatLevel.HIGH
        elif score >= 3:
            return ThreatLevel.ELEVATED
        return ThreatLevel.NORMAL
 
    def allow_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}"
        if self.redis.exists(block_key):
            ttl = self.redis.ttl(block_key)
            return False, {
                "blocked": True,
                "threat_level": threat_level.value,
                "retry_after": ttl,
                "reason": "Temporarily blocked due to suspicious activity"
            }
 
        # Apply rate limit with threat-adjusted parameters
        self.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)
 
        if not allowed and limits.block_duration_seconds > 0:
            # Block the client for the threat-level duration
            self.redis.setex(block_key, limits.block_duration_seconds, threat_level.value)
 
        info["threat_level"] = threat_level.value
        return allowed, info
 
    def record_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()

NGINX Rate Limiting Configuration

# Define rate limit zones
limit_req_zone $binary_remote_addr zone=api_general:10m rate=10r/s;
limit_req_zone $binary_remote_addr zone=api_auth:10m rate=3r/s;
limit_req_zone $binary_remote_addr zone=api_sensitive:10m rate=1r/s;
 
# Apply rate limits to API routes
server {
    listen 443 ssl;
 
    # General API endpoints - 10 req/s with burst of 20
    location /api/v1/ {
        limit_req zone=api_general burst=20 nodelay;
        limit_req_status 429;
        proxy_pass http://api_backend;
    }
 
    # Authentication endpoints - strict 3 req/s
    location /api/v1/auth/ {
        limit_req zone=api_auth burst=5;
        limit_req_status 429;
        proxy_pass http://api_backend;
    }
 
    # Sensitive data endpoints - 1 req/s
    location /api/v1/admin/ {
        limit_req zone=api_sensitive burst=3;
        limit_req_status 429;
        proxy_pass http://api_backend;
    }
 
    # Custom 429 response with Retry-After header
    error_page 429 = @rate_limited;
    location @rate_limited {
        add_header Retry-After 30;
        add_header X-RateLimit-Limit $limit_req_status;
        return 429 '{"error": "rate_limit_exceeded", "retry_after": 30}';
    }
}

Response Headers

Always include standard rate limit headers:

HTTP/1.1 429 Too Many Requests
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 1672531200
Retry-After: 30
Content-Type: application/json
 
{"error": "rate_limit_exceeded", "retry_after": 30}

References

Source materials

References and resources

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

References 1

api-reference.md2.0 KB

API Reference: Implementing API Abuse Detection with Rate Limiting

Redis Token Bucket (Python)

import redis, time
r = redis.Redis()
 
# Lua-based atomic token bucket
lua = """
local tokens = tonumber(redis.call('HGET', KEYS[1], 'tokens') or ARGV[1])
local last = tonumber(redis.call('HGET', KEYS[1], 'last') or ARGV[3])
local elapsed = ARGV[3] - last
tokens = math.min(tonumber(ARGV[1]), tokens + elapsed * tonumber(ARGV[2]))
if tokens >= 1 then
    tokens = tokens - 1
    redis.call('HMSET', KEYS[1], 'tokens', tokens, 'last', ARGV[3])
    return 1
end
return 0
"""
allowed = r.eval(lua, 1, f"rl:{client_ip}", max_tokens, refill_rate, time.time())

Rate Limit Response Headers

Header Description
X-RateLimit-Limit Maximum requests allowed
X-RateLimit-Remaining Requests remaining
X-RateLimit-Reset Unix timestamp when limit resets
Retry-After Seconds until client can retry

NGINX Rate Limiting

limit_req_zone $binary_remote_addr zone=api:10m rate=10r/s;
location /api/ {
    limit_req zone=api burst=20 nodelay;
    limit_req_status 429;
}

Abuse Detection Thresholds

Attack Type Indicator Threshold
Brute Force Auth failures/IP > 10 in 5 min
Credential Stuffing Unique users/IP > 20
API Scraping Requests/IP > 500/hr
Rate Bypass User-Agent rotation > 10 unique UAs

Flask-Limiter

from flask import Flask
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
 
app = Flask(__name__)
limiter = Limiter(get_remote_address, app=app, default_limits=["100/minute"])
 
@app.route("/api/login")
@limiter.limit("5/minute")
def login():
    pass

References

Scripts 1

agent.py6.9 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Agent for implementing API abuse detection with rate limiting analysis."""

import json
import argparse
from datetime import datetime
from collections import defaultdict, Counter


def load_access_logs(log_path):
    """Load API access logs from JSON lines."""
    entries = []
    with open(log_path) as f:
        for line in f:
            try:
                entries.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return entries


def detect_brute_force(logs, failure_threshold=10, window_minutes=5):
    """Detect brute force attacks by counting auth failures per IP."""
    ip_failures = defaultdict(list)
    for entry in logs:
        status = entry.get("status_code", entry.get("status", 0))
        if int(status) in (401, 403):
            ip = entry.get("client_ip", entry.get("ip", ""))
            ts = entry.get("timestamp", "")
            ip_failures[ip].append(ts)
    findings = []
    for ip, timestamps in ip_failures.items():
        if len(timestamps) >= failure_threshold:
            findings.append({
                "client_ip": ip,
                "auth_failures": len(timestamps),
                "severity": "CRITICAL" if len(timestamps) > 50 else "HIGH",
                "category": "brute_force",
                "first_seen": timestamps[0],
                "last_seen": timestamps[-1],
            })
    return sorted(findings, key=lambda x: x["auth_failures"], reverse=True)


def detect_api_scraping(logs, threshold=500):
    """Detect API scraping by high request volume per IP."""
    ip_counts = Counter()
    ip_endpoints = defaultdict(set)
    for entry in logs:
        ip = entry.get("client_ip", entry.get("ip", ""))
        endpoint = entry.get("path", entry.get("endpoint", ""))
        ip_counts[ip] += 1
        ip_endpoints[ip].add(endpoint)
    findings = []
    for ip, count in ip_counts.items():
        if count >= threshold:
            findings.append({
                "client_ip": ip,
                "total_requests": count,
                "unique_endpoints": len(ip_endpoints[ip]),
                "severity": "HIGH",
                "category": "api_scraping",
            })
    return sorted(findings, key=lambda x: x["total_requests"], reverse=True)


def detect_credential_stuffing(logs, threshold=20):
    """Detect credential stuffing: many unique usernames from single IP."""
    ip_users = defaultdict(set)
    for entry in logs:
        if entry.get("path", "").endswith(("/login", "/auth", "/signin")):
            ip = entry.get("client_ip", entry.get("ip", ""))
            user = entry.get("username", entry.get("user", ""))
            if user:
                ip_users[ip].add(user)
    findings = []
    for ip, users in ip_users.items():
        if len(users) >= threshold:
            findings.append({
                "client_ip": ip,
                "unique_usernames": len(users),
                "severity": "CRITICAL",
                "category": "credential_stuffing",
            })
    return sorted(findings, key=lambda x: x["unique_usernames"], reverse=True)


def detect_rate_limit_bypass(logs):
    """Detect attempts to bypass rate limiting."""
    findings = []
    ip_ua_combos = defaultdict(set)
    for entry in logs:
        ip = entry.get("client_ip", entry.get("ip", ""))
        ua = entry.get("user_agent", "")
        ip_ua_combos[ip].add(ua)
    for ip, agents in ip_ua_combos.items():
        if len(agents) >= 10:
            findings.append({
                "client_ip": ip,
                "unique_user_agents": len(agents),
                "severity": "HIGH",
                "category": "ua_rotation",
                "reason": "Rotating User-Agent to bypass rate limits",
            })
    ip_429_count = Counter()
    for entry in logs:
        if int(entry.get("status_code", entry.get("status", 0))) == 429:
            ip = entry.get("client_ip", entry.get("ip", ""))
            ip_429_count[ip] += 1
    for ip, count in ip_429_count.items():
        if count >= 50:
            findings.append({
                "client_ip": ip,
                "rate_limit_hits": count,
                "severity": "MEDIUM",
                "category": "rate_limit_persistence",
                "reason": "Continuing requests after rate limiting",
            })
    return findings


def generate_rate_limit_config(logs):
    """Generate recommended rate limit configuration based on traffic patterns."""
    endpoint_counts = Counter()
    for entry in logs:
        path = entry.get("path", entry.get("endpoint", ""))
        endpoint_counts[path] += 1
    auth_endpoints = [p for p in endpoint_counts if any(
        k in p for k in ["login", "auth", "signin", "register", "password"])]
    config = {
        "global": {"requests_per_minute": 100, "burst": 20},
        "auth_endpoints": {
            "endpoints": auth_endpoints,
            "requests_per_minute": 10,
            "burst": 3,
            "block_duration_seconds": 300,
        },
        "sensitive_endpoints": {
            "endpoints": ["/api/admin", "/api/users", "/api/export"],
            "requests_per_minute": 30,
            "burst": 5,
        },
    }
    return config


def main():
    parser = argparse.ArgumentParser(description="API Abuse Detection Agent")
    parser.add_argument("--log", required=True, help="API access log (JSON lines)")
    parser.add_argument("--output", default="api_abuse_report.json")
    parser.add_argument("--action", choices=[
        "brute_force", "scraping", "stuffing", "bypass", "config", "full_analysis"
    ], default="full_analysis")
    args = parser.parse_args()

    logs = load_access_logs(args.log)
    report = {"generated_at": datetime.utcnow().isoformat(), "total_requests": len(logs),
              "findings": {}}
    print(f"[+] Loaded {len(logs)} API requests")

    if args.action in ("brute_force", "full_analysis"):
        f = detect_brute_force(logs)
        report["findings"]["brute_force"] = f
        print(f"[+] Brute force sources: {len(f)}")

    if args.action in ("scraping", "full_analysis"):
        f = detect_api_scraping(logs)
        report["findings"]["api_scraping"] = f
        print(f"[+] Scraping sources: {len(f)}")

    if args.action in ("stuffing", "full_analysis"):
        f = detect_credential_stuffing(logs)
        report["findings"]["credential_stuffing"] = f
        print(f"[+] Credential stuffing sources: {len(f)}")

    if args.action in ("bypass", "full_analysis"):
        f = detect_rate_limit_bypass(logs)
        report["findings"]["bypass_attempts"] = f
        print(f"[+] Rate limit bypass attempts: {len(f)}")

    if args.action in ("config", "full_analysis"):
        config = generate_rate_limit_config(logs)
        report["findings"]["recommended_config"] = config
        print("[+] Rate limit config generated")

    with open(args.output, "w") as fout:
        json.dump(report, fout, indent=2, default=str)
    print(f"[+] Report saved to {args.output}")


if __name__ == "__main__":
    main()
Keep exploring
Implementing API Gateway Security ControlsImplements security controls at the API gateway layer including authentication enforcement, rate limiting, request validation, IP allowlisting, TLS termination, and threat protection. The engineer configures API gateways (Kong, AWS API Gateway, Azure APIM, Apigee) to act as a centralized security enforcement point that validates, throttles, and monitors all API traffic before it reaches backend services. Activates for requests involving API gateway security, API management security, gateway authentication, or centralized API protection.Implementing API Rate Limiting and ThrottlingImplements API rate limiting and throttling controls using token bucket, sliding window, and fixed window algorithms to protect against brute force attacks, credential stuffing, resource exhaustion, and API abuse. The engineer configures per-user, per-IP, and per-endpoint rate limits using Redis-backed counters, API gateway plugins, or application middleware, and implements proper HTTP 429 responses with Retry-After headers. Activates for requests involving rate limiting implementation, API throttling setup, request quota management, or API abuse prevention.Detecting API Enumeration AttacksDetect and prevent API enumeration attacks including BOLA and IDOR exploitation by monitoring sequential identifier access patterns and authorization failures.Implementing API Schema Validation SecurityImplement API schema validation using OpenAPI specifications and JSON Schema to enforce input/output contracts and prevent injection, data exposure, and mass assignment attacks.Performing API Rate Limiting BypassTests API rate limiting implementations for bypass vulnerabilities by manipulating request headers, IP addresses, HTTP methods, API versions, and encoding schemes to circumvent request throttling controls. The tester identifies rate limit headers, determines enforcement mechanisms, and attempts bypasses including X-Forwarded-For spoofing, parameter pollution, case variation, and endpoint path manipulation. Maps to OWASP API4:2023 Unrestricted Resource Consumption. Activates for requests involving rate limit bypass, API throttling evasion, brute force protection testing, or API abuse prevention assessment.Detecting Broken Object Property Level AuthorizationDetect and test for OWASP API3:2023 Broken Object Property Level Authorization vulnerabilities including excessive data exposure and mass assignment attacks.