PythonでJWT認証を実装しよう

この章では、FastAPIとSQLModelでJWT（JSON Web Token）による認証を扱い、ハンズオン形式で学習しながら実装します。これにより、ログインAPIからアクセストークンの発行・検証・リフレッシュ・ログアウトによる失効管理までを含む、実用的な認証機構が構築できるようになります。

1. 本章の概要

1.1 本章の目的

前章 PythonでCORS対応をしよう までで、ブラウザから呼び出せるAPIサーバを準備できました。実際のアプリケーションでは、これに加えて「誰からのリクエストなのか」を判定する認証の仕組みが必要になります。本章では、認証方式としてよく使われるJWT（JSON Web Token）を採り上げます。JWTはトークン自体に検証可能な情報が含まれる仕組みで、サーバがセッション状態を保持しなくてもリクエスト元を判定できるため、REST APIとの相性が良い方式です。本章では、ユーザ登録・ログインでトークンを発行し、保護されたエンドポイントで検証したうえで、リフレッシュトークンによる再発行とログアウトによる失効管理までを一通り扱います。

1.2 ハンズオンの流れ

MySQLにusersテーブルとrefresh_tokensテーブルを用意し、FastAPIとSQLModelで認証APIを実装します。まずサインアップ・ログインでアクセストークンとリフレッシュトークンを発行し、Dependsで作成した認証依存で保護されたGET /notesを叩けるところまで作ります。そのうえで、リフレッシュトークンでアクセストークンを再発行するPOST /refreshと、リフレッシュトークンを失効させるPOST /logoutを実装します。

本章で構築する認証フローを以下に示します。

sequenceDiagram
    participant C as クライアント
    participant S as APIサーバ
    participant DB as DB (users / refresh_tokens)
    C->>S: POST /login (email, password)
    S->>DB: パスワード検証
    S->>DB: refresh_token 保存
    S-->>C: access_token + refresh_token
    C->>S: GET /notes (Authorization: Bearer access_token)
    S-->>C: ノート一覧
    C->>S: POST /refresh (refresh_token)
    S->>DB: 旧 refresh_token を失効し、新規発行
    S-->>C: 新しい access_token + refresh_token
    C->>S: POST /logout (refresh_token)
    S->>DB: refresh_token を失効
    S-->>C: 204 No Content

1.3 事前準備

必要なツール

この章では、以下のツールを使用します。まだインストールしていない場合は、リンク先の手順に沿って準備をお願いします。

2. JWT認証の仕組み

実装に入る前に、そもそもなぜ認証が必要かを押さえたうえで、JWTがどのように「サーバがセッションを持たずに認証を成立させるか」を確認します。JWTの構造、アクセストークンとリフレッシュトークンの役割分担、そしてサーバ側で持つべき責務の範囲を整理します。

2.1 認証の必要性

Webアプリケーションでは、リクエストの送信元が「誰なのか」を判定できないと、以下のような問題が発生します。

なりすましの防止

リクエストにユーザ本人であることを示す情報が付いていなければ、他人のIDを騙って自由にリクエストを送れる状態になります。攻撃者は他人のIDでAPIを呼び出すだけで、その人のデータを閲覧したり、勝手に変更・削除したりできてしまいます。SNSであれば他人になりすまして投稿する、ECサイトであれば他人のアカウントで注文するといった被害に直結するため、リクエスト元が本人であることを検証する認証の仕組みが必要になります。

責任追跡性の担保

誰がいつどの操作を行ったかを特定できない状態だと、不正な操作が発生しても追跡や責任の切り分けができません。監査ログ自体は残っていても、「その操作をしたのが本当に本人か」を保証する仕組みが無ければ、ログの信頼性が下がります。認証の仕組みを通してリクエスト元を特定できて初めて、操作ログが証跡として機能します。

機密情報の保護

個人情報・課金情報・非公開ドラフトなど、本来は本人だけが見られるべき情報を扱うAPIに認証が無いと、URLを推測されるだけで他人の情報を取得されてしまいます。公開情報しか扱わないサービスでも、ユーザごとの設定・履歴・下書きといった非公開情報を持つ場面はほぼ必ずあるため、認証を前提として本人以外からのアクセスを弾く設計が欠かせません。

こうした問題を防ぐために、Webアプリケーションでは「そのリクエストが誰から送られたものか」を検証する認証の仕組みが必要になります。認証が成立して初めて、ユーザごとに「何をしてよいか」を判定する認可の判断に進めます。

2.2 JWT認証とは

認証を成立させる方式には複数の選択肢があります。従来はサーバ側にセッション情報を保持し、Cookieでセッションキーを受け渡すステートフルな方式が使われてきました。この方式では、以下のようなリスクがありました。

• ロードバランサー越しに複数のAPIサーバを並べる構成の場合、リクエストごとに違うサーバに振り分けられるため、認証情報を全台で共有する仕組みが別途必要になる
• オートスケーリングでサーバが増減したり、サーバが再起動したりすると、そのサーバが持っていた認証情報が失われ、ユーザがログアウトされてしまう
• Cookieを前提とした仕組みのため、Cookieを扱いにくい環境からは同じ方式で認証しにくい

JWTは、必要な情報をトークン自身に埋め込み、サーバの署名で改竄されていないことを検証できる形にしたステートレスな方式です。JWTを採用することで、上記のリスクは以下のように解消できます。

• サーバ側で認証状態を保持しないため、ロードバランサーがどのサーバに振り分けても認証情報の共有が不要になる
• 認証情報がトークン自体に含まれるため、オートスケーリングでのサーバの増減や再起動があっても認証状態が失われない
• HTTPヘッダに載せてやり取りする方式なので、Cookieが扱いにくい環境からも同じ方式で認証できる

一方で、「サーバ側にセッションが無い＝サーバから任意のトークンを即時に無効化できない」という性質もあります。これを補うために、後述のリフレッシュトークンと失効管理の仕組みを組み合わせて運用します。

2.3 JWTの構造

JWTは、以下の3つのパートを.で連結した文字列です。

ヘッダ.ペイロード.署名

実際に発行されたJWTは、以下のようにそれぞれのパートがBase64URLエンコードされた文字列で並びます。

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOjEsImV4cCI6MTcyNDMzMTIwMCwiaWF0IjoxNzI0MzMwMzAwfQ.dQw4w9WgXcQ_1a2B3c4D5e6F7g8H9i0J

各パートを順に見ていきます。

ヘッダ

ヘッダは、署名アルゴリズムやトークンの種別を示すJSONを、Base64URLでエンコードしたものです。実際の値の例は以下のとおりです。

{
    "alg": "HS256",
    "typ": "JWT"
}

これをBase64URLエンコードすると、以下のようになります。

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9

algは署名アルゴリズム（ここではHMAC-SHA256）、typはトークンの種別（JWT）を示します。

ペイロード

ペイロードは、発行者・有効期限・ユーザIDなどのクレーム（Claims）を表すJSONを、Base64URLでエンコードしたものです。

クレーム（Claims）は、ペイロードに含まれる個々の情報要素です。「このトークンについて主張（claim）したい事実」をJSONのキーと値のペアで表現するもので、たとえば「このトークンはユーザID=1のもの」「有効期限はUNIX時刻1724331200まで」といった一つひとつの主張がクレームにあたります。JWTの仕様で名前が予約されているものを標準クレーム、アプリケーション独自に定義したものをプライベートクレームと呼びます。

ペイロードには、以下のような標準クレームがあります。

実際のペイロードの例は以下のとおりです。

{
    "sub": 1,
    "exp": 1724331200,
    "iat": 1724330300
}

これをBase64URLエンコードすると、以下のようになります。

eyJzdWIiOjEsImV4cCI6MTcyNDMzMTIwMCwiaWF0IjoxNzI0MzMwMzAwfQ

