如何基于Flask-HTTPAuth实现无状态REST API的端点级权限控制？

我来帮你解决这个无状态API的权限控制问题，核心思路是在基础认证通过后，利用请求上下文保存用户信息，再通过自定义装饰器做细粒度的权限校验，完全适配无状态场景。

步骤1：优化认证函数，保存用户到请求上下文 #

你的verify_password目前只验证账号密码是否正确，我们可以在验证通过后，把用户对象存入Flask的g对象（请求上下文专属，每个请求独立存在），这样后续的权限检查就能直接拿到当前用户：

from flask import g
from flask_httpauth import HTTPBasicAuth

auth = HTTPBasicAuth()

@auth.verify_password
def verify_password(username, password):
    query_set = models.User.objects(username=username)
    if not query_set:
        return False
    user = query_set[0]
    # 验证密码
    if helpers.verify(password, user.password):
        # 将用户对象存入请求上下文g
        g.current_user = user
        return True
    return False

步骤2：实现权限检查装饰器 #

接下来写一个自定义装饰器，用来检查当前用户是否拥有对应端点/方法的权限。假设你的数据库里有Permission模型，记录了用户能访问的endpoint和method（比如endpoint是Flask-RESTful资源的endpoint名称，method是GET/PUT等）：

from functools import wraps
from flask import request

def require_permission(endpoint=None, method=None):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            # 如果没指定endpoint/method，自动从当前请求获取
            target_endpoint = endpoint or request.endpoint
            target_method = method or request.method.upper()
            
            # 从数据库检查用户是否有对应权限
            # 这里假设Permission模型有user、endpoint、method字段
            has_permission = models.Permission.objects(
                user=g.current_user,
                endpoint=target_endpoint,
                method=target_method
            ).exists()
            
            if not has_permission:
                # 返回403禁止访问
                return {"message": "Insufficient permissions"}, 403
            return f(*args, **kwargs)
        return decorated_function
    return decorator

步骤3：在Flask-RESTful资源中应用权限控制 #

现在你可以在需要保护的资源方法上，同时使用@auth.login_required和@require_permission装饰器，实现细粒度控制：

from flask_restful import Resource

class ExampleResource(Resource):
    # 只有拥有该endpoint GET权限的用户才能访问
    @auth.login_required
    @require_permission(method='GET')
    def get(self):
        return {"data": "This is protected GET data"}
    
    # 只有拥有该endpoint PUT权限的用户才能访问
    @auth.login_required
    @require_permission(method='PUT')
    def put(self):
        # 处理PUT请求逻辑
        return {"message": "Resource updated successfully"}

补充：权限模型示例（如果还没定义） #

如果你的数据库还没有权限相关模型，可以参考下面的MongoEngine模型定义（适配你用的models.User.objects的风格）：

from mongoengine import Document, ReferenceField, StringField

class Permission(Document):
    user = ReferenceField('User', required=True)
    endpoint = StringField(required=True)  # 比如"exampleresource"（对应Flask-RESTful资源的endpoint）
    method = StringField(required=True, choices=['GET', 'PUT', 'POST', 'DELETE'])
    
    meta = {
        'unique_with': ['user', 'endpoint', 'method']  # 避免重复权限记录
    }

关键注意点 #

• 无状态场景下，每次请求都会重新执行verify_password和权限检查，完全符合RESTful无状态要求
• g.current_user只在当前请求周期内有效，不会残留到下一个请求
• 可以根据实际需求调整权限检查逻辑，比如支持通配符端点、角色组权限等（比如用户属于某个角色，角色拥有权限）

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