Есть ли способ авторизации пользователя с помощью Flask Socket IO?

Я использую Flask с Flask SocketIO и Flask Login .

Моя проблема в том, что когда я пытался использовать функцию login_user внутри события socketio.on , ошибки не было, но пользователь не вошел в систему. Мой пример может помочь проиллюстрировать это. my example event будет вызван для входа пользователя, а затем /usersonly будет после того, как это событие /usersonly .

@socketio.on('my example event')
def my_func(json):
    ...
    login_user(user)

@app.route('/usersonly')
@login.user_required
def my_call():
    print(current_user)

Как сделать так, чтобы my_func мог войти в систему пользователя?

python flask flask-login flask-socketio

Всего 1 ответ

Это невозможно с Flask Login, потому что в документации говорится, что Flask Login использует Cookies или ключи заголовка для аутентификации, и невозможно получить к нему доступ через WebSocket (Socket IO), поэтому я рекомендую использовать систему на основе токенов, как в API , используя токены JWT.

@socket.on('login')
def on_login(message):
    user = User.query.filter_by(email=message['email']).first()
    if not user:
       socket.emit('login', {'msg': "User not found"})

    if not user.check_password(form['password']):
       socket.emit('login', {'msg': "User not found"}) 
       return 
    user = UserSchema().dump(user).data
    token = jwt.encode(
            {'id': user['id'], 'user': user['name']})
    user['token'] = token.decode('UTF-8')
    socket.emit('login', {'user': user})

И создать декоратор для входа

def login_requeried(f):
    @wraps(f)
    def decorated(message):

        try:
            data = jwt.decode(message['token'], globals.current_app.config['SECRET_KEY'])
            try:
                user = User.query.filter_by(id=data['id']).first()
                Globals.current_user = UserSchema().dump(user).data
            except Exception as e:
                return Error.api_response(Error.USER_NOT_FOUND)
        except Exception as e:

            return Error.api_response(Error.INVALID_TOKEN)
        return f(message)

    return decorated

и использовать как это

@socketio.on('my example event')
@login_requeried
def my_func(json):
    ...

Но помните, что вы всегда должны передавать токен в свой сокет, чтобы сделать проверку сервера, я рекомендую прочитать этот блог от Мигеля Гринберга.