🛡️

Cloudflare TurnstileでDiscord認証Botを作った

に公開

Discordサーバー向けに、人間認証Botを作りました。

ユーザーがDiscord上の認証ボタンを押すと、本人だけに認証URLが送られます。

Discord上で人間認証Botが認証パネルを表示し、認証ボタンを押したユーザーに認証URLを送信している画面

そのURLを開いてCloudflare Turnstileの認証を通過すると、Botが認証済みロールを付与します。

全体の流れはこんな感じです。

Discordの認証ボタン

JWT付きの認証URLを発行

Cloudflare Turnstileで人間確認

Flask側で検証

DiscordBotが認証済みロールを付与

データベースは使っていません。

認証に必要な情報はJWTに入れて、短い有効期限を付けています。

  • DiscordユーザーID
  • サーバーID

作ったもの

今回作ったのは、Discordサーバーに参加したユーザーへ「認証済みロール」を付与するためのBotです。

ユーザー側の流れはシンプルです。

  1. Discord上の認証ボタンを押す
  2. Botから認証URLを受け取る
  3. WebページでTurnstile認証をする
  4. 認証に成功するとロールが付く

裏側では、次のものを組み合わせています。

  • DiscordBot
  • Flask
  • JWT
  • Cloudflare Turnstile
役割 使うもの
Discord側の操作 DiscordBot
認証ページ Flask
人間確認 Cloudflare Turnstile
一時的な認証情報 JWT
ロール付与 DiscordBot

小さめの認証システムなので、DBを持たずにJWTで状態を渡す形にしました。

認証フロー

認証の流れは次のようになります。

処理の順番はこうです。

  1. Bot起動時に認証パネルを用意する
  2. ユーザーが「認証する」ボタンを押す
  3. BotがDiscordユーザーIDとサーバーIDを含むJWTを生成する
  4. BotがJWT付きの認証URLを本人だけに送る
  5. ユーザーが認証URLを開く
  6. FlaskがJWTを検証する
  7. Turnstile付きの認証ページを表示する
  8. ユーザーがTurnstile認証を完了する
  9. FlaskがTurnstileの結果をサーバー側で検証する
  10. 検証成功後、Botが認証済みロールを付与する

すでに認証済みロールを持っている場合は、再付与せず成功扱いにしています。

Discord側の処理

Discord側では、認証チャンネルに認証パネルを投稿します。

パネルはEmbedとボタンで構成しています。

ボタンには固定のcustom_idを設定します。

discord.ui.Button(
    label="認証する",
    style=discord.ButtonStyle.primary,
    custom_id=VERIFY_BUTTON_CUSTOM_ID
)

custom_idは、Bot再起動後も既存メッセージ上のボタンを動かすために必要です。

Bot起動時には、永続Viewを登録します。

async def setup_hook(self):
    self.add_view(VerifyButton())

これにより、Botを再起動しても認証ボタンの処理を維持できます。

認証URLの発行

ユーザーが認証ボタンを押すと、BotはJWTを生成します。

token = generate_token(
    interaction.user.id,
    interaction.guild.id
)

そのJWTを使って認証URLを作ります。

url = f"{Config.FLASK_BASE_URL}/verify/{token}"

認証URLはephemeral messageで送信します。

await interaction.response.send_message(
    f"以下のURLから認証してください。\n{url}",
    ephemeral=True
)

ephemeral messageは、ボタンを押した本人だけに見えるメッセージです。

認証URLを通常メッセージで送ると他のユーザーにも見えてしまうため、この用途ではephemeralがおすすめです。

JWTの中身

JWTには、認証に必要な最小限の情報だけを入れています。

フィールド 内容
user_id DiscordユーザーID
guild_id DiscordサーバーID
exp 有効期限
iat 発行時刻

生成処理のイメージです。

from datetime import datetime, timedelta
import jwt

def generate_token(user_id: int, guild_id: int) -> str:
    now = datetime.utcnow()

    payload = {
        "user_id": str(user_id),
        "guild_id": str(guild_id),
        "iat": now,
        "exp": now + timedelta(minutes=Config.TOKEN_EXPIRATION_MINUTES),
    }

    return jwt.encode(
        payload,
        Config.FLASK_SECRET_KEY,
        algorithm="HS256"
    )

署名アルゴリズムはHS256です。

署名キーにはFLASK_SECRET_KEYを使用しています。

認証URLが漏れた際のリスクを下げるため、JWTの有効期限を1分に設定しました。

