[Python]Fast APIでTodoリストに認証機能をつけてみた #Python

概要

Zennに投稿した下記記事で作ったTodoリストのアプリに認証機能をつけてみました(具体的には新規登録・ログイン)
[Python]Fast APIでTodoリストをつくってみた

※上記記事ではフロントエンドも作成しましたが、本対応に追従していません😢

方法

FastAPI公式チュートリアルのSecurityという章を参考にしました。
※日本語訳で参照すると、Securityの章で参照できるページが5→2に減ってしまうので、英語でご覧いただく方がオススメです。

OAuth2PasswordBearer

from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/items/")
async def read_items(token: str = Depends(oauth2_scheme)):
    return {"token": token}

OAuth2PasswordBearerというクラスを使って、oauth2_schemeを生成し、Depends()に渡します。
こうすることで、例えば上記コードの場合は、/itemsにGETでアクセスする際にoauth2_schem()が実行され、認証情報が要求されるようになります。
具体的には、リクエストヘッダーのAuthorizationヘッダーにBearer+文字列（トークン）が存在するかどうかをチェックし、トークンを返します。
また、Authorizationヘッダーがなかったり、Bearer文字列がない場合は、401 Unauthorized Errorレスポンスを返します。

さらに、自動で生成されるOpenAPIドキュメントには、「Authorize」というボタンが表示されます。
認証情報が入力できるようになり、oauth2_schemaをDependsに持つ、認証で保護されているリソースにアクセスできるようになります。

リクエストにおける認証情報の要求とトークンの取得、ドキュメントでの認証機能を追加してくれるというわけですね。

ちなみに、今回作ったTodoリストアプリでは、下記のように、@router.get("/tasks")リソースの依存関数get_current_active_userの依存関数→get_current_userの中で、依存関数としてoauth2_schemaを渡しています。
(取得したtokenを元にDBに問い合わせて、有効なユーザー情報を取得しています)

/src/routers/task.py

@router.get("", response_model=List[task_schema.Task])
async def list_tasks(
    current_user: user_schema.User = Depends(authenticate.get_current_active_user),
    db: AsyncSession = Depends(get_db)
):

/src/libs/authenticate.py


async def get_current_user(
    token: str = Depends(oauth2_schema),
    db: AsyncSession = Depends(get_db)
):
...

async def get_current_active_user(current_user: user_schema.User = Depends(get_current_user)):
...

OAuth2PasswordRequestForm

参考: Code to get the username and password

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):

次に、認証情報を受け付けてログインできるようにするためにOAuth2PasswordRequestFormをパスオペレーション関数(ここではasync def login)の引数の型ヒントに設定します。

こうすることで、リクエストボディにusername passwordが要求されるようになります。
また、オプションとしてscopeとgrant_typeも受け付けることができるようになっています(OAuth2ではgrant_typeが必須らしい)

今回は、下記のように、ログイン時に認証(authenticate.authenticate_user)して入力されたusernameとpassword(DBではハッシュ値)に一致するユーザーを取得、usernameを元にトークンを生成し、レスポンスボディとして返しています。

src/routers/user.py

@router.post("/token", response_model=user_schema.Token)
async def login_for_access_token(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: AsyncSession = Depends(get_db)
):
    user = await authenticate.authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"}
        )
    access_token = authenticate.create_access_token(user.username)
    return {"access_token": access_token, "token_type": "bearer"}

src/libs/authenticate.py

async def authenticate_user(db: AsyncSession, username: str, password: str):
    user = await user_crud.get_user(db, username)
    if not user:
        return False
    if not verify_password(password, user.password):
        return False
    return user

def create_access_token(username: str):
    expire = datetime.now() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {
        "sub": username,
        "exp": expire
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

トークン

トークンはJWT(ジョット)トークンを使用しています。
JWTトークンに関しては、【JWT】入門が詳しいです。

JWTとは
...
JSON Web Tokenの略　　
電子署名により、改ざん検知できる。
認証用のトークンなどで用いられる。

構成
ヘッダ、ペイロード、署名の3つから成る。
それぞれは、Base64でエンコードされている
それぞれは、 . (ドット) で結合されている。

ペイロードには任意の情報を含めることができます。
今回の場合はusernameを含めており、トークンをデコードしてusernameを取得、DBに問い合わせてユーザー情報を取得しています。

authenticate.py

async def get_current_user(
    token: str = Depends(oauth2_schema),
    db: AsyncSession = Depends(get_db)
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        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
        token_data = user_schema.TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = await user_crud.get_user(db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

その他

パスワードのハッシュ化には、Passlibというライブラリを使用しています(アルゴリズムはbcrypt)。
参考: Install passlib　Passlib
JWTの生成・デコードには、python-joseというライブラリを使用しています。Cryptographic Backendsというものを指定する必要があるらしく、今回はFastAPIチュートリアルに従ってcryptographyを選択しています。
参考: Install python-jose　python-jose
私の理解が曖昧なのですっ飛ばしましたが、今回の認証の流れはOAuth2の「パスワード」フローを採用しているようで、概要がチュートリアルの最初で説明されています
参考: The password flow
(OAuth 2.0 全フローの図解と動画だとどれに該当するんだろう...？)
今回はusernameを認証情報としていますが、代わりにemailを必須にしたかったらどうすればいいんだろう（それってOAuth2のフローに準拠していないことになるのかな、どうなんだろう...?）