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

Django と Flask

Django または Flask に Bot Protection を導入します。ベーステンプレートのタグ、requests を使った呼び出し用モジュール、プロキシの背後での訪問者の IP を説明します。

1 つのモジュールで両方に対応します。IP、リクエストのヘッダー、_kz Cookie を引数に取るので、Django と Flask はそれぞれ自分の値を渡します。requests をプロセスごとに 1 つの Session で使うため、接続は維持されます。timeout は接続と各読み取りに適用されます。

タグ

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

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

呼び出し

# 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 はリクエストのヘッダーが届いた順序です。ブラウザの値をコピーするスクリプトでも、その順序まで真似ることはまれです。gunicorn はこの順序を WSGI の environ の HTTP_* キー(Django では request.META、Flask では request.environ)に保ち、_ を含む名前は捨てます。nginx、ロードバランサー、Cloudflare の後ろでは、ブラウザの順序ではなく、その中継が転送する順序になります。

Django では、アカウントを作成するビューで呼び出します。

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

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

訪問者の IP

Django にはプロキシ用の設定がありません。REMOTE_ADDR は接続元のアドレスで、プロキシがあればプロキシのアドレスになります。上の client_ip は X-Forwarded-For の最後の値、つまりプロキシが追加した値を使います。自前のプロキシが 2 つある場合は、その 1 つ前の値を使ってください。Cloudflare の背後では CF-Connecting-IP を読みます。Flask では ProxyFix(x_for=1) がプロキシ 1 つ分の同じ処理を行い、request.remote_addr を設定します。

アクションを適用する

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

request_id を保存する

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

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

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

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

よくある間違い

  • X-Forwarded-For の最初の値を使う。訪問者が書き込める値なので、bot が自分の IP を好きに選べてしまいます。
  • 実際のプロキシの数より大きい x_for を ProxyFix に指定する。Flask が訪問者の書いた値を信頼してしまいます。
  • ページの読み込み直後、タグが報告する前に JavaScript でフォームを送信する。送信前に await window.korzen?.ready() を呼んでください。
  • ブラウザーから /v1/decide を呼び出す。API key がすべての訪問者に渡ってしまいます。呼び出しはサーバーで行ってください。
  • タイムアウトがない、またはエラーで新規登録が止まる。200 以外の応答や、時間内に応答がない場合は、訪問者を通すという意味です。