Uso¶

Montaje mínimo¶

from fastapi import FastAPI
from gatekeeper import Config, GatekeeperMiddleware

app = FastAPI()
app.add_middleware(GatekeeperMiddleware, config=Config())

Config() con sus defectos ya es una configuración sensata: 100 peticiones por minuto por IP, inspección activa, body hasta 1 MiB.

La combinación de ruta/query con valores textuales del body es opt-in: Config(inspect_cross_carrier=True). Puede detectar firmas divididas entre ambos componentes, pero combinar contextos también puede aumentar falsos positivos. Pruébala primero con monitor_only=True sobre tráfico representativo. Los campos excluidos no participan en la combinación; las claves JSON se inspeccionan individualmente, pero no se unen al objetivo de query. La métrica CRS publicada mide el modo normal; no representa el rendimiento de esta opción.

Rate limiting¶

Campo	Defecto	Notas
`rate_limit`	`100`	Peticiones por ventana. `0` desactiva
`rate_window`	`60.0`	Segundos de la ventana deslizante
`rate_backend`	`"memory"`	`"memory"` (un proceso) o `"redis"` (todos los workers)
`redis_url`	`redis://localhost:6379/0`	Con `rate_backend="redis"`
`rate_fail_open`	`True`	Qué hacer si Redis no responde (ver abajo)
`rate_max_keys`	`10000`	IPs distintas que el backend en memoria recuerda (LRU al llegar al tope)
`rate_ipv6_prefix`	`64`	Bits de prefijo IPv6 que se agrupan al contar

Qué cuenta como el mismo cliente

La clave de conteo se normaliza: ::ffff:1.2.3.4 cuenta como 1.2.3.4, y las IPv6 del mismo /64 cuentan juntas. Sin eso, quien tiene un /64 (el prefijo delegable más pequeño) evade el límite rotando direcciones. En la auditoría siempre se registra la IP cruda.

rate_max_keys acota la memoria del backend en memoria: sin tope, inundar de IPs distintas hacía crecer el proceso sin límite (medido: 300.000 IPs → ~245 MB).

Un worker, un contador

Con el backend en memoria cada proceso cuenta lo suyo: 4 workers con rate_limit=100 admiten ~400/min. Para un límite global usa Redis:

Config(rate_backend="redis", redis_url="redis://localhost:6379/0")
# o en YAML: rate_backend: redis

La ventana se actualiza con un script Lua atómico (sin carreras leer-modificar-escribir) y el cliente es redis.asyncio: no bloquea el event loop.

Política si Redis no responde¶

`rate_fail_open`	Comportamiento	Para quién
`True` (defecto)	Se admite la petición, se audita `rate_backend_error`	Disponibilidad primero
`False`	503 con `{"detail": "Service Unavailable"}`	Seguridad estricta

En ambos casos el fallo queda en la auditoría: no hay forma de que Redis caiga en silencio.

Qué inspecciona¶

En este orden, y el primero que salta manda:

IP bloqueada (blocked_ips: IP exacta o CIDR) → 403.
Rate limit → 429.
Ruta + query string decodificadas contra las reglas → 403.
Cabeceras de inspect_headers (User-Agent, Referer, Cookie, X-Forwarded-For) → 403.
Body con cualquier content-type textual (JSON, form, text/plain, XML o sin content-type) → 403. Si el body supera max_body_size, por tamaño declarado o al leer el stream → 413. Si la petición agota el presupuesto de campos (max_fields) o de tiempo de inspección (max_inspection_ms) → 413 también.
Body multipart/form-data: se inspeccionan los nombres de campo y de fichero (los ../../etc/passwd van ahí) → 403. El contenido del fichero no: es binario.
Body application/octet-stream: el content-type no decide, decide el contenido. Si el cuerpo es JSON, se inspecciona como JSON.
Handshake de WebSocket: la ruta, la query y las cabeceras del handshake se inspeccionan, y el rate limit se aplica igual que en HTTP. El handshake rechazado se cierra con el código 1008 (el servidor responde 403).