ペイロードは暗号化されず、Base64URLでエンコードされているだけなので、誰でも中身を復号して見られます。パスワードなどの秘匿情報を入れてはいけません。改竄検知は署名で担保され、秘密鍵を知らない第三者が中身を書き換えると、署名の検証に失敗してサーバが弾きます。

署名

署名は、Base64URLエンコード済みのヘッダとペイロードを.で連結した文字列に対して、ヘッダで指定したアルゴリズム（例: HMAC-SHA256）とサーバの秘密鍵でハッシュ署名し、その結果をBase64URLエンコードしたものです。実際の値の例は以下のとおりです。

dQw4w9WgXcQ_1a2B3c4D5e6F7g8H9i0J

同じ秘密鍵を持っているサーバのみ、この署名を再計算して一致するかを検証できるため、トークンが改竄されていないことをステートレスに確認できます。

2.4 アクセストークン

JWT自体はトークンの形式であり、運用の設計は別途決める必要があります。本章では、以下の2種類のトークンを組み合わせる構成を採用します。

アクセストークンは、APIリクエストの認証に使う短命なトークンです。JWT形式で発行し、以降のAPIリクエストごとにAuthorization: Bearer <アクセストークン>ヘッダに載せてサーバへ送ります。サーバは署名の検証と有効期限のチェックだけでリクエスト元を判定できるため、DBへの問い合わせなしにステートレスに扱えます。

アクセストークンを短命にすることで、盗まれた場合の影響時間を狭められます。ただし短命にすると再ログインの頻度が増えてユーザ体験が悪くなるため、後述のリフレッシュトークンを合わせて発行し、期限が切れる前にアクセストークンだけを再発行する構成にします。

2.5 リフレッシュトークン

リフレッシュトークンは、アクセストークンの再発行に使う長命なトークンです。JWT形式にはせず、secretsモジュールで生成した乱数文字列をサーバ側のDBに保管し、以下の管理を行います。

• ログアウト時に該当のリフレッシュトークンを失効させる
• リフレッシュのたびに古いリフレッシュトークンを失効させ、新しいものを発行する（ローテーション）

この構成にすると、アクセストークンはステートレスに検証しつつ、リフレッシュトークンによって「特定のユーザ・端末のセッションを即時に切る」ことが可能になります。

2.6 認証の全体フロー

ここまで説明した仕組みを踏まえ、実際にクライアントとサーバがやり取りする流れを追います。

1. ユーザ登録

ユーザ登録を表すエンドポイント（例: POST /signup）を用意し、クライアントからメールアドレスとパスワードを受け取ってユーザレコードを作成します。この段階ではトークンは発行されません。

リクエスト (POST /signup):
{"email": "test@example.com", "password": "password12345"}

処理内容:
- パスワードをハッシュ化して users テーブルにユーザレコードを作成

レスポンス:
{"id": 1, "email": "test@example.com", "created_at": "2026-08-24T10:00:00Z"}

2. ログイン

ログインを表すエンドポイント（例: POST /login）を用意し、受け取ったパスワードを検証したうえで、2種類のトークンを返します。アクセストークンはJWT形式の文字列、リフレッシュトークンは乱数由来の16進文字列で、サーバはリフレッシュトークンの値を同じ内容でDBにも保管しておきます。

リクエスト (POST /login):
{"email": "test@example.com", "password": "password12345"}

処理内容:
- メールアドレスでユーザを検索
- 受け取ったパスワードと DB のハッシュを照合
- アクセストークン（JWT）を生成
- リフレッシュトークン（乱数）を生成して DB に保存

レスポンス:
{
  "access_token": "eyJhbGciOiJIUzI1NiIs...",
  "refresh_token": "3f4c8d2a6b..."
}

3. 保護されたAPI呼び出し

認証済みユーザだけが呼び出せるAPI（例: GET /notes）にアクセスする際は、アクセストークンをリクエストに添えて送ります。クライアントは受け取ったアクセストークンをAuthorizationヘッダにBearerスキームで載せて送り、サーバはFastAPIのDependsで作成した認証依存で署名と有効期限を検証したうえで、正当ならハンドラに処理を渡します。

リクエスト (GET /notes):
Authorization: Bearer eyJhbGciOiJIUzI1NiIs...

処理内容:
- Authorization ヘッダからアクセストークンを取り出す
- 署名と有効期限を検証
- 検証に成功したらハンドラに処理を渡し、対応するデータを返す

レスポンス:
[{"id": 1, "title": "買い物"}, {"id": 2, "title": "打ち合わせ資料"}]

4. アクセストークンの再発行

アクセストークンは短命なので、期限が切れそう／切れたら再発行が必要になります。再発行を表すエンドポイント（例: POST /refresh）を用意し、受け取ったリフレッシュトークンがDB上で失効していないことを確認したうえで、古いリフレッシュトークンを失効させて新しいアクセストークンとリフレッシュトークンを返します。以降クライアントは、新しく受け取ったトークンだけを使う運用になります。

リクエスト (POST /refresh):
{"refresh_token": "3f4c8d2a6b..."}

処理内容:
- DB からリフレッシュトークンを検索
- 失効・期限切れになっていないことを確認
- 古いリフレッシュトークンを失効させる（ローテーション）
- 新しいアクセストークンとリフレッシュトークンを発行

レスポンス:
{
  "access_token": "eyJhbGciOiJIUzI1NiIs...新しい...",
  "refresh_token": "a91b7e3d..."
}

5. ログアウト

ログアウトを表すエンドポイント（例: POST /logout）を用意し、受け取ったアクセストークンとリフレッシュトークンの両方を無効化します。アクセストークンはJWT側で即時無効化ができない（ステートレスなので）ため、DBのブラックリスト（revoked_access_tokensテーブル）に登録し、認証依存で照合することで即時拒否させます。リフレッシュトークンはrefresh_tokensテーブルのrevoked_atを更新して、以降の/refreshを拒否します。

リクエスト (POST /logout):
Authorization: Bearer eyJhbGciOiJIUzI1NiIs...
{"refresh_token": "a91b7e3d..."}

処理内容:
- アクセストークンを revoked_access_tokens に登録して即時無効化
- 対応するリフレッシュトークンを無効化

レスポンス:
204 No Content

この流れを本章の後半で1エンドポイントずつ実装します。

3. 認証機能の実装準備

ここからハンズオンに入ります。プロジェクトフォルダの初期化、依存パッケージの導入、データベース・モデル・接続設定の準備までを行います。

3.1 プロジェクトの準備

任意の場所にjwt-auth-hands-onフォルダを作成し、Visual Studio Codeの「ファイル」→「フォルダーを開く」から作成したjwt-auth-hands-onフォルダを開きます。以降の操作は、Visual Studio Codeのターミナルから行います。

jwt-auth-hands-on/  ← このフォルダを作成

以下のコマンドで、Pythonがインストールされていることを確認します。

Windowsの場合:

python --version

Macの場合:

python3 --version

以下のようにPythonのバージョンが表示されれば、インストールは確認できています。

Python 3.12.x

バージョンが表示されない場合は、Pythonのインストールを先に実施してください。

続いて、必要なパッケージをまとめてインストールします。

Windowsの場合:

pip install fastapi uvicorn sqlmodel pymysql python-dotenv pyjwt bcrypt

Macの場合:

pip3 install fastapi uvicorn sqlmodel pymysql python-dotenv pyjwt bcrypt

インストールした各ライブラリの役割は以下のとおりです。

3.2 データベースの作成

認証APIから接続するデータベースを作成します。テーブルはSQLModelから自動生成するため、ここではデータベースだけ用意します。

以下のコマンドで、MySQLがインストールされていることを確認します。

mysql --version

