Flask JWT для пользовательского объекта SQLAlchemy?

Question

У меня есть приложение, в котором данные пользователя передаются в виде JWT, содержащего информацию о текущем пользователе и его ролях. Каждый раз, когда пользователь входит в систему (через экземпляр KeyCloak), информация из JWT анализируется на моем конце в функции, которая обновляет объект пользователя через SQLAlchemy. Однако, поскольку в бэкэнде нет объекта пользователя, передаваемого туда и обратно, я должен анализировать JWT на наличие ролей для каждого действия, которое этого требует. У меня также есть потребность в аудите, и из-за структуры приложения этот модуль не обязательно имеет доступ к объектам запроса во время ведения журнала.

Я ищу изящный способ сделать что-то вроде flask_user s current_user() работает путем сопоставления JWT -> объекта пользователя ORM, чтобы иметь возможность прозрачно получить текущего пользователя. Есть ли способ сообщить об этом go? Регистрация пользователя и т. Д. Полностью отделена от приложения, и Flask знает только, какой пользователь он основан на токенах в отправляемых запросах.

TLDR; Есть ли способ загрузить пользователя из БД на основе уже выпущенного JWT (который содержит информацию для сопоставления с пользователем), и, возможно, уже существует библиотека или расширение для flask, которое поддерживает это?

Answer 1

Поддерживается flask -jwt-extended: https://flask-jwt-extended.readthedocs.io/en/stable/complex_objects_from_token/

Answer 2

Я использую декоратор для анализа токена JWT, используя pyjwt.

Затем из проанализированного токена вы можете получить пользователя и выполнить надлежащую авторизацию.

Если вы этого не сделаете хотите добавить декоратор ко всем вашим функциям, требующим авторизации, вы можете использовать Flasks before_request.

from functools import wraps

from flask import Response, current_app, request
from jwt import decode
from jwt.exceptions import (DecodeError, ExpiredSignatureError,
                            InvalidSignatureError)

def authorize(func):

    @wraps(func)
    def check_authorization(*args, **kwargs):
        try:
            jwt_token = request.cookies.get('auth_token') # get token from request
            if jwt_token is None:
                return Response(status=401, headers={'WWW-Authenticate': 'Bearer'})
            
            token = decode(
                jwt_token,
                key='pub_key', # public key to validate key
                algorithms=['RS256'], # list of algs the key could be signed 
                verify=True
            )
            
            # you can call another function to do check user roles here
            # e.g authorize(token['sub'])
            
            return func(*args, **kwargs)
        except (InvalidSignatureError, DecodeError, ExpiredSignatureError):
            return Response(
                response='{ "error": "token_invalid"}',
                status=401,
                headers={'WWW-Authenticate': 'Bearer'})
    return check_authorization