如何阻止未登录用户访问Flask+Pyrebase4应用的tasks页面？

问题背景 #

已基于Flask和Pyrebase4实现基础登录登出功能，但未做访问权限校验，未认证用户可直接通过URL访问需要登录的/tasks页面，需实现未登录用户无法进入该页面的限制。

解决方案：自定义登录验证装饰器 #

Flask支持通过自定义装饰器校验用户登录状态，在需要权限的路由上挂载该装饰器，即可拦截未认证请求。

步骤1：添加登录验证装饰器 #

在app.py中新增以下装饰器函数：

from functools import wraps

def login_required(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        # 检查session中是否存在已登录用户标识
        if 'user' not in session:
            # 未登录则重定向到登录页
            return redirect(url_for('index'))
        return f(*args, **kwargs)
    return decorated_function

步骤2：为tasks路由挂载装饰器 #

修改/tasks路由的定义，添加@login_required装饰器实现权限拦截：

@app.route('/tasks', methods=['POST', 'GET'])
@login_required
def tasks():
    # 改用session存储用户专属数据，避免全局变量导致多用户数据冲突
    if request.method == 'POST':
        texto = request.form['texto']
        session['texto_length'] = len(texto)
    # 从session读取数据，默认值为0
    texto_length = session.get('texto_length', 0)
    return render_template('tasks.html', texto=texto_length)

步骤3：优化全局变量问题（可选但推荐） #

原代码中使用全局变量texto会导致多用户访问时数据混乱，改用Flask的session存储用户专属数据，能保证数据隔离性。

修改后的完整app.py代码 #

from flask import Flask, session, render_template, request, redirect, url_for
from functools import wraps
import pyrebase

app = Flask(__name__)

config = {
    'apiKey': "here",
    'authDomain': "here",
    'projectId': "here",
    'storageBucket': "here",
    'messagingSenderId': "here",
    'appId': "here",
    'measurementId': "here",
    'databaseURL':""
}

firebase = pyrebase.initialize_app(config)
auth = firebase.auth()

app.secret_key='hufrhhfhrjfbrekjfberufureye4674656486435hjjjfhhkjsh'

# 登录验证装饰器
def login_required(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if 'user' not in session:
            return redirect(url_for('index'))
        return f(*args, **kwargs)
    return decorated_function

@app.route('/', methods=['POST', 'GET'])
def index():
    if request.method == 'POST':
        email = request.form.get('email')
        password = request.form.get('password')
        try:
            user = auth.sign_in_with_email_and_password(email, password)
            session['user'] = email
            # 清空之前的文本长度记录
            session.pop('texto_length', None)
        except:
            return '密码或账号错误'
        
        return redirect(url_for("tasks"))
    
    return render_template('home.html')
      
@app.route('/tasks', methods=['POST', 'GET'])
@login_required
def tasks():
    if request.method == 'POST':
        texto = request.form['texto']
        session['texto_length'] = len(texto)
    texto_length = session.get('texto_length', 0)
    return render_template('tasks.html', texto=texto_length)       
        
@app.route('/logout')
def logout():
    session.pop('user', None)
    session.pop('texto_length', None)
    return redirect('/')

if __name__=='__main__':
    app.run(debug=True)

模板文件说明 #

home.html和tasks.html无需修改，保持原代码即可。

效果验证 #

1. 未登录状态下直接访问/tasks，会自动重定向到登录页
2. 登录成功后可正常访问/tasks页面
3. 登出后再次访问/tasks，仍会被拦截到登录页

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