Flask-JWT-ExtendedでSPA(Vue.js+Flask)のAPI認可を実装する

こんにちは。CTO室の神沼(@t_kanuma)です。
本稿はPythonのWebフレームワーク”Flask“のJWT周りのライブラリである
Flask-JWT-Extended“を使ったAPI認可の実装に関する記事です。

題材

過去にVuetifyを試したときに作った簡単なメモアプリを流用しました。
そのバックエンドをFlaskで実装します。


アプリケーションの構成

バージョン

・ Python : 3.9.8
・ Flask : 2.0.2
・ Flask-JWT-Extended : 4.3.1

インストール

筆者、pipenvを使っています。

pipenv install flask-jwt-extended

初期化

SQLAlchemyやMarshmallowと同様です。

# main/__init__.py (app.pyです)
from flask_jwt_extended import JWTManager

jwt = JWTManager()
def create_app():
    app = Flask(__name__)
    app.config.from_object('config')
    # omit...
    jwt.init_app(app)
    return app

コンフィグ

このアプリをデプロイして公開する気は全くないのですが、XSSとCSRF対策のためJWTをCookieに保管します。(保管場所のセキュリティについては、この記事がとてもためになりました。)

# config.py

# 署名鍵(対称鍵)
# デフォルトのアルゴリズムはHS256
JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY')

# デフォルトは15分
JWT_ACCESS_TOKEN_EXPIRES = datetime.timedelta(minutes=10)

JWT_TOKEN_LOCATION = ['cookies']

# CSRF対策 - same siteにのみCookieを送信する。
JWT_COOKIE_SAMESITE = 'Strict'

# XSS対策 - デフォルトでhttponly=Trueのため、設定不要

# httpsでのみCookieを送信する。ローカル環境のためFalseの設定。
JWT_COOKIE_SECURE = os.getenv('JWT_COOKIE_SECURE')

ログイン処理

JWT生成

# auth/controller.py
from flask_jwt_extended import (create_access_token, set_access_cookies,
                                unset_access_cookies, verify_jwt_in_request)
from main import jwt   # 上で宣言済み

@auth.route('/login', methods=['POST'])
def authenticate() -> Tuple[Response, int]:
    app.logger.info('login started')

    req_data: Any = request.get_json()

    user_name_to_auth: Optional[str] = req_data['user_name']
    password_to_auth: Optional[str] = req_data['password']

    user = User.query.filter(User.user_name == user_name_to_auth)\
        .filter(User.password == password_to_auth).one_or_none()

    if user is not None:
        app.logger.info('login succeeded')
        # アクセストークンの生成
        access_token = create_access_token(
            identity=user_schema.dump(user))
        response = make_response()
        # クッキーにアクセストークンを配置する。
        set_access_cookies(response, access_token)
        return response, 200
    else:
        app.logger.info('login failed')
        return jsonify({'msg': 'Wrong login id or password.'}), 401

コールバック登録

このライブラリではアクセストークン作成時やリソースアクセス時に起動するデコレーター群を用意しています。(詳細はこちらをご確認ください。)

# auth/controller.py

# JWTのsubject claimを決める。
# returnの内容がjwtの"sub"に入る。
# create_access_token(..)実行時に起動する。
@jwt.user_identity_loader
def identity_user(user):
    return user['id']

# claimを追加する。
# create_access_token(..)実行時に起動する。
@jwt.additional_claims_loader
def add_claims_to_access_token(identity):
    return {
        'department': 'CTO Office',
        'manager_role_attached': False
    }

# @jwt_requiredが付与されている関数の実行前に起動する。
# (つまりリソースにアクセスした時)
# @jwt_requiredを付与した関数の中で、current_user変数でここでreturnする値を参照できる。
@jwt.user_lookup_loader
def lookup_user(_jwt_header, jwt_data):
    identity = jwt_data["sub"]
    return User.query.filter_by(id=identity).one_or_none()

その他

CORS対応

Flask-CORSをインストールして使います。

# python側
# main/__init__.py (app.pyです)
from flask_cors import CORS

def create_app():
    app = Flask(__name__)

    # omit...

    # origins -> レスポンスヘッダにAccess-Control-Allow-Originを付与する。
    # supports_credentials -> レスポンスヘッダにAccess-Control-Allow-Credentialsを付与する。 フロントエンドがCookie(アクセストークン)をAPIに送信できるようになる。
    CORS(app, resources={
     '/*': {'origins': os.getenv('FRONTEND_ORIGIN'), 'supports_credentials': True}})

    # omit...
// Vue側のログイン処理
// views/Login.vue

