Flask API认证中间件对POST请求失效问题求助

搭建基础Flask API时，使用Firebase Admin SDK实现的Authorization请求头校验中间件对GET请求正常工作，但处理POST请求的/is_subscribed端点时失效。移除中间件后接口可正常运行，但失去认证保护且无法获取user数据。服务器日志返回400状态码，前端请求始终进入catch分支。

相关代码 #

后端/is_subscribed端点代码 #

@app.route('/is_subscribed', methods=['POST'])
@check_auth
def is_subscribed():
    if request.method == 'POST':
        user = request.user
        try:
            itemId = request.get_json()['item_id']
            
            subscriptions = db. \
                                collection('users'). \
                                document(user['uid']). \
                                get(). \
                                to_dict()['subscriptions']

            if itemId in subscriptions:
                return {
                    'message': 'User is subscribed to this item',
                    'status': 'success',
                    'data': True
                }, 200
        except Exception as e:
            return {
                'message': 'Error checking subscription', 
                'status': 'error'
            }, 400

认证中间件代码 #

def check_auth(f):
    @wraps(f)
    def wrap(*args,**kwargs):
        if not request.headers.get('authorization'):
            return {'message': 'No authorization token provided'}, 400
        try:
            user = auth.verify_id_token(request.headers['authorization'])
            request.user = user
        except:
            return {'message':'Invalid token provided.'}, 400
        return f(*args, **kwargs)
    return wrap

前端请求代码 #

const BASE_URL = 'http://127.0.0.1:5000';

const api = axios.create({
    baseURL: BASE_URL,
});

export const isSubscribed = async ({
    itemId,
    token,
}: {
    itemId: string;
    token: string;
}) => {
    try {
        const response = await api.post(
            '/is_subscribed',
            {
                item_id: itemId,
            },
            {
                headers: {
                    Accept: 'application/json',
                    'Content-Type': 'application/json',
                    authorization: `${token}`,
                },
            },
        );
        return response.data;
    } catch (error) {
        showToast('Error checking subscription status', 'error');
        return null;
    }
};

排查思路 #

• 检查Authorization头格式：Firebase ID Token在请求头中需要以Bearer 为前缀，前端可能直接传递token字符串，缺少前缀导致校验失败。
• 捕获具体异常信息：当前中间件的except块未指定异常类型，无法定位是token格式错误、过期还是其他问题，需打印异常详情。
• 验证请求头接收情况：在中间件中打印request.headers['authorization']的值，确认前端token是否正确传递到后端。

解决方案 #

1. 修正前端请求的Authorization头格式 #

给token添加Bearer 前缀：

headers: {
    Accept: 'application/json',
    'Content-Type': 'application/json',
    authorization: `Bearer ${token}`,
},

2. 优化后端中间件，增加格式校验和异常日志 #

def check_auth(f):
    @wraps(f)
    def wrap(*args,**kwargs):
        auth_header = request.headers.get('authorization')
        if not auth_header:
            return {'message': 'No authorization token provided'}, 400
        # 校验Authorization头格式
        if not auth_header.startswith('Bearer '):
            return {'message': 'Invalid authorization header format. Use "Bearer <token>"'}, 400
        # 提取实际token
        token = auth_header.split(' ')[1]
        try:
            user = auth.verify_id_token(token)
            request.user = user
        except Exception as e:
            # 打印异常到日志，方便调试
            print(f"Authentication error: {str(e)}")
            return {'message':'Invalid token provided.'}, 400
        return f(*args, **kwargs)
    return wrap

3. 优化后端端点的异常处理，提升错误信息准确性 #

@app.route('/is_subscribed', methods=['POST'])
@check_auth
def is_subscribed():
    user = request.user
    try:
        req_data = request.get_json()
        # 校验请求体是否包含item_id
        if not req_data or 'item_id' not in req_data:
            return {
                'message': 'Missing item_id in request body',
                'status': 'error'
            }, 400
        itemId = req_data['item_id']
        
        user_doc = db.collection('users').document(user['uid']).get()
        # 校验用户文档是否存在
        if not user_doc.exists:
            return {
                'message': 'User not found',
                'status': 'error'
            }, 404
        
        user_data = user_doc.to_dict()
        # 处理用户无subscriptions字段的情况
        if 'subscriptions' not in user_data:
            return {
                'message': 'User has no subscriptions',
                'status': 'success',
                'data': False
            }, 200
        
        subscriptions = user_data['subscriptions']
        is_subscribed = itemId in subscriptions
        return {
            'message': 'User is subscribed to this item' if is_subscribed else 'User is not subscribed to this item',
            'status': 'success',
            'data': is_subscribed
        }, 200
    except Exception as e:
        print(f"Subscription check error: {str(e)}")
        return {
            'message': 'Error checking subscription', 
            'status': 'error'
        }, 400

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