Как аутентифицировать пользователя приложения при подключении к RabbitMQ по протоколу WEB-STOMP?

Question

[Денис Бондарь]

Исходные данные
Дано: WEB-приложение на PHP с фронтендом на JS, а также RabbitMQ, который в том числе используется как websocket сервер, работающий по протоколу STOMP (web-stomp-plugin).

Схема взаимодействия стандартная: фронтенд приложение, используя stomp.js, подключается к RabbitMQ, авторизуясь при помощи логина и пароля. На RabbitMQ созданы exchange типа fanout для каждого зарегистрированного пользователя веб-приложения - они используются для передачи push-сообщений в сторону фронтенда. Фронтенд после загрузки сразу же подключается к RabbitMQ, создает временную автоубиваемую очередь и подключает ее к конкретному exhange, ассоциированному с текущим пользователем web-приложения. Теперь, если бекенд отправит сообщение в exchange пользователя - оно сразу же попадет во все очереди, подключенные к нему и будет передано в сторону фронтенд по протоколу stomp. Эта схема работает отлично.

Проблема
Дело в том, что изучив исходный js-код на фронтенде или же трафик, можно найти там логин и пароль подключения к RabbitMQ, а также наименование exchange, к которой подключается очередь, созданная фронтендом. Имея эти данные можно не санкционировано подключаться к RabbitMQ, а также создавать очередь и подключать ее к любому exchange, перехватывая информацию, которая предназначается другим пользователям.

Вопрос
Каким образом можно организовать доступ к конкретному exchange только в пределах известной сессии? Или же каким-то другим способом? Наверняка есть стандартные решения этого вопроса безопасности, реализуемые средствами в RabbitMQ.

Comments

[RidgeA]

Решение на вскидку - сделать небольшой сервис, который будет авторизоват пользователя по токену или по куке и прокидывать запросы на rmq. Но это выглядит как велосипед, возможно есть более адекватные способы решения проблемы, может кто напишет...

Answer 1

[Boris Köln]

Создавать очереди должен только бэкенд (PHP). Он же выдает уникальные ID (например, session_id или md5) клиенту. Далее клиент подключается к ранее созданной очереди с этим ID.

Теоретически клиент может брутфорсить очереди, но можете сами посчитать, сколько там будет возможных комбинаций.

P.S. Логин/пароль на клиенте нет смысла использовать, ибо он все легко равно перехватывается.

Comments

[Денис Бондарь]

Большое спасибо!

Answer 2

[Сергей]

Boris Korobkov дал ценное замечание по поводу того, что всем должен рулить сервер и нужно добавить элемент случайности в имена, дополню его несколькими рекомендациями.

1) Если очередь будет создана сервером, то auto-delete не будет работать, если клиент не подпишется на неё.
2) Не забывайте, что если два клиента (например, две вкладки или две сессии одного пользователя) подключатся к одной очереди, то сообщение получит только один из них. В своих решениях я использовал подход, что очередь создавалась клиентом, а обменник сервером, но необходимо убедиться, что до RabbitMQ доходят только легитимные запросы (от авторизованных клиентов) и пользователь ограничен в правах создавать только нужные очереди и подписывать их на нужные обменники.
3) Рекомендовал бы не выставлять порт web-stomp плагина без доп.защиты (например, HAProxy), которая позволит защитить RabbitMQ от перегрузки подключениями.

P.S. Писал статью на Хабр год назад о подобной системе, может быть что-то будет интересное в ней ещё - https://habr.com/post/341068/

Comments

[Денис Бондарь]

