FastAPIでJWTによる認証の実装

はじめに

現代のウェブアプリケーション開発において、認証はシステムのセキュリティを確保するための重要な要素です。JSON Web Token（JWT）は、JSON（RFC 7519）に基づくオープンスタンダードとして、そのコンパクトさとセキュリティから、分散サイトにおけるシングルサインオン（SSO）シナリオで広く使用されています。FastAPIは、Pythonに基づく高性能なウェブフレームワークです。この記事では、提供されたJWT実装コードと組み合わせて、FastAPIでJWTを使用して認証を行う方法を紹介します。

JWTの基本

JWTとは？

JSON Web Token（JWT）は、ネットワークアプリケーション環境間でクレームを伝送するように設計されています。コンパクトで安全であり、特に分散サイトでのシングルサインオンシナリオに適しています。JWTクレームは、リソースサーバーからリソースを取得するために、認証されたユーザーのID情報をIDプロバイダーとサービスプロバイダー間で伝送するためによく使用され、ビジネスロジックに必要な他のクレーム情報を追加することもできます。このトークンは、認証または暗号化に直接使用できます。

JWTの構成

JWTは、3つのセグメントの情報で構成されています。

ヘッダー: 通常、ファイルタイプと暗号化アルゴリズムを宣言します。
ペイロード: JSON形式で伝送されるデータを記録します。
署名: 検証に使用される署名です。

base64UrlEncodeメソッド

JWTを生成する際、署名にはbase64UrlEncodeメソッドが使用されます。このメソッドの基本的な機能は以下のとおりです。

UTF-8エンコードされた文字列s1を入力します。
文字列s1をbase64でエンコードして文字列s2を取得します。
s2の末尾に等号がある場合、末尾のすべての等号を削除します。
s2にプラス記号（+）がある場合、すべてのプラス記号をマイナス記号（-）に置き換えます。
s2にスラッシュ（/）がある場合、すべてのスラッシュをアンダースコア（_）に置き換えます。

以下は、このメソッドを実装するPythonコードです。

import hmac
import base64
from hashlib import sha256
from urllib import parse as urlp

def b64url(str1):
    if type(str1) == str:
        return str(base64.b64encode(str1.encode('utf-8')), encoding="utf-8").strip('=').replace('+', '-').replace('/', '_')
    elif type(str1) == bytes:
        return str(base64.b64encode(str1), encoding="utf-8").strip('=').replace('+', '-').replace('/', '_')
    else:
        raise TypeError("The type of given argument must be string or bytes")

HS256で暗号化されたJWTの生成

以下は、上記のb64url関数を使用してHS256で暗号化されたJWTを生成するコードです。

# Enter Your Infomation Here ...
header = b64url('{"alg":"HS256","typ":"JWT"}')
payload = b64url('{"sub":"1234567890","name":"John Doe","iat":1516239022}')
secret = 'happynewyear'.encode('utf-8')
# ###

sig = b64url(hmac.new(
    secret, (header + '.' + payload).encode('utf-8'), digestmod=sha256
).digest())

jwt = header + '.' + payload + '.' + sig
print(jwt)

FastAPIでのJWTの統合

依存関係のインストール

開始する前に、fastapiとuvicornをインストールする必要があります。次のコマンドを使用してインストールできます。

pip install fastapi uvicorn

実装手順

1. FastAPIアプリケーションの作成

from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import hmac
import base64
from hashlib import sha256

app = FastAPI()

# Define the b64url function
def b64url(str1):
    if type(str1) == str:
        return str(base64.b64encode(str1.encode('utf-8')), encoding="utf-8").strip('=').replace('+', '-').replace('/', '_')
    elif type(str1) == bytes:
        return str(base64.b64encode(str1), encoding="utf-8").strip('=').replace('+', '-').replace('/', '_')
    else:
        raise TypeError("The type of given argument must be string or bytes")

# Secret key
SECRET_KEY = 'happynewyear'.encode('utf-8')

# Function to generate JWT
def generate_jwt():
    header = b64url('{"alg":"HS256","typ":"JWT"}')
    payload = b64url('{"sub":"1234567890","name":"John Doe","iat":1516239022}')
    sig = b64url(hmac.new(
        SECRET_KEY, (header + '.' + payload).encode('utf-8'), digestmod=sha256
    ).digest())
    return header + '.' + payload + '.' + sig

# Function to verify JWT
def verify_jwt(token: str):
    parts = token.split('.')
    if len(parts) != 3:
        return False
    header, payload, received_sig = parts
    new_sig = b64url(hmac.new(
        SECRET_KEY, (header + '.' + payload).encode('utf-8'), digestmod=sha256
    ).digest())
    return new_sig == received_sig

# Define the JWT verification dependency
class JWTBearer(HTTPBearer):
    def __init__(self, auto_error: bool = True):
        super(JWTBearer, self).__init__(auto_error=auto_error)

    async def __call__(self, request):
        credentials: HTTPAuthorizationCredentials = await super(JWTBearer, self).__call__(request)
        if credentials:
            if not credentials.scheme == "Bearer":
                raise HTTPException(status_code=401, detail="Invalid authentication scheme.")
            if not verify_jwt(credentials.credentials):
                raise HTTPException(status_code=401, detail="Invalid or expired token.")
            return credentials.credentials
        else:
            raise HTTPException(status_code=401, detail="Invalid authorization code.")

2. ログインインターフェースの作成

# Simulated login interface
@app.post("/login")
def login():
    token = generate_jwt()
    return {"access_token": token, "token_type": "bearer"}

3. 保護されたインターフェースの作成

# Protected interface
@app.get("/protected", dependencies=[Depends(JWTBearer())])
def protected_route():
    return {"message": "This is a protected route."}

アプリケーションの実行

上記のコードをmain.pyとして保存し、ターミナルで次のコマンドを実行してアプリケーションを起動します。

uvicorn main:app --reload

インターフェースのテスト

ログインインターフェース: curlやPostmanなどのツールを使用して、http://localhost:8000/loginにPOSTリクエストを送信すると、JWTが返されます。

curl -X POST http://localhost:8000/login

保護されたインターフェース: 返されたJWTを使用して、http://localhost:8000/protectedにGETリクエストを送信し、リクエストヘッダーにAuthorization: Bearer <your_token>を追加します。

curl -X GET http://localhost:8000/protected -H "Authorization: Bearer <your_token>"

結論

上記のステップを通じて、FastAPIでJWTを使用して認証を行う方法を学びました。JWTはユーザーIDを管理するための安全で便利な方法を提供し、分散システム間の認証をより効率的にします。実際のアプリケーションでは、有効期限の追加やクレームのカスタマイズなど、必要に応じてJWTの生成および検証ロジックを調整できます。