FastAPI

Diese Anleitung gilt für FastAPI auf Python 3.11 oder neuer, ausgeliefert von uvicorn, und funktioniert in jeder Starlette-App gleich. Der Aufruf verwendet httpx mit einem AsyncClient pro Prozess, der die Verbindung offen hält; asyncio.timeout begrenzt den ganzen Aufruf, die Verbindung eingeschlossen, denn das eigene Timeout von httpx gilt für jeden Schritt einzeln.

Der Tag

Setzen Sie den Tag in den <head> des Templates, das Ihre Registrierungs-, Anmelde- und Checkout-Seiten erweitern:

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

Der Aufruf

# 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 ist die Reihenfolge, in der die Header der Anfrage ankamen: Ein Skript, das die Werte eines Browsers kopiert, kopiert selten ihre Reihenfolge. Anders als WSGI hält ASGI sie, in request.scope["headers"]. Hinter nginx, einem Load Balancer oder Cloudflare ist es die Reihenfolge, in der dieser Hop sie weitergibt, nicht die des Browsers.

In der Route, die das Konto anlegt (Form braucht 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)

Die IP des Besuchers

request.client.host ist die Adresse der Verbindung, solange uvicorn dem Proxy davor nicht vertraut. uvicorn liest X-Forwarded-For von den Adressen, die --forwarded-allow-ips nennt (standardmäßig 127.0.0.1), und setzt den Client auf den letzten Eintrag, den sie nicht angehängt haben: Starten Sie es mit der Adresse Ihres Load Balancers, etwa uvicorn main:app --forwarded-allow-ips 10.0.0.5, oder mit dem Netz, in dem er steht. Hinter Cloudflare lesen Sie stattdessen request.headers.get("cf-connecting-ip"), wenn Ihr Server nur Traffic von Cloudflare annimmt.

Die Aktion anwenden

Lesen Sie nur action. Bei block antworten Sie mit demselben Fehler, den die Aktion zeigt, wenn sie aus einem anderen Grund scheitert, damit ein Bot nichts lernt. Bei challenge nutzen Sie die Prüfung, die Sie schon haben (ein CAPTCHA, ein Code per E-Mail oder Telefon); ohne eine machen Sie weiter. Das Konto beginnt im Monitor-Modus, in dem action immer allow ist: Für Besucher ändert sich nichts, bis Sie den Block-Modus einschalten.

Die request_id aufbewahren

Jede Antwort hat eine request_id. Bewahren Sie sie bei dem auf, was die Aktion erstellt hat, in einer Spalte des Kontos oder der Bestellung, oder in einer Logzeile mit deren ID. Sie benennt diese Entscheidung, wenn Sie uns dazu etwas fragen, und lässt Sie später herausfinden, welche Antwort ein Fake-Konto durchgelassen hat.

Auf Ihrem Rechner testen

Starten Sie uvicorn main:app --reload und öffnen Sie es auf localhost. Seiten auf localhost akzeptieren den Tag immer und erscheinen nie als Domain. Lokal ist die IP 127.0.0.1 oder ::1, und das ist in Ordnung. Läuft die App auf Ihrem Rechner in Docker, sieht sie stattdessen die Adresse der Bridge, etwa 172.17.0.1: Die Challenge-Seite bringt Sie trotzdem zu localhost zurück, denn der Tag hat Ihre Seite dort gesehen.

1. Setzen Sie KORZEN_API_KEY in der Umgebung und öffnen Sie die Registrierungsseite. In den Entwicklertools des Browsers erscheint nach ein paar Sekunden das _kz-Cookie.
2. Registrieren Sie sich. Die Entscheidung erscheint auf der Seite Entscheidungen des Dashboards, unter dem Pfad, den Sie gesendet haben.
3. Prüfen Sie fail-open: Setzen Sie einen falschen Schlüssel und registrieren Sie sich erneut. Der Aufruf antwortet 401, und die Registrierung geht durch.

Häufige Fehler

• --forwarded-allow-ips '*'. Es vertraut jedem Hop, sodass der Client zum ersten Eintrag von X-Forwarded-For wird, den der Besucher schreibt.
• Ein neuer httpx.AsyncClient() für jeden Aufruf. Jeder öffnet seine eigene Verbindung, mit einem TLS-Handshake innerhalb Ihres Timeouts.
• timeout= allein in httpx. Es begrenzt jeden Schritt einzeln (Verbinden, Schreiben, Lesen), sodass der Aufruf ein Vielfaches dauern kann; asyncio.timeout begrenzt ihn ganz.
• /v1/decide aus dem Browser aufrufen. Der API key würde jeden Besucher erreichen; der Aufruf gehört auf Ihren Server.
• Kein Timeout, oder ein Fehler, der die Registrierung stoppt. Jede andere Antwort als 200, und keine Antwort rechtzeitig, bedeutet: den Besucher durchlassen.