はじめに

個人で AI 議事録サービスを開発・運用しています。構成はフロントが Next.js、認証が Supabase Auth、API が FastAPI（AWS 東京リージョンの EC2 上で docker compose 運用）です。

ブラウザは Supabase でログインし、受け取った access token を Authorization: Bearer で FastAPI に送ります。FastAPI 側は リクエストごとに Supabase へ問い合わせず、JWKS（公開鍵）で署名だけを自前で検証 しています。

この記事は、その実装を手順として書き起こしたものです。本番で実際に踏んだ障害（全リクエストが 500、テストがたまに落ちる、ヘッダの alg が鍵の種類を選べてしまう）も、そのまま載せます。

環境

パッケージ	バージョン
fastapi	0.141.1
PyJWT[crypto]	2.13.0
cryptography	50.0.1
httpx	0.27.2

うちの本番・ステージング両方の Supabase プロジェクトは、JWKS に ES256 の鍵を 1 本ずつ 公開しています（2026-09-02 に JWKS を取得して確認）。古いプロジェクトで共有シークレット（HS256）署名のままの場合も、後述の設定で扱えるようにしています。

全体の流れ

ブラウザ ──(Supabaseでログイン)──> access token (ES256署名のJWT)
   │
   └─ Authorization: Bearer <token> ──> FastAPI
                                          ├─ ヘッダの alg を許可リストで確認
                                          ├─ kid に対応する公開鍵を JWKS キャッシュから取得
                                          └─ 署名・exp・aud を検証 → sub をユーザーIDとして使う

JWKS の URL は {SUPABASE_URL}/auth/v1/.well-known/jwks.json です。

手順1: 依存関係は `PyJWT[crypto]` で入れる

# requirements.txt
PyJWT[crypto]==2.13.0
cryptography==50.0.1

[crypto] なしの PyJWT だけだと、ES256 / RS256 の検証に必要な cryptography が入りません。これは後で「つまずいた点」に書く 500 障害の原因そのものなので、明示的に [crypto] を付ける のがおすすめです。

手順2: JWKS を取得してキャッシュする

公開鍵は毎回取りに行かず、1 時間の TTL でプロセス内にキャッシュします（抜粋・簡略化）。

import asyncio, time, logging
from datetime import datetime, timedelta, timezone
from typing import Optional

import httpx
import jwt
from fastapi import HTTPException, status

logger = logging.getLogger(__name__)

_jwks_cache: dict = {}
_jwks_fetched_at: Optional[datetime] = None
_JWKS_TTL = timedelta(hours=1)


def _fetch_jwks_sync() -> dict:
    url = f"{settings.SUPABASE_URL}/auth/v1/.well-known/jwks.json"
    for attempt in range(3):
        try:
            with httpx.Client(timeout=10) as client:
                resp = client.get(url)
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            logger.warning("JWKS fetch attempt %d/3 failed: %s", attempt + 1, e)
            if attempt < 2:
                time.sleep(0.5 * (attempt + 1))
    raise RuntimeError("JWKS fetch failed after 3 attempts")


async def _refresh_jwks_if_stale() -> None:
    global _jwks_cache, _jwks_fetched_at
    stale = _jwks_fetched_at is None or (
        datetime.now(timezone.utc) - _jwks_fetched_at
    ) > _JWKS_TTL
    if not stale:
        return
    try:
        # 同期 httpx をスレッドに逃がし、イベントループを止めない
        _jwks_cache = await asyncio.to_thread(_fetch_jwks_sync)
        _jwks_fetched_at = datetime.now(timezone.utc)
    except Exception as e:
        logger.error("JWKS refresh failed: %s", e)
        if not _jwks_cache:
            # 一度も取れていない → 認証不能なので 503
            raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, "認証サービスに接続できません")
        # 取得済みなら古いキャッシュで続行（Supabase 側の一時障害で全員ログアウトにしない）


def _resolve_key_from_cache(kid: Optional[str]):
    for key_data in _jwks_cache.get("keys", []):
        if kid is None or key_data.get("kid") == kid:
            if key_data.get("kty", "RSA") == "EC":
                return jwt.algorithms.ECAlgorithm.from_jwk(key_data)
            return jwt.algorithms.RSAAlgorithm.from_jwk(key_data)
    raise jwt.InvalidTokenError(f"No matching JWKS key for kid={kid}")