Qué NO inspecciona (por diseño)

El contenido de los ficheros adjuntos, los frames de un WebSocket (solo el handshake) y los binarios que no son JSON (imagen, PDF, zip…): no son cadenas que inspeccionar sin comerse el fichero entero. Son límites conscientes y están declarados en la medición del corpus externo: no se cuentan como fallos, se cuentan como exclusiones.

Bloqueo y modo observación¶

La respuesta de bloqueo es 403 con JSON:

{"detail": "Forbidden"}

Para medir falsos positivos con tu tráfico real antes de bloquear nada:

Config(monitor_only=True)   # todo se audita, nada se bloquea

También puedes apagar la inspección completa (inspect_requests=False) o desactivar reglas concretas (disabled_rules={"sqli"}).

Packs de reglas¶

Por defecto se evalúan las once familias. Si tu app expone menos superficie —una API JSON no renderiza HTML, un backend sin SQL— puedes activar solo lo que corresponde:

from gatekeeper import Config

Config(rule_packs=["api"])                          # sin xss, ssti ni proto_pollution
Config(rule_packs=["api"], disabled_rules=["cmdi"])  # además, sin comandos

Pack	Para qué
`strict`	las once familias (equivale al defecto)
`web`	web pública con HTML: `api` más `xss`
`api`	APIs JSON/GraphQL: SQLi, NoSQL, traversal, ficheros sensibles, SSRF, Log4Shell, CRLF y comandos
`minimal`	solo SQLi y traversal
`infrastructure`	solo ataques que llegan por destinos, cabeceras o logs

Los packs son opt-in: si no configuras rule_packs, se evalúa todo. Un nombre desconocido es ValueError al arrancar, nunca una app sin protección.

Excluir un campo concreto¶

Si tu app acepta por diseño un valor que una firma bloquearía, exímeno el campo, no la familia:

Config(exclude_fields=["template", "bio", "doc_*"])

Sirve para parámetros de query, cabeceras y punteros JSON (/user/bio o bio). Un payload repartido entre a y b con a exento se sigue detectando: Gatekeeper reconstruye el objetivo sin a y vigila b.

Medir el tráfico real: `gatekeeper report`¶

Antes de bloquear en producción, pon monitor_only=True y deja correr el tráfico. Cuando tengas el log:

gatekeeper report gatekeeper.log          # ranking de reglas, rutas y campos
gatekeeper report gatekeeper.log --json   # salida legible por máquina

Agrupa por (regla, campo): lo que se repite es ruido. El informe propone un exclude_fields o un disabled_rules listo para copiar a tu configuración. Acepta el JSON del logger de Gatekeeper o un log de acceso genérico (GET /ruta?q=... 200). Como la auditoría no registra el contenido de los payloads, este informe tampoco puede filtrar lo que el log no trae.

Rutas exentas¶

Config(allow_paths=["/docs*", "/openapi.json", "/health"])

