Korzen
JA
サインイン無料で始める
Bot Protection

FastAPI

FastAPI に Bot Protection を導入します。タグ、1 つのタイムアウトで httpx を使う呼び出し用モジュール、ASGI の scope から得るヘッダーの順序、プロキシの背後の uvicorn を説明します。

このガイドは、uvicorn で配信する Python 3.11 以降の FastAPI を対象にしており、どの Starlette アプリでも同じように動きます。呼び出しには httpx をプロセスごとに 1 つの AsyncClient で使うため、接続は維持されます。httpx 自体のタイムアウトは各段階に別々に適用されるので、asyncio.timeout で接続を含む呼び出し全体を制限します。

タグ

新規登録、ログイン、チェックアウトの各ページが継承するテンプレートの <head> にタグを置きます。

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

呼び出し

# 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 はリクエストのヘッダーが届いた順序です。ブラウザの値をコピーするスクリプトでも、その順序まで真似ることはまれです。WSGI と違い、ASGI はこの順序を request.scope["headers"] に保ちます。nginx、ロードバランサー、Cloudflare の後ろでは、ブラウザの順序ではなく、その中継が転送する順序になります。

アカウントを作成するルートで呼び出します（Form には 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)

訪問者の IP

uvicorn が前段のプロキシを信頼しない限り、request.client.host は接続元のアドレスになります。uvicorn は --forwarded-allow-ips が指定するアドレス（既定は 127.0.0.1）からの X-Forwarded-For を読み、それらが追加していない最後の値をクライアントにします。uvicorn main:app --forwarded-allow-ips 10.0.0.5 のようにロードバランサーのアドレス、またはそれがあるネットワークを指定して起動してください。Cloudflare の背後にあり、サーバーが Cloudflare からのトラフィックしか受け付けない場合は、代わりに request.headers.get("cf-connecting-ip") を読んでください。

アクションを適用する

読むのは action だけです。block の場合は、その操作がほかの理由で失敗したときと同じエラーを返し、bot に何も悟らせないようにします。challenge の場合は、すでにある確認手段（CAPTCHA、メールや電話の確認コード）を使います。何もなければそのまま続けます。アカウントは monitor mode で始まり、その間 action は常に allow です。block mode をオンにするまで、訪問者にとっては何も変わりません。

request_id を保存する

すべての応答には request_id が含まれます。操作で作成されたもの（アカウントや注文）のカラムに保存するか、その ID と一緒にログに残してください。お問い合わせの際にその判定を特定でき、後からどの応答が偽アカウントを通したのかを調べられます。

手元のマシンでテストする

uvicorn main:app --reload を実行し、localhost で開きます。localhost 上のページは常にタグを受け付け、ドメインとして一覧に載ることはありません。ローカルでは IP が 127.0.0.1 または ::1 になりますが、問題ありません。自分のマシンの Docker でアプリを動かすと、代わりにブリッジのアドレス（172.17.0.1 など）になりますが、タグがそこであなたのページを見ているので、チャレンジページは localhost へ戻してくれます。

  1. 環境変数に KORZEN_API_KEY を設定し、新規登録ページを開きます。数秒後、ブラウザーの開発者ツールに _kz Cookie が表示されます。
  2. 新規登録します。判定が dashboard の判定ページに、送信したパスで表示されます。
  3. fail-open を確認します。誤ったキーを設定してもう一度新規登録すると、呼び出しは 401 を返し、新規登録はそのまま完了します。

よくある間違い

  • --forwarded-allow-ips '*'。すべてのホップを信頼するため、クライアントが訪問者の書く X-Forwarded-For の最初の値になります。
  • 呼び出しのたびに新しい httpx.AsyncClient() を作る。それぞれが自分の接続を開き、タイムアウトの中で TLS ハンドシェイクが発生します。
  • httpx の timeout= だけに頼る。各段階（接続、書き込み、読み取り）を別々に制限するので、呼び出しが何倍もかかることがあります。asyncio.timeout は全体を制限します。
  • ブラウザーから /v1/decide を呼び出す。API key がすべての訪問者に渡ってしまいます。呼び出しはサーバーで行ってください。
  • タイムアウトがない、またはエラーで新規登録が止まる。200 以外の応答や、時間内に応答がない場合は、訪問者を通すという意味です。