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 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,

…