目次

5日目 - ユーザの認証の追加

公式のリファレンスに則ってJWT認証を設定する。

パスワードのハッシュ化

ライブラリのインストール

bash

poetry add passlib ["bcrypt"]# パスワードのハッシュするため

認証用のファイルを作成

api/v1/cruds/auth.py

from passlib.context import CryptContext
from sqlalchemy.orm import Session
from api.v1.models.user import User

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# パスワードのハッシュ化
def get_password_hash(password):
  return pwd_context.hash(password)

# パスワードの検証
def verify_password(plain_password,hashed_password):
  return pwd_context.verify(plain_password,hash_password)

# ユーザーの認証
def authenticate_user(db:Session,login_id:str,password:str):
  user = db.query(User).filter_by(login_id=login_id).first()
  if user is None:
    return None
  if not verify_password(password,user.password_hash):
    return None
  return user

ユーザーの作成時の処理の変更

api/v1/cruds/user.py

# ~~~
from api.v1.cruds.auth import get_password_hash
# ~~~
def create_user(db:Session,user:user_schema.UserCreateRequest):
  password_hash = get_password_hash(user.password)
  new_user = User(
    login_id      = user.login_id,
    name          = user.name,
    description   = user.description,
    password_hash = password_hash
  )
  db.add(new_user)
  db.commit()
  db.refresh(new_user)
  return new_user

これでパスワードのハッシュ化とハッシュ化したパスワードの認証を行うためのメソッドが用意できた。

JWTの認証の追加

JWT認証のためのSECRETキーなどの保存

以下よりsecretキーを生成

bash

openssl rand -hex 32

生成したキーを環境変数として読み込むためにdocker-compose.ymlと同じディレクトリに.envを生成
(環境変数の読み込み方を前回から変更)

.env

MYSQL_ROOT_PASSWORD=...
SECRET_KEY=...

環境変数をdocker-compose.yml内に定義

docker-compose.yaml

# ~~~
  backend:
  # ~~~
    environment:
      - MYSQL_ROOT_PASSWORD=$MYSQL_ROOT_PASSWORD
      - SECRET_KEY=$SECRET_KEY

これで環境変数は.envファイル書き換えるだけでオッケー

ライブラリのインストール

bash

poetry add python-jose # JWTの認証用
poetry add python-multipart # multipart/form-dataのリクエストを受け付けられるように

CRUD処理の追加

access_tokenの作成

api/v1/cruds/auth.py

from datetime import datetime, timedelta
from jose import jwt
import os
# ~~~
SECRET_KEY = os.environ.get("SECRET_KEY")
ALGORITHM = "HS256"
ACCESS_TOEKN_EXPIRE_MINUTES = 30
# ~~~
def create_access_token(data:dict):
  to_encode = data.copy()
  expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  to_encode.update({"exp":expire})
  encode_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  return encode_jwt

tokenからのユーザー情報の取得

api/v1/cruds/auth.py

from fastapi import OAuth2PasswordBearer
from jose import JWTError
# ~~~
def get_current_user(db:Session,token:str):
  try:
    payload = jwt.decode(token,SECRET_KEY,algorithm=ALGORITHM)
    login_id:str = payload.get("sub")
    if login_id is None:
      return None
  except JWTError:
    return None
  current_user = db.query(User).filter_by(login_id=login_id).first()
  if current_user is None:
    return None
  return current_user

schemaファイルの作成

api/v1/schemas/auth.py

from pydantic import BaseModel

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

auth用のルーター作成

api/v1/routers/auth.py

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from fastapi.security import OAuth2PasswordRequestForm

import api.v1.schemas.auth as auth_schema
import api.v1.schemas.user as user_schema
import api.v1.cruds.auth as auth_crud

from api.v1.db import get_db

router = APIRouter()

@router.post('/token', response_model=auth_schema.Token)
def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(),db:Session = Depends(get_db)):
  user = auth_crud.authenticate_user(
    db=db,
    login_id=form_data.username,
    password=form_data.password
  )
  if user is None:
    raise HTTPException(
      status_code=status.HTTP_401_UNAUTHORIZED,
      detail="Incorrect login_id or password",
      headers={"WWW-Authenticate":"Bearer"}
    )
  access_token = auth_crud.get_current_user(db=db,token=token)
  return {"access_token":access_token,"token_type": "bearer"}

auth用のルーター反映

api/v1/main.py

# ~~~
from api.v1.routers import user,post,auth
# ~~~
app.include_router(auth.router)

これで認証用のエンドポイントができた。

ユーザー認証が必要なエンドポイントに反映

api/v1/router/post.py

# ~~~
import api.v1.cruds.auth as auth_crud
# ~~~
@router.post('/posts',response_model=post_schema.Post)
def create_post(post:post_schema.PostBase,token: str = Depends(auth_crud.oauth2_scheme),db:Session = Depends(get_db)):
  current_user = auth_crud.get_current_user(db=db,token=token)
  post = post_crud.create_post(db=db,post=post,user_id=current_user.id)
  return post
# ~~~
@router.post('/post/{id}/lgtm')
def add_lgtm(id:int,token: str = Depends(auth_crud.oauth2_scheme),db:Session = Depends(get_db)):
  current_user = auth_crud.get_current_user(db=db,token=token)
  lgtm_crud.add_lgtm(db=db,post_id=id,user_id=current_user.id)
  return None
# ~~~
@router.delete('/post/{id}/lgtm')
def delete_lgtm(id:int,token: str = Depends(auth_crud.oauth2_scheme),db:Session = Depends(get_db)):
  current_user = auth_crud.get_current_user(db=db,token=token)
  lgtm = lgtm_crud.get_lgtm(db=db,post_id=id,user_id=current_user.id)
  if lgtm is None:
    raise HTTPException(status_code = 400, detail = "Lgtm not Found")
  lgtm_crud.remove_lgtm(db=db,lgtm=lgtm)
  return None

確認

/docsにアクセスすると認証用のボタンが右上に設置される

このボタンで認証を行なうと、認証が必要なエンドポイントのお試しができる。
とりあえずこれで認証が必要なルーティングが完成。