以下のようにMySQLのバージョンが表示されれば、インストールは確認できています。

mysql  Ver 8.0.xx for macos on arm64 (MySQL Community Server - GPL)

バージョンが表示されない場合は、MySQLのインストールを先に実施してください。

続いて、auth_hands_onデータベースを作成します。

mysql -u root -p -e "CREATE DATABASE auth_hands_on;"

パスワードを入力してエラーが出なければ、データベースが作成されています。

3.3 環境変数ファイル（.env）の作成

DB接続情報とJWTの署名鍵を.envファイルにまとめます。Visual Studio Codeのエクスプローラーでjwt-auth-hands-onフォルダを右クリックし、「新しいファイル」を選択して.envという名前で作成します。

jwt-auth-hands-on/
└── .env  ← このファイルを作成

作成したファイルに以下の内容を記述して保存します。

DB_HOST=localhost
DB_USER=root
DB_PASSWORD=your_password
DB_NAME=auth_hands_on
JWT_SECRET=dev-secret-please-change

3.4 モデル定義（models.py）

usersテーブルとrefresh_tokensテーブルに対応するSQLModelを定義します。同時に、リクエスト・レスポンスで使う非テーブルモデルもここに置きます。

Visual Studio Codeのエクスプローラーでjwt-auth-hands-onフォルダを右クリックし、「新しいファイル」を選択してmodels.pyを作成します。

jwt-auth-hands-on/
├── .env
└── models.py  ← このファイルを作成

作成したファイルに以下の内容を記述して保存します。

from datetime import datetime
from typing import Optional

from sqlmodel import Field, SQLModel


class User(SQLModel, table=True):
    __tablename__ = "users"
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str = Field(unique=True, index=True, max_length=255)
    password_hash: str = Field(max_length=255)
    created_at: datetime = Field(default_factory=datetime.utcnow)


class RefreshToken(SQLModel, table=True):
    __tablename__ = "refresh_tokens"
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(foreign_key="users.id", index=True)
    token: str = Field(unique=True, index=True, max_length=255)
    expires_at: datetime
    revoked_at: Optional[datetime] = None
    created_at: datetime = Field(default_factory=datetime.utcnow)


class SignupRequest(SQLModel):
    email: str
    password: str


class LoginRequest(SQLModel):
    email: str
    password: str


class TokenResponse(SQLModel):
    access_token: str
    refresh_token: str


class RefreshRequest(SQLModel):
    refresh_token: str

コードを解説します。

class User(SQLModel, table=True):
    __tablename__ = "users"
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str = Field(unique=True, index=True, max_length=255)
    password_hash: str = Field(max_length=255)
    created_at: datetime = Field(default_factory=datetime.utcnow)

ユーザを表すテーブルモデルです。emailにはunique=Trueとindex=Trueを付けて重複登録を防ぎ、検索を高速化しています。password_hashにはハッシュ化された文字列を保存する前提のカラムです。

class RefreshToken(SQLModel, table=True):
    __tablename__ = "refresh_tokens"
    ...
    expires_at: datetime
    revoked_at: Optional[datetime] = None

リフレッシュトークンを表すテーブルモデルです。tokenカラムに発行済みのトークン文字列を保存し、expires_atで有効期限、revoked_atで失効時刻を管理します。revoked_atをOptional[datetime]にすることで、「まだ失効していない状態（NULL）」と「失効済み（時刻あり）」を区別できます。

class SignupRequest(SQLModel):
    email: str
    password: str

リクエストボディで使う非テーブルモデルです。table=Trueを指定していないため、DBには対応せずバリデーションのためだけに使われます。テーブルモデルとリクエストモデルを分けておくと、レスポンスにpassword_hashを含めるといった事故を防げます。

class TokenResponse(SQLModel):
    access_token: str
    refresh_token: str

ログイン・リフレッシュのレスポンスで返すトークンの組を表すモデルです。エンドポイント側で戻り値の型として使うことで、Swagger UIにスキーマが表示されるようになります。

3.5 DB接続設定（database.py）

続いて、DB接続とSessionの生成を扱うdatabase.pyを作成します。

jwt-auth-hands-on/
├── .env
├── models.py
└── database.py  ← このファイルを作成

作成したファイルに以下の内容を記述して保存します。

import os

from dotenv import load_dotenv
from sqlmodel import Session, SQLModel, create_engine

load_dotenv()

DB_HOST = os.getenv("DB_HOST", "localhost")
DB_USER = os.getenv("DB_USER", "root")
DB_PASSWORD = os.getenv("DB_PASSWORD", "")
DB_NAME = os.getenv("DB_NAME", "auth_hands_on")

DATABASE_URL = f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}/{DB_NAME}"

engine = create_engine(DATABASE_URL)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session

前章までで扱った内容と同じ構成で、.envから接続情報を読み込み、create_engineでengineを作成しています。create_db_and_tablesはSQLModel.metadata.create_allでmodels.pyで定義したテーブルを一括作成し、get_sessionはDependsで注入するためのSessionのジェネレータです。

3.6 テーブル作成スクリプト（setup_db.py）

create_db_and_tablesを呼び出してテーブルを作成する独立スクリプトを作成します。アプリケーションの起動時ではなく、初回セットアップ時に一度だけ実行する形にします。

jwt-auth-hands-on/
├── .env
├── models.py
├── database.py
└── setup_db.py  ← このファイルを作成

作成したファイルに以下の内容を記述して保存します。

from database import create_db_and_tables
import models  # SQLModel.metadata にテーブル定義を登録するためのインポート

if __name__ == "__main__":
    create_db_and_tables()
    print("tables created.")

以下のコマンドでスクリプトを実行します。

Windowsの場合:

python setup_db.py

Macの場合:

python3 setup_db.py

以下のような実行結果が表示されます。

tables created.

MySQL CLIでusersテーブルとrefresh_tokensテーブルが作成されていることを確認します。

mysql -u root -p -e "USE auth_hands_on; SHOW TABLES;"

以下のような実行結果が表示されます。

+-------------------------+
| Tables_in_auth_hands_on |
+-------------------------+
| refresh_tokens          |
| users                   |
+-------------------------+

usersとrefresh_tokensが並んでいれば、モデルからのテーブル作成は成功しています。

4. ユーザ登録

ユーザ登録を行うPOST /signupエンドポイントを実装します。メールアドレスとパスワードを受け取り、ハッシュ化してusersテーブルにユーザレコードを作成する処理です。

4.1 パスワードハッシュヘルパーの追加

パスワードをハッシュ化するヘルパー関数を追加します。

パスワードのハッシュ化はsignupハンドラだけで呼び出す処理ですが、以降で追加するパスワード検証（verify_password）やJWT発行と合わせて「認証まわりの共通処理」として、auth.pyという専用ファイルにまとめて置きます。ハンドラから独立させておくと、認証ロジックの読みやすさが上がり、bcryptから別のハッシュアルゴリズムへの入れ替えなどの変更も1箇所で済むようになります。

Visual Studio Codeのエクスプローラーでjwt-auth-hands-onフォルダを右クリックし、「新しいファイル」を選択してauth.pyを作成します。

jwt-auth-hands-on/
├── .env
├── auth.py  ← このファイルを作成
├── database.py
├── models.py
└── setup_db.py

作成したauth.pyに以下の内容を記述して保存します。

import bcrypt


def hash_password(password: str) -> str:
    salt = bcrypt.gensalt()
    return bcrypt.hashpw(password.encode("utf-8"), salt).decode("utf-8")

コードを解説します。

def hash_password(password: str) -> str:

関数のシグネチャです。引数として生のパスワード文字列を受け取り、戻り値としてハッシュ化した文字列を返します。

salt = bcrypt.gensalt()

