FastAPI

Este guia é para FastAPI no Python 3.11 em diante, servido pelo uvicorn, e funciona igual em qualquer app Starlette. A chamada usa httpx com um AsyncClient por processo, que mantém a conexão aberta; o asyncio.timeout limita a chamada inteira, conexão incluída, já que o timeout do próprio httpx vale para cada etapa em separado.

A tag

Coloque a tag no <head> do template que as páginas de cadastro, login e checkout estendem:

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

A chamada

# 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 é a ordem em que os headers da requisição chegaram: um script que copia os valores de um navegador raramente copia a ordem deles. Ao contrário do WSGI, o ASGI guarda essa ordem, em request.scope["headers"]. Atrás do nginx, de um load balancer ou do Cloudflare, é a ordem em que esse salto os repassa, não a do navegador.

Na rota que cria a conta (Form precisa de 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)

O IP do visitante

request.client.host é o endereço da conexão, a menos que o uvicorn confie no proxy da frente. O uvicorn lê o X-Forwarded-For dos endereços que --forwarded-allow-ips nomeia (127.0.0.1 por padrão) e define o client como a última entrada que eles não acrescentaram: rode-o com o endereço do seu load balancer, como uvicorn main:app --forwarded-allow-ips 10.0.0.5, ou com a rede em que ele está. Atrás do Cloudflare, leia request.headers.get("cf-connecting-ip"), se o seu servidor só recebe tráfego do Cloudflare.

Siga a action

Leia só a action. Em block, responda com o mesmo erro que a ação mostra quando falha por qualquer outro motivo, para o bot não aprender nada. Em challenge, use a verificação que você já tem (um CAPTCHA, um código por email ou telefone); sem nenhuma, siga em frente. A conta começa em monitor mode, em que a action é sempre allow: nada muda para os visitantes até você ligar o block mode.

Guarde o request_id

Toda resposta tem um request_id. Guarde-o junto do que a ação criou, numa coluna da conta ou do pedido, ou numa linha de log com o id deles. Com ele, você acha a decisão quando falar com a gente e descobre depois qual resposta deixou passar uma conta falsa.

Teste na sua máquina

Rode uvicorn main:app --reload e abra em localhost. Páginas em localhost sempre aceitam a tag e nunca entram na lista de domínios. Localmente o IP é 127.0.0.1 ou ::1, e não tem problema. No Docker na sua máquina, o app vê em vez disso o endereço da bridge, como 172.17.0.1: a página de desafio ainda traz você de volta para localhost, porque a tag viu sua página lá.

1. Defina KORZEN_API_KEY no ambiente e abra a página de cadastro. Nas ferramentas de desenvolvedor do navegador, o cookie _kz aparece depois de alguns segundos.
2. Cadastre-se. A decisão aparece na página Decisões do painel, com o path que você enviou.
3. Confira o fail open: coloque uma chave errada e cadastre-se de novo. A chamada responde 401, e o cadastro passa.

Erros comuns

• --forwarded-allow-ips '*'. Confia em todo salto, e o client vira a primeira entrada do X-Forwarded-For, que quem escreve é o visitante.
• Um httpx.AsyncClient() novo a cada chamada. Cada um abre a própria conexão, com um handshake TLS dentro do seu timeout.
• Só o timeout= do httpx. Ele limita cada etapa em separado (conexão, escrita, leitura), então a chamada pode levar várias vezes mais; o asyncio.timeout limita a chamada inteira.
• Chamar /v1/decide do navegador. A API key ficaria exposta para todo visitante; a chamada sai do seu servidor.
• Sem timeout, ou um erro que trava o cadastro. Se a resposta não for 200, ou não chegar a tempo, deixe o visitante passar.