What should MQTT plugin behavior be for existing consumers be when their JWT token expires?

[arkadipdigite]

Describe the bug

I created a PahoMQTT Python MQTT consumer.
What I am noticing is that even after the JWT has expired, the client is getting new messages published on the broker.

The broker is not issuing any DISCONNECT messages, on top of that the consumer with expired JWT is getting all the messages.

This is a very critical issue and a security vulnerability. I don't know if this issue is from the broker, or if it's from any of my configurations.

Please help me to resolve the issue as I think it is a major drawback.

Reproduction steps

1. Using rabbitmq_auth_backend_oauth2 and rabbitmq_mqtt plugin.
2. Issuing JWT with a small expiration time.
3. Creating a MQTT consumer with the JWT in the password.
4. Client is still connected even after the JWT is expired.

Expected behavior

The broker should not send any new message to the client and also should send a DISCONNECT message when the JWT expires.

Additional context

No response

[lukebakken]

Hello, thanks for using RabbitMQ.

This issue report is woefully inadequate - you haven't even mentioned the versions of the software you are using.

If you would like us to assist you, you must do the following:

• Provide the version of all software components in play in this environment (RabbitMQ, Erlang, Python, etc)
• Provide RUNNABLE MQTT client code.
• Provide your complete RabbitMQ configuration files.
• Provide details on your OAuth2 setup.

cc @MarcialRosales @ansd @michaelklishin

[michaelklishin]

@arkadipdigite if this is a true security vulnerability, you should have reported the details in private but you've decided to start a public discussion with bold claims and zero evidence. Not cool.

When a JWT token expires, connections will not be closed. However, individual operations such as publishing, registration of consumers (SUBSCRIBE in MQTT speak) will be refused, and that in practice will lead to a closed connection.

New connections with an expired JWT token will be refused (there is a dedicated integration test for that).

This is true for all protocols, not just MQTT.

The OAuth 2 plugin will also try to renew the token (when it has a refresh token to do so) shortly before its expiration, by design. If that fails, eventually in many cases the connection will be closed, here are more integration tests that prove this claim.

None of the protocol specs explicitly state what should happen when a JWT token expires. They simply do not cover JWT at all. For RabbitMQ's OAuth 2 plugin it means that all new operations performed by clients will be refused, including new subscriptions.

It's a good question as to what should happen to existing connections, and yes, there can be multiple opinions about it.

Finally, there is an emergecy "fix" that should work equally for all protocols: deleting a user will always close all of its connections.

[arkadipdigite]

@michaelklishin

When a JWT token expires, connections will not be closed. However, individual operations such as publishing, registration of consumers (SUBSCRIBE in MQTT speak) will be refused, and that in practice will lead to a closed connection.

Even after 30 minutes of the exp of JWT, the MQTT client consumer is still connected and consuming messages like charm.

New connections with an expired JWT token will be refused (there is a dedicated integration test for that).

This is true for all protocols, not just MQTT.

Yes, this is what I agree with. Any new connection is refused, but the old connection is maintained.

The OAuth 2 plugin will also try to renew the token (when it has a refresh token to do so) shortly before its expiration, by design. If that fails, eventually in many cases the connection will be closed, here are more integration tests that prove this claim.

I don't think, this is happening in my case.

Finally, there is an emergecy "fix" that should work equally for all protocols: deleting a user will always close all of its connections.

How Can I delete the user? Any suggestion for that?

[michaelklishin]

@arkadipdigite

How Can I delete the user? Any suggestion for that?

There is a dedicated doc section for that in the Access Control guide.

[arkadipdigite]

@michaelklishin

As I am not using basic authentication, but oauth2, the user's info is stored separately. So in that case,

Do rabbitmq create any user for the same provided by the JWT?

[michaelklishin]

@arkadipdigite the two things are not mutually exclusive. In any case, rabbitmqctl help | grep close, rabbitmqctl help close_all_user_connections. Please study what commands CLI tools and the HTTP API have to offer, this is RabbitMQ 101.

[michaelklishin]