1. Я наткнулся сразу же на проблему автоуничтожения очередей. Оказалось, что все работает как нужно, если бекенд создает очередь типа autodelete, а фронтенд подключается к ней с параметрами: autodelete: true, durable: false, exclusive: false. В такой конфигурации очередь удаляется после отключения клиента или после истечения таймаута (stomp об этом заботится, как я понимаю). Как раз понял разницу между autodelete и exclusive. Нагрузка у нас будет не такой большой, чтобы отказаться от autodelete. Единственное, что я пока не решил, это то, каким образом подчищать очереди без консумеров. Теоретически их не должно быть, но на практике вполне могут появиться, например, когда клиент открыл приложение, запрос на создание очереди для его копии ушел в бекенд, но изза плохого соединения клиент отключился. В итоге очередь была создана, но подписываться на нее уже некому.
2. Это само собой. Мы будем создавать по одной очереди на каждую копию клиента (вкладки браузера). Пока что идея только в голове, а не в реализации, так что, возможно, еще наткнемся на подводные камни. Но в целом она кажется нормальной. Вариант, когда очередь создается клиентом, а обменник сервером - стандартный из примеров web-stomp-examples. Именно с ним я и не разобрался в плане прав доступа и задал этот вопрос на тостере. Вариант, когда контроль за созданием очередей лежит на бекенде мне понравился больше.
3. Спасибо. Обязательно воспользуемся nginx для проксирования вебсокетов.

За Вашу статью отдельное огромное спасибо. Именно она помогла найти правильный подход. Я остановился на варианте с отдельным обменником fanout для каждого юзера + двумя обменниками direct, с которыми работает клиентский код. (Рис. 8)

Я с удовольствием пообщался бы с Вами на тему RabbitMQ. Мне интересно всё, что касается этого брокера. Мы долго выбирали и решили остановиться именно на нем. Ну и вебсокеты тоже решили реализовать на нем, чтобы не придумывать еще что-то. Так что есть огромное желание изучить его работу досконально.

Спасибо!

[Сергей]

На очередь можно навесить TTL через аргумент x-expires (при создании или через политику), и RabbitMQ удалит её через указанный интервал времени, если не будет консьюмеров.
Можете написать мне на почту или в телеграм, но я уже почти год не работаю с RabbitMQ, так что смогу ответить, наверное, только на какие-то базовые вопросы.

[serega_chem]

Денис Бондарь, как удалось решить основной вопрос с аутентификацией?
Я пишу чат с использованием того же стека. У меня все сообщения проходят через один обменник с типом topic. Есть постоянная очередь, в которую попадают все сообщения (оттуда они пишутся в Redis, а затем в СУБД). Очереди для каждой страницы создаются на клиенте.
Вот так выглядит подписка:

stompClient.subscribe("/exchange/topic_message/currentUserId", callback);

Отправка сообщения:

stompClient.send('/exchange/topic_message/toUserId' , {}, message);

Буду рад совету)

[Денис Бондарь]

serega_chem, Клиент (WEB-STOMP) сначала по АПИ запрашивает имя временной очереди (и настройки этой очереди). АПИ метод генерирует это имя случайным образом, например `push-subscription-13-sEQEccPHmM479GBw9Mg3PA` и выдает его в ответ клиенту. Здесь 13 - это ИД пользователя - проще обслуживать брокер, видя ИД пользователя. При этом на стороне сервера создается эта временная очередь, которая затем подключается к exchange этого конкретного пользователя. То есть у пользователя один exchange, куда попадают сообщения для него, но при этом может быть запущено несколько клиентов (несколько вкладок браузера, например) и каждому клиенту будет создана своя отдельная временная очередь.



Временная очередь — автоудаляемая (установлен флаг autodelete), а также с установленным аргументом `x-expires` (у меня он равен 60 секундам). Это значит, что если клиент не подключится к очереди в течение этого времени - она автоматически удалится, а также если клиент отключится от очереди - она тоже автоматически удалится. Это решает проблему накопления зомби-очередей. Тут еще можно сказать о том, что протокол STOMP контролирует наличие клиента через heartbeat и, если клиент не ответил, то очередь также будет удалена.

Клиент, после получения имени временной очереди и ее параметров — подключается к ней и слушает ее события.

При обновлении страницы браузера происходит и перезапуск web-stomp клиента, что автоматически приводит к отключению клиента от временной очереди (она тут же удаляется) и затем созданию запроса на новую очередь и подключению к ней. То есть не нужно хранить имя очереди между перезапусками клиента (между обновлениями страницы) - в этом нет смысла. К тому же это упрощает схему и делает ее надежней.

