FastAPI

Esta guía es para FastAPI en Python 3.11 o posterior, servido por uvicorn, y funciona igual en cualquier app Starlette. La llamada usa httpx con un AsyncClient por proceso, que mantiene abierta la conexión; asyncio.timeout acota la llamada entera, conexión incluida, porque el timeout propio de httpx se aplica a cada paso por separado.

El tag

Pon el tag en el <head> de la plantilla que extienden tus páginas de registro, inicio de sesión y checkout:

<script src="https://korzen.dev/v1/tag.js" data-site-key="YOUR_SITE_KEY" async></script>

La llamada

# korzen.py
import asyncio
import os

import httpx

korzen = httpx.AsyncClient()  # one per process: it keeps the connection open

# The headers Bot Protection reads: those one browser build sends alike, and Accept-Language.
FORWARDED = ["accept", "accept-encoding", "accept-language", "sec-ch-ua", "sec-ch-ua-mobile", "sec-ch-ua-platform",
             "sec-fetch-dest", "sec-fetch-mode", "sec-fetch-site", "upgrade-insecure-requests"]


async def decide(request, path):
    """The answer for this request, or None on any error: then go on as on allow (fail open)."""
    try:
        async with asyncio.timeout(0.2):  # the whole call, the connection included
            r = await korzen.post(
                "https://korzen.dev/v1/decide",
                headers={"Authorization": f"Bearer {os.environ['KORZEN_API_KEY']}", "Korzen-Deadline-Ms": "200"},
                json={
                    "ip": request.client.host if request.client else None,  # see "The visitor's IP"
                    "user_agent": request.headers.get("user-agent"),
                    "headers": {h: request.headers[h] for h in FORWARDED if h in request.headers},
                    # The order the request's headers came in, as the ASGI server keeps it.
                    "header_order": [name.decode("latin-1").lower() for name, _ in request.scope["headers"][:64]],
                    "token": request.cookies.get("_kz"),
                    "path": path,
                    "method": "POST",
                },
            )
        return r.json() if r.status_code == 200 else None
    except Exception:
        return None

header_order es el orden en que llegaron los headers de la petición: un script que copia los valores de un navegador rara vez copia su orden. A diferencia de WSGI, ASGI lo guarda, en request.scope["headers"]. Detrás de nginx, un balanceador de carga o Cloudflare, es el orden en que ese salto los reenvía, no el del navegador.

En la ruta que crea la cuenta (Form necesita python-multipart):

# main.py
from typing import Annotated

from fastapi import FastAPI, Form, Request
from fastapi.responses import PlainTextResponse, RedirectResponse

import korzen

app = FastAPI()


@app.post("/signup")
async def signup(request: Request, email: Annotated[str, Form()]):
    d = await korzen.decide(request, "/signup")
    if d and d["action"] == "block":
        return PlainTextResponse("We could not create your account.", status_code=400)
    # "challenge": your CAPTCHA, or an email or phone check, before going on
    user = await create_account(email)  # your own code
    await save_request_id(user, d and d["request_id"])  # see "Keep the request_id"
    return RedirectResponse("/welcome", status_code=303)

La IP del visitante

request.client.host es la dirección de la conexión, salvo que uvicorn confíe en el proxy que tiene delante. uvicorn lee X-Forwarded-For de las direcciones que nombra --forwarded-allow-ips (127.0.0.1 por defecto) y pone como cliente la última entrada que ellas no añadieron: arráncalo con la dirección de tu load balancer, como uvicorn main:app --forwarded-allow-ips 10.0.0.5, o con la red en la que está. Detrás de Cloudflare, lee en su lugar request.headers.get("cf-connecting-ip"), si tu servidor solo recibe tráfico de Cloudflare.

Actúa según la action

Lee solo action. Con block, responde con el mismo error que la acción muestra cuando falla por cualquier otro motivo, para que un bot no aprenda nada. Con challenge, usa la verificación que ya tienes (un CAPTCHA, un código por email o teléfono); si no tienes ninguna, sigue adelante. La cuenta empieza en monitor mode, donde action es siempre allow: nada cambia para los visitantes hasta que actives el block mode.

Guarda el request_id

Cada respuesta trae un request_id. Guárdalo con lo que la acción creó, en una columna de la cuenta o del pedido, o en una línea de log con su id. Sirve para identificar esa decisión cuando nos preguntes por ella y para averiguar después qué respuesta dejó entrar una cuenta falsa.

Pruébalo en tu máquina

Ejecuta uvicorn main:app --reload y ábrelo en localhost. Las páginas en localhost siempre aceptan el tag y nunca aparecen como dominio. En local la IP es 127.0.0.1 o ::1, y es normal. En Docker en tu máquina, la app ve en su lugar la dirección del bridge, como 172.17.0.1: la página de desafío te devuelve igual a localhost, porque el tag vio tu página allí.

1. Define KORZEN_API_KEY en el entorno y abre la página de registro. En las herramientas de desarrollo del navegador, la cookie _kz aparece tras unos segundos.
2. Regístrate. La decisión aparece en la página Decisiones del panel, con el path que enviaste.
3. Comprueba el fail open: pon una clave incorrecta y regístrate de nuevo. La llamada responde 401, y el registro pasa.

Errores comunes

• --forwarded-allow-ips '*'. Confía en todos los saltos, así que el cliente pasa a ser la primera entrada de X-Forwarded-For, que escribe el visitante.
• Un httpx.AsyncClient() nuevo en cada llamada. Cada uno abre su propia conexión, con un handshake TLS dentro de tu timeout.
• Solo timeout= en httpx. Acota cada paso por separado (conexión, escritura, lectura), así que la llamada puede tardar varias veces más; asyncio.timeout la acota entera.
• Llamar a /v1/decide desde el navegador. La API key quedaría expuesta a cada visitante; la llamada se hace desde tu servidor.
• No poner timeout, o dejar que un error detenga el registro. Cualquier respuesta distinta de 200, o la falta de respuesta a tiempo, significa: deja pasar al visitante.