JWTは使い捨てではない

この構成ではDBを使っていないため、JWTを一度使ったかどうかを記録していません。

つまり、このJWTは厳密には使い捨てトークンではありません。

有効期限内であれば、同じトークンを再利用できる可能性があります。

今回はシンプルにするため、短い有効期限でリスクを抑える形にしました。

より厳密に一度きりにしたい場合は、DBに使用済みトークンを記録する構成が必要になります。

今回は面倒だったんで、短い有効期限にして解決しました。

Flask側の認証ページ

GET /verify/<token>では、まずJWTを検証します。

@auth_bp.route("/verify/<token>", methods=["GET"])
def verify(token):
    payload = verify_token(token)

    if payload is None:
        return render_template("error.html"), 400

    return render_template(
        "verify.html",
        site_key=Config.TURNSTILE_SITE_KEY,
        token=token
    )

無効なトークンや期限切れのトークンであれば、エラーページを返します。

エラーページの画像

有効なトークンであれば、Turnstileウィジェット付きの認証ページを表示します。

Turnstileの表示

認証ページでは、Cloudflare Turnstileのスクリプトを読み込みます。

<script
  src="https://challenges.cloudflare.com/turnstile/v0/api.js"
  async
  defer
></script>

<div
  class="cf-turnstile"
  data-sitekey="{{ site_key }}"
  data-callback="onTurnstileSuccess"
></div>

Turnstile認証に成功すると、ブラウザ側でレスポンスを受け取ります。

そのレスポンスを/verify/<token>/completeへPOSTします。

<script>
async function onTurnstileSuccess(token) {
  const formData = new FormData();
  formData.append("cf-turnstile-response", token);

  const response = await fetch("/verify/{{ token }}/complete", {
    method: "POST",
    body: formData
  });

  if (response.ok) {
    window.location.href = "/success";
  } else {
    alert("認証に失敗しました。もう一度お試しください。");
  }
}
</script>

ブラウザ側で受け取ったTurnstileレスポンスは、そのまま信用しません。

サーバー側でCloudflareのSiteverify APIを呼び出して検証します。

Turnstileのサーバー側検証

Turnstileの検証処理はservices/turnstile.pyに分けています。

import requests

def verify_turnstile(response: str, remote_ip: str | None = None) -> bool:
    data = {
        "secret": Config.TURNSTILE_SECRET_KEY,
        "response": response,
    }

    if remote_ip:
        data["remoteip"] = remote_ip

    try:
        result = requests.post(
            "https://challenges.cloudflare.com/turnstile/v0/siteverify",
            data=data,
            timeout=10
        )

        json_data = result.json()
        return json_data.get("success") is True

    except requests.RequestException:
        return False

successtrueの場合だけ、認証成功として扱います。

Turnstileはフロントに表示するだけでは不十分なので、ここは必ずサーバー側で確認します。

認証完了処理

POST /verify/<token>/completeでは、JWTとTurnstileの両方を検証します。

@auth_bp.route("/verify/<token>/complete", methods=["POST"])
def complete(token):
    payload = verify_token(token)

    if payload is None:
        return {"error": "Invalid token"}, 400

    turnstile_response = request.form.get("cf-turnstile-response")

    if not turnstile_response:
        return {"error": "Missing Turnstile response"}, 400

    ok = verify_turnstile(
        turnstile_response,
        request.remote_addr
    )

    if not ok:
        return {"error": "Turnstile verification failed"}, 400

    user_id = int(payload["user_id"])
    guild_id = int(payload["guild_id"])

    # ここからDiscordロール付与へ進む

認証ページを表示するときだけでなく、完了時にもJWTを検証しています。

そのため、認証ページを開いたまま時間を置きすぎると、完了時に期限切れになります。

有効期限を1分にしているため、認証URLを開いたらそのまま認証する前提です。

FlaskからDiscordBotへ処理を渡す

Flaskは同期的なWebアプリとして動いています。

一方で、DiscordBotはasyncioのイベントループで動いています。

そのため、Flaskのリクエスト処理の中で直接await member.add_roles(role)はできません。

そこで、asyncio.run_coroutine_threadsafe()を使っています。

future = asyncio.run_coroutine_threadsafe(
    member.add_roles(role),
    bot.loop
)

future.result(timeout=10)

これで、FlaskのスレッドからDiscordBotのイベントループへロール付与処理を投げられます。

同期処理から非同期処理へ橋渡しする部分です。

ロール付与