bcrypt.gensaltでソルトを生成します。ソルトはパスワードと組み合わせてハッシュ化する際に使う一意のランダム値で、レインボーテーブル攻撃を防ぐ役割があります。

return bcrypt.hashpw(password.encode("utf-8"), salt).decode("utf-8")

bcrypt.hashpwで、生のパスワードをソルト付きのハッシュにします。bcryptはstrではなくbytesを受け取るため、encode("utf-8")で変換して渡しています。生成されたハッシュ文字列にはソルトも埋め込まれているため、後の検証時には追加でソルトを渡す必要がありません。DBにはstrとして保存するため、decode("utf-8")で文字列化しています。

4.2 main.py と handlers.py の作成、signup ハンドラの追加

続いて、エントリポイントとなるmain.pyと、HTTPハンドラを配置するhandlers.pyを作成します。以降のセクションで追加するハンドラはすべてhandlers.pyに配置し、main.pyはappの初期化とhandlers.pyのルーターの取り込みに専念する構成にします。

まず、main.pyを作成します。

jwt-auth-hands-on/
├── .env
├── auth.py
├── database.py
├── main.py  ← このファイルを作成
├── models.py
└── setup_db.py

作成したmain.pyに以下の内容を記述して保存します。

from fastapi import FastAPI

from handlers import router

app = FastAPI()
app.include_router(router)

続いて、handlers.pyを作成します。

jwt-auth-hands-on/
├── .env
├── auth.py
├── database.py
├── handlers.py  ← このファイルを作成
├── main.py
├── models.py
└── setup_db.py

作成したhandlers.pyに以下の内容を記述して保存します。APIRouterを作成し、そこにsignupハンドラを登録する形です。

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select

from auth import hash_password
from database import get_session
from models import SignupRequest, User

router = APIRouter()


@router.post("/signup", status_code=status.HTTP_201_CREATED)
def signup(
    payload: SignupRequest,
    session: Annotated[Session, Depends(get_session)],
):
    if len(payload.password) < 8:
        raise HTTPException(status_code=400, detail="password は 8 文字以上にしてください")
    existing = session.exec(select(User).where(User.email == payload.email)).first()
    if existing is not None:
        raise HTTPException(status_code=400, detail="このメールアドレスは既に登録されています")

    user = User(email=payload.email, password_hash=hash_password(payload.password))
    session.add(user)
    session.commit()
    session.refresh(user)
    return {"id": user.id, "email": user.email, "created_at": user.created_at}

main.py側でapp.include_router(router)を呼んでいるので、handlers.pyにrouter.post(...)で登録したエンドポイントはすべてappに自動で取り込まれます。

4.3 コードの解説

router = APIRouter()

APIRouterはFastAPIアプリケーションと同じデコレータ（.post・.getなど）を提供するルーター用のクラスです。main.pyからapp.include_router(router)で取り込むことで、app.post(...)で直接登録した場合と同じ扱いになります。ハンドラを別ファイルに切り出すためのFastAPIの標準的な手段です。

@router.post("/signup", status_code=status.HTTP_201_CREATED)
def signup(
    payload: SignupRequest,
    session: Annotated[Session, Depends(get_session)],
):

signupハンドラのシグネチャです。payload: SignupRequestでリクエストボディを型付きで受け取り、Depends(get_session)でDBセッションを依存注入します。status_code=status.HTTP_201_CREATEDで成功時に201 Createdを返す設定です。

if len(payload.password) < 8:
    raise HTTPException(status_code=400, detail="password は 8 文字以上にしてください")
existing = session.exec(select(User).where(User.email == payload.email)).first()
if existing is not None:
    raise HTTPException(status_code=400, detail="このメールアドレスは既に登録されています")

サインアップ時のバリデーションです。パスワードの最小長チェックと、メールアドレスの重複チェックを行っています。重複時に400で拒否することで、UNIQUE制約違反の生SQLエラーが利用者に返るのを防いでいます。

user = User(email=payload.email, password_hash=hash_password(payload.password))
session.add(user)
session.commit()
session.refresh(user)

hash_passwordでパスワードをハッシュ化してからUserを作り、DBに保存します。session.refresh(user)でDB側から自動採番されたidやcreated_atをuserに取り込みます。

return {"id": user.id, "email": user.email, "created_at": user.created_at}

作成したユーザを返します。レスポンスはハッシュを含めない安全な形（id・email・created_at）で作って返します。テーブルモデルをそのまま返すとpassword_hashが漏れるため、明示的に取り出しています。

4.4 動作確認

APIサーバを起動します。

uvicorn main:app --reload

以下のような実行結果が表示されます。

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

別のターミナルから、ユーザを作成します。

curl -X POST http://localhost:8000/signup -H "Content-Type: application/json" -d '{"email":"test@example.com","password":"password12345"}'

以下のような実行結果が表示されます（IDと日時は環境により変わります）。

{"id":1,"email":"test@example.com","created_at":"2026-08-22T12:00:00"}

password_hashは返却せず、id・email・created_atだけがレスポンスに含まれています。念のため、MySQL側でもusersテーブルに実際にレコードが登録されていることを確認します。別のターミナルを開き、rootユーザでMySQLにログインします。

mysql -u root -p

ログイン後、auth_hands_onデータベースを使用します。

USE auth_hands_on;

続いて、usersテーブルの中身を確認します。

SELECT id, email, password_hash, created_at FROM users;

以下のような実行結果が表示されます（password_hashの値はハッシュ化された結果なので毎回異なります）。

+----+-------------------+--------------------------------------------------------------+---------------------+
| id | email             | password_hash                                                | created_at          |
+----+-------------------+--------------------------------------------------------------+---------------------+
|  1 | test@example.com  | $2b$12$abc...（bcryptのハッシュ文字列）                       | 2026-08-22 12:00:00 |
+----+-------------------+--------------------------------------------------------------+---------------------+

password_hashカラムに$2b$12$から始まる文字列が入っていれば、bcryptによるハッシュ化と保存が正しく行えています。確認できたら、exit;でMySQLのプロンプトを抜けます。

exit;

5. ログイン

ログインを行うPOST /loginエンドポイントを実装します。メールアドレスとパスワードを検証し、成功時にアクセストークン（JWT）とリフレッシュトークンを返す処理です。

5.1 auth.py の書き換えとJWT発行ヘルパーの追加

auth.pyにJWT発行・パスワード検証などのヘルパー関数を追加します。冒頭のimportと定数、および関数群を含めて、以下の内容に書き換えます。既存のhash_passwordはそのまま残し、verify_password・generate_access_token・parse_access_token・generate_refresh_token・refresh_token_expires_atを追記した形です。

import os
import secrets
import time
from datetime import datetime, timedelta

import bcrypt
import jwt

ACCESS_TOKEN_TTL_SECONDS = 15 * 60
REFRESH_TOKEN_TTL_DAYS = 30
JWT_ALGORITHM = "HS256"


def get_jwt_secret() -> str:
    secret = os.getenv("JWT_SECRET")
    if not secret:
        raise RuntimeError("JWT_SECRET が設定されていません")
    return secret


def hash_password(password: str) -> str:
    salt = bcrypt.gensalt()
    return bcrypt.hashpw(password.encode("utf-8"), salt).decode("utf-8")


def verify_password(password: str, hashed: str) -> bool:
    return bcrypt.checkpw(password.encode("utf-8"), hashed.encode("utf-8"))


def generate_access_token(user_id: int) -> str:
    now = int(time.time())
    payload = {
        "sub": str(user_id),
        "iat": now,
        "exp": now + ACCESS_TOKEN_TTL_SECONDS,
    }
    return jwt.encode(payload, get_jwt_secret(), algorithm=JWT_ALGORITHM)


