Un solo modulo serve a entrambi: riceve l'IP, gli header della richiesta e il cookie _kz, così Django e Flask passano ciascuno i propri. Usa requests con una Session per processo, che tiene aperta la connessione; timeout vale per la connessione e per ogni lettura.
Il tag
Metti il tag nel <head> del tuo template base (base.html), che le pagine di registrazione, login e checkout estendono:
<script src="https://korzen.dev/v1/tag.js" data-site-key="YOUR_SITE_KEY" async></script>
La chiamata
# 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 è l'ordine in cui sono arrivati gli header della richiesta: uno script che copia i valori di un browser ne copia di rado l'ordine. gunicorn lo tiene nelle chiavi HTTP_* dell'environ WSGI (request.META in Django, request.environ in Flask), e scarta un nome con _. Dietro nginx, un load balancer o Cloudflare, è l'ordine in cui quel passaggio li inoltra, non quello del browser.
In Django, nella view che crea l'account:
# 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")
In 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 del visitatore
Django non ha un'impostazione per i proxy: REMOTE_ADDR è l'indirizzo della connessione, quello del tuo proxy se ne hai uno. La client_ip qui sopra prende l'ultima voce di X-Forwarded-For, quella aggiunta dal tuo proxy; con due tuoi proxy davanti, prendi quella precedente. Dietro Cloudflare, leggi CF-Connecting-IP. In Flask, ProxyFix(x_for=1) fa lo stesso per un proxy, e imposta request.remote_addr.
Applica l'azione
Leggi solo action. Su block, rispondi con lo stesso errore che l'azione mostra quando fallisce per qualsiasi altro motivo, così il bot non impara niente. Su challenge, usa la verifica che hai già (un CAPTCHA, un codice via email o telefono); se non ne hai, prosegui. L'account parte in monitor mode, dove action è sempre allow: per i visitatori non cambia niente finché non attivi il block mode.
Conserva il request_id
Ogni risposta ha un request_id. Conservalo insieme a ciò che l'azione ha creato, in una colonna dell'account o dell'ordine, o in una riga di log con il loro id. Identifica quella decisione quando ci chiedi informazioni, e ti permette di scoprire in seguito quale risposta ha fatto entrare un account falso.
Provalo sulla tua macchina
Avvia python manage.py runserver o flask run e aprilo su localhost. Le pagine su localhost accettano sempre il tag e non compaiono mai tra i domini. In locale l'IP è 127.0.0.1 o ::1, e va bene così. In Docker sulla tua macchina, l'app vede invece l'indirizzo del bridge, come 172.17.0.1: la pagina di challenge ti riporta comunque su localhost, perché il tag ha visto lì la tua pagina.
- Imposta
KORZEN_API_KEYnell'ambiente (o nel tuo.env, se ne carichi uno) e apri la pagina di registrazione. Negli strumenti per sviluppatori del browser, il cookie_kzcompare dopo qualche secondo. - Registrati. La decisione compare nella pagina Decisioni della dashboard, sotto il path che hai inviato.
- Verifica il fail-open: imposta una chiave sbagliata e registrati di nuovo. La chiamata risponde
401, e la registrazione va a buon fine.
Errori comuni
- La prima voce di
X-Forwarded-For. La scrive il visitatore, quindi un bot si sceglie il proprio IP. ProxyFixcon unx_forpiù alto dei proxy che hai davvero. Flask si fida allora di una voce scritta dal visitatore.- Un form inviato via JavaScript appena la pagina si carica, prima che il tag abbia inviato i dati: chiama
await window.korzen?.ready()prima di inviarlo. - Chiamare
/v1/decidedal browser. L'API key arriverebbe a ogni visitatore; la chiamata va fatta dal tuo server. - Nessun timeout, o un errore che blocca la registrazione. Qualsiasi risposta diversa da
200, o nessuna risposta in tempo, significa: lascia passare il visitatore.