認証に成功したら、JWTのguild_iduser_idを使って、対象のサーバーとメンバーを取得します。

guild = bot.get_guild(guild_id)

if guild is None:
    return {"error": "Guild not found"}, 404

member = guild.get_member(user_id)

if member is None:
    return {"error": "Member not found"}, 404

role = guild.get_role(Config.VERIFIED_ROLE_ID)

if role is None:
    return {"error": "Role not found"}, 404

すでに認証済みロールを持っている場合は、再付与せず成功扱いにします。

if role in member.roles:
    return render_template("success.html", already_verified=True)

まだ持っていなければ、ロールを付与します。

future = asyncio.run_coroutine_threadsafe(
    member.add_roles(role),
    bot.loop
)

future.result(timeout=10)

これで認証完了です。

Discord側で必要な設定

コード側の実装が正しくても、Discord側の設定が不足しているとロール付与に失敗します。

特に重要なのは、次の2つです。

  • Botの権限
  • ロール階層

Botには、対象ロールを付与できる権限が必要です。

サーバーにBotを入れるとき、ロール付与の権限を与えておきましょう。

また、DiscordではBotのロールより上にあるロールは付与できません。

下の画像のような状態だと、オーナーのロールを付与することはできないんです。

ロールの設定画面

そのため、Botのロールは認証済みロールより上に配置しておく必要があります。

この設定が抜けていると、認証処理自体は通っているのにロール付与だけ失敗します。

失敗時の挙動

代表的な失敗ケースは次の通りです。

失敗ケース 挙動
JWTが無効または期限切れ 認証ページではエラー、完了APIでは400
Turnstileレスポンスがない 400
Turnstile検証失敗 400
guildが見つからない 404
memberが見つからない 404
roleが見つからない 404
すでにroleを持っている 認証済みとして成功扱い
ロール付与に失敗 500

ユーザー向けのエラーでは、内部例外や環境変数名をそのまま出さないようにしています。

詳細はログに出し、画面では必要最低限のメッセージにします。

特に理由はないですが、生の情報を出す意味もないかなと思ったので、このような設計にしました。

この構成でよかったところ

この構成で良かったところを書いていきます。

DBなしで作れるところ

DBなしで作れるのは扱いやすかったです。

認証セッションを保存する必要がなく、JWTだけで必要な情報を渡せます。

DiscordユーザーIDとサーバーIDをJWTに入れておけば、Flask側で検証したあとに、どのユーザーへロールを付ければいいか分かります。

認証URLを本人のみに送れる

認証URLをephemeral messageで送れるのも便利でした。

認証URLが本人以外に見えないので、Discord上の導線として扱いやすいです。

間違って他人の認証をすることがないのは安心ですね。

Turnstileも、サーバー側でSiteverify APIを呼ぶ形にすれば、人間確認として組み込みやすいです。

気をつけるところ

この構成で気を付けるところを書いていきます。

JWTが使い捨てではない

DBやRedisを使って使用済みトークンを管理していないため、有効期限内なら同じURLを再利用できる可能性があります。

今回は有効期限を1分にして、そのリスクを小さくしています。

FlaskとDiscordBotを同じプロセス内で動かしている

小さく作るには楽ですが、規模が大きくなるならWebアプリとBotを分けたほうが管理しやすいと思います。

今回は小さめの構成なので、同一プロセスで十分と判断しました。

まとめ:Discord認証は小さく作っても十分実用的

Discordサーバー向けに、人間認証Botを作りました。

構成は次の通りです。

  • DiscordBot
  • Flask
  • Cloudflare Turnstile
  • JWT

ユーザーがDiscordの認証ボタンを押すと、BotがJWT付きの認証URLをephemeral messageで送ります。

ユーザーはそのURLから認証ページを開き、Cloudflare Turnstileで人間確認をします。

Flask側でJWTとTurnstileの検証を行い、成功したらDiscordBotが認証済みロールを付与します。

DBは使っていないので、JWTの有効期限を短くすることで、認証URLが長く使われないようにしています。

実装としてはシンプルですが、次の点は重要でした。

  • 認証URLはephemeral messageで本人だけに送る
  • JWTには必要最小限の情報だけを入れる
  • Turnstileはサーバー側で検証する
  • FlaskからDiscordBotのイベントループへ処理を渡す
  • Botの権限とロール階層を正しく設定する

小さめのDiscordサーバーに人間認証を入れる場合には十分だと思います。

ぜひ参考にしてみてください。

Discussion