はじめに

以下のチュートリアルをハンズオンします。

JWT（JSON Web Token）とは？

JWT（JSON Web Token）とは、JSON形式で表現された認証情報などをURL文字列などとして安全に送受信できるよう、符号化やデジタル署名の仕組みを規定した標準規格。
https://e-words.jp/w/JWT.html#google_vignette

要するに、「JSON ベースのデータを暗号化してつくられる文字列」のこと。

以下の解説も分かりやすい。

セットアップ

requirements.txt

# パスワードのハッシュを処理するため
bcrypt==4.0.1
passlib[bcrypt]

# JWTトークンの生成と検証を行うため
pyjwt

pip install -r requirements.txt

ドキュメントでは、以下のように書かれていたが、上記モジュールでないとサンプルは動作しなかった。

PythonでJWTトークンの生成と検証を行うために、python-joseをインストールする必要があります

下準備

JWTトークンの署名に使用されるランダムな秘密鍵を生成

openssl rand -hex 32

c4c573b0ba30dbf7aed6614d74372409ec0e1f10c4746e558f84aff97d7222ee

本来はこのキーはソースコードに含めてはならず、環境変数などで読み込むようにするなど、管理を徹底する必要がある。

ソース

app.py

from datetime import datetime, timedelta, timezone
from typing import Union

import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from passlib.context import CryptContext
from pydantic import BaseModel

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "c4c573b0ba30dbf7aed6614d74372409ec0e1f10c4746e558f84aff97d7222ee"
# JWTトークンの署名に使用するアルゴリズム
ALGORITHM = "HS256"
# トークンの有効期限
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# passwordは「secret」（ハッシュ化されている）
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    }
}

# トークンのデータモデル
class Token(BaseModel):
    access_token: str
    token_type: str

# トークンに含まれるデータモデル
class TokenData(BaseModel):
    username: Union[str, None] = None

# ユーザーのデータモデル
class User(BaseModel):
    username: str
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None

# データベース内のユーザーデータモデル
class UserInDB(User):
    hashed_password: str

# パスワードのハッシュ化に使用するコンテキスト
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2スキーマの定義
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# FastAPIアプリケーションの初期化
app = FastAPI()

# パスワードの検証関数
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

# パスワードのハッシュ化関数
def get_password_hash(password):
    return pwd_context.hash(password)

# ユーザー名からユーザー情報を取得する関数
def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

# ユーザーの認証関数
def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

# 新しいアクセストークンを生成する
def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# 受け取ったトークンを復号して検証し、現在のユーザーを返す
async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except InvalidTokenError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

# 現在のアクティブなユーザーを取得する関数
async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

# トークンの有効期限を設定
# JWTアクセストークンを作成
# usernameとpasswordが正しければ、access_tokenをreturnする。
@app.post("/token")
async def login_for_access_token(
    form_data: OAuth2PasswordRequestForm = Depends(),
) -> Token:
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")

# 現在のユーザー情報を取得するエンドポイント
@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

# 現在のユーザーのアイテムを取得するエンドポイント
@app.get("/users/me/items/")
async def read_own_items(current_user: User = Depends(get_current_active_user)):
    return [{"item_id": "Foo", "owner": current_user.username}]

実行・動作確認

FastAPI起動

uvicorn app:app --reload

ログインしてJWTトークンを取得

curl -X 'POST' \
  'http://localhost:8000/token' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/x-www-form-urlencoded' \
  -d 'grant_type=&username=johndoe&password=secret&scope=&client_id=&client_secret='

以下のようにトークンが返ってくるればOK。

{"access_token":"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJqb2huZG9lIiwiZXhwIjoxNzIyODY5OTY2fQ.vL3deQZIoHFyliLBSycQVXHlH2gNlKBOkQSD0EMGl-4","token_type":"bearer"}

JWTトークンを使用してデータ取得

上記トークンを利用する。

curl -X 'GET' \
  'http://localhost:8000/users/me/' \
  -H 'accept: application/json' \
  -H 'Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJqb2huZG9lIiwiZXhwIjoxNzIyODY5OTY2fQ.vL3deQZIoHFyliLBSycQVXHlH2gNlKBOkQSD0EMGl-4'

以下のように結果が返ってくれば成功。

{"username":"johndoe","email":"johndoe@example.com","full_name":"John Doe","disabled":false}

参考