Flask升级后roles_required装饰器失效问题求助

问题背景 #

将基于Flask 1.0.2、Flask-Login 0.4.1的项目升级至最新版本后，登录功能正常，但@roles_required('Read')装饰器完全失效：调用带该装饰器的接口时返回“Unauthorized Access”，移除装饰器则接口可正常访问。旧版本中load_user方法会被触发，新版本却无法进入该方法。

相关代码片段如下：

models.py #

...
from flask_user import UserMixin

class ApiUserRoles(db.Model):
    __tablename__ = 'apiuser_roles'
    user_id = db.Column("user_id", db.Text, db.ForeignKey("public.apiuser.id", use_alter=True), nullable=False,
                        primary_key=True)
    role_id = db.Column("role_id", db.Integer, db.ForeignKey("public.apiroles.id", use_alter=True), nullable=False,
                        primary_key=True)
    description = db.Column("description", db.String(50))


class ApiUser(db.Model, UserMixin):
    __tablename__ = 'apiuser'
    id = db.Column("id", db.Text, nullable=False, primary_key=True)
    roles = db.relationship('ApiRoles', secondary='public.apiuser_roles')
    password_hash = db.Column("password_hash", db.Text, nullable=False)
    description = db.Column("description", db.String(50))

auth.py #

from flask import session
from flask_httpauth import HTTPBasicAuth
from flask_user import UserManager


class ApiUserManager(UserManager):
    auth = None

    def __init__(self, app, db, User, auth):
        super(ApiUserManager, self).__init__(app, db, User)
        self.auth = auth

    def unauthenticated_view(self):
        return self.auth.auth_error_callback(401)

    def unauthorized_view(self):
        return self.auth.auth_error_callback(403)


class ApiAuth(HTTPBasicAuth):
    def authenticate(self, auth, stored_password):
        result = super(ApiAuth, self).authenticate(auth, stored_password)
        if result and auth:
            session['user_id'] = auth.get('username')
        return result


auth = ApiAuth()

init.py #

from auth import ApiUserManager, auth as auth_object
from models import ApiUser

def init(app):
    app.app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('database_url')
    db.init_app(app.app)
    db.app = app.app

    # Flask-User settings
    app.app.config['USER_APP_NAME'] = 'api'
    app.app.config['USER_ENABLE_EMAIL'] = False
    ...
    
    user_manager = ApiUserManager(app.app, db, ApiUser, auth_object)
    
    # 以下代码不再生效
    login_manager = user_manager.login_manager

    @login_manager.user_loader
    def load_user(user_id):
        try:
            return ApiUser.query.filter(ApiUser.id == user_id).first()
        except:
            return None
...

def create_app(config_name):
    app = connexion.FlaskApp(__name__)
    with app.app.app_context():
        init(app)
    return app.app

接口示例 #

@auth.login_required
@roles_required('Read')
def getFoo():
...

问题原因 #

1. Flask-User与Flask-Login协同逻辑变更：新版本Flask-User不再依赖手动设置的user_loader跟踪用户，仅在session存入user_id无法让Flask-User识别已登录状态，导致roles_required无法获取用户角色信息。
2. 角色验证方法缺失：roles_required依赖用户模型的has_role()方法判断权限，当前ApiUser未显式实现该方法，新版本Flask-User不再自动推导角色判断逻辑。

解决方案 #

1. 修改认证逻辑，让Flask-User识别登录状态 #

在ApiAuth的authenticate方法中，认证成功后调用Flask-User的login_user方法，替代仅存入session的操作：

修改auth.py中的ApiAuth类：

class ApiAuth(HTTPBasicAuth):
    def authenticate(self, auth, stored_password):
        result = super(ApiAuth, self).authenticate(auth, stored_password)
        if result and auth:
            user = ApiUser.query.filter_by(id=auth.get('username')).first()
            if user:
                from flask_user import login_user
                login_user(user)
        return result

2. 移除无效的user_loader代码 #

新版本Flask-User会自动处理用户加载，删除init.py中以下代码：

# 以下代码不再生效
login_manager = user_manager.login_manager

@login_manager.user_loader
def load_user(user_id):
    try:
        return ApiUser.query.filter(ApiUser.id == user_id).first()
    except:
        return None

3. 为用户模型添加角色判断方法 #

在models.py的ApiUser类中新增has_role方法：

class ApiUser(db.Model, UserMixin):
    __tablename__ = 'apiuser'
    id = db.Column("id", db.Text, nullable=False, primary_key=True)
    roles = db.relationship('ApiRoles', secondary='public.apiuser_roles')
    password_hash = db.Column("password_hash", db.Text, nullable=False)
    description = db.Column("description", db.String(50))

    # 新增角色判断方法
    def has_role(self, role_name):
        # 假设ApiRoles模型有name字段存储角色名称
        return any(role.name == role_name for role in self.roles)

4. 确认Flask-User配置完整性 #

检查init.py中的Flask-User配置，确保角色相关设置正确：

# Flask-User settings
app.app.config['USER_APP_NAME'] = 'api'
app.app.config['USER_ENABLE_EMAIL'] = False
# 绑定角色模型（如果ApiRoles是自定义角色模型）
app.app.config['USER_ROLE_MODEL'] = ApiRoles

验证 #

完成修改后重启服务，调用带@roles_required('Read')装饰器的接口，此时装饰器应能正确识别用户角色并完成权限验证，用户加载逻辑由Flask-User自动处理。

内容的提问来源于stack exchange，提问作者saromba