@ansd has mentioned it to me in our team's chat that the MQTT plugin expires its permission cache after 1 second of client inactivity.

@arkadipdigite so your bold claim of a security vulnerability comes down to a very specific set of circumstances that are present in every protocol implementation in RabbitMQ for efficiency reasons:

• When a token expires, all new operations will be refused, and that covers effectively all operations except for existing consumers that have a constant stream of deliveries
• Connections (channels) use a cache of permissions for obvious efficiency reasons that is cleared in certain scenarios
• Checking the token for validity for every outgoing messages would completely destroy throughput on the happy code path
• As an emergency solution, deleting a user will close all of its connections, for any protocol

So the treatment of expired JWT tokens in RabbitMQ 3.13 (and in earlier versions) is fairly reasonable and does not cut corners but it also does not ruin the hot message delivery path throughput by checking for token validity after every message delivered.

The only solution I can think of is introducing a periodic timer that would close connections with expired and non-renewed tokens. For environments with 100s of thousands or millions of connections it will come with a resource footprint price in terms of both per-connection RAM usage and CPU (context switching).

The higher is the interval, the more resources will be wasted 99% (if not 99.9%) of the time (because most tokens are reasonably long lived, e.g. at least a few hours), and proportionally more so the more connections a cluster has.

Update. @ansd has exactly this consideration in mind during the MQTT plugin redesign for 3.13: how not to waste resources 99% of the time by expiring the permissions cache too often or even for every message.

[michaelklishin]

A set of relevant changes in the RabbitMQ Stream protocol: #10299. Note that with streams, there usually aren't millions of connections and we fully control the protocol and clients, which is not the case of MQTT or AMQP 1.0, so certain solutions for streams won't be feasible for other protocols.

[arkadipdigite]

@lukebakken

Here is the running MQTT code

import logging
import certifi

import paho.mqtt.client as mqtt
from paho.mqtt.enums import MQTTProtocolVersion

logging.basicConfig(level=logging.DEBUG)


# The callback for when the client receives a CONNACK response from the server.
def on_connect(client, userdata, flags, reason_code, properties):
    print(f"Connected with result code {reason_code}")
    # Subscribing in on_connect() means that if we lose the connection and
    # reconnect then subscriptions will be renewed.
    client.subscribe("std/request/356303488760727")


# The callback for when a PUBLISH message is received from the server.
def on_message(client, userdata, msg):
    print(msg.topic+" "+str(msg.payload))


mqttc = mqtt.Client(
    mqtt.CallbackAPIVersion.VERSION2,
    client_id="356303488760727",
    protocol=MQTTProtocolVersion.MQTTv5,
)
mqttc.username_pw_set('', '<JWT TOKEN>')
mqttc.tls_set(
    ca_certs=certifi.where()
)

mqttc.on_connect = on_connect
mqttc.on_message = on_message

mqttc.connect("broker.url.com", 8883, 60)

# Blocking call that processes network traffic, dispatches callbacks and
# handles reconnecting.
# Other loop*() functions are available that give a threaded interface and a
# manual interface.
mqttc.loop_forever()

Here are the version details

 ##  ##      RabbitMQ 3.13.2
  ##  ##
  ##########  Copyright (c) 2007-2024 Broadcom Inc and/or its subsidiaries
  ######  ##
  ##########  Licensed under the MPL 2.0. Website: https://rabbitmq.com

  Erlang:      26.2.5 [jit]
  TLS Library: OpenSSL - OpenSSL 3.1.5 30 Jan 2024

Python Version: v3.10.12

RabbitMQ Configuration

apiVersion: rabbitmq.com/v1beta1
kind: RabbitmqCluster
metadata:
  name: rabbitmq
  namespace: rabbitmq
