Flask中无需第三方库实现flask-socketio用户会话认证

核心思路 #

Flask-SocketIO会自动携带客户端的Flask会话Cookie，因此你可以在Socket连接的on_connect事件中直接检查现有会话中的loggedin字段，以此判断用户是否已登录，未通过认证则拒绝连接。

具体实现步骤 #

1. 在Socket命名空间的on_connect方法中添加认证逻辑
   在你的get_patients_namespace类的on_connect方法最开始，加入会话检查逻辑，未通过认证则直接拒绝连接：

   class get_patients_namespace(Namespace):
       def on_connect(self):
           # 会话认证：检查loggedin字段是否存在且有效
           if "loggedin" not in session or not session["loggedin"]:
               # 向客户端发送认证失败提示（可选）
               self.emit("auth_error", {"message": "请先登录后再尝试连接"})
               # 返回False直接拒绝连接
               return False
           
           # 以下是原有的业务逻辑
           patients_data = []
   
           def get_patients_data():
               mydb = MySQLdb.connect(
                   host=app.config["MYSQL_HOST"],
                   user=app.config["MYSQL_USER"],
                   passwd=app.config["MYSQL_PASSWORD"],
                   db=app.config["MYSQL_DB"],
                   charset="utf8",
               )
   
               cursor = mydb.cursor(MySQLdb.cursors.DictCursor)  # type: ignore
               nonlocal patients_data
               today_date = datetime.today().strftime('%Y-%m-%d')
   
               cursor.execute("SELECT * FROM patients WHERE visit_date = %s", (today_date,))
               patients = cursor.fetchall()
               if patients != patients_data:  # type: ignore
                   patients_data = patients
   
                   self.emit(
                       "data", json.dumps(patients_data, indent=4, sort_keys=True, default=str)
                   )
               else:
                   self.emit("data", "No new data")
   
               mydb.commit()
               cursor.close()
               mydb.close()
   
           k = ThreadJob(get_patients_data, event, 5)
           k.start()
   
       def on_disconnect(self):
           pass
   

2. 客户端侧处理认证失败
   在前端JavaScript中监听连接错误和认证失败事件，引导用户跳转至登录页面：

   // 连接指定Socket命名空间
   const socket = io('/get_patients');
   
   // 监听连接错误（包含认证被拒绝的场景）
   socket.on('connect_error', () => {
       alert('连接失败，请先登录');
       window.location.href = '/login';
   });
   
   // 监听服务器主动发送的认证失败消息
   socket.on('auth_error', (res) => {
       alert(res.message);
       window.location.href = '/login';
   });
   

关键注意事项 #

• 确保你的Flask应用已正确设置secret_key（你已在代码中配置app.secret_key = "key"，这是会话加密的必要条件）。
• 客户端需在同域名下访问Socket服务，否则Cookie无法自动携带（跨域场景需额外配置CORS和Cookie传递策略）。
• 认证逻辑必须放在所有业务代码之前，避免未登录用户触发数据查询等敏感操作。

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