Korzen
Bot Protection

Django y Flask

Bot Protection en Django o Flask: el tag en la plantilla base, un módulo para la llamada con requests y la IP del visitante detrás de un proxy.

Un mismo módulo sirve para los dos: recibe la IP, las cabeceras de la petición y la cookie _kz, y cada framework pasa las suyas. Usa requests con una Session por proceso, que mantiene la conexión abierta; timeout vale para la conexión y para cada lectura.

El tag

Pon el tag en el <head> de tu plantilla base (base.html), que extienden las 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 os

import requests

korzen = requests.Session()  # 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"]


def header_order(environ):
    """The order the request's headers came in, as gunicorn keeps it in the WSGI environ."""
    names = [k for k in environ if k.startswith("HTTP_") or k in ("CONTENT_TYPE", "CONTENT_LENGTH")]
    return [k.removeprefix("HTTP_").lower().replace("_", "-") for k in names[:64]]


def decide(ip, headers, environ, token, path):
    """The answer for this request, or None on any error: then go on as on allow (fail open)."""
    try:
        r = korzen.post(
            "https://korzen.dev/v1/decide",
            timeout=0.2,
            headers={"Authorization": f"Bearer {os.environ['KORZEN_API_KEY']}"},
            json={
                "ip": ip,
                "user_agent": headers.get("User-Agent"),
                "headers": {h: headers[h] for h in FORWARDED if h in headers},
                "header_order": header_order(environ),
                "token": token,
                "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. gunicorn lo guarda en las claves HTTP_* del environ de WSGI (request.META en Django, request.environ en Flask), y descarta un nombre con _. 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 Django, en la vista que crea la cuenta:

# views.py
from django.shortcuts import redirect, render

from . import korzen


def client_ip(request):
    # Behind Cloudflare: return request.headers.get("CF-Connecting-IP").
    # Behind one proxy of yours that appends to X-Forwarded-For: its last entry.
    forwarded = request.headers.get("X-Forwarded-For")
    if forwarded:
        return forwarded.split(",")[-1].strip()
    return request.META.get("REMOTE_ADDR")


def signup(request):
    if request.method == "POST":
        d = korzen.decide(client_ip(request), request.headers, request.META, request.COOKIES.get("_kz"), "/signup")
        if d and d["action"] == "block":
            return render(request, "signup.html", {"error": "We could not create your account."}, status=400)
        # "challenge": your CAPTCHA, or an email or phone check, before going on
        user = create_account(request.POST)  # your own code
        save_request_id(user, d and d["request_id"])  # see "Keep the request_id"
        return redirect("welcome")
    return render(request, "signup.html")

En Flask:

# app.py
from flask import Flask, redirect, render_template, request
from werkzeug.middleware.proxy_fix import ProxyFix

import korzen

app = Flask(__name__)
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1)  # one proxy of yours in front


@app.post("/signup")
def signup():
    d = korzen.decide(request.remote_addr, request.headers, request.environ, request.cookies.get("_kz"), "/signup")
    if d and d["action"] == "block":
        return render_template("signup.html", error="We could not create your account."), 400
    # "challenge": your CAPTCHA, or an email or phone check, before going on
    user = create_account(request.form)  # your own code
    save_request_id(user, d and d["request_id"])  # see "Keep the request_id"
    return redirect("/welcome")

La IP del visitante

Django no tiene ajuste para proxies: REMOTE_ADDR es la dirección de la conexión, la de tu proxy si tienes uno. El client_ip de arriba toma la última entrada de X-Forwarded-For, la que añadió tu proxy; con dos proxies tuyos delante, toma la anterior. Detrás de Cloudflare, lee CF-Connecting-IP. En Flask, ProxyFix(x_for=1) hace lo mismo para un proxy y rellena request.remote_addr.

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 python manage.py runserver o flask run 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 (o tu .env, si cargas uno) 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

  • La primera entrada de X-Forwarded-For. La escribe el visitante, así que un bot elige su propia IP.
  • ProxyFix con un x_for mayor que los proxies que tienes. Flask pasa a confiar en una entrada que escribió el visitante.
  • Un formulario enviado por JavaScript nada más cargar la página, antes de que el tag informe: llama a await window.korzen?.ready() antes de enviarlo.
  • 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.