FastAPI 認証・認可 本番実装ガイド：OAuth2パスワードフロー × JWT（PyJWT）× Security scopesで守るAPI

from fastapi.security import OAuth2PasswordBearer

# tokenUrl は「トークンを取りに行く先」を OpenAPI に伝えるだけ。
# このエンドポイント自体を作るわけではない（作るのは第4章）。
# 相対URLにするのが重要：プロキシ配下でもパスが壊れない。
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

from typing import Annotated
from fastapi import Depends, FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items(token: Annotated[str, Depends(oauth2_scheme)]):
    # ここに来た時点で token は「Bearerヘッダから取れた文字列」。
    # ただし、まだ"中身が正しいか"は検証していない（第5章で検証する）。
    return {"token": token}

pip install "pwdlib[argon2]"

from pwdlib import PasswordHash

# recommended() は安全な既定（Argon2）を選んでくれる。
# pwdlib は bcrypt も使えるが、レガシー（古い）ハッシュ互換が要るときだけ passlib を使う。
password_hash = PasswordHash.recommended()

def get_password_hash(password: str) -> str:
    # 登録時：平文を受け取って即ハッシュ化。平文はここから先へ持ち出さない。
    return password_hash.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    # ログイン時：入力とDBのハッシュを照合（定数時間比較で行われる）。
    return password_hash.verify(plain_password, hashed_password)

from pydantic import BaseModel

class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None

class UserInDB(User):
    # DB から読むときだけ存在するフィールド。API レスポンス（User）には現れない。
    hashed_password: str

pip install pyjwt
# RSA/ECDSA（RS256など非対称鍵）を使うなら cryptography も:  pip install "pyjwt[crypto]"

# 署名鍵は十分なエントロピーで生成する。コードに書かない。
openssl rand -hex 32
# 例: 09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    secret_key: str                       # 環境変数 SECRET_KEY（openssl rand -hex 32）
    access_token_expire_minutes: int = 30
    model_config = SettingsConfigDict(env_file=".env")

settings = Settings()      # 環境変数/.env から読み込む。鍵はここにしか存在しない。
ALGORITHM = "HS256"        # 対称鍵署名。単一アプリならこれで十分（後述：マルチサービスは RS256）

from datetime import datetime, timedelta, timezone
import jwt    # PyJWT

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    # exp は「timezone-aware な UTC」で入れる。naive datetime は事故の元。
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    # 署名鍵で署名する。これにより改ざんは検知可能になる（中身は"暗号化"ではなく"署名"）。
    return jwt.encode(to_encode, settings.secret_key, algorithm=ALGORITHM)

# 例：ユーザーの一意IDを文字列化して sub に入れる
access_token = create_access_token(data={"sub": f"username:{user.username}"})

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    # トークンから復元する「検証済みの中身」。usernameは None 許容にしておく。
    username: str | None = None

from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm

app = FastAPI()

DUMMY_HASH = password_hash.hash("a-dummy-password-for-constant-time")

def authenticate_user(username: str, password: str) -> UserInDB | None:
    user = get_user(username)          # DB から UserInDB | None
    if user is None:
        # ユーザーが居なくても必ず検証を走らせ、応答時間を一定に保つ（タイミング攻撃対策）
        password_hash.verify(password, DUMMY_HASH)
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

@app.post("/token")
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        # 「ユーザー名が違う」「パスワードが違う」を区別して返さない（情報を与えない）
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="ユーザー名またはパスワードが正しくありません",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(
        data={"sub": f"username:{user.username}"},
        expires_delta=timedelta(minutes=settings.access_token_expire_minutes),
    )
    return Token(access_token=access_token, token_type="bearer")

from jwt.exceptions import InvalidTokenError

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],   # ヘッダから取れたトークン文字列
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="資格情報を検証できませんでした",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        # algorithms を明示（第3.4章の必須対策）。署名・exp が自動検証される。
        payload = jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM])
        subject = payload.get("sub")
        if subject is None:
            raise credentials_exception
        username = subject.removeprefix("username:")   # sub の名前空間を外す
        token_data = TokenData(username=username)
    except InvalidTokenError:
        # 署名不正・期限切れ・改ざん・alg不一致などはすべてここに落ちる
        raise credentials_exception
    user = get_user(username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)],
) -> User:
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="無効化されたユーザーです")
    return current_user

# 型エイリアスに畳んで再利用（公式推奨の Annotated パターン）
CurrentUser = Annotated[User, Depends(get_current_active_user)]

@app.get("/users/me", response_model=User)
async def read_users_me(current_user: CurrentUser) -> User:
    # 認証ロジックはハンドラに一切ない。current_user は"検証済み"で渡ってくる。
    return current_user

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    # scopes は {スコープ名: 説明} の dict。/docs にチェックボックスとして表示される。
    scopes={"me": "自分自身の情報を読む", "items": "アイテムを読む"},
)

