Rate limiting

Instalación de slowapi

Para implementar rate limiting en nuestras aplicaciones FastAPI, utilizaremos la biblioteca slowapi, que es una adaptación de Flask-Limiter específicamente diseñada para FastAPI. Esta biblioteca nos permite controlar la frecuencia de las peticiones de manera sencilla y eficiente.

slowapi se integra perfectamente con FastAPI y proporciona decoradores que podemos aplicar directamente a nuestras rutas para establecer límites de peticiones. La biblioteca utiliza un sistema de ventanas deslizantes para contar las peticiones y aplicar las restricciones correspondientes.

Instalación mediante pip

La instalación de slowapi es directa utilizando el gestor de paquetes pip. Ejecuta el siguiente comando en tu terminal:

pip install slowapi

Para proyectos que requieren un control de versiones más estricto, puedes especificar una versión concreta:

pip install slowapi==0.1.9

Verificación de la instalación

Una vez completada la instalación, podemos verificar que slowapi está correctamente instalado creando un pequeño script de prueba:

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

print("slowapi instalado correctamente")
print(f"Versión disponible: {Limiter.__module__}")

Configuración básica en FastAPI

Para comenzar a utilizar slowapi en nuestra aplicación FastAPI, necesitamos importar los componentes principales y configurar un limitador básico:

from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

# Crear el limitador usando la IP del cliente
limiter = Limiter(key_func=get_remote_address)

# Crear la aplicación FastAPI
app = FastAPI()

# Asociar el limitador con la aplicación
app.state.limiter = limiter

# Configurar el manejador de errores para rate limiting
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

La función get_remote_address extrae automáticamente la dirección IP del cliente desde la petición HTTP, que será utilizada como clave para identificar y limitar las peticiones de cada usuario.

Dependencias adicionales

En algunos casos, especialmente cuando trabajamos con aplicaciones que utilizan proxies o balanceadores de carga, podríamos necesitar instalar dependencias adicionales para el manejo de almacenamiento en memoria:

pip install slowapi[memory]

Esta instalación incluye componentes adicionales que mejoran el rendimiento del sistema de rate limiting cuando se manejan grandes volúmenes de peticiones.

Con slowapi correctamente instalado y configurado, nuestra aplicación FastAPI estará preparada para implementar restricciones de velocidad que protejan nuestros endpoints contra el uso excesivo y los ataques de fuerza bruta.

Aplicar límites por IP y por usuario

Una vez configurado slowapi en nuestra aplicación, podemos implementar diferentes estrategias de limitación según nuestras necesidades de seguridad. Los límites por IP protegen contra ataques automatizados, mientras que los límites por usuario autenticado ofrecen un control más granular sobre el uso de la API.

Límites por dirección IP

La limitación por IP es la forma más básica y efectiva de proteger nuestra API contra ataques de fuerza bruta y uso excesivo. Utilizamos decoradores que se aplican directamente a nuestras rutas:

from fastapi import FastAPI, Request, HTTPException
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter

@app.get("/api/datos")
@limiter.limit("10/minute")
async def obtener_datos(request: Request):
    return {"mensaje": "Datos obtenidos correctamente"}

@app.post("/api/login")
@limiter.limit("5/minute")
async def login(request: Request, credenciales: dict):
    # Lógica de autenticación
    return {"token": "ejemplo_token"}

En este ejemplo, el endpoint /api/datos permite 10 peticiones por minuto por IP, mientras que /api/login tiene un límite más restrictivo de 5 peticiones por minuto, ideal para prevenir ataques de fuerza bruta contra el sistema de autenticación.

Configuración de diferentes ventanas temporales

slowapi admite múltiples formatos de tiempo para definir las ventanas de limitación:

@app.get("/api/publico")
@limiter.limit("100/hour")  # 100 peticiones por hora
async def endpoint_publico(request: Request):
    return {"data": "información pública"}