ポイントは 2 つです。

取得失敗時は古いキャッシュで続行。「一度も取れていない」ときだけ 503 にします。
起動時にプリウォーム。FastAPI の lifespan で一度取得しておき、最初のリクエストが JWKS 取得待ちにならないようにしています。失敗しても起動は止めず、初回リクエストで再試行します。

@asynccontextmanager
async def lifespan(app):
    try:
        _get_jwks_public_key(None)  # 同期版で1回取得してキャッシュ
    except Exception as e:
        logger.warning("JWKS pre-warm failed (auth will retry on first request): %s", e)
    yield

手順3: 検証本体 — `alg` は許可リストで固定する

_HMAC_ALGS = frozenset({"HS256", "HS384", "HS512"})

# config: SUPABASE_JWT_ALGS="ES256"（カンマ区切り）
# settings.supabase_jwt_algs -> frozenset({"ES256"})

async def verify_supabase_token(token: str) -> dict:
    try:
        header = jwt.get_unverified_header(token)
        alg = header.get("alg")
        if alg not in settings.supabase_jwt_algs:
            logger.warning("JWT rejected: alg %r not allowed", alg)
            raise jwt.InvalidTokenError(f"Unsupported algorithm: {alg}")

        options = {"leeway": 30}  # コンテナとサーバーの時計ずれ対策
        if alg in _HMAC_ALGS:
            key = settings.SUPABASE_JWT_SECRET
        else:
            await _refresh_jwks_if_stale()
            key = _resolve_key_from_cache(header.get("kid"))

        return jwt.decode(
            token, key,
            algorithms=[alg],          # 許可リストを通った alg だけ
            audience="authenticated",  # Supabase のログインユーザー向けトークン
            options=options,
        )
    except HTTPException:
        raise
    except jwt.ExpiredSignatureError:
        raise HTTPException(401, "セッションの有効期限が切れています")
    except jwt.PyJWTError as e:  # InvalidTokenError ではなく PyJWTError で受ける
        logger.warning("JWT verify failed: %s: %s", type(e).__name__, e)
        raise HTTPException(401, "認証が必要です")

get_unverified_header() の結果は まだ誰も検証していない入力 です。以前はこの alg だけで「共有シークレットで検証するか、JWKS で検証するか」を分岐していました。ES256 で署名しているプロジェクトなら、HS 系は来るはずがありません。そこで、鍵を引く前に許可リストで弾くようにしました（詳しくは後述）。

audience="authenticated" を付けないと、aud クレームを持つ Supabase のトークンは PyJWT がエラーにします。

手順4: FastAPI の依存性にする

from fastapi import Request, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def get_current_user(
    request: Request,
    credentials: HTTPAuthorizationCredentials = Security(security),
) -> dict:
    payload = await verify_supabase_token(credentials.credentials)
    user_id = payload.get("sub")
    if not user_id:
        raise HTTPException(401, "ユーザーが見つかりません")
    # nginx の後ろでは IP が全員同じになるので、レート制限はユーザーIDで行う
    request.state.user_id = user_id
    return {"id": user_id, "email": payload.get("email"), "payload": payload}

ルートでは user = Depends(get_current_user) と書くだけです。request.state.user_id は slowapi のキー関数から読み、アップロードなどコストのかかる API のレート制限をユーザー単位で行っています。

手順5: テストは固定鍵 + 32 バイトにパディングした JWK で

def _ec_jwk(private_key, kid):
    """P-256 public JWK with x/y padded to 32 bytes (RFC 7518 §6.2.1)."""
    nums = private_key.public_key().public_numbers()
    return {
        "kty": "EC", "crv": "P-256",
        "x": base64url_encode(nums.x.to_bytes(32, "big")).decode(),
        "y": base64url_encode(nums.y.to_bytes(32, "big")).decode(),
        "kid": kid,
    }

テスト用の鍵は毎回ランダム生成せず、固定の秘密鍵スカラーから作っています。理由は次の章に書きます。テストは「正しいトークンが通る」だけでなく、次のケースも入れています。