from fastapi import Security
from fastapi.security import SecurityScopes
from pydantic import ValidationError

class TokenData(BaseModel):
    username: str | None = None
    scopes: list[str] = []        # トークンが持つスコープ

async def get_current_user(
    security_scopes: SecurityScopes,                 # ← 要求されたスコープ群を受け取る
    token: Annotated[str, Depends(oauth2_scheme)],
) -> User:
    # 失敗時の WWW-Authenticate ヘッダに、必要なスコープを載せる（OAuth2準拠）
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="資格情報を検証できませんでした",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        payload = jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM])
        subject = payload.get("sub")
        if subject is None:
            raise credentials_exception
        username = subject.removeprefix("username:")
        token_scopes = payload.get("scope", "").split()    # "scope" claim は空白区切り
        token_data = TokenData(scopes=token_scopes, username=username)
    except (InvalidTokenError, ValidationError):
        raise credentials_exception
    user = get_user(username=token_data.username)
    if user is None:
        raise credentials_exception
    # 要求された全スコープを、このトークンが持っているか検証する
    for scope in security_scopes.scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="権限が不足しています",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user

@app.post("/token")
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="ユーザー名またはパスワードが正しくありません")
    # ⚠️ 本番の肝：要求スコープを鵜呑みにせず「ユーザーが実際に許可されたスコープ」と突き合わせる
    granted = sorted(set(form_data.scopes) & set(user_allowed_scopes(user)))
    access_token = create_access_token(
        data={"sub": f"username:{user.username}", "scope": " ".join(granted)},
        expires_delta=timedelta(minutes=settings.access_token_expire_minutes),
    )
    return Token(access_token=access_token, token_type="bearer")

# 「自分の情報」: me スコープが要る
@app.get("/users/me", response_model=User)
async def read_users_me(
    current_user: Annotated[User, Security(get_current_active_user, scopes=["me"])],
) -> User:
    return current_user

# 「自分のアイテム」: items スコープが要る（依存ツリーで me も累積される）
@app.get("/users/me/items/")
async def read_own_items(
    current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])],
):
    return [{"item_id": "Foo", "owner": current_user.username}]

from fastapi import APIRouter

# /admin 配下の全エンドポイントに admin スコープを強制する（書き忘れが構造的に起きない）
admin_router = APIRouter(
    prefix="/admin",
    dependencies=[Security(get_current_active_user, scopes=["admin"])],
)

@admin_router.get("/reports")        # ← ここに認可を書かなくても admin が要求される
async def list_reports():
    ...

app.include_router(admin_router)

from fastapi import Form

@app.post("/token/refresh")
async def refresh(refresh_token: Annotated[str, Form()]) -> Token:
    # リフレッシュトークンはDBで検証する（=即時失効できる）。
    session = lookup_refresh_token(refresh_token)         # 無効/失効済みなら None
    if session is None:
        raise HTTPException(status_code=401, detail="再ログインが必要です")
    rotate_refresh_token(session)        # 使ったら作り直す（盗用検知＝ローテーション）
    new_access = create_access_token(data={"sub": f"username:{session.username}"})
    return Token(access_token=new_access, token_type="bearer")

# 発行（鍵を持つ認証サーバーだけ）
jwt.encode(payload, private_key, algorithm="RS256")
# 検証（各サービス。公開鍵だけでよい）。aud/iss も必ず検証する。
jwt.decode(token, public_key, algorithms=["RS256"], audience="my-api", issuer="https://issuer.example.com")

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],   # ← Cookie併用時は "*" 不可。明示する
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

from fastapi.testclient import TestClient
from app.main import app
from app.security import get_current_active_user

# 1) ハンドラ単体：認証を偽装して、ビジネスロジックに集中する
def fake_user() -> User:
    return User(username="tester", disabled=False)

app.dependency_overrides[get_current_active_user] = fake_user
client = TestClient(app)

def test_read_me_returns_injected_user():
    res = client.get("/users/me")
    assert res.status_code == 200
    assert res.json()["username"] == "tester"

app.dependency_overrides.clear()    # テスト間の汚染を防ぐため必ずクリア

# 2) 認証フロー自体：トークン無し → 401、正しいログイン → トークン取得 → 保護APIが通る
def test_auth_flow_end_to_end():
    real_client = TestClient(app)
    assert real_client.get("/users/me").status_code == 401          # トークン無しは弾かれる
    token = real_client.post(
        "/token", data={"username": "johndoe", "password": "secret", "scope": "me"}
    ).json()["access_token"]
    res = real_client.get("/users/me", headers={"Authorization": f"Bearer {token}"})
    assert res.status_code == 200