FastAPIでJWT認証を実装する方法

プログラミング

FastAPIでのJWT認証実装

FastAPIでJWT(JSON Web Token)認証を実装することは、RESTful APIにおけるセキュリティを確保する上で非常に一般的かつ効果的な手法です。JWTは、ユーザーの認証情報や権限などの情報を、署名付きのJSON形式で安全にやり取りするための標準規格です。FastAPIはPythonのフレームワークであり、その非同期処理能力と高いパフォーマンスは、JWT認証の実装にも適しています。

JWT認証の基本

JWTは、主に3つの部分から構成されます。それぞれがドット(.)で区切られており、HeaderPayloadSignatureです。

  • Header: トークンのタイプ(例: JWT)と署名に使用されるアルゴリズム(例: HS256、RS256)が記述されます。
  • Payload: クレーム(Claims)と呼ばれる、エンティティ(通常はユーザー)に関する情報が含まれます。公開クレーム(Registered Claims)とプライベートクレーム(Private Claims)があります。公開クレームには、iss(発行者)、exp(有効期限)、sub(サブジェクト)などがあります。
  • Signature: ヘッダーとペイロードをBase64エンコードし、秘密鍵(Secret Key)または公開鍵/秘密鍵ペアを使用して署名されます。これにより、トークンの改ざんを防ぎ、発行者を確認することができます。

JWT認証の基本的な流れは以下のようになります。

  1. ユーザーは、ユーザー名とパスワードなどの認証情報を使用してAPIにログインリクエストを送信します。
  2. サーバー(FastAPIアプリケーション)は、送信された認証情報を検証します。
  3. 認証が成功した場合、サーバーはユーザーIDや権限などの情報をペイロードに含めたJWTを生成します。
  4. 生成されたJWTは、クライアントに返されます。
  5. クライアントは、以降のAPIリクエストのAuthorizationヘッダーにBearerトークンとしてJWTを含めて送信します。
  6. サーバーは、リクエストヘッダーからJWTを受け取り、署名を検証してトークンが改ざんされていないか、有効期限が切れていないかなどを確認します。
  7. 検証が成功した場合、サーバーはJWTに含まれる情報(ユーザーIDなど)を使用して、リクエストを処理します。

FastAPIでのJWT実装に必要なライブラリ

FastAPIでJWT認証を実装するには、主に以下のライブラリが利用されます。

  • fastapi: FastAPIフレームワーク本体です。
  • uvicorn: ASGIサーバーとして、FastAPIアプリケーションを実行するために使用します。
  • python-jose: JWTのエンコード、デコード、署名、検証を行うためのライブラリです。
  • passlib: パスワードのハッシュ化に利用します。
  • python-multipart: フォームデータ(ユーザー名・パスワードの送信など)を扱う際に必要になることがあります。

これらのライブラリは、pipを使用してインストールできます。

pip install fastapi uvicorn python-jose[cryptography] passlib python-multipart

JWT認証の実装手順

1. 設定ファイルの準備

JWTの生成と検証に使用する秘密鍵やアルゴリズムなどを設定ファイルや環境変数で管理します。秘密鍵は推測されにくい、十分に複雑な文字列を使用し、安全に保管することが重要です。

# main.py (例)
from passlib.context import CryptContext

# JWT設定
SECRET_KEY = "your-super-secret-key-change-this" # 実際には環境変数から取得することを推奨
ALGORITHM = "HS256"

# パスワードハッシュ設定
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def get_password_hash(password: str):
    return pwd_context.hash(password)

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

2. ユーザーモデルの定義

ユーザー情報を管理するためのPydanticモデルを定義します。認証時には、ユーザー名とパスワードを受け取ります。

# models.py (例)
from pydantic import BaseModel

class UserBase(BaseModel):
    username: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int

    class Config:
        orm_mode = True # SQLAlchemyなどのORMと連携する場合に必要

3. 認証エンドポイントの実装

ユーザーからのログインリクエストを受け付け、認証後にJWTを生成するエンドポイントを作成します。ここでは、TokenDataというペイロードの型を定義し、create_access_token関数でJWTを生成します。

# auth.py (例)
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from datetime import datetime, timedelta
from pydantic import BaseModel
from typing import Optional

# 上記 main.py から SECRET_KEY, ALGORITHM, pwd_context, get_password_hash, verify_password をインポート
from main import SECRET_KEY, ALGORITHM, verify_password, get_password_hash

