Flask RESTful API -特定用户或admin用户的授权

Question

我正在使用Flask Restful开发一个Flask API。我想知道是否有一种干净的方式来授权用户，而不会强迫我进行代码复制。我在API中使用Flask-JWT-Extended进行身份验证。

我有一些端点，我希望只有具有admin角色的用户或与给定资源相关的用户才能访问。

因此，假设我想允许用户获取有关其帐户的信息，并且我想阻止其他用户访问此信息。现在，我用这种方法来解决它:

from flask import request, Response
from flask_restful import Resource, reqparse
from flask_jwt_extended import (create_access_token, create_refresh_token, jwt_required, get_jwt_identity)
from app.mod_auth.models import User
[...]
class UserApi(Resource):
@jwt_required()
def get(self, name):
current_user = User.find_by_login(get_jwt_identity())
if current_user.login==name or current_user.role==1:
user = User.query.filter_by(login=name).first_or_404(description="User not found")
return user.json()
else:
return {'message': 'You are not authorized to access this data.'}, 403 
[...]

因此，首先检查请求中是否有正确有效的JWT令牌，然后根据令牌检查与令牌相关的用户是否与返回其数据的用户相同。访问数据的另一种方式是具有角色1的用户，我将其视为管理角色。

这是我的用户模型的一部分:

[...]
class User(Base):
login = db.Column(db.String(128), nullable=False)
email = db.Column(db.String(128), unique=True, nullable=False)
password = db.Column(db.String(192), nullable=False)
active = db.Column(db.Boolean(), default=True, nullable=False)
role = db.Column(db.SmallInteger, default=0, nullable=False)
[...]

当然，很快我就会有一些特定于用户的数据端点。

我在Flask-JWT-Extended中找到了一个自定义操作符的例子，它为管理员用户提供了授权:Flask-JWT-Extended admin authz -但另一方面，它不支持特定于用户的授权。我不知道如何改进该代码片段以验证，请求资源的用户是否是对资源具有权限的特定用户。

我如何定义一个自定义操作符，它将提供对用户特定数据的正确访问?

也许我应该在每个DB模型中包含某种owner

数据，应该支持授权，并在请求中验证，如上面的示例?Thanks in advance

Answer 1

你可以创建一个检查用户身份的装饰器:

def validate_user(role_authorized:list() = [1]):
def decorator(fn):
@wraps(fn)
def wrapper(*args, **kwargs):
name = request.path.rsplit('/', 1)[-1]
current_user = User.find_by_login(get_jwt_identity())
if (current_user.login == name) or (current_user.role in role_authorized):
kwargs["logged_user"] = current_user # If you need to use the user object in the future you can use this by passing it through the kwargs params
return fn(*args, **kwargs)
else:
return {'message': 'You are not authorized to access this data.'}, 403                            
return wrapper
return decorator

class Test(Resource):
@jwt_required()
#list of the roles authorized for this endpoint = [1]
@validate_user([1])
def post(self, name, **kwargs):
#logged_user = kwargs["logged_user"] # Logged in User object 
user = User.query.filter_by(login=name).first_or_404(description="User not found")
return user.json()