В реббите создаем пользователя stomp с паролем stomp и разрешаем ему доступ только к очередям:
rabbitmqctl add_user stomp stomp
rabbitmqctl set_permissions -p / stomp "^push-subscription-.*" "" "^push-subscription-.*"

Вот такая реализация клиента (упрощенно, конечно же):

// Тут сначала выполняются два АПИ запроса - на получение имени очереди и на получение настроек
// Имя временной очереди получаем по API
var queue_name = "push-subscription-2-ZWM3YzY1YjgyOTIwZGQzZDRlYzlkYzU1NmVkOTQ3MTI";

// Эти настройки тоже получаем по API от сервера
var broker_username = "stomp";
var broker_password = "stomp";
// Очередь с этими  параметрами изначально создается на сервере, но чтобы подключиться к ней клиентом, параметры этой очереди должны совпадать с теми, что на сервере.
var queue_expires = 60000;  // Это время жизни очереди, отведенное на подключение клиента
var queue_message_ttl = 20000;  // Это ТТЛ собощений в очереди по умолчанию. Через 20 секунд сообщения, которые не доставлены клиенту - будут удалены, так как теряется их актуальность. Для сообщений с другой актуальностью можно установить конкретный ТТЛ
var queue_max_priority = 10;  // Приоритет по умолчанию. Для сообщений с другим приоритетом его можно выставить индивидуально

// Далее код клиента
var client = Stomp.client('ws://' + window.location.hostname + ':15674/ws');
client.heartbeat.outgoing = 30000;  // 30 секунд - исходящий heartbeat
client.heartbeat.incoming = 0;  // входящий отключаем (мы так решили для себя)
client.reconnect_delay = 5000;  // ожидание соединения
//client.debug = null;  // для отладки раскомментировать или установить в качестве значения какой-то ИД блока, куда будет выводиться отладка. По умолчанию - в консоль.

var on_connect = function () {
    client.subscribe(queue_name, function (d) {
        $("#websocket").append("Получено из очереди: " + d.body + "<br />")
    }, {
        durable: false,
        exclusive: false,
        'x-expires': queue_expires,
        'x-message-ttl': queue_message_ttl,
        'x-max-priority': queue_max_priority
    });
    console.log('WEB STOMP Connected');
};

var on_error = function () {
    console.log('WEB STOMP error');
};

client.connect(broker_username, broker_password, on_connect, on_error, '/');

[serega_chem]

Денис Бондарь, спасибо за развёрнутый ответ, мне очень пригодится пример вашей реализации.
Но у вас клиент только принимает сообщения. Я бы хотел и отправлять. Скорее всего придется это делать Ajax'ом, т.к. не вижу как можно ограничить доступ клиента к обменникам

[Алексей Кузнецов]

Денис Бондарь, а можешь показать, как примерно это выглядит на стороне бэкенда? Что именно делается при запросе имени очереди?

[Денис Бондарь]

Алексей Кузнецов, выше пример фронта, в котором уже есть имя очереди. Чтобы получить имя очереди, а также получить имя пользователя и пароль для подключения к RabbitMQ, нужно выделить на бэкенде эндпоинт с аутентификацией, который создает очередь и возвращает имя очереди, имя пользователя и пароль в json во фронт. Помимо этого, этот эндпоинт, если это необходимо, должен создать все необходимые exchange (но это зависит от конкретной архитектуры).

Простой пример сервиса, который все это длеает (и вызывается из АПИ-контроллера):

public function createAutodeleteQueueForUserId(string $userId): string
{
    $this->createDirectExchanges();

    $queueName = sprintf(
        'webstomp-%s-%s',
        $userId,
        $this->queueSuffixGenerator->generate()
    );

    $this->amqpChannel->queue_declare(
        $queueName,
        false,
        false,
        false,
        true,
        false,
        new AMQPTable(
            [
                'x-expires' => 60000,
                'x-message-ttl' => 20000,
                'x-max-priority' => 10,
            ]
        )
    );

    $fanoutExchange = $this->createUserFanoutExchange($userId);
    $this->amqpChannel->queue_bind($queueName, $fanoutExchange);

    return $queueName;
}