Globs fnmatch (sensible a mayúsculas). La garantía de seguridad: una ruta que contenga .., barra invertida o codificación residual nunca se exime, así que /docs/../admin no se cuela por /docs/*. El rate limit y el bloqueo de IP siguen aplicando en las rutas exentas.

Auditoría¶

Todo se registra en el logger gatekeeper (nivel INFO para eventos, WARNING para backend caído), como JSON. Por defecto no incluye el contenido del payload: attack_detected trae la regla que saltó y la IP, no el texto.

import logging
logging.basicConfig(level=logging.INFO)

Eventos: attack_detected, rate_limited, blocked_ip, body_too_large, too_many_fields, inspection_timeout y rate_backend_error.

Cabeceras de seguridad¶

Se añaden sin pisar las tuyas. Por defecto: X-Content-Type-Options, X-Frame-Options, Referrer-Policy, Permissions-Policy y HSTS. Para cambiarlas por completo, Config(security_headers={...}).

HSTS

Strict-Transport-Security se envía siempre, también en http://. No rompe el desarrollo local: según el RFC 6797 (§7.2), el navegador tiene que ignorar un STS recibido por una conexión no segura, así que en http:// no hace nada.

Y omitirlo según el esquema de la petición sería peor: en un despliegue con un proxy que termina TLS, la app ve http y el navegador ve https, así que HSTS desaparecería sin avisar justo donde hace falta. Si prefieres controlarlo, quítalo de security_headers.

Reglas propias¶

En código:

Config(custom_rules={
    "token_interno": r"\btk_[a-z0-9]{24}\b",   # se filtra una clave interna
})

O por YAML (extra [yaml]):

# gatekeeper.yaml
rate_limit: 60
rate_window: 60
custom_rules:
  token_interno: '\btk_[a-z0-9]{24}\b'

app.add_middleware(GatekeeperMiddleware, config=Config.from_yaml("gatekeeper.yaml"))

Las reglas personalizadas se evalúan sobre texto decodificado y en minúsculas, iguales que las integradas. A diferencia de las integradas, las tuyas se validan al arrancar: una regex con backtracking exponencial se rechaza con un ValueError que dice exactamente qué parte del patrón es la problema.

Tus reglas pueden colgar el proceso (por eso ahora se comprueban)

Las reglas integradas tienen todos los cuantificadores acotados (\s{0,16}), y un test vigila que nadie introduzca un \s* sin cota: un \s* adyacente sobre una racha de espacios es O(n²).

Las tuyas las escribes tú, y una regex con backtracking exponencial (por ejemplo (a+)+$) puede tardar minutos con una entrada crafted, y eso para el proceso entero, no solo para esa petición. Gatekeeper lo comprueba al arrancar, de forma estática (sin ejecutar el regex: medirlo sería el propio DoS).

Se rechaza (ValueError al arrancar) el riesgo exponencial: cuantificadores anidados sin nada obligatorio detrás ((a+)+, (.*)*), alternativas que se solapan dentro de un grupo repetido ((a|a)+, (a|ab)+) y referencias atrás repetidas ((\w+)\1+). Con esos, 60 caracteres ya bastan para colgar el proceso.
Se avisa (warning en el log; la app arranca) el riesgo cuadrático: cuantificadores anidados con un separador detrás ((?:\w+\.)+\w+). No cuelga con una entrada corta, pero se degrada con el tamaño: medido, 7,4 s con 50.000 caracteres.

Si tu patrón es legítimo y prefieres aceptar el riesgo:

Config(custom_rules={"mia": r"(a+)+$"}, allow_unsafe_custom_rules=True)

Y si puedes acotarlo, mejor: \s{0,16}, [a-z]{1,32}, no \s* ni +. También puedes sacarlo del escaneo con exclude_fields.

Fallar al arrancar, no en producción

Una clave con typo (rate_limt) o un patrón inválido rompe el arranque con ValueError. Nunca te levantas con una protección que crees tener y no tienes.

Ejemplo completo¶

from fastapi import FastAPI
from gatekeeper import Config, GatekeeperMiddleware

app = FastAPI()
app.add_middleware(
    GatekeeperMiddleware,
    config=Config(
        rate_limit=120,
        rate_window=60,
        rate_backend="redis",
        redis_url="redis://localhost:6379/0",
        rule_packs=["web"],                 # families (opt-in; por defecto, todas)
        exclude_fields=["template"],        # ese campo sí puede llevar HTML
        allow_paths=["/docs*", "/openapi.json", "/health"],
        max_body_size=262_144,               # 256 KiB; sube si tu app lo necesita
        max_fields=1000,                     # textos inspeccionados por peticion
        max_inspection_ms=1000,              # presupuesto de CPU por peticion
        inspect_multipart=True,              # nombres de campo y de fichero
        inspect_websockets=True,             # handshake (ruta, query, cabeceras)
    ),
)

@app.get("/health")
async def health() -> dict[str, str]:
    return {"status": "ok"}

Siguiente: Despliegue → · Detección y medición →