# --- ユーザー関連の定義 (models.py から抜粋、または統合) ---
class UserBase(BaseModel):
    username: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int

    class Config:
        orm_mode = True

# --- JWT関連の定義 ---
ACCESS_TOKEN_EXPIRE_MINUTES = 30

class TokenData(BaseModel):
    username: Optional[str] = None

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# OAuth2PasswordBearer を使って、Authorization: Bearer  ヘッダーからトークンを取得
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# --- 認証エンドポイント ---
# 実際には、データベースからユーザー情報を取得する処理が必要です
# ここではダミーのユーザーリストを使用します
fake_users_db = {
    "testuser": get_password_hash("password123")
}

async def authenticate_user(username: str, password: str):
    hashed_password = fake_users_db.get(username)
    if not hashed_password:
        return None
    if verify_password(password, hashed_password):
        return User(id=1, username=username) # ダミーのユーザーオブジェクトを返す
    return None

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") # 'sub'はSubject、ユーザー識別子として一般的
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    # ここで、トークンに紐づくユーザー情報をデータベースなどから取得し、
    # further validation (e.g., user is active) を行うことができます。
    # 今回はダミーなので、usernameのみを返します。
    user = User(id=1, username=username) # ダミーのユーザーオブジェクト
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)):
    # ここで、ユーザーがアクティブかなどをチェックするロジックを追加できます。
    # 例: if current_user.is_active == False: raise HTTPException(...)
    return current_user

# FastAPIインスタンス (main.pyで定義されていると仮定)
# app = FastAPI()

# @app.post("/token", response_model=Token) # Token モデルは別途定義が必要
# async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
#     user = await authenticate_user(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 {"access_token": access_token, "token_type": "bearer"}

# --- protected endpoint example ---
# @app.get("/users/me/")
# async def read_users_me(current_user: User = Depends(get_current_active_user)):
#     return current_user

4. protected エンドポイントの作成

JWT認証が必要なエンドポイントを作成します。ここではDependsを使用して、get_current_active_user関数を依存関係として注入します。これにより、リクエストヘッダーから有効なJWTが渡されない場合、自動的に401 Unauthorizedエラーが返されます。

上記のauth.pyのコメントアウトされた部分が、実際のprotectedエンドポイントの例になります。

JWTのカスタマイズと高度な利用

Payloadには、ユーザーIDやロール、権限などのカスタムクレームを追加できます。これにより、認証されたユーザーが特定のリソースにアクセスする権限を持っているかなどを、API側で細かく制御できます。例えば、scopesというクレームにユーザーの権限リストを含めることで、APIリクエストごとに権限チェックを行うことが可能です。

python-joseライブラリは、RS256などの公開鍵/秘密鍵ペアを使用した署名アルゴリズムもサポートしています。これは、複数のサービス間でJWTを共有する場合などに、より安全な認証メカニズムを提供します。

セキュリティに関する注意点

  • 秘密鍵の管理: JWTの秘密鍵は、最も重要なセキュリティ要素です。絶対にコードに直接埋め込まず、環境変数やキー管理サービスを使用して安全に管理してください。
  • 有効期限: JWTには適切な有効期限を設定してください。長すぎる有効期限は、トークンが漏洩した場合のリスクを高めます。
  • HTTPSの使用: JWTは通常、HTTPS経由で送信されるべきです。これにより、通信経路での盗聴を防ぎます。
  • トークンのブラックリスト: ユーザーがログアウトした場合や、トークンが不正利用された場合に、そのトークンを無効化するための仕組み(ブラックリストやレボケーションリスト)を導入することを検討してください。
  • アルゴリズムの選択: HS256は秘密鍵のみを使用するため、サーバー側でのみ署名と検証が行われます。RS256などの公開鍵/秘密鍵ペアを使用するアルゴリズムは、署名と検証を異なるキーで行えるため、より柔軟で安全なシナリオに対応できます。

まとめ

FastAPIでJWT認証を実装することは、fastapipython-josepasslibといったライブラリを活用することで、比較的容易に行えます。ユーザー認証、JWTの生成、そしてProtectedエンドポイントでの認証トークンの検証といった一連のプロセスを、FastAPIの依存性注入システムを効果的に利用することで、クリーンかつ再利用可能なコードとして構築できます。セキュリティ上の注意点を理解し、適切に実装することで、堅牢なAPIセキュリティを実現することが可能です。