Korzen
Bot Protection

Django et Flask

Bot Protection dans Django ou Flask : le tag dans le template de base, un module pour l'appel avec requests, et l'IP du visiteur derrière un proxy.

Un seul module sert aux deux : il prend l'IP, les en-têtes de la requête et le cookie _kz, donc Django et Flask passent chacun les leurs. Il utilise requests avec une Session par processus, qui garde la connexion ouverte ; timeout s'applique à la connexion et à chaque lecture.

Le tag

Placez le tag dans le <head> de votre template de base (base.html), qu'étendent les pages d'inscription, de connexion et de checkout :

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

L'appel

# 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 est l'ordre dans lequel les headers de la requête sont arrivés : un script qui copie les valeurs d'un navigateur en copie rarement l'ordre. gunicorn le garde dans les clés HTTP_* de l'environ WSGI (request.META dans Django, request.environ dans Flask), et écarte un nom avec _. Derrière nginx, un load balancer ou Cloudflare, c'est l'ordre dans lequel ce saut les transmet, pas celui du navigateur.

Dans Django, dans la vue qui crée le compte :

# 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")

Dans 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")

L'IP du visiteur

Django n'a pas de réglage pour les proxies : REMOTE_ADDR est l'adresse de la connexion, celle de votre proxy si vous en avez un. Le client_ip ci-dessus prend la dernière entrée de X-Forwarded-For, celle que votre proxy a ajoutée ; avec deux de vos proxies devant, prenez celle d'avant. Derrière Cloudflare, lisez CF-Connecting-IP. Dans Flask, ProxyFix(x_for=1) fait de même pour un proxy, et définit request.remote_addr.

Appliquer l'action

Ne lisez que action. Avec block, répondez avec la même erreur que l'action affiche quand elle échoue pour toute autre raison, pour qu'un bot n'apprenne rien. Avec challenge, utilisez la vérification que vous avez déjà (un CAPTCHA, un code par e-mail ou téléphone) ; sans aucune, continuez. Le compte commence en mode monitor, où action est toujours allow : rien ne change pour les visiteurs tant que vous n'activez pas le mode block.

Garder le request_id

Chaque réponse a un request_id. Gardez-le avec ce que l'action a créé, dans une colonne du compte ou de la commande, ou dans une ligne de log avec leur identifiant. Il désigne cette décision quand vous nous interrogez à son sujet, et vous permet de retrouver plus tard quelle réponse a laissé entrer un faux compte.

Tester sur votre machine

Lancez python manage.py runserver ou flask run et ouvrez-le sur localhost. Les pages sur localhost acceptent toujours le tag et n'apparaissent jamais comme domaine. En local, l'IP est 127.0.0.1 ou ::1, et c'est très bien. Dans Docker sur votre machine, l'app voit à la place l'adresse du bridge, comme 172.17.0.1 : la page de challenge vous ramène quand même sur localhost, puisque le tag y a vu votre page.

  1. Définissez KORZEN_API_KEY dans l'environnement (ou votre .env, si vous en chargez un) et ouvrez la page d'inscription. Dans les outils de développement du navigateur, le cookie _kz apparaît après quelques secondes.
  2. Inscrivez-vous. La décision apparaît sur la page Décisions du tableau de bord, sous le chemin que vous avez envoyé.
  3. Vérifiez le fail-open : mettez une mauvaise clé et inscrivez-vous à nouveau. L'appel répond 401, et l'inscription passe.

Erreurs fréquentes

  • La première entrée de X-Forwarded-For. Le visiteur l'écrit, donc un bot choisit sa propre IP.
  • ProxyFix avec un x_for plus élevé que vos proxies. Flask fait alors confiance à une entrée écrite par le visiteur.
  • Un formulaire envoyé par JavaScript dès le chargement de la page, avant que le tag n'ait rapporté : appelez await window.korzen?.ready() avant de l'envoyer.
  • Appeler /v1/decide depuis le navigateur. L'API key atteindrait chaque visiteur ; l'appel a sa place sur votre serveur.
  • Pas de timeout, ou une erreur qui bloque l'inscription. Toute réponse autre que 200, ou l'absence de réponse à temps, signifie : laissez passer le visiteur.