Метод возвращает имя созданной очереди. Например, для пользователя с ID 12345 будет создана очередь с именем вроде webstomp-12345-SGVsbG8sIFdvcmxkISBIZWxsbywgSGFiciE. И это имя очереди нужно передать во фронтенд, чтобы фронтенд приложение подключилось к ней в течение x-expires мс.

Архитектура моего push модуля такая:

• Один Direct exchange для широковещательных сообщений (для всех пользоватлей)
  
• Один Direct exchange для персональных сообщений конкретного пользователя
  
• Множество Fanout exchange, создаваемых для каждого пользователя (выше в коде видно его создание в конце метода) — после создания этот exchange безусловно биндится с широковещательным direct-exchange, а также биндится с персональным direct-exchange, но при помощи ключа маршрутизации. На каждого пользователя создается один такой fanout exchange
  
• Множество очередей, которые создаются приведенным выше кодом и биндятся на fanout-exchange этого пользователя. Для каждого запроса создается своя очередь. Это могут быть как разные вкладки браузера, так и другие каналы получения уведомлений (например, мобильные приложения)
  

В итоге в системе получается по одному direct-exchange каждого типа, множество fanout-exchange по одному для каждого пользователя и множество очередей по одной на каждую вкладку браузера.

После того, как очередь создана, есть 60 секунд чтобы фронтенд приложение подключилась к ней — иначе она будет автоудалена. Если фронтенд приложение по какой-то причине отключается от очереди — она также автоматически удаляется. Если фронтенд приложение не ответело на служебный ping (webstomp), то очередь также будет автоудалена.

При такой архитектуре новая копия приложения (открыта в новой вкладке браузера) будет получать только новые уведомления параллельно (fanout-exchange распараллеливает все сообщения во все подключенные к нему очереди). Если нужно другое поведение (например, чтобы новая копия клиентского приложения получала вообще все уведомления за все время), то придется изменить архитектуру.

[Алексей Кузнецов]

Кажется всё запуталось в голове ещё сильней ))) Вот смотри, да, мне интересна тема с получением уведомлений на каждой вкладке отдельно с момента подключения к уникальной для вкладки очереди. Какие-то широковещательные уведомления для всех мне не нужны. Я правильно понимаю, что мне достаточно создавать только фанаут эксченджи? Или в каком-то виде всё-таки нужны директы?
И ещё покажи пожалуйста пример отправки сообщения какому-либо пользователю. Он их получает на каждой вкладке или на конкретной? Извиняюсь, если вопросы глупые ))

[Денис Бондарь]

Алексей Кузнецов, да. можно создать только fanout и отправлять в ту, которая закреплена за конкретным пользователем.

Пользователь открывает несколько копий системы в разных вкладках. Для каждой вкладки создается отдельная очередь, которая подключается к fanout-exchange этого пользователя. Фишка fanout-exchange в том, что этот тип exchange дублирует сообщения во все подключенные очереди — значит пользователь будет получать одни и те же сообщения во всех открытых вкладках.

Пример отправки тривиальный — создаешь сообщение и отправляешь в exchange. Всё просто.
Под "сообщением" здесь понимается не просто текстовое сообщение, а какой-то объект (json) произвольной структуры, который можно обработать на фронтенде. Ну, напрмер, чтобы отделить всплывающие уведомления от других типов информации, передаваемых во фронт.

Упрощенный пример кода отправки сообщения если есть только fanout-exchange с именем вроде fanout-exchange-name-12345

$message = new AMQPMessage(json_encode($messageBody), ['content_type' => 'application/json']);
$this->amqpChannel->basic_publish($message, 'fanout-exchange-name-' . $userId);

[Алексей Кузнецов]

Денис Бондарь, крутяк, вроде завелось! Спасибо, ты очень помог!

[Денис Бондарь]

Алексей Кузнецов, пожалуйста :)