認証とJWT — 安全なAPIアイデンティティシステム
JWTはホテルのルームキーのようなものです。Access Tokenはデイパス (短時間有効), Refresh Tokenは月間パス (長期間有効だがいつでも取り消し可能)です。フロントデスク (認証サービス)がキーの発行と検証を担当します。
1. 学ぶ内容
- JWTの原理と構造:Header, Payload, Signatureの解説
python-joseによるAccess TokenとRefresh Tokenの発行・検証 (デュアルトークン)OAuth2PasswordBearerとFastAPIセキュリティツールの統合- パスワードハッシュ:
passlib+bcryptのベストプラクティス - Aliceシナリオ:PriceTrackerのSaaSマルチテナント認証 — 各サブスクリプションプランの権限マトリックス
2. Aliceのリアルストーリー
(1) ペインポイント:未認証APIが悪意あるスクレイピングにさらされている
AliceのPriceTracker APIは完全にオープンで, 誰でもすべてのエンドポイントを呼び出せます。競合他社がスクリプトを書いて1分間に100万件の価格データをスクレイピングしています。さらに悪いことに, 誰かがAPIを使って偽の価格データを送信し, データベース全体を汚染しました。Aliceはユーザー登録・ログイン, API認証, 各サブスクリプションレベルに基づくアクセス制御を安全に実装する必要がありますが, その方法がわかりません。
(2) JWT認証のソリューション
JWT (JSON Web Token)はステートレスな認証ソリューションです。ユーザーはログイン時にトークンを取得し, 以降のリクエストにトークンを含め, サーバーが署名を検証します。セッションを保存する必要がなく, 分散システムに本質的に対応しています。
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
@app.get("/me")
async def get_me(token: str = Depends(oauth2_scheme)):
user = verify_token(token)
return user
(3) 収益
API認証を有効にすると, 未登録ユーザーは保護されたエンドポイントにアクセスできなくなり, 悪意あるスクレイピングはレート制限ミドルウェアと認証の両方でブロックされます。Proユーザーはデータを一括インポートでき, Freeユーザーは1,000件に制限されます。競合他社は公開データしか閲覧できません。
3. JWTの原理と構造
(1) JWTの3つの構成要素
JWTは3つの部分で構成されます:Header (アルゴリズム), Payload (データ), Signature (署名)。.で区切られます。
sequenceDiagram
participant Client as Bob Frontend
participant Auth as /auth/login
participant API as Protected API
participant Verify as Token Verifier
Client->>Auth: POST email + password
Auth->>Auth: Verify credentials
Auth-->>Client: Access Token + Refresh Token
Client->>API: GET /products (Bearer Token)
API->>Verify: Decode & verify signature
Verify-->>API: User payload
API-->>Client: 200 OK + data
Note over Client,Auth: Access Token expires after 30 min
Client->>Auth: POST /auth/refresh (Refresh Token)
Auth-->>Client: New Access Token
| セクション | 内容 | 例 |
|---|---|---|
| Header | アルゴリズムタイプ | {"alg": "HS256", "typ": "JWT"} |
| Payload | ユーザーデータ (Claims) | {"sub": "alice", "role": "admin", "exp": 1700000000} |
| Signature | 署名 | HMACSHA256(header.payload, secret) |
(1) ▶サンプル:JWTのエンコードとデコード
from jose import jwt, JWTError
from datetime import datetime, timedelta
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
def create_access_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=30))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def verify_token(token: str) -> dict:
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
raise ValueError("Invalid token")
# テスト
token = create_access_token({"sub": "alice", "role": "admin"})
print(f"Token: {token[:50]}...")
payload = verify_token(token)
print(f"Payload: {payload}")
出力:
# 関数定義成功
(2) Access TokenとRefresh Tokenの比較
| 項目 | Access Token | Refresh Token |
|---|---|---|
| 有効期間 | 30分 | 7日間 |
| 用途 | APIリソースへのアクセス | Access Tokenの更新 |
| 保存先 | メモリ (フロントエンド) | HttpOnly Cookie |
| 漏洩リスク | 高 (全リクエストに含まれる) | 低 (更新時のみ使用) |
| 取消方法 | 有効期限切れを待つ | サーバーのブラックリスト |
4. OAuth2PasswordBearerの統合
(1) ▶サンプル:完全な認証設定
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from datetime import datetime, timedelta
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE = timedelta(minutes=30)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
app = FastAPI()
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=401,
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
except JWTError:
raise credentials_exception
# 本番環境:データベースからユーザーを検索
user = {"username": username, "role": payload.get("role", "user")}
return user
@app.post("/auth/login")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
# 本番環境:データベースからパスワードを検証
if form_data.username != "alice" or form_data.password != "secret":
raise HTTPException(status_code=401, detail="Incorrect credentials")
access_token = create_access_token(
data={"sub": form_data.username, "role": "admin"},
expires_delta=ACCESS_TOKEN_EXPIRE,
)
return {"access_token": access_token, "token_type": "bearer"}
@app.get("/me")
async def read_me(current_user: dict = Depends(get_current_user)):
return current_user
出力:
# 関数定義成功
5. パスワードハッシュ
(1) passlib + bcrypt
パスワードは決して平文で保存せず, bcryptアルゴリズムでソルト付きハッシュ化します。
(1) ▶サンプル:パスワードハッシュと認証
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
# テスト
hashed = hash_password("my-secret-password")
print(f"Hashed: {hashed[:30]}...")
print(f"Verify correct: {verify_password('my-secret-password', hashed)}")
print(f"Verify wrong: {verify_password('wrong-password', hashed)}")
出力:
Hashed: $2b$12$K3YQ8z9wB5eF7gH1j...
Verify correct: True
Verify wrong: False
(2) パスワードセキュリティのベストプラクティス
| プラクティス | 説明 |
|---|---|
| bcryptの使用 | アダプティブソルトハッシュでレインボーテーブルを防止 |
| MD5/SHA256は使用しない | ソルトなし, 計算が速すぎてブルートフォース攻撃に脆弱 |
| パスワード長さ≥8 | 最小長を強制 |
| 検証前にハッシュ化 | タイミング攻撃を防止 (passlibに内蔵) |
| SECRET_KEYは十分に長く | 最低32文字のランダム文字列, 環境変数に保存 |
6. SaaSマルチテナント権限マトリックス
(1) サブスクリプションレベルの権限設計
(1) ▶サンプル:サブスクリプションベースの権限定義
from fastapi import Depends, HTTPException
SUBSCRIPTION_LIMITS = {
"free": {"max_import": 1000, "websocket": False, "export": False},
"pro": {"max_import": 100000, "websocket": True, "export": True},
"enterprise": {"max_import": 1000000, "websocket": True, "export": True},
}
def require_subscription(min_level: str):
LEVELS = {"free": 0, "pro": 1, "enterprise": 2}
async def check_subscription(user: dict = Depends(get_current_user)):
user_level = user.get("subscription", "free")
if LEVELS.get(user_level, 0) < LEVELS.get(min_level, 0):
raise HTTPException(
status_code=403,
detail=f"Requires {min_level} subscription. Current: {user_level}",
)
return user
return check_subscription
@app.get("/api/v1/analytics")
async def get_analytics(user=Depends(require_subscription("pro"))):
return {"total_products": 1000000, "active_users": 5000}
@app.post("/api/v1/prices/bulk")
async def bulk_import(
prices: list[PriceCreate],
user=Depends(require_subscription("free")),
):
limits = SUBSCRIPTION_LIMITS[user.get("subscription", "free")]
if len(prices) > limits["max_import"]:
raise HTTPException(
status_code=403,
detail=f"Import limit: {limits['max_import']} for {user['subscription']} plan",
)
return {"imported": len(prices)}
出力:
# 関数定義成功
| エンドポイント | Free | Pro | Enterprise |
|---|---|---|---|
| GET /products | 1,000/日 | 無制限 | 無制限 |
| POST /prices | 1,000/バッチ | 100,000/バッチ | 1,000,000/バッチ |
| WebSocket /ws/prices | - | ✓ | ✓ |
| GET /analytics | - | ✓ | ✓ |
| CSVエクスポート | - | ✓ | ✓ |
| APIレート制限 | 100/分 | 1,000/分 | 無制限 |
❓ よくある質問
📖 まとめ
- JWTはHeader, Payload, Signatureの3部分で構成されます。ステートレス認証は分散システムに本質的に対応しています。
- Access Token (短期)はAPIアクセスに, Refresh Token (長期)は更新に使用します。この2トークンシステムによりセキュリティが向上します。
OAuth2PasswordBearerがFastAPIセキュリティツールと統合されており, Swagger UIが自動的にログインフォームを表示します- パスワードはbcryptハッシュで保存され, passlibがソルティングと検証ロジックを処理します。
- PriceTrackerの3層サブスクリプション権限マトリックス:Free/Pro/Enterpriseは
require_subscriptionDIで実装されます
📝 練習問題
- 基本問題 (難易度 ⭐):
/auth/loginエンドポイントを実装し, ユーザー名とパスワードを受け取り, 認証後にJWT Access Token (30分間有効)を返します。/meエンドポイントでDepends(oauth2_scheme)を使ってユーザーを解析します。ヒント:OAuth2PasswordRequestForm+jwt.encode - 応用問題 (難易度 ⭐⭐):Refresh Tokenの仕組みを追加します。Access Tokenは15分間, Refresh Tokenは7日間有効です。
/auth/refreshエンドポイントを実装し, Refresh Tokenと新しいAccess Tokenを交換します。ヒント:2つのcreate_token関数 + 異なる有効期限 - チャレンジ (難易度 ⭐⭐⭐):完全な登録/ログイン/権限システムを実装します。登録エンドポイントではbcryptハッシュでパスワードを保存し,
require_subscription(min_level)DIチェーンでサブスクリプションレベルを確認します。Freeユーザーは1,000件までインポート, Proユーザーは無制限にします。ヒント:hash_password+verify_password+require_subscriptionDI
---|