許可リスト外の alg（HS256 に書き換えたトークンなど）が 401 になる
壊れた JWKS 鍵で 500 ではなく 401 になる

つまずいた点

1. python-jose を外したら、認証付きリクエストが全部 500 に（2026-06）

以前は python-jose を使っていて、依存の整理で削除しました。すると、python-jose が 推移的に入れていた cryptography も一緒に消えました。ES256 の検証に使う jwt.algorithms.ECAlgorithm が存在しなくなり、module jwt.algorithms has no attribute ECAlgorithm で全リクエストが 500 になりました。

本番は稼働中のコンテナに pip install してその場で直しました。そのあと PyJWT[crypto] を requirements に明記し、イメージを作り直しています。使っている機能の依存は、推移的依存に任せず直接書く。当たり前ですが、身をもって学びました。

2. テストが約 1/128 の確率で落ちる（2026-07）

PyJWT の ECAlgorithm.to_jwk は、P-256 の座標を最小バイト長で出力します。先頭バイトが 0 の座標だと 31 バイトになり、from_jwk 側は 32 バイトを要求するので InvalidKeyError になります。x・y それぞれ 1/256 なので、ランダムな鍵だと合わせて約 1/128 で踏みます。

実際の Supabase の JWKS は RFC 7518 どおり 32 バイトにパディングされているので、テスト側を固定鍵 + パディング済み JWK に揃えました。直したあと 20 回連続で通ることを確認しています。

3. 壊れた鍵だと 401 ではなく 500 になっていた（2026-07）

2 のテストの失敗を調べているうちに、本番コードのバグにも気づきました。当初は except jwt.InvalidTokenError で受けていましたが、InvalidKeyError は InvalidTokenError の子クラスではありません（どちらも PyJWTError の子）。JWKS の鍵が壊れていると、すべての認証リクエストが未処理例外で 500 になる状態でした。最後の except を jwt.PyJWTError に広げ、壊れた鍵のテストで 401 になることを固定しています。

4. 検証前のヘッダが鍵の種類を選べていた（2026-09）

手順3の話です。攻撃が成立していたわけではありませんが、「未検証の入力が検証方法を決める」構造は JWT で繰り返し問題になってきたパターンです。SUPABASE_JWT_ALGS（既定 ES256）の外の alg は、鍵を引く前に warning ログを出して 401 にしました。HS256 が必要な古いプロジェクトは env で追加します。

5. Next.js 側は `getSession()` ではなく `getUser()`

同じ見直しで、Next.js の Route Handler にも穴が見つかりました。getSession() は Cookie を読むだけで、中身を Supabase に確かめません。そのため、Route 自身の認可チェックには getUser() を使い、FastAPI に転送するトークンは検証が通ったあとで読むように変えました。

よくある疑問

Q. 毎回 Supabase にユーザーを問い合わせたほうが確実では？
確実です。その代わり、全 API リクエストが Supabase への往復を 1 回待つことになり、Supabase が不調なときは API も巻き込まれます。うちでは API は署名検証だけにしています。ただし Next.js 側の重要な Route は getUser() で問い合わせています。署名検証だけだと、ログアウトしても exp まではトークンが有効なままになる点は理解した上で選んでいます。

Q. Supabase の鍵がローテーションされたら？
今の実装では、キャッシュにない kid のトークンは、次の TTL 更新（最大 1 時間）まで 401 になります。今後の改善点で、「知らない kid が来たらレート制限付きで JWKS を取り直す」形にするのが素直だと考えています。

Q. leeway 30 秒は長くない？
コンテナとホストの時計ずれで iat/exp 判定がぶれるのを避ける値です。短い有効期限のトークンに対して 30 秒の余裕を持たせる程度なので、許容範囲と判断しています。

まとめ

PyJWT[crypto] を明示する（推移的依存に頼らない）
JWKS は TTL キャッシュ＋起動時プリウォーム＋失敗時は古いキャッシュで継続
alg は検証前の入力。許可リストで固定してから鍵を選ぶ
例外は PyJWTError で受けて、壊れた鍵でも 500 にしない
テストの EC 鍵は固定値＋32 バイトパディング

この実装は、個人開発している AI 議事録サービス Kaigi AI で本番運用しているものです。