Implementando Detecção de Abuso de API com Limitação de Taxa

Visão Geral

A limitação de taxa de API é um controle de segurança crítico que restringe o número de requisições que um cliente pode fazer em um período de tempo definido. Ela defende contra ataques de negação de serviço (DDoS), tentativas de login por força bruta, credential stuffing, raspagem de API e ataques de exaustão de recursos. Implementações modernas usam algoritmos como token bucket, janela deslizante e contadores de janela fixa, frequentemente apoiados por armazenamentos distribuídos como Redis. A limitação de taxa adaptativa aperta dinamicamente os limites durante ataques detectados e relaxa durante a operação normal, alcançando uma redução de 94% nas tentativas de DDoS bem-sucedidas em comparação com abordagens estáticas baseadas em IP.

Quando Usar

• Ao implantar ou configurar a detecção de abuso de API com capacidades de limitação de taxa em seu ambiente
• Ao estabelecer controles de segurança alinhados aos requisitos de conformidade
• Ao construir ou melhorar a arquitetura de segurança para este domínio
• Ao conduzir avaliações de segurança que exigem esta implementação

Pré-requisitos

• Gateway de API (Kong, AWS API Gateway, Apigee) ou proxy reverso (NGINX, Envoy)
• Redis ou Memcached para contadores de limite de taxa distribuídos
• Infraestrutura de monitoramento e alerta (Prometheus, Grafana ou SIEM)
• Compreensão dos padrões de tráfego de API normais e linhas de base
• Python 3.8+ ou Node.js para implementação personalizada

Algoritmos de Limitação de Taxa

Algoritmo Token Bucket

O token bucket atribui a cada cliente um "balde" com uma capacidade fixa de tokens. Os tokens são reabastecidos a uma taxa constante. Cada requisição consome um token. Quando o balde está vazio, as requisições são rejeitadas. Isso permite rajadas controladas enquanto mantém limites médios.

"""Limitador de Taxa Token Bucket com Backend Redis

Implementa um algoritmo distribuído de token bucket para limitação de taxa de API
com permissão para rajadas e reabastecimento automático.
"""

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 por segundo
        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]:
        """Verifica se uma requisição deve ser permitida sob o limite de taxa.

        Retorna (permitido, info) onde info contém tokens restantes
        e segundos de retry-after.
        """
        key = self._get_key(client_id)
        now = time.time()

        # Operação atômica de token bucket usando script Lua
        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])

        -- Inicializa o balde se ele não existir
        if tokens == nil then
            tokens = max_tokens
            last_refill = now
        end

        -- Calcula tokens reabastecidos
        local elapsed = now - last_refill
        local refilled = elapsed * refill_rate
        tokens = math.min(max_tokens, tokens + refilled)

        -- Verifica se há tokens suficientes disponíveis
        local allowed = 0
        if tokens >= requested then
            tokens = tokens - requested
            allowed = 1
        end

        -- Atualiza o estado do balde
        redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
        redis.call('EXPIRE', key, 3600)  -- TTL para limpeza

        -- Calcula retry-after se negado
        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,