🐱 うさねこ教室 Python と可観測性の教室

認証と認可

🐱 この章の目次

認証と認可の違い

認証(Authentication) は「誰であるか」を確認するプロセスです。 認可(Authorization) は「何を許可するか」を判定するプロセスです。 API 設計では、この 2 つを明確に分離し、異なるレイヤーで実装します。

OAuth 2.0 と JWT

OAuth 2.0 は、サードパーティアプリケーションにリソースへのアクセス権を委譲するための認可フレームワークです。 アクセストークンの形式としてよく使われるのが JWT(JSON Web Token) です。 JWT はヘッダー・ペイロード・署名の 3 部分からなり、署名によって改竄を検知できます。

# JWT の生成例
from datetime import datetime, timedelta, timezone
from jose import jwt

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"


def create_access_token(sub: str, expires_delta: timedelta = timedelta(hours=1)) -> str:
    payload = {
        "sub": sub,
        "exp": datetime.now(timezone.utc) + expires_delta,
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

API キー認証

API キー は、シンプルなサービス間認証に適した方式です。 ヘッダー(X-API-Key)やクエリパラメータで送信しますが、セキュリティ上はヘッダーが推奨されます。 ユーザー単位の認証には OAuth 2.0 を、マシン間通信には API キーを使い分けます。

FastAPI のセキュリティユーティリティ

FastAPI は fastapi.security モジュールに認証スキームの定義を提供しています。 これらを使うと OpenAPI ドキュメントにセキュリティ要件が自動反映されます。

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")


async def get_current_user(token: str = Depends(oauth2_scheme)) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    return {"id": user_id}

ミドルウェアによる認可

ルートごとに認可ロジックを書くとコードが散らばるため、共通の認可処理はミドルウェアまたは DI で集約します。 ロールベースアクセス制御(RBAC)を実装する場合は、DI でユーザー情報を取得し、デコレータやパーミッションチェッカーで権限を判定します。

from functools import wraps
from fastapi import Depends


def require_role(role: str):
    async def checker(current_user: dict = Depends(get_current_user)):
        if role not in current_user.get("roles", []):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
        return current_user
    return checker


@app.get("/admin/users", dependencies=[Depends(require_role("admin"))])
async def list_all_users():
    return {"users": []}