def parse_access_token(token: str) -> int:
    payload = jwt.decode(token, get_jwt_secret(), algorithms=[JWT_ALGORITHM])
    return int(payload["sub"])


def generate_refresh_token() -> str:
    return secrets.token_hex(32)


def refresh_token_expires_at() -> datetime:
    return datetime.utcnow() + timedelta(days=REFRESH_TOKEN_TTL_DAYS)

コードを解説します。

ACCESS_TOKEN_TTL_SECONDS = 15 * 60
REFRESH_TOKEN_TTL_DAYS = 30
JWT_ALGORITHM = "HS256"

アクセストークンとリフレッシュトークンの有効期限、および署名アルゴリズムを定数で定義します。アクセストークンは短命（15分）、リフレッシュトークンは長め（30日）にすることで、盗難時の影響と再ログイン頻度のバランスを取ります。

def get_jwt_secret() -> str:
    secret = os.getenv("JWT_SECRET")
    if not secret:
        raise RuntimeError("JWT_SECRET が設定されていません")
    return secret

JWTの署名鍵を環境変数JWT_SECRETから読み込むヘルパーです。ここをコードにハードコードすると、リポジトリを見られた時点でトークンの偽造が可能になるため、.envや本番環境の環境変数として外部から注入します。未設定時にRuntimeErrorで即座に停止させることで、鍵が空のまま起動してしまう事故を防ぎます。

def verify_password(password: str, hashed: str) -> bool:
    return bcrypt.checkpw(password.encode("utf-8"), hashed.encode("utf-8"))

verify_passwordは、入力パスワードとDBに保存したハッシュをbcrypt.checkpwで照合する関数です。ハッシュ側にソルトが埋め込まれているため、呼び出し側でソルトを別途保持・受け渡しする必要はありません。

def generate_access_token(user_id: int) -> str:
    now = int(time.time())
    payload = {
        "sub": str(user_id),
        "iat": now,
        "exp": now + ACCESS_TOKEN_TTL_SECONDS,
    }
    return jwt.encode(payload, get_jwt_secret(), algorithm=JWT_ALGORITHM)

generate_access_tokenは、JWTのアクセストークンを発行する関数です。subにユーザID、iatに発行時刻、expに有効期限を含めます。HS256はHMAC-SHA256の共通鍵署名で、JWT_SECRETを秘密鍵として署名します。同じ秘密鍵を持っているサーバのみ、後から署名を検証できます。

def parse_access_token(token: str) -> int:
    payload = jwt.decode(token, get_jwt_secret(), algorithms=[JWT_ALGORITHM])
    return int(payload["sub"])

parse_access_tokenはトークンの検証と復号を行う関数です。algorithms=[JWT_ALGORITHM]で許容する署名アルゴリズムを明示することで、algフィールドをnoneなどに偽装したトークンを弾けるようにしています。ここを省略してalgorithms=Noneにすると、署名検証をバイパスされる恐れがあります。

def generate_refresh_token() -> str:
    return secrets.token_hex(32)

generate_refresh_tokenは、リフレッシュトークン文字列を生成する関数です。secrets.token_hexで生成した暗号学的乱数を16進文字列にしたものを使います。JWTのように自己完結した情報を持たせず、DBに保存された値と一致するかで判定するため、失効管理をDB操作だけで完結できます。

def refresh_token_expires_at() -> datetime:
    return datetime.utcnow() + timedelta(days=REFRESH_TOKEN_TTL_DAYS)

リフレッシュトークンの有効期限（現在時刻＋TTL日数）を計算するヘルパーです。ハンドラ側から呼び出してRefreshTokenレコードのexpires_atに入れる用途で使います。

5.2 handlers.py への login ハンドラの追加

handlers.pyのimportとrouter宣言を以下のように書き換えます。前のimportにLoginRequest・RefreshToken・TokenResponseのモデル、そしてauth.pyから新たにgenerate_access_token・generate_refresh_token・refresh_token_expires_at・verify_passwordを追加した形です。

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select

from auth import (
    generate_access_token,
    generate_refresh_token,
    hash_password,
    refresh_token_expires_at,
    verify_password,
)
from database import get_session
from models import (
    LoginRequest,
    RefreshToken,
    SignupRequest,
    TokenResponse,
    User,
)

router = APIRouter()

続いて、handlers.pyの末尾（signup関数の下）に、リフレッシュトークン発行の共通ヘルパーissue_refresh_tokenとloginハンドラを追加します。

def issue_refresh_token(user_id: int, session: Session) -> str:
    token = generate_refresh_token()
    rt = RefreshToken(
        user_id=user_id,
        token=token,
        expires_at=refresh_token_expires_at(),
    )
    session.add(rt)
    session.commit()
    return token


@router.post("/login", response_model=TokenResponse)
def login(
    payload: LoginRequest,
    session: Annotated[Session, Depends(get_session)],
):
    user = session.exec(select(User).where(User.email == payload.email)).first()
    if user is None or not verify_password(payload.password, user.password_hash):
        raise HTTPException(status_code=401, detail="メールアドレスまたはパスワードが不正です")

    return TokenResponse(
        access_token=generate_access_token(user.id),
        refresh_token=issue_refresh_token(user.id, session),
    )

5.3 コードの解説

def issue_refresh_token(user_id: int, session: Session) -> str:
    token = generate_refresh_token()
    rt = RefreshToken(
        user_id=user_id,
        token=token,
        expires_at=refresh_token_expires_at(),
    )
    session.add(rt)
    session.commit()
    return token

リフレッシュトークンをDBに保存する共通処理を関数として切り出しています。ログイン時とリフレッシュ時の両方から呼び出すため、重複を避けるためのヘルパーです。ハンドラ間で共有する処理はモジュールレベルの関数として切り出しておくと、責務の切り分けがしやすくなります。

@router.post("/login", response_model=TokenResponse)
def login(
    payload: LoginRequest,
    session: Annotated[Session, Depends(get_session)],
):

loginハンドラのシグネチャです。signupと同じく、payload: LoginRequestでリクエストボディを型付きで受け取り、Depends(get_session)でDBセッションを注入します。response_model=TokenResponseにより、レスポンスがTokenResponseスキーマにキャストされます。

user = session.exec(select(User).where(User.email == payload.email)).first()
if user is None or not verify_password(payload.password, user.password_hash):
    raise HTTPException(status_code=401, detail="メールアドレスまたはパスワードが不正です")

入力されたメールアドレスでusersテーブルを検索し、見つかったUserのハッシュと入力パスワードをverify_passwordで照合します。「ユーザが存在しない」場合と「パスワードが違う」場合の両方で共通の401メッセージを返しています。区別できるメッセージを返してしまうと、「そのメールアドレスがサービスに登録されているか」を第三者に推測されやすくなります。認証に関わる失敗は、原因を明示しないメッセージにそろえるのが基本です。

return TokenResponse(
    access_token=generate_access_token(user.id),
    refresh_token=issue_refresh_token(user.id, session),
)

パスワード検証を通過したユーザに対して、generate_access_tokenでアクセストークンを、issue_refresh_tokenでリフレッシュトークンを発行してまとめて返します。クライアントはこの2つを受け取り、以降のAPIリクエストではアクセストークンを、期限切れ時にはリフレッシュトークンを使います。

5.4 動作確認

uvicornは--reloadで動いているので、handlers.pyとauth.pyの保存で自動再起動されます。先ほど作成したユーザでログインしてトークンを取得します。別のターミナルから以下を実行します。

curl -X POST http://localhost:8000/login -H "Content-Type: application/json" -d '{"email":"test@example.com","password":"password12345"}'

以下のような実行結果が表示されます（トークンの値は環境により変わります）。

{"access_token":"eyJhbGciOiJIUzI1NiIs...","refresh_token":"3f4c8d2a6b..."}

