はじめに
個人で 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 で本番運用しているものです。