@app.get("/api/sensible")
@limiter.limit("1/second")  # 1 petición por segundo
async def endpoint_sensible(request: Request):
    return {"data": "información sensible"}

@app.post("/api/upload")
@limiter.limit("3/day")  # 3 peticiones por día
async def subir_archivo(request: Request):
    return {"mensaje": "Archivo procesado"}

Límites por usuario autenticado

Para aplicaciones que requieren autenticación, podemos implementar límites específicos por usuario utilizando una función personalizada que extraiga el identificador del usuario:

from fastapi import Depends
from fastapi.security import HTTPBearer

security = HTTPBearer()

def get_user_id(request: Request) -> str:
    """Extrae el ID del usuario desde el token de autorización"""
    try:
        # Obtener el token del header Authorization
        auth_header = request.headers.get("authorization")
        if not auth_header:
            return get_remote_address(request)  # Fallback a IP
        
        # Extraer el user_id del token (simplificado)
        token = auth_header.replace("Bearer ", "")
        # Aquí decodificarías el JWT y extraerías el user_id
        # Por simplicidad, usamos el token como identificador
        return f"user_{hash(token) % 10000}"
    except:
        return get_remote_address(request)

# Crear limitador específico para usuarios
user_limiter = Limiter(key_func=get_user_id)
app.state.user_limiter = user_limiter

Aplicación de límites por usuario

Con el limitador configurado para usuarios, podemos aplicar restricciones personalizadas basadas en la identidad del usuario autenticado:

@app.get("/api/usuario/perfil")
@user_limiter.limit("50/hour")
async def obtener_perfil(request: Request, token: str = Depends(security)):
    return {"perfil": "datos del usuario"}

@app.post("/api/usuario/actualizar")
@user_limiter.limit("10/hour")
async def actualizar_perfil(request: Request, datos: dict, token: str = Depends(security)):
    return {"mensaje": "Perfil actualizado"}

Combinación de límites IP y usuario

En escenarios más complejos, podemos aplicar múltiples limitadores a un mismo endpoint para proporcionar capas adicionales de protección:

@app.post("/api/operacion-critica")
@limiter.limit("20/hour")  # Límite por IP
@user_limiter.limit("5/hour")  # Límite por usuario
async def operacion_critica(request: Request, datos: dict, token: str = Depends(security)):
    return {"resultado": "operación completada"}

Manejo de excepciones personalizado

Para mejorar la experiencia del usuario, podemos personalizar las respuestas cuando se exceden los límites:

from slowapi.errors import RateLimitExceeded
from fastapi.responses import JSONResponse

@app.exception_handler(RateLimitExceeded)
async def rate_limit_handler(request: Request, exc: RateLimitExceeded):
    response = JSONResponse(
        status_code=429,
        content={
            "error": "Límite de peticiones excedido",
            "detalle": f"Límite: {exc.detail}",
            "reintentar_en": "60 segundos"
        }
    )
    response.headers["Retry-After"] = "60"
    return response

Límites condicionales

También podemos implementar límites dinámicos que varíen según el contexto o tipo de usuario:

def get_dynamic_limit(request: Request) -> str:
    """Determina el límite basado en el tipo de usuario"""
    auth_header = request.headers.get("authorization", "")
    
    if "premium_user" in auth_header:
        return "100/hour"
    elif "basic_user" in auth_header:
        return "20/hour"
    else:
        return "5/hour"  # Usuario no autenticado

@app.get("/api/contenido")
async def obtener_contenido(request: Request):
    # Aplicar límite dinámico
    limite = get_dynamic_limit(request)
    # Lógica para aplicar el límite correspondiente
    return {"contenido": "datos según tipo de usuario"}

Esta aproximación nos permite crear un sistema de rate limiting flexible que se adapta a diferentes tipos de usuarios y escenarios de uso, proporcionando la protección necesaria sin comprometer la funcionalidad para usuarios legítimos.