spec:
  replicas: 1 # Number of replica
  terminationGracePeriodSeconds: 60 # Termination Grace Period Timeout

  service:
    type: ClusterIP
  
  # TLS certificate
  tls:
    secretName: broker-url-com-tls
    disableNonTLSListeners: false
  
  # Persistence Storage
  persistence:
    storageClassName: managed-premium-retain
    storage: 32Gi
  
  # Resource request and limits
  resources:
    requests:
      cpu: 100m
      memory: 500Mi
    limits:
      cpu: 1000m
      memory: 2Gi
  
  # rabbitmq config
  rabbitmq:
    # Additional Config
    additionalConfig: |
      log.default.level = debug
      default_user = admin
      default_pass = Password
      auth_backends.1 = rabbit_auth_backend_oauth2
      auth_oauth2.resource_server_id = rabbitmq
      auth_oauth2.additional_scopes_key = permissions
      management.oauth_disable_basic_auth = false


    advancedConfig: |
      [
        {rabbitmq_auth_backend_oauth2, [
          {key_config, [
            {default_key, <<"id1">>},
            {signing_keys,
              #{<<"id1">> => {pem, <<"-----BEGIN PUBLIC KEY-----
            Key Goes Here
            -----END PUBLIC KEY-----">>}
              }
            }
          ]}
        ]}
      ].
    # Env config
    # envConfig: |
    #   RABBITMQ_DISTRIBUTION_BUFFER_SIZE=some_value

    # Additional RabbitMQ plugins
    additionalPlugins:
      - rabbitmq_management
      - rabbitmq_mqtt
      - rabbitmq_event_exchange
      - rabbitmq_auth_backend_oauth2

[michaelklishin]

@arkadipdigite we do not have much to add to the findings above. RabbitMQ's treatment of expired JWT tokens is reasonable, perfectly secure for new connections or new SUBSCRIBE frames, and the way it works for pre-registered MQTT consumers (subscribers) is not arbitrary: checking for token validity for every message or periodically will have substantial effects on throughput or non-trivial affects on per-connection resource consumption.

Do you have a solution to suggest?

[arkadipdigite]

I understand the current behavior, but It's not 100% accurate.

I don't have much idea on RabbitMQ's internal mechanism. So can't provide very insightful solution. But here are my ideas.

The simple solution can be for the broker to send a DISCONNECT by looking at the expire time, or might be check the JWT before forwarding any messages.

[michaelklishin]

@arkadipdigite how is that "not 100% accurate"? You were provided a detailed description of the behavior and the design decision by the core team. If that's not accurate enough for you, go read the source code.

The simple solution can be for the broker to send a DISCONNECT by looking at the expire time

It was explained above that this means using a timer per connection, and with 100s of thousands or
millions of connections will be a non-trivial waste of resources, in particular CPU resources but also memory, and reducing per-connection memory consumption was the key design goal of the MQTT plugin redesign in 3.13.

or might be check the JWT before forwarding any messages

@arkadipdigite sorry, do you actually read our responses? We do not do exactly that by design because this would completely ruin the throughput for the "regular operations" path. It is not a coincidence that the validity of the token is not checked before delivering a message or a batch of messages. This directly affects the hot code path.

Engineering is all about making trade-offs and optimizing for how the system operates most of the time. An expired JWT token of an already connected subscriber is an edge case, not the norm.

[michaelklishin]

This discussion is going in circles. We have provided an explanation of

• How expired JWT tokens are treated in different situations: all new connections, new subscriptions and newly published messages will be rejected, for example
• Why the current MQTT plugin design explicit avoids verifying token validity before every delivery
• Why a per-connection timer solution was not adopted in the MQTT plugin so far
• What tooling exists for closing connections of a certain user, in a certain virtual host, etc in
  case of an emergency

RabbitMQ's OAuth 2 and MQTT implementations do not cut corners and the only scenario where an expired JWT token would not very quickly result in client disconnection has been considered, weighted against its throughput and resource consumption downsides, and left as is.

If someone has a specific design in mind that has minimal resource and throughput effects, they are welcome to start a new discussion or submit a PR.

[michaelklishin]

#11862 proposes a specific (and hardly elegant) solution that would neither affect the throughput nor will introduce an additional per-connection timer.

This is not a promise of delivery: the above issue may or may not be investigated, or work out well, or ship in any specific release.

[ansd]

#11867 fixes this issue.