Authenticating user sessions in Flask while using flask-socketio

Question

I have a Flask application that uses flask-socketio for real time communication between the server and client. I need to know how I can authenticate user sessions in flask-socketio. I.e., not allowing anyone to connect to the socket without being logged in.

Currently, what I do to authenticate if the user is logged in or not is checking the session variable for the loggedin value, that is supposed to be available in the session of all requests from a logged in user.

For example,

from flask import Flask, session

@app.route("/view_patient")
def view_patient():
    if "loggedin" in session:
        # Do stuff you'd do with a logged in user...
    else:
        # Do stuff you'd do with a non-logged in user, like showing the login form.

Now my question is, how can I go by authenticating a client that connects to my socketio server the same way, i.e., by checking their session cookie if they are logged in or not. Because I don't really want anyone opening a connection to my server and grabbing data that only connections from logged in users should have access to.

I have heard of flask-session, and flask-login implementation that is also endorsed by the maker of flask-socketio over here: https://blog.miguelgrinberg.com/post/flask-socketio-and-the-user-session

However, to me, flask-login and flask-session seem complicated to set up and I really don't want to replace my existing authentication with another one.

Now can someone guide me on how I can go about integrating user sessions with socketio so that only logged in users can open a websocket connection without using any third-party solution like flask-login and flask-session? If needed, my socket usage is something like this:

from flask import Flask, render_template, session, request
from staff.staff import staff
from patient.patient import patient
from flask_mysqldb import MySQL
import MySQLdb
from flask_socketio import SocketIO, emit, Namespace
import threading
import json
from datetime import datetime
from flask_session import Session

app.secret_key = "key"
app.config["app_name"] = "appname"
app.config["MYSQL_HOST"] = "host"
app.config["MYSQL_USER"] = "user"
app.config["MYSQL_PASSWORD"] = ""
app.config["MYSQL_DB"] = "dbname"
mysql = MySQL(app)
app.config["mysql"] = mysql

class get_patients_namespace(Namespace):
    def on_connect(self):
        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

socketio.on_namespace(get_patients_namespace('/get_patients'))

Answer 1

You can do this in the connect handler. Just return False if your user is not logged in.

class get_patients_namespace(Namespace):
    def on_connect(self):
        if "loggedin" not in session:
            return False

        # your logged in user stuff here