Implementando Limitação e Estrangulamento de Taxa de API

Quando Usar

• Protegendo endpoints de autenticação contra ataques de força bruta e roubo de credenciais
• Prevenindo abuso de API e esgotamento de recursos por scripts automatizados e bots
• Implementando cotas de uso justo para diferentes níveis de consumidores de API (gratuito, premium, empresarial)
• Defendendo contra ataques de negação de serviço na camada de aplicação
• Atendendo a requisitos de conformidade que exigem controles de prevenção de abuso de API

Não use a limitação de taxa como única defesa contra ataques. Combine com autenticação, autorização e regras de WAF.

Pré-requisitos

• Redis 6.0+ para contadores de limite de taxa distribuídos (ou em memória para implantações de instância única)
• Framework de API (Express.js, FastAPI, Spring Boot ou Django REST Framework)
• Sistema de monitoramento para métricas de limite de taxa (Prometheus, CloudWatch, Datadog)
• Compreensão dos padrões de tráfego normais e uso de pico da API
• Ferramenta de teste de carga (k6, Gatling ou Locust) para validar o comportamento do limite de taxa

Fluxo de Trabalho

Etapa 1: Design da Estratégia de Limitação de Taxa

Defina limites de taxa por categoria de endpoint e nível de usuário:

# Rate limit configuration
RATE_LIMITS = {
    # Authentication endpoints (most restrictive)
    "auth": {
        "login": {"requests": 5, "window_seconds": 60, "by": "ip"},
        "register": {"requests": 3, "window_seconds": 300, "by": "ip"},
        "forgot_password": {"requests": 3, "window_seconds": 3600, "by": "ip"},
        "verify_mfa": {"requests": 5, "window_seconds": 300, "by": "user"},
    },
    # Standard API endpoints
    "api": {
        "free": {"requests": 60, "window_seconds": 60, "by": "user"},
        "premium": {"requests": 300, "window_seconds": 60, "by": "user"},
        "enterprise": {"requests": 1000, "window_seconds": 60, "by": "user"},
    },
    # Resource-intensive endpoints
    "expensive": {
        "search": {"requests": 10, "window_seconds": 60, "by": "user"},
        "export": {"requests": 5, "window_seconds": 3600, "by": "user"},
        "bulk_import": {"requests": 2, "window_seconds": 3600, "by": "user"},
    },
    # Global limits
    "global": {
        "per_ip": {"requests": 1000, "window_seconds": 60, "by": "ip"},
        "per_user": {"requests": 5000, "window_seconds": 3600, "by": "user"},
    },
}

Etapa 2: Limitador de Taxa de Janela Deslizante (Redis)

import redis
import time
import hashlib
from functools import wraps
from flask import Flask, request, jsonify, g

app = Flask(__name__)
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

class SlidingWindowRateLimiter:
    """Limitador de taxa de janela deslizante usando conjuntos ordenados do Redis."""

    def __init__(self, redis_conn):
        self.redis = redis_conn

    def is_allowed(self, key, max_requests, window_seconds):
        """Verifica se a solicitação é permitida e a registra."""
        now = time.time()
        window_start = now - window_seconds
        pipe = self.redis.pipeline()

        # Remove entradas expiradas
        pipe.zremrangebyscore(key, 0, window_start)
        # Conta as solicitações na janela atual
        pipe.zcard(key)
        # Adiciona a solicitação atual
        pipe.zadd(key, {f"{now}:{hashlib.md5(str(now).encode()).hexdigest()[:8]}": now})
        # Define o TTL para a chave
        pipe.expire(key, window_seconds + 1)

        results = pipe.execute()
        current_count = results[1]

        if current_count >= max_requests:
            # Calcula o retry-after
            oldest = self.redis.zrange(key, 0, 0, withscores=True)
            if oldest:
                retry_after = int(oldest[0][1] + window_seconds - now) + 1
            else:
                retry_after = window_seconds
            return False, current_count, max_requests, retry_after

        return True, current_count + 1, max_requests, 0

rate_limiter = SlidingWindowRateLimiter(redis_client)

def rate_limit(max_requests, window_seconds, key_func=None):
    """Decorador para limitar a taxa de endpoints de API."""
    def decorator(f):
        @wraps(f)
        def wrapped(*args, **kwargs):
            # Determina a chave de limite de taxa
            if key_func:
                identifier = key_func()
            elif hasattr(g, 'user_id'):
                identifier = f"user:{g.user_id}"
            else:
                identifier = f"ip:{request.remote_addr}"

            key = f"ratelimit:{request.endpoint}:{identifier}"
            allowed, current, limit, retry_after = rate_limiter.is_allowed(

…