// omit...
methods: {
    async login() {
      if (this.$refs.form.validate()) {
        try {
          const response = await fetch(
            `${process.env.VUE_APP_API_BASE_URL}/auth/login`,
            {
              method: "POST",
             // Cookie(アクセストークン)をAPIに送るため
              credentials: "include",
              headers: {
                "Content-Type": "application/json",
              },
              body: JSON.stringify({
                user_name: this.loginName,
                password: this.password,
              }),
            }
          );

          const status = response.status;
          if (status === 200) {
            this.$router.push("/");
          } else {
            throw new Error(`login resulted in ${status}`);
          }
        } catch (error) {
          console.log(error);
        }
      }
    },
    // omit...

ログイン処理の検証

ログイン処理成功後、APIは以下の2つのCookie項目を返します。

access_token_cookie=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJmcmVzaCI6ZmFsc2UsImlhdCI6MTYzODg1MDU0NiwianRpIjoiYmZjZDdkZGItNGI0NC00MzI4LWJlZDAtZDdjN2I3ZGRmN2U1IiwidHlwZSI6ImFjY2VzcyIsInN1YiI6MSwibmJmIjoxNjM4ODUwNTQ2LCJjc3JmIjoiNmE2YWVkZDktNjgxMi00NzU0LTljOTQtNTYxMDBkMWI4YTFmIiwiZXhwIjoxNjM4ODUxMTQ2LCJkZXBhcnRtZW50IjoiQ1RPIE9mZmljZSIsIm1hbmFnZXJfcm9sZV9hdHRhY2hlZCI6ZmFsc2V9.awA_wTX9mLnOBVHvVGCbVDmP0xBgOiKYVyUtwZVVrkQ; Secure; HttpOnly; Path=/; SameSite=Strict

csrf_access_token=6a6aedd9-6812-4754-9c94-56100d1b8a1f; Secure; Path=/; SameSite=Strict

JWT形式のアクセストークン

1つ目はJWT形式のアクセストークンです。以下、このサイトでデコードした結果です。  
(ヘッダーとペイロード)

{
  "typ": "JWT",
  "alg": "HS256"
}
{
  "fresh": false,
  "iat": 1638850546,
  "jti": "bfcd7ddb-4b44-4328-bed0-d7c7b7ddf7e5",
  "type": "access",
  "sub": 1,
  "nbf": 1638850546,
  "csrf": "6a6aedd9-6812-4754-9c94-56100d1b8a1f",
  "exp": 1638851146,
  "department": "CTO Office",
  "manager_role_attached": false
}

subにユーザIDが入っていること、追加したclaimが下部に入っていることが確認できます。

CSRFトークン

このライブラリはデフォルトでCSRFトークンを付与してくれます。
またJWTの方にも同じCSRFトークンが入っていることが確認できます。

”CSRFトークンをCookieで返して大丈夫なの?Form送信で渡ってしまわない?”と思われた方もいるかも知れません。実はこれを次のリクエストでヘッダーの”X-CSRF-TOKEN”に設定して送信する必要があります。このライブラリは、リクエストヘッダーとJWTの中のCSRFトークンが同じであることチェックします。ここがミソで、悪意のあるサイトがJavaScriptでこのヘッダーにトークンを設定しようと思っても、違うドメインを持つため、そもそもこのクッキーを読み取ることができないわけです。これによりそのリクエストがクライアントにとって意図したものであると確証が取れます。 (JWTはXSS対策でHttpOnlyがついていますが、CSRFトークンにはついていないため、このメモアプリサイトのJavaScriptから取得可能です。)(また余談ですが、JWTにCSRFトークンを入れ込むことでAPIのステートレス性を維持できており、まさにRESTfulの一端を担っていると言えると思います。)

今回のケースで言えば、Same Siteからのアクセスしかない上でSameSite=Strictを設定しているので不要ではありますが、Same Site以外のクライアントを持つ場合、このCSRFトークンが必要になると考えます。

リソースアクセス

認可処理

jwt_requiredアノテーションを付与することで、それを付与した関数の実行前に、 JWTが有効あるかを検証してくれます。

またcurrent_userで上記のコールバック(jwt.user_lookup_loader)で返却する値を取得できます。以下ではJWTの中のユーザIDを元にDBアクセスをしています。

コメントアウトしていますが(e.g. authorizationの部分)、Claimを取得してその内容を元に権限チェックができます。

# Python側
# memo/controller.py
from flask_jwt_extended import current_user, jwt_required

# omit...

@memo.route('', methods=['GET'])
@jwt_required()
def get_memos() -> Tuple[Response, int]:
    app.logger.info('get_memos started')

    # e.g. authorization
    # claims = get_jwt()
    # if claims['manager_role_attached'] is False:
    #     return jsonify({'msg': 'access to this resource is not allowed'}), 403

    memos: List[Memo] = Memo.query.filter(Memo.user_id == current_user.id).\
        filter(Memo.archived == false()).order_by(Memo.id.asc()).all()
    return jsonify(memo_schema.dump(memos)), 200
// Vue側
// views/MemoList.vue

// omit...
  methods: {
    async showMemos() {
      try {
        const response = await fetch(
          `${process.env.VUE_APP_API_BASE_URL}/memos`,
          {
            cache: "no-store",
            method: "GET",
            credentials: "include",
            headers: {
              "X-CSRF-TOKEN": this.$cookies.get("csrf_access_token"),
            },
          }
        );
        // TODO 401ならログイン画面に遷移
        this.memos = await response.json();
        // omit...
      } catch (error) {
        // TODO エラー画面に遷移
        console.log(error);
      }
    },

カスタムデコレーター

各関数に同じ認可処理を実装するためにはカスタムデコレータが有用です。
自作のアノテーションを付与することで認可処理をAOPできます。
詳細はここをご確認ください。

ログアウト(JWTの破棄)

ログアウトしたのにCookieにまだ有効なJWTを残すのはセキュリティ上よくないと考えます。
ここでXSS対策によりJavaScriptからCookie(JWT)を読み取ることはできません。
フロントエンドから要求して、API側で破棄します。

# auth/controller.py
from flask_jwt_extended import (create_access_token, set_access_cookies,
                                unset_access_cookies, verify_jwt_in_request)

# omit...

@auth.route('/logout', methods=['POST'])
def logout():
    app.logger.info('logout started.')
    response: Response = make_response()
    # cookieから取り除く。
    unset_access_cookies(response)
    return response, 200

画面遷移などのイベントでの有効性チェック

JWTが無効(有効期限切れ、未発行など)な状態での画面遷移、URL直打ちについてもログイン画面に遷移させています。(ここではそうではありませんが、良いUXのために有効期限切れの場合はリフレッシュトークンでアクセストークンをリフレッシュするのが通例だと思います。)

以下、Vueで画面遷移またはURL直打ちの際に、遷移先の処理を開始する前にインターセプトしてAPI側で有効性をチェックしています。無効の場合、ログイン画面にリダイレクトします。

// router/index.js
import auth from "@/modules/auth.js";

// omit...
router.beforeEach((to, from, next) => {
  if (to.matched.some((record) => record.meta.requiresAuth)) {
    (async () => {
      const result = await auth.isloggedIn();
      if (!result) {
        next({
          path: "/login",
          query: { redirect: to.fullPath },
        });
      } else {
        next();
      }
    })().catch((error) => {
      console.log(error);
    });
  } else {
    next();
  }
});

// modules/auth.js
export default {
  _getCookie: (name) => {
    const value = `; ${document.cookie}`;
    const parts = value.split(`; ${name}=`);
    if (parts.length === 2) return parts.pop().split(";").shift();
  },
  async isloggedIn() {
    try {
      const response = await fetch(
        `${process.env.VUE_APP_API_BASE_URL}/auth/status`,
        {
          cache: "no-store",
          method: "GET",
          credentials: "include",
          headers: {
            "X-CSRF-TOKEN": this._getCookie("csrf_access_token"),
          },
        }
      );
      return response.status === 200;
    } catch (error) {
      console.log(error);
    }
  },
// omit...
# auth/controller.py
from flask_jwt_extended import (create_access_token, set_access_cookies,
                                unset_access_cookies, verify_jwt_in_request)

# omit...

@auth.route('/status', methods=['GET'])
def isloggedIn():
    app.logger.info('isloggedIn started.')
    # アノテーションでなく、以下のメソッド呼び出しで検証する。
    verify_jwt_in_request()
    return make_response(), 200

公開鍵ペアによる署名と検証

OAuthのような認可サーバとリソースサーバが別れている場合は、対称鍵ではなく公開鍵ペアでの署名と検証をすると思います。そのような場合は、コンフィグのJWT_PRIVATE_KEY
とJWT_PUBLIC_KEY項目で鍵を設定できます。

リフレッシュトークン

本稿ではリフレッシュトークンについてはスコープ外としました。

以上でこの記事を終わります。
どなたかのお役に立てたならば幸いです。
最後までお読みいただき、ありがとうございました。