LoginSignup
6
6

More than 1 year has passed since last update.

@sand(sand mypress)

FastAPI OAuth2 クライアント

Last updated at Posted at 2021-11-25

==>【続編を書きました 2021 12/15】FastAPIでつくるVue SPAアプリの雛形 - Qiita
FastAPI OAuth2パスワード認証 - Qiita

前回、QiitaにてFastAPIのパスワード認証について書きましたが、クライアント側はSwagger UIを利用したおかげで、パスワードフローの動きのほとんどが隠蔽されていました。プログラム的には以下の公式サイトの前者の方を掲載させていただきました。これはパスワードフローの枠組みをコーディングしているだけで、JWTの暗号的要素を無視したものでした。わかり易いが完全ではない。後者の公式サイトの方はJWTをサポートした完全なパスワードフローです。今回はこれを利用させていただきます。

【公式サイト】Simple OAuth2 with Password and Bearer
【公式サイト】OAuth2 with Password (and hashing), Bearer with JWT tokens

今回の注目点はクライアント側の動きです。「FastAPI OAuth2パスワード認証 - Qiita」で記したフロー図で示したようにJWT tokenを生成し、検証するのはひたすらサーバ側です。クライアント側はtokenを単なる文字列として受け取り、保管し、リクエスト時にヘッダーに付加して送信するだけです。クライアントにとってはJWT tokenは単なる文字列にすぎません。

今回は簡単なクライアントをVueで作成しました。パスワードフローのクライアント実装の参考になればと思います。

1, プログラムとディレクトリ

プログラムの詳細に入る前にディレクトリ構成について説明します。

FastAPIのプログラムファイル jwttoken.py の置いてある場所をルートディレクトリとして、その直下にstatic サブディレクトリを作成しindex.htmlとindex.jsを置きます。プログラム的にはFastAPIの静的ファイルを扱うStaticFilesを利用します。以下の2文をサーバに追加します。

jwttoken.py
---
from fastapi.staticfiles import StaticFiles
---
app.mount("/static", StaticFiles(directory="static"), name="static")

起動はルートディレクトリで以下のコマンドで行います。

uvicorn jwttoken:app --reload

ブラウザからのアクセスは以下の通りです。

http://localhost:8000/static/index.html

2. クライアント

プログラムの簡素化のためにVue.jsを使用します。ファイルはindex.htmlとindex.jsの2つです。

2-1. HTMLファイル - index.html

画面は１ページだけで、「ログイン」のLoginフォームと、「リソースにアクセス」するGet Resourceボタン２つから成ります。

  • ログインは、OAuth2パスワード認証の規則に従ってusername/passwordをForm bodyで送信しなければなりません。結果としてJWT tokenが返ってきます。
  • リソースへのアクセスは、Authorization headerが必要で、その値が「Bearer ＋ token」の形であることが求められます。

FastAPI OAuth2パスワード認証 - Qiita

index.html
<html>
    <head>
        <link rel="stylesheet" href="index.css">
        <script src="https://cdn.jsdelivr.net/npm/vue/dist/vue.js"></script>
        <script src="https://cdn.jsdelivr.net/npm/axios/dist/axios.min.js"></script>
    </head>
    <body>

        <div id="app">
            <h1>Login</h1>
            <p><input v-model="username"></p>
            <p><input v-model="password"></p>
            <button v-on:click="doLogin">Login</button>
            <p>Current token = {{ token }}</p>
            <hr>
            <h1>Get Resource</h1>
            <button v-on:click="getMe">Get Me</button>
            <button v-on:click="getItems">Get Items</button>
            <p>Response = {{ response }}</p>
        </div>

        <script src="index.js"></script>
    </body>
</html>

2-2. クライアント画面

少し早いですが、クライアント画面の動きをレビューしておきたいと思います。

Loginの「Current token」と、Get Resourceの「Response」の初期値が～文字列であることに注意してください。

image.png

ログイン画面で以下の入力を行います。

username=johndoe, passowrd=secret

Current tokenが取得でき表示されます。Responseはまだ空のままです。

image.png

「Grt Me」ボタンをクリックします。

image.png

ちゃんとResponseが表示されました。JWT tokenは正しく作られているようです。

2-3. Vueプログラム - index.js

とてもシンプルなものです。3つのメソッドから成ります。ログインのハンドラーdoLogin()、リソースアクセスのgetMe()とgetItems()の3つです。

