Bearer スキームでトークン認証する API を FastAPI で実装した

勢いで書いたはいいものの特に使うこともないコードが発生してしまったので Qiita に書いて供養する。

FastAPI のお試しと Bearer スキームの仕様理解のための練習という感じ。

大いに参考にした記事： トークンを利用した認証・認可 API を実装するとき Authorization: Bearer ヘッダを使っていいのか調べた - Qiita

作ったもの

• パスワード認証
• トークンを用いた認証・認可
  • Bearer スキーム = Authorization: Bearer ヘッダを見る
• バックエンド API のみでフロントエンドは作っていない
  • 認証通ったら LocalStorage に token 入れておいて API 叩くときはヘッダ付けてくれって感じ
  • @nuxt/auth とかでシュッとできそう
• JWT ではなく古典的なセッションストレージ (メモリ) なのでステートフル

コード

main.py

from fastapi import FastAPI
from routers import auth

app = FastAPI()

app.include_router(auth.router, prefix="/auth")

routers/auth.py

from os import getenv
from hashlib import pbkdf2_hmac
from secrets import token_hex
from fastapi import APIRouter, Header, Depends, HTTPException
from pydantic import BaseModel

router = APIRouter()

users = {
    "foo": "d7277d65c13dbd72e4b2d3ce66a56dd70fb5a0fea0bec058dc8e313fc743c12c",
    "bar": "466b21c4b915616bf9d69696bc1da7c0f819daa7371dadbecfd4d542d7e8b8a2",
}

session = {}


class Credentials(BaseModel):
    username: str
    password: str


def hash_password(password: str) -> str:
    return pbkdf2_hmac("sha256", password.encode(), getenv("SALT").encode(), 100000).hex()


def get_token_header(authorization: str = Header(None)):
    if authorization is None:
        raise HTTPException(401, headers={"WWW-Authenticate": 'Bearer realm=""'})
    if authorization[:7] != "Bearer ":
        raise HTTPException(400)
    return authorization[7:]


def verify_token(token: str = Depends(get_token_header)):
    if token not in session:
        raise HTTPException(401, headers={"WWW-Authenticate": 'Bearer error="invalid_token"'})
    return session[token]


@router.post("/signup")
def post_auth_signup(c: Credentials):
    if c.username in users:
        raise HTTPException(409)
    # TODO check password policy
    users[c.username] = hash_password(c.password)
    token = token_hex()
    session[token] = c.username
    return {"token": token}


@router.post("/signin")
def post_auth_signin(c: Credentials):
    if c.username not in users:
        raise HTTPException(404)
    if users[c.username] != hash_password(c.password):
        raise HTTPException(401, headers={"WWW-Authenticate": 'Bearer error="invalid_request"'})
    token = token_hex()
    session[token] = c.username
    return {"token": token}


@router.post("/signout")
def post_auth_signout(token: str = Depends(get_token_header)):
    if token in session:
        del session[token]


@app.get("/user")
def get_index(username: str = Depends(verify_token)):
    return {"username": username}

作りが雑すぎるので実用的ではないけど、何かやるときの参考にはなりそう。