access_tokenはJWT（.で3つに区切られた文字列）、refresh_tokenはsecrets.token_hexで生成した16進文字列です。次のセクションで、このアクセストークンをAuthorizationヘッダに載せて保護されたエンドポイントを叩きます。

6. 保護されたAPI呼び出し

発行したアクセストークンを検証する仕組みを実装し、認証済みユーザのみが叩けるGET /notesエンドポイントを作ります。FastAPIのDependsとHTTPBearerスキームを組み合わせて、認証を必要とするエンドポイントに認証依存を注入します。

6.1 dependencies.py の作成と認証依存の実装

続いて、認証依存を配置するdependencies.pyを作成します。認証依存は複数のハンドラの前段で共通に使う横断的な処理なので、handlers.pyと分けて専用ファイルに置くと役割が明確になります。

Visual Studio Codeのエクスプローラーでjwt-auth-hands-onフォルダを右クリックし、「新しいファイル」を選択してdependencies.pyを作成します。

jwt-auth-hands-on/
├── .env
├── auth.py
├── database.py
├── dependencies.py  ← このファイルを作成
├── handlers.py
├── main.py
├── models.py
└── setup_db.py

作成したdependencies.pyに以下の内容を記述して保存します。HTTPBearerスキームと、認証済みユーザを取り出すget_current_user依存関数を定義しています。

from typing import Annotated

from fastapi import Depends, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlmodel import Session

from auth import parse_access_token
from database import get_session
from models import User

security = HTTPBearer()


def get_current_user(
    credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
    session: Annotated[Session, Depends(get_session)],
) -> User:
    try:
        user_id = parse_access_token(credentials.credentials)
    except Exception:
        raise HTTPException(status_code=401, detail="トークンが無効です")
    user = session.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=401, detail="ユーザが見つかりません")
    return user

コードを解説します。

security = HTTPBearer()

HTTPBearerは、Authorization: Bearer <トークン>形式のヘッダを受け取るスキームです。Swagger UI（/docs）に「Authorize」ボタンが表示され、そこから入力したトークンが以降のリクエストに自動で付与されます。

def get_current_user(
    credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
    session: Annotated[Session, Depends(get_session)],
) -> User:

get_current_userの関数シグネチャです。Depends(security)でBearerヘッダから取り出したトークン情報を、Depends(get_session)でDBセッションを注入します。戻り値の型はUserで、保護エンドポイント側はcurrent_user: Userとして直接扱えます。

try:
    user_id = parse_access_token(credentials.credentials)
except Exception:
    raise HTTPException(status_code=401, detail="トークンが無効です")

credentials.credentialsにトークン本体（Bearer を除いた部分）が入っているので、parse_access_tokenで検証・復号します。検証失敗時は401で早期リターンします。呼び出し元にどの理由で失敗したかは伝えず、抽象化したエラーだけを返します。

user = session.get(User, user_id)
if user is None:
    raise HTTPException(status_code=401, detail="ユーザが見つかりません")
return user

user_idをキーにDBからUserを取得します。トークンは有効でもユーザが削除されている場合は401で拒否します。取得したUserインスタンスを返すことで、認証依存を使うハンドラはcurrent_userとして直接ユーザ情報を扱えます。

6.2 handlers.py への notes ハンドラの追加

handlers.pyのimportにget_current_userを追加し、notesハンドラを末尾に追記します。importを以下のように書き換えます。

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select

from auth import (
    generate_access_token,
    generate_refresh_token,
    hash_password,
    refresh_token_expires_at,
    verify_password,
)
from database import get_session
from dependencies import get_current_user
from models import (
    LoginRequest,
    RefreshToken,
    SignupRequest,
    TokenResponse,
    User,
)

続いて、handlers.pyの末尾（login関数の下）にnotesハンドラを追加します。ここでは認証の動作確認が主眼のため、DBからノートを取得する処理は入れず、current_user.idを含んだ固定のノート一覧を返す形にしています。

@router.get("/notes")
def notes(current_user: Annotated[User, Depends(get_current_user)]):
    return [
        {"id": 1, "title": "買い物", "user_id": current_user.id},
        {"id": 2, "title": "打ち合わせ資料", "user_id": current_user.id},
    ]

6.3 コードの解説

@router.get("/notes")
def notes(current_user: Annotated[User, Depends(get_current_user)]):

notesハンドラのシグネチャです。Depends(get_current_user)で認証依存を注入するだけで、そのエンドポイントは自動的に「有効なトークンが必要」な保護エンドポイントになります。ハンドラ側はミドルウェアの登録などを気にせず、current_userを使うだけで済みます。

return [
    {"id": 1, "title": "買い物", "user_id": current_user.id},
    {"id": 2, "title": "打ち合わせ資料", "user_id": current_user.id},
]

本章では認証の動作確認が主眼のため、DBからノートを取得する処理は入れず、固定のノート一覧を作って返しています。実際のアプリケーションでは、current_user.idをキーにnotesテーブルから該当ユーザのレコードを取得する形になります。

6.4 保護されたエンドポイントの動作確認

uvicornは--reloadオプションで動いているので、ファイル保存で自動再起動されます。

まず、Authorizationヘッダを付けずに/notesを叩いて、401が返ることを確認します。

curl -i http://localhost:8000/notes

以下のような実行結果が表示されます。

HTTP/1.1 401 Unauthorized
www-authenticate: Bearer
...
{"detail":"Not authenticated"}

HTTPBearerが働き、Authorizationヘッダの無いリクエストが弾かれています。www-authenticate: Bearer ヘッダは「Bearer 認証で再送してほしい」ことをクライアントに伝える標準的な認証チャレンジで、HTTPBearer が自動で付与します。

次に、先ほど/loginで取得したアクセストークンをAuthorizationヘッダに付けて再度叩きます。トークンの値は自分のログイン結果に置き換えてください。

curl -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIs..." http://localhost:8000/notes

以下のような実行結果が表示されます。

[{"id":1,"title":"買い物","user_id":1},{"id":2,"title":"打ち合わせ資料","user_id":1}]

トークンが検証を通過し、get_current_userで取得されたUserのIDを使ってノート一覧が返されています。認証済みユーザのみが叩けるエンドポイントの雛形ができました。

7. アクセストークンの再発行

アクセストークンの再発行を行うPOST /refreshエンドポイントを実装します。リクエストで受け取ったリフレッシュトークンがDBに存在し、失効・期限切れになっていないことを確認したうえで、古いリフレッシュトークンを失効させて新しいアクセストークンとリフレッシュトークンを返す処理です。

7.1 refresh ハンドラの追加

handlers.pyのimportにdatetimeとRefreshRequestを追加します。

from datetime import datetime
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select

from auth import (
    generate_access_token,
    generate_refresh_token,
    hash_password,
    refresh_token_expires_at,
    verify_password,
)
from database import get_session
from dependencies import get_current_user
from models import (
    LoginRequest,
    RefreshRequest,
    RefreshToken,
    SignupRequest,
    TokenResponse,
    User,
)

続いて、handlers.pyの末尾（notes関数の下）にrefreshハンドラを追加します。

@router.post("/refresh", response_model=TokenResponse)
def refresh(
    payload: RefreshRequest,
    session: Annotated[Session, Depends(get_session)],
):
    rt = session.exec(
        select(RefreshToken).where(RefreshToken.token == payload.refresh_token)
    ).first()
    if rt is None or rt.revoked_at is not None or rt.expires_at <= datetime.utcnow():
        raise HTTPException(status_code=401, detail="refresh token が無効または失効しています")

    rt.revoked_at = datetime.utcnow()
    session.add(rt)
    session.commit()

    return TokenResponse(
        access_token=generate_access_token(rt.user_id),
        refresh_token=issue_refresh_token(rt.user_id, session),
    )

