Contents
JWT の基本概念と Flask‑JWT‑Extended が提供する機能概要
JSON Web Token(JWT) は「ヘッダー・ペイロード・シグネチャ」の三部構造で、署名されたトークン自体に認可情報を持たせることでステートレスな認証が可能になります。本章では JWT の基礎と、Flask アプリにおける access/refresh トークン管理・カスタムクレーム付与・ブラックリスト機能 を概観します。
Access Token と Refresh Token の違いと活用シーン
| トークン種別 | 推奨有効期限 | 主な利用シーン |
|---|---|---|
| Access Token | 5〜15 分 | API エンドポイントへの認可チェック |
| Refresh Token | 7 日以上(HTTPS・Secure Cookie が必須) | /refresh エンドポイントで新しい Access Token を取得 |
- 短命な Access Token は漏洩時の被害範囲を最小化し、長命な Refresh Token は安全に保管すれば再認証不要でトークンローテーションが可能です。
- トークンは 「すぐに使う」 と 「安全に保管する」 の二軸で設計すると、ユーザー体験とセキュリティの両立が実現します。
カスタムクレーム(role)で実装できるシンプル RBAC
JWT に role クレームを埋め込むだけで、サーバ側はデータベース参照なしにロール判定が行えます。以下は典型的なペイロード例です。
|
1 2 3 4 5 6 7 |
{ "sub": "user_123", "role": ["admin", "editor"], "iat": 1729856400, "exp": 1729860000 } |
@jwt_required() デコレータで保護したエンドポイント内で get_jwt()["role"] を取得すれば、ロールごとの処理分岐がシンプルに実装できます。
ポイント:トークンは改ざん防止のため署名されているものの、クライアント側でデコード可能です。したがってサーバ側で必ず再チェックし、信頼できない情報として扱うことが重要です。
プロジェクト構成と必須パッケージのインストール方法
このセクションでは 2025 年以降推奨の Flask‑JWT‑Extended 4.x と Logto SDK を用いた開発環境を、コードベースから機密情報を分離した形で構築する手順を示します。
仮想環境と依存パッケージのインストール
|
1 2 3 4 5 6 7 8 9 10 |
# 1️⃣ 仮想環境作成(推奨) python -m venv .venv source .venv/bin/activate # Windows: .venv\Scripts\activate # 2️⃣ 必要パッケージをインストール pip install "flask-jwt-extended>=4.0" logto-sdk python-dotenv redis flask-cors Flask-Talisman # 3️⃣ バージョン確認(インストール成功の目安) pip list | grep -E "Flask-JWT-Extended|logto-sdk" |
環境変数で安全に設定する項目
プロジェクトルートに .env を作成し、機密情報は .gitignore に追加 してリポジトリから除外します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
FLASK_ENV=development SECRET_KEY=your_flask_secret_key # RS256 用の鍵パス(後述の設定例を参照) PRIVATE_KEY_PATH=/path/to/private_key.pem PUBLIC_KEY_PATH=/path/to/public_key.pem ACCESS_TOKEN_EXPIRES_MINUTES=10 REFRESH_TOKEN_EXPIRES_DAYS=7 LOGTO_ENDPOINT=https://YOUR_TENANT.logto.io LOGTO_M2M_APP_ID=your_m2m_app_id LOGTO_M2M_APP_SECRET=your_m2m_secret |
ディレクトリ構成(ベストプラクティス)
|
1 2 3 4 5 6 7 8 9 10 |
my_flask_jwt_logto/ ├─ app/ │ ├─ __init__.py # アプリファクトリーと設定ロード │ ├─ routes.py # エンドポイント実装 │ └─ utils.py # JWT / Logto 補助関数 ├─ config/ │ └─ settings.py # dotenv 読み込みロジック ├─ tests/ # pytest 用テストコード └─ .env # 環境変数(git 管理外) |
この構成は 「設定はコードから分離」・「テスト容易性」 を同時に満たし、実務プロジェクトで広く採用されています。
認証エンドポイントでの Access / Refresh Token 発行コード例
認証ロジックは POST /login に集約し、成功時に access_token と refresh_token を JSON 形式で返します。以下ではパスワードハッシュ比較とロール情報(カスタムクレーム)を組み込んだ実装例を示します。
POST /login の実装
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 |
# app/routes.py from datetime import timedelta from flask import Blueprint, request, jsonify, current_app from flask_jwt_extended import ( create_access_token, create_refresh_token, ) from werkzeug.security import check_password_hash from .utils import get_user_by_username auth_bp = Blueprint("auth", __name__) @auth_bp.post("/login") def login(): """ユーザー名とパスワードで認証し、トークンを発行する""" payload = request.get_json(silent=True) or {} username = payload.get("username") password = payload.get("password") user = get_user_by_username(username) if not user or not check_password_hash(user.password_hash, password): return jsonify({"code": "AUTH001", "msg": "認証に失敗しました"}), 401 # カスタムクレームとしてロール情報を埋め込む additional_claims = {"role": user.roles} # 例: ["admin"] access_expires = timedelta(minutes=current_app.config["ACCESS_TOKEN_EXPIRES_MINUTES"]) access_token = create_access_token( identity=user.id, additional_claims=additional_claims, expires_delta=access_expires, ) refresh_token = create_refresh_token(identity=user.id) return jsonify( { "access_token": access_token, "refresh_token": refresh_token, "expires_in": access_expires.total_seconds(), } ), 200 |
- ポイント:
create_access_tokenにadditional_claimsを渡すだけでロール情報がトークンに埋め込まれます。 - ベストプラクティス
- Access Token は HTTPOnly・Secure Cookie に保存し、JavaScript から直接参照できないようにします。
- Refresh Token は同様に
SameSite=Laxの Secure Cookie として保持し、CSRF 対策を併用してください。
トークン生成ロジックの再利用
|
1 2 3 4 5 6 7 8 9 10 |
# app/utils.py from flask_jwt_extended import create_access_token, create_refresh_token def generate_tokens(user_id: int, roles: list): """共通化したトークン発行ユーティリティ""" claims = {"role": roles} access = create_access_token(identity=user_id, additional_claims=claims) refresh = create_refresh_token(identity=user_id) return access, refresh |
この関数はユーザー登録後やパスワードリセット時にも流用でき、コードの重複を防止します。
保護ルート実装と Logto 連携による RBAC の具体的手順
本章では @jwt_required() デコレータで認証済みリクエストを保護しつつ、Logto SDK を利用して最新ロール情報を取得・同期するフローを示します。Logto の API エンドポイントはテナント設定によりパスが変わる可能性があるため、実装時は公式ドキュメントでエンドポイントとレスポンス形式を必ず確認してください。
JWT による認可チェックの基本形
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
# app/routes.py(続き) from flask_jwt_extended import jwt_required, get_jwt from .utils import require_roles @auth_bp.get("/admin") @jwt_required() def admin_panel(): """JWT の role クレームだけでシンプルに管理者判定を行う例""" token_claims = get_jwt() if "admin" not in token_claims.get("role", []): return jsonify({"code": "AUTH002", "msg": "管理者権限が必要です"}), 403 return jsonify({"msg": "Welcome, admin!"}) |
get_jwt()が返す辞書にrole配列 があれば、シンプルな Python 条件で認可判断できます。
Logto SDK を用いたロール取得(実装例)
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 |
# app/utils.py import os, json from flask import current_app from logto import ManagementClient _logto_client = ManagementClient( endpoint=os.getenv("LOGTO_ENDPOINT"), client_id=os.getenv("LOGTO_M2M_APP_ID"), client_secret=os.getenv("LOGTO_M2M_APP_SECRET"), ) def fetch_roles_from_logto(user_id: str) -> list: """ Logto の管理 API から指定ユーザーのロール一覧を取得する。 実装例は /api/users/:id/roles エンドポイント(実際のパスはテナントに合わせて調整)。 """ try: resp = _logto_client.get(f"/api/users/{user_id}/roles") data = resp.json() # 期待されるレスポンス例: {"data": [{"id":"...","name":"admin"}, ...]} return [role["name"] for role in data.get("data", [])] except Exception as exc: current_app.logger.error(f"Logto fetch error ({user_id}): {exc}") return [] |
注意:上記エンドポイントは Logto のバージョンやテナント設定に依存します。実装前に
GET /api/users/:id/rolesが有効か、レスポンスのキー (data,nameなど) を必ず確認してください。
before_request フックでロールをリアルタイム同期
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
# app/__init__.py from flask import Flask, g, request from flask_jwt_extended import verify_jwt_in_request, get_jwt_identity from .utils import fetch_roles_from_logto def create_app(): app = Flask(__name__) # 設定読み込みは省略(config/settings.py を参照) @app.before_request def sync_user_roles(): """JWT が付与されたリクエストだけ Logto からロールを取得""" try: verify_jwt_in_request() user_id = get_jwt_identity() g.roles = fetch_roles_from_logto(str(user_id)) except Exception: # 認証が無い/失敗時は空配列にしておく g.roles = [] return app |
g.roles に最新ロールを格納すれば、ハンドラ内で if "admin" in g.roles: と書くだけで権限判定が完了します。
デコレータ形式の汎用 RBAC ユーティリティ
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
# app/utils.py(続き) from functools import wraps from flask import jsonify, g def require_roles(*required): """指定ロールがひとつでもあれば通過させるデコレータ""" def decorator(fn): @wraps(fn) def wrapper(*args, **kwargs): user_roles = getattr(g, "roles", []) if not set(required).intersection(user_roles): return jsonify({"code": "AUTH003", "msg": "必要な権限がありません"}), 403 return fn(*args, **kwargs) return wrapper return decorator # 使用例 @auth_bp.get("/admin") @jwt_required() @require_roles("admin") def admin_panel(): return jsonify({"msg": "管理者専用データ"}) |
エンドポイントごとにロール要件を宣言的に記述でき、コードの可読性が大幅に向上します。
トークンリフレッシュ・失効・エラーハンドリングと最新セキュリティ対策
本節では Refresh Token の安全な使用方法 と ブラックリスト(ブロックリスト)による即時失効 を実装し、さらに RS256 への移行手順を詳述します。
Refresh Token によるアクセストークン再発行 (POST /refresh)
Flask‑JWT‑Extended 4.x では @jwt_required(refresh=True) が推奨されます。以下は更新エンドポイントの実装例です。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
# app/routes.py(続き) from flask_jwt_extended import ( jwt_required, get_jwt_identity, create_access_token, ) @auth_bp.post("/refresh") @jwt_required(refresh=True) # ← 旧 @jwt_refresh_token_required() の代替 def refresh(): """Refresh Token が有効なら新しい Access Token を発行し、ロールは Logto から再取得""" user_id = get_jwt_identity() roles = fetch_roles_from_logto(str(user_id)) new_access = create_access_token( identity=user_id, additional_claims={"role": roles}, ) return jsonify({"access_token": new_access}), 200 |
- ポイント:Refresh Token は
@jwt_required(refresh=True)によって保護され、漏洩リスクが低減します。 - ベストプラクティス:新しい Access Token を生成するたびに Logto からロールを取得すれば、権限変更が即時に反映されます。
Redis を用いたトークンブロックリスト実装(本番向け)
インメモリ set はプロセス終了で消失し、水平スケーリングにも対応できません。以下は Redis に JTI と TTL を保存する永続化ブロックリストのサンプルです。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 |
# app/__init__.py(追加) import redis from flask_jwt_extended import JWTManager, get_jwt app = Flask(__name__) jwt = JWTManager(app) # Redis クライアント初期化(環境変数で接続情報を管理) redis_client = redis.StrictRedis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379/0")) # JTI をキー、"revoked" 文字列を値にし、有効期限はトークンの exp と同じに設定 @jwt.token_in_blocklist_loader def check_if_token_revoked(jwt_header, jwt_payload): jti = jwt_payload["jti"] return redis_client.get(jti) == b"revoked" # logout エンドポイント(access と refresh 両方を失効させる例) @auth_bp.post("/logout") @jwt_required() def logout(): jti = get_jwt()["jti"] exp_timestamp = get_jwt()["exp"] ttl = exp_timestamp - int(time.time()) # Redis に revocation 記録 redis_client.setex(jti, ttl, "revoked") return jsonify({"msg": "ログアウトしました"}), 200 |
- TTL の設定:トークンの有効期限と同じ TTL を付与すれば、Redis のメモリ使用量が自然に削減されます。
- 水平スケーリング:複数 Flask ワーカー・コンテナから同一 Redis にアクセスできるため、どのインスタンスでも失効状態を即座に参照できます。
RS256 署名アルゴリズムへの移行手順
非対称鍵(RS256)へ切り替えることで、秘密鍵がサーバ側だけに残り、公開鍵は検証用に配布可能 になるため、キー管理の安全性が向上します。
鍵ファイルを環境変数で読み込む例
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
# config/settings.py import os BASE_DIR = os.path.abspath(os.path.dirname(__file__)) def load_key(path_env: str) -> str: """PEM ファイルの内容を文字列として返す。パスが未設定の場合は空文字""" path = os.getenv(path_env) if not path: return "" with open(path, "r", encoding="utf-8") as f: return f.read() JWT_PRIVATE_KEY = load_key("PRIVATE_KEY_PATH") JWT_PUBLIC_KEY = load_key("PUBLIC_KEY_PATH") # Flask‑JWT‑Extended の設定 JWT_ALGORITHM = "RS256" |
|
1 2 3 4 5 |
# app/__init__.py(設定ロード部) app.config["JWT_PRIVATE_KEY"] = JWT_PRIVATE_KEY app.config["JWT_PUBLIC_KEY"] = JWT_PUBLIC_KEY app.config["JWT_ALGORITHM"] = "RS256" |
- 環境変数でパスを管理すれば、Docker イメージや CI/CD パイプラインに鍵ファイルを埋め込む必要がなく、安全性が保たれます。
- 本番環境では 権限最小化されたサービスアカウント が鍵ファイルへアクセスできるよう、OS のファイルパーミッションも合わせて設定してください。
統一された JSON エラー形式と Flask の例外ハンドラ
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
# app/__init__.py(エラーハンドリング) from flask import jsonify def json_error(code, msg, status): response = jsonify({"code": code, "msg": msg, "detail": None}) response.status_code = status return response @app.errorhandler(401) def handle_401(e): return json_error("AUTH001", "認証が必要です", 401) @app.errorhandler(403) def handle_403(e): return json_error("AUTH002", "権限が不足しています", 403) @app.errorhandler(422) def handle_422(e): return json_error("VALIDATION_ERR", "入力値の検証に失敗しました", 422) |
エラーコードは AUTH001~ 系列で統一し、フロントエンド側が簡単にメッセージマッピングできるようにします。
最新セキュリティ対策まとめ
| 対策 | 推奨設定例 |
|---|---|
| 署名アルゴリズム | RS256(公開鍵は PUBLIC_KEY_PATH、秘密鍵は PRIVATE_KEY_PATH) |
| HTTPS 強制 | Flask‑Talisman → Talisman(app, force_https=True) |
| CORS 設定 | flask-cors → CORS(app, resources={r"/api/*": {"origins": "https://myfrontend.example.com"}}) |
| Cookie セキュリティ | set_cookie(..., httponly=True, secure=True, samesite='Lax') |
| トークン有効期限 | Access: 10 分、Refresh: 7 日 |
| ブロックリスト永続化 | Redis(TTL=トークン残存時間) |
開発・本番導入チェックリスト
- 環境変数の確認
PRIVATE_KEY_PATH/PUBLIC_KEY_PATHが正しい PEM ファイルを指すか。-
REDIS_URLが本番 Redis インスタンスに接続できるか。 -
HTTPS・CORS のテスト
- ローカルでも
mkcertで自己署名証明書を作成し、curl -k https://localhost/api/...が成功すること。 -
フロントエンドからのリクエストが CORS エラーなく通過するか。
-
トークンフローの検証(Postman / curl)
/login→access_tokenとrefresh_tokenが取得できる。Authorization: Bearer <access>で保護エンドポイントにアクセスし、期待通りステータスが返る。/refreshに Refresh Token を送って新しい Access Token が生成される。-
/logout後に同じ Access Token で再度リクエストすると 401 が返る。 -
Logto ロール同期の確認
-
任意ユーザーのロールを Logto コンソールで変更し、即座に
/adminのアクセス結果が変わるか。 -
Redis ブロックリスト動作の確認
- 複数ワーカー(Docker Compose で
scale)を起動し、どのインスタンスでも logout 後にトークンが失効していること。
まとめ
- JWT と Flask‑JWT‑Extended を正しく設定すれば、access/refresh の自動管理・カスタムクレーム付与・ブラックリスト機能がシンプルに実装できます。
- Logto SDK によるロール取得を
before_requestで同期すれば、権限変更がリアルタイムに反映され、RBAC が安全かつ柔軟になります。 - RS256 署名 と Redis 永続化ブロックリスト を導入することで、本番環境でも鍵管理・トークン失効の要件を満たします。
- エラーハンドリングと統一フォーマット、HTTPS/CORS/Cookie のベストプラクティス も併せて実装すれば、全体として堅牢かつ保守しやすい認証基盤が完成します。
このガイドをそのままプロジェクトに組み込めば、数時間で「安全な JWT 認証 + Logto 連携」の最小構成が手に入ります。ぜひ実装・テストを繰り返し、運用上の要件に合わせて微調整してください。