index.js
var app = new Vue({
    el: '#app',
    data: {
      username: 'anonymouse',
      password: 'abc_123',
      token: '',
      response: ''
    },
    methods: {
      doLogin: function() {
          console.log("doLogin:: username=", this.username, " passowrd=", this.password)
          const params = new URLSearchParams();
          params.append('username', this.username);
          params.append('password', this.password);

          let config = {
            headers: {
              'Content-Type': 'application/x-www-form-urlencoded'
            }
          };

          axios.post('/token', params, config)
          .then((response) => {
            console.log("response.data = ", response.data)
            this.token = response.data.access_token
          })
          .catch((err) => {
            console.log("Error = ", err)
          })
        },
      getMe: function() {
        let config = {
          headers: {
            Authorization: `Bearer ${this.token}`
          }
        };
        axios.get('/users/me', config)
        .then((response) => {
          console.log("response.data = ", response.data)
          this.response = JSON.stringify(response.data)
        })
        .catch((err) => {
          console.log("Error = ", err)
          this.response = JSON.stringify(err)
        })
      },
      getItems: function() {
        let config = {
          headers: {
            Authorization: `Bearer ${this.token}`
          }
        };
        axios.get('/users/me/items', config)
        .then((response) => {
          console.log("response.data = ", response.data)
          this.response = JSON.stringify(response.data)
        })
        .catch((err) => {
          console.log("Error = ", err)
          this.response = JSON.stringify(err)
        })
      }
    }
  })

2-3-1. ログイン

username/passwordはform bodyとしてサーバに渡す必要があります。以下のコードで実現できます。

index.js
          const params = new URLSearchParams();
          params.append('username', this.username);
          params.append('password', this.password);

          let config = {
            headers: {
              'Content-Type': 'application/x-www-form-urlencoded'
            }
          };

          axios.post('/token', params, config)

【参考サイト】Axios post method requesting with x-www-form-urlencoded content type

2-3-2. リソース取得

Authorizationヘッダーにtokenを添付します。以下のコードで実現します。

index.js
        let config = {
          headers: {
            Authorization: `Bearer ${this.token}`
          }
        };
        axios.get('/users/me', config)

3. サーバ - jwttoken.py

JWT抜きのプログラムは以下の記事にありますので、まだの方は大変申し訳ありませんが、まずそちらを読んでください。
FastAPI OAuth2パスワード認証 - Qiita

これからはJWT tokenに絞って、以下の公式サイトのドキュメントをなぞることになります。
【公式サイト】OAuth2 with Password (and hashing), Bearer with JWT tokens

3-1. サーバプログラム - jwttoken.py

jwttoken.py
from datetime import datetime, timedelta
from typing import Optional

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

from fastapi.staticfiles import StaticFiles

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    }
}

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

class TokenData(BaseModel):
    username: Optional[str] = None

class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None


class UserInDB(User):
    hashed_password: str

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

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme)):
    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])
        print(f'### token = {token},  payload = {payload}')
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_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_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

@app.get("/users/me/items/")
async def read_own_items(current_user: User = Depends(get_current_active_user)):
    return [{"item_id": "Foo", "owner": current_user.username}]

3-1. PythonでJWT - python-jose

pythonでJWTを扱うためにpython-joseをインストールします。

pip install "python-jose[cryptography]"

python-joseの利用方法は以下の通り。

>>> from jose import jwt
>>> token = jwt.encode({'key': 'value'}, 'secret', algorithm='HS256')
u'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJrZXkiOiJ2YWx1ZSJ9.FG-8UppwHaFp1LgRYQQeS6EDQF7_6-bMFegNucHjmWg'

>>> jwt.decode(token, 'secret', algorithms=['HS256'])
{u'key': u'value'}

python-jose - GIT

secretは以下のコマンドで取得できます。

openssl rand -hex 32
566fad6d8df44eef2c4be2727b6693153bf61f6bae30b3e8e4fb5697e1953d0a

JWTに関係のある個所を抜粋すると以下のようになります。

jwttoken.py
from jose import JWTError, jwt

---

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

---

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

---

    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )

decodeの結果をプリントしているので確かめてみます。

jwttoke.py
async def get_current_user(token: str = Depends(oauth2_scheme)):
    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])
        print(f'### token = {token},  payload = {payload}')

コンソールには以下のプリント出力があります。正しいですね。

### token = eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJqb2huZG9lIiwiZXhwIjoxNjM3ODIwNjU5fQ.TU5EI6Y8bfshkT1h4hY7IW3z6_cx2mkF-yaqq3j2KhE,  
payload = {'sub': 'johndoe', 'exp': 1637820659}

3-2. Hash を使った passwords の検証 - passlib

password は平文では保存されず Hash（暗号）化されて保存されます。ここではpasslibを使ってHash化を行います。

pip install "passlib[bcrypt]"

passlibに関係しているところを抜粋しています。ここではDBに保存されているパスワード hashed_password は既にハッシュ化されていると仮定しているので、ログイン時に送られてきた平文パスワードとverify関数で比較しています。

jwttoken.py
from passlib.context import CryptContext
---
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
---
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

今回は以上です。

6
6
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
6
6