7.2 コードの解説

@router.post("/refresh", response_model=TokenResponse)
def refresh(
    payload: RefreshRequest,
    session: Annotated[Session, Depends(get_session)],
):

refreshハンドラのシグネチャです。payload: RefreshRequestでリフレッシュトークンをリクエストボディから受け取り、Depends(get_session)でDBセッションを注入します。response_model=TokenResponseにより、レスポンスはTokenResponseスキーマにキャストされて返ります。

rt = session.exec(
    select(RefreshToken).where(RefreshToken.token == payload.refresh_token)
).first()
if rt is None or rt.revoked_at is not None or rt.expires_at <= datetime.utcnow():
    raise HTTPException(status_code=401, detail="refresh token が無効または失効しています")

DBからリフレッシュトークンを検索し、以下のいずれかに該当する場合はまとめて401で拒否します。

• トークンがDBに存在しない
• revoked_atが設定されている（＝失効済み）
• expires_atが現在時刻を過ぎている

「存在しない」と「失効している」でメッセージを分けず、詳細を漏らさない構成にしています。

rt.revoked_at = datetime.utcnow()
session.add(rt)
session.commit()

古いリフレッシュトークンをrevoked_atで失効させます。再発行のたびに古いトークンを失効させるローテーション運用にしておくと、盗まれた古いトークンで永続的に新しいアクセストークンを取得され続けるリスクを下げられます。

return TokenResponse(
    access_token=generate_access_token(rt.user_id),
    refresh_token=issue_refresh_token(rt.user_id, session),
)

古いトークンを失効させたうえで、新しいアクセストークンとリフレッシュトークンを発行してレスポンスに返します。クライアント側は、以降のリクエストで受け取った最新のリフレッシュトークンだけを使う運用になります。

7.3 動作確認

APIサーバは--reloadで動いているので、handlers.pyの保存で自動再起動されます。ログイン時に取得しておいたrefresh_tokenを/refreshに投げて、新しいアクセストークンとリフレッシュトークンを取得します。

curl -X POST http://localhost:8000/refresh -H "Content-Type: application/json" -d '{"refresh_token":"3f4c8d2a..."}'

以下のような実行結果が表示されます。

{"access_token":"eyJhbGciOi...新しい...","refresh_token":"a91b7e3d...新しい..."}

アクセストークンとリフレッシュトークンの両方が新しい値になっています。ここで、古いリフレッシュトークンで再度リフレッシュを試みると失効エラーになることを確認します。

curl -X POST http://localhost:8000/refresh -H "Content-Type: application/json" -d '{"refresh_token":"3f4c8d2a..."}'

以下のような実行結果が表示されます。

{"detail":"refresh token が無効または失効しています"}

/refreshのローテーションが働き、古いリフレッシュトークンは1度しか使えない状態になっています。次のセクションのログアウトでは、この新しく取得したリフレッシュトークン（a91b7e3d...）を使います。

8. ログアウト

ログアウトを行うPOST /logoutエンドポイントを実装します。JWTはステートレスな設計上、発行済みのアクセストークンをサーバ側から即時に無効化することは本来できません。「ログアウトしても、そのアクセストークンが有効期限まで生き残ってしまう」という状態は本来のログアウトの意図とかみ合わないため、DBにアクセストークンの無効化リスト（ブラックリスト）を持ち、認証依存で照合する形で即時無効化を実現します。

本章のログアウトでは、以下の2つを同時に行います。

• アクセストークンをrevoked_access_tokensテーブルに登録し、認証依存で拒否させる
• リフレッシュトークンをrefresh_tokensテーブルのrevoked_atで失効させ、以降の/refreshを拒否させる

8.1 無効化用モデルの追加

まず、失効させたアクセストークンを管理するモデルをmodels.pyに追加します。

class RevokedAccessToken(SQLModel, table=True):
    __tablename__ = "revoked_access_tokens"
    id: Optional[int] = Field(default=None, primary_key=True)
    token: str = Field(sa_column=Column(String(512), unique=True, nullable=False, index=True))
    expires_at: datetime
    revoked_at: datetime = Field(default_factory=datetime.utcnow)

続いて、setup_db.pyでSQLModel.metadata.create_allが実行される際に新しいモデルが認識されるよう、setup_db.pyのimport側にもRevokedAccessTokenが含まれていることを確認します（from models import ...で*または対象モデルを列挙している場合、RevokedAccessTokenも併せてimportします）。準備できたら、テーブル作成スクリプトを実行してrevoked_access_tokensテーブルを作成します。

Windowsの場合:

python setup_db.py

Macの場合:

python3 setup_db.py

コードを解説します。

token: str = Field(sa_column=Column(String(512), unique=True, nullable=False, index=True))

TokenにはJWTの文字列そのものを保管します。JWTはヘッダ.ペイロード.署名の3パートを連結した長めの文字列なので、String(512)で十分な長さを確保します。unique=Trueとindex=Trueにより、同じトークンを重複登録しない制約と、検索の高速化を同時に指定しています。

expires_at: datetime

このトークンの有効期限（JWT側のexpと同じ値）を保存します。今回のハンズオンでは使いませんが、有効期限を過ぎたレコードは検証時に既に弾かれるため、バッチ処理で古いレコードを削除するためのカラムになります。

8.2 認証依存への無効化チェック追加

続いて、get_current_userに「アクセストークンがブラックリストに載っていないか」を確認する処理を追加します。dependencies.pyのimportにselectとRevokedAccessTokenを追加します。

from typing import Annotated

from fastapi import Depends, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlmodel import Session, select

from auth import parse_access_token
from database import get_session
from models import RevokedAccessToken, User

続いて、get_current_userを以下のように書き換えます。

def get_current_user(
    credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
    session: Annotated[Session, Depends(get_session)],
) -> User:
    try:
        user_id = parse_access_token(credentials.credentials)
    except Exception:
        raise HTTPException(status_code=401, detail="トークンが無効です")

    revoked = session.exec(
        select(RevokedAccessToken).where(
            RevokedAccessToken.token == credentials.credentials
        )
    ).first()
    if revoked is not None:
        raise HTTPException(status_code=401, detail="トークンが無効化されています")

    user = session.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=401, detail="ユーザが見つかりません")
    return user

コードを解説します。

revoked = session.exec(
    select(RevokedAccessToken).where(
        RevokedAccessToken.token == credentials.credentials
    )
).first()
if revoked is not None:
    raise HTTPException(status_code=401, detail="トークンが無効化されています")

JWTの検証（署名・有効期限）を通過したあとに、revoked_access_tokensテーブルにそのトークンが登録されていないかを確認します。見つかった場合は無効化済みとして401 Unauthorizedで拒否します。この照合が入ることで、ログアウト直後のアクセストークンを即座に拒否できるようになります。

8.3 logout ハンドラの追加

handlers.pyのimportにtimedelta、HTTPAuthorizationCredentials、ACCESS_TOKEN_TTL_SECONDS、RevokedAccessToken、dependenciesのsecurityを追加します。

from datetime import datetime, timedelta
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials
from sqlmodel import Session, select

from auth import (
    ACCESS_TOKEN_TTL_SECONDS,
    generate_access_token,
    generate_refresh_token,
    hash_password,
    refresh_token_expires_at,
    verify_password,
)
from database import get_session
from dependencies import get_current_user, security
from models import (
    LoginRequest,
    RefreshRequest,
    RefreshToken,
    RevokedAccessToken,
    SignupRequest,
    TokenResponse,
    User,
)

続いて、handlers.pyの末尾（refresh関数の下）にlogoutハンドラを追加します。アクセストークンをrevoked_access_tokensテーブルに登録し、あわせてリフレッシュトークンも失効させます。Depends(get_current_user)を付けることで、認証依存を通過したリクエストだけがこのハンドラに到達するようになります。

@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
def logout(
    payload: RefreshRequest,
    credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
    current_user: Annotated[User, Depends(get_current_user)],
    session: Annotated[Session, Depends(get_session)],
):
    revoked = RevokedAccessToken(
        token=credentials.credentials,
        expires_at=datetime.utcnow() + timedelta(seconds=ACCESS_TOKEN_TTL_SECONDS),
    )
    session.add(revoked)

    rt = session.exec(
        select(RefreshToken).where(
            RefreshToken.token == payload.refresh_token,
            RefreshToken.revoked_at.is_(None),
        )
    ).first()
    if rt is not None:
        rt.revoked_at = datetime.utcnow()
        session.add(rt)

    session.commit()

8.4 コードの解説

@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
def logout(
    payload: RefreshRequest,
    credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
    current_user: Annotated[User, Depends(get_current_user)],
    session: Annotated[Session, Depends(get_session)],
):

logoutハンドラのシグネチャです。payloadでリフレッシュトークンを、credentialsで生のアクセストークン文字列（Bearer を除いた部分）を取り出し、get_current_userで認証必須にします。get_current_userを通過するのは有効かつ無効化されていないトークンだけなので、このハンドラに到達する時点でトークンは有効です。status_code=status.HTTP_204_NO_CONTENTにより、戻り値なしでも204 No Contentが返る設定です。

revoked = RevokedAccessToken(
    token=credentials.credentials,
    expires_at=datetime.utcnow() + timedelta(seconds=ACCESS_TOKEN_TTL_SECONDS),
)
session.add(revoked)

取り出したアクセストークンをrevoked_access_tokensテーブルに登録します。以降このトークンで保護されたAPIを叩くと、get_current_userのブラックリスト照合で弾かれるようになります。

rt = session.exec(
    select(RefreshToken).where(
        RefreshToken.token == payload.refresh_token,
        RefreshToken.revoked_at.is_(None),
    )
).first()
if rt is not None:
    rt.revoked_at = datetime.utcnow()
    session.add(rt)

続けて、リクエストボディで受け取ったリフレッシュトークンをrefresh_tokensテーブルのrevoked_atで失効させます。revoked_at IS NULLの条件で、まだ失効していないレコードだけを対象にします。

session.commit()

アクセストークンのブラックリスト登録とリフレッシュトークンの失効を同じトランザクションでコミットします。status_code=status.HTTP_204_NO_CONTENTの指定により、戻り値なしでも204 No Contentが返る設定です。

8.5 動作確認

前のセクションで取得した最新のアクセストークン（eyJhbGci...）とリフレッシュトークン（a91b7e3d...）を/logoutに送り、両方を失効させます。アクセストークンはAuthorizationヘッダに、リフレッシュトークンはリクエストボディに載せます。

curl -i -X POST http://localhost:8000/logout \
  -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIs..." \
  -H "Content-Type: application/json" \
  -d '{"refresh_token":"a91b7e3d..."}'

以下のような実行結果が表示されます。

HTTP/1.1 204 No Content

204 No Contentが返り、アクセストークンはrevoked_access_tokensテーブルに登録され、リフレッシュトークンはrefresh_tokensテーブルのrevoked_atに時刻が入りました。

即時無効化が働いていることを確認するため、先ほどログアウトに使ったアクセストークンで、もう一度/notesを叩いてみます。

curl -i -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIs..." http://localhost:8000/notes

以下のような実行結果が表示されます。

HTTP/1.1 401 Unauthorized
...
{"detail":"トークンが無効化されています"}

アクセストークン自体はまだ有効期限内ですが、get_current_userのブラックリスト照合で拒否されました。JWTが本来持てなかった「即時無効化」を、DBのブラックリストで実現できていることが確認できます。

念のため、リフレッシュトークン側も無効化されていることを確認します。同じリフレッシュトークンで/refreshを叩くと、失効エラーで拒否されます。

curl -X POST http://localhost:8000/refresh -H "Content-Type: application/json" -d '{"refresh_token":"a91b7e3d..."}'

{"detail":"refresh token が無効または失効しています"}

アクセストークン・リフレッシュトークンの両方が無効化された状態になり、そのセッションからは新規のトークン発行もAPI呼び出しもできなくなりました。これで、DB管理方式による即時無効化を含めたログアウトの仕組みが一通り動くことが確認できます。

9. マネージド認証サービスの活用

本章ではJWT認証を独自に実装しましたが、実運用のアプリケーションでは、パスワードの保存・多要素認証（MFA）・パスワードリセット・ソーシャルログイン（Google・Appleなど）といった機能をゼロから実装せず、マネージドな認証サービスを使うことも一般的です。代表的な選択肢は以下のとおりです。

• Amazon Cognito: AWSが提供する認証サービス。ユーザプール機能でサインアップ・ログイン・MFA・ソーシャルログインを一通り扱える
• Auth0: 認証・認可のマネージドサービスとして広く使われる
• Firebase Authentication: GoogleのBaaS。モバイルとの相性が良い
• Google Identity Platform: エンタープライズ向けのGCPマネージド認証

これらのサービスを使うと、本章で実装したパスワードハッシュ・JWT発行・リフレッシュトークン管理・失効管理などをサービス側に委ねられます。ただし、サービス側から発行されるIDトークン・アクセストークンの検証はAPIサーバ側で行う必要があるため、本章で扱ったJWT検証の考え方はそのまま活かせます。

独自に実装するか、マネージドサービスに寄せるかは、要件やチームの状況次第で判断します。認証まわりに独自の要件が多い場合や既存システムと密に連携する場合は独自実装、逆に短期間で本番運用を立ち上げたい場合やチームに認証実装の経験者がいない場合は、実装ミスによる脆弱性を避けるためにもマネージドサービスの利用が現実的です。

本章の内容自体は、いずれのケースでも「トークンベース認証の仕組み」を理解するための基礎知識として役立ちます。

10. 不要リソースの削除

APIサーバをCtrl + Cで停止します。ハンズオンで使ったauth_hands_onデータベースは、残しておく必要はないため削除します。

mysql -u root -p -e "DROP DATABASE auth_hands_on;"

パスワードを入力してエラーが出なければ、データベースの削除が完了しています。作業用のフォルダ（jwt-auth-hands-on/）はそのまま残しておいても支障ありませんが、不要であれば削除してください。

11. まとめ

この章では、JWT認証の仕組みを学びつつ、実際にサインアップ・ログイン・保護されたエンドポイント・リフレッシュ・ログアウトまでの一通りをFastAPIで実装しました。

• JWTは、トークン自体に検証可能な情報を埋め込むステートレスな認証方式である
• JWTはヘッダ・ペイロード・署名の3パートを.で連結した文字列で、ペイロードは暗号化されず誰でも復号できる
• パスワードはbcryptでソルト付きハッシュ化して保存し、レスポンスに含めない
• 認証系のエラーメッセージは、原因を区別しない共通の文言にそろえる
• アクセストークンは短命（例: 15分）にして盗難時の影響を狭め、リフレッシュトークンで再発行する構成にする
• リフレッシュトークンはDBで管理し、ローテーションと失効管理ができる
• FastAPIのDependsとHTTPBearerを組み合わせて、認証済みユーザを注入する共通処理を作れる
• ログアウトはリフレッシュトークンをrevoked_atで失効させることで実現し、以降の/refreshを拒否できる
• JWTの即時無効化を実現するため、revoked_access_tokensテーブルでアクセストークンのブラックリストを管理し、get_current_userで照合することでログアウト直後から拒否できる

次の章では、Pythonの静的解析ツールをハンズオン形式で体験します。