Repository: rabbitmq/rabbitmq-server (language: javascript)

Issue:
Support OAuth 2.0 token renewal in AMQP 1.0
### Is your feature request related to a problem? Please describe.

In order to fully support OAuth2 authentication in AMQP 1.0 protocol, it should be possible to refresh a token before it expires. Today we use PLAIN SASL authentication mechanism to pass the token in the password field. SASL handle-shake  only occurs during the connection establishment phase. Which means, we cannot replace the first token with a new one.


 


### Describe the solution you'd like

In order to fully support token refresh it seems the spec extension, called [AMQP Claims-based Security](https://docs.oasis-open.org/amqp/amqp-cbs/v1.0/csd01/amqp-cbs-v1.0-csd01.html) is the way to go. Without having this feature, today our users have to wait until the token expires to reconnect with a new token.

The first course of action is to implement the **SASL AMQPCBS Mechanism** which is a SASL mechanism meant to transfer one or many, potentially large, tokens to the SASL/AMPQ server. The AMQP 1.0 frame size should be small, e.g. 512 bytes, except for transferring tokens which is allowed to go up to 8192 bytes and send more than one frame until all tokens are transferred. 

The second course of action is to handle token refresh. For that, the AMQP client sends [set-token](https://docs.oasis-open.org/amqp/amqp-cbs/v1.0/csd01/amqp-cbs-v1.0-csd01.html#_Toc67999906) messages over a link to a CBS node which holds all the tokens on a connection-basis. 

**Question 1**: According to the spec, a client may send multiple tokens, however RabbitMQ, at least until today, only works with one token, not with many. For instance, if a client refreshes a token on an AMQP 0.9.1 connection via the command `update_secret`, RabbitMQ drops the old token and use the new one. In AMQP 0.9.1, RabbitMQ supports oauth2 authentication via the SASL PLAIN mechanism. The password field carries the token and the username field is ignored.

**Question 2**: Is it worth implementing an RFC like `SASL AMQPCBS` if there are no brokers currently implementing it? All investigated brokers and client libraries (see table below) have adopted what is called as XOAUTH2 which is essentially what RabbitMQ does which is tunneling the token in the password field. 
 [Solace](https://docs.solace.com/Security/Client-Authentication-Overview.htm#OAuth) mentions that for MQTT connections, the tokens must be less than 4Kbytes.
 [Azure](https://pkg.go.dev/github.com/Azure/go-amqp#SASLTypeXOAUTH2) says that to support oauth2 tokens, the maxframesize must be increased. This is what RabbitMQ does at the moment. 
 [IBM MQ](https://www.ibm.com/docs/en/ibm-mq/9.3?topic=SSFKSJ_9.3.0/secure/using_auth_tokens_in_an_application.htm) does not support yet JWT tokens for AMQP 1.0 clients however it does support it over its proprietary API and it does also tunnel the JWT token over the password field. Channels must be configured, like RabbitMQ, to take a token rather than a password, in the password field. 

**Question 3**: Despite being XOAUTH2 the adopted mechanism, it does not address token refresh without reconnecting. Therefore, XOAUTH2 by itself does not give us much because we were looking for a standardized mechanism to refresh tokens and obviously a better protocol support for tokens which allowed for large tokens which could extend beyond the max frame size.

Conclusion: *SASL AMQPCBS Mechanism* plus AMQP Claim-Based-security (cbs) seem like an ideal solution to handle large tokens and refresh them however there is no evidence that these specifications have been adopted yet. Thus, for now, we have to live with what we have today which tunnelling the token over the password field and wait until the token expires to reconnect with a new obtained token.

### Describe alternatives you've considered

In the stream protocol we allowed the exchange of the SASL frame after the connection was established. But I do not think we can do that in AMQP 1.0 protocol unless we want to support it in RabbitMQ. In that case, we would have to modify the amqp10_client too which would only work with RabbitMQ amqp1.0 plugin.


### Additional context

https://learn.microsoft.com/en-us/azure/service-bus-messaging/service-bus-amqp-protocol-guide#claims-based-authorization

### Other Considerations

AMQP 1.0 client libraries should be able to use this new SASL mechanism for OAuth2. Up until today, tokens have been tunneled via SASL PLAIN mechanism. Clients currently using this SASL PLAIN mechanism for oauth2 authentication have to change in order to continue authenticating with OAuth2 tokens and this mechanism must be readily available in the AMQP 1.0 client library.

| Library | OAuth2 support |
| ------------- | ------------- |
| Qpid Proton | Supports a different mechanism called [XOAUTH2](https://cwiki.apache.org/confluence/display/qpid/XOAUTH2+SASL+Mechanism+and+token+expiration) |
| Qpid JMS | Supports [XOAuth2](https://qpid.apache.org/releases/qpid-jms-2.4.0/docs/index.html#amqp-configuration-options) |
| Go AMQP (from [Azure](https://github.com/Azure/go-amqp?tab=readme-ov-file)) | Supports [XOAUTH2](https://pkg.go.dev/github.com/Azure/go-amqp#SASLTypeXOAUTH2) |
| Solace | Supports [XOAUTH2](https://docs.solace.com/Security/Client-Authentication-Overview.htm#OAuth) |
| Javascript [rhea](https://github.com/amqp/rhea) | [XOAUTH2](https://github.com/amqp/rhea/issues/393) SASL Mechanism | 
| [RabbitMQ.Client](https://github.com/rabbitmq/rabbitmq-dotnet-client?tab=readme-ov-file) for .Net | It currently uses PLAIN mechanism. We would need to implement a new mechanism. Either SASLCBS or XOAUTH2 |  
| Kafka(broker) | Supports [XOAUTHBEARER](https://docs.confluent.io/platform/current/kafka/authentication_sasl/authentication_sasl_oauth.html). I have read that this is the new name for XOAUTH2. I need to confirm it. This is the [RFC](https://datatracker.ietf.org/doc/html/rfc7628#section-3) | 
| .Net [amqpnetlite](https://github.com/Azure/amqpnetlite) | It only supports PLAIN, External and Anonymous. There is an opened [issue](https://github.com/Azure/amqpnetlite/issues/509) asking for oauth2 support |


Discussion / hints:
Which clients implement [AMQP Claims-based Security](https://docs.oasis-open.org/amqp/amqp-cbs/v1.0/csd01/amqp-cbs-v1.0-csd01.html) today?
@ansd so far i have not found any

Fix patch:
diff --git a/deps/rabbit/src/rabbit_fifo.erl b/deps/rabbit/src/rabbit_fifo.erl
index c99b361c21c0..6a61d1d2e87f 100644
--- a/deps/rabbit/src/rabbit_fifo.erl
+++ b/deps/rabbit/src/rabbit_fifo.erl
@@ -853,6 +853,8 @@ overview(#?STATE{consumers = Cons,
     Conf = #{name => Cfg#cfg.name,
              resource => Cfg#cfg.resource,
              dead_lettering_enabled => undefined =/= Cfg#cfg.dead_letter_handler,
+             dead_letter_handler => Cfg#cfg.dead_letter_handler,
+             overflow_strategy => Cfg#cfg.overflow_strategy,
              max_length => Cfg#cfg.max_length,
              max_bytes => Cfg#cfg.max_bytes,
              consumer_strategy => Cfg#cfg.consumer_strategy,
diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl
index 6c967b396d7a..67c308a1810a 100644
--- a/deps/rabbit/src/rabbit_quorum_queue.erl
+++ b/deps/rabbit/src/rabbit_quorum_queue.erl
@@ -316,9 +316,8 @@ declare_queue_error(Error, Queue, Leader, ActingUser) ->
 ra_machine(Q) ->
     {module, rabbit_fifo, ra_machine_config(Q)}.
 
-ra_machine_config(Q) when ?is_amqqueue(Q) ->
+gather_policy_config(Q, IsQueueDeclaration) ->
     QName = amqqueue:get_name(Q),
-    {Name, _} = amqqueue:get_pid(Q),
     %% take the minimum value of the policy and the queue arg if present
     MaxLength = args_policy_lookup(<<"max-length">>, fun min/2, Q),
     OverflowBin = args_policy_lookup(<<"overflow">>, fun policy_has_precedence/2, Q),
@@ -327,28 +326,42 @@ ra_machine_config(Q) when ?is_amqqueue(Q) ->
     DeliveryLimit = case args_policy_lookup(<<"delivery-limit">>,
                                             fun resolve_delivery_limit/2, Q) of
                         undefined ->
-                            rabbit_log:info("~ts: delivery_limit not set, defaulting to ~b",
-                                             [rabbit_misc:rs(QName), ?DEFAULT_DELIVERY_LIMIT]),
+                            case IsQueueDeclaration of
+                                true ->
+                                    rabbit_log:info(
+                                              "~ts: delivery_limit not set, defaulting to ~b",
+                                              [rabbit_misc:rs(QName), ?DEFAULT_DELIVERY_LIMIT]);
+                                false ->
+                                    ok
+                            end,
                             ?DEFAULT_DELIVERY_LIMIT;
                         DL ->
                             DL
                     end,
     Expires = args_policy_lookup(<<"expires">>, fun min/2, Q),
     MsgTTL = args_policy_lookup(<<"message-ttl">>, fun min/2, Q),
-    #{name => Name,
-      queue_resource => QName,
-      dead_letter_handler => dead_letter_handler(Q, Overflow),
-      become_leader_handler => {?MODULE, become_leader, [QName]},
+    DeadLetterHandler = dead_letter_handler(Q, Overflow),
+    #{dead_letter_handler => DeadLetterHandler,
       max_length => MaxLength,
       max_bytes => MaxBytes,
-      single_active_consumer_on => single_active_consumer_on(Q),
       delivery_limit => DeliveryLimit,
       overflow_strategy => Overflow,
-      created => erlang:system_time(millisecond),
       expires => Expires,
       msg_ttl => MsgTTL
      }.
 
+ra_machine_config(Q) when ?is_amqqueue(Q) ->
+    PolicyConfig = gather_policy_config(Q, true),
+    QName = amqqueue:get_name(Q),
+    {Name, _} = amqqueue:get_pid(Q),
+    PolicyConfig#{
+      name => Name,
+      queue_resource => QName,
+      become_leader_handler => {?MODULE, become_leader, [QName]},
+      single_active_consumer_on => single_active_consumer_on(Q),
+      created => erlang:system_time(millisecond)
+     }.
+
 resolve_delivery_limit(PolVal, ArgVal)
   when PolVal < 0 orelse ArgVal < 0 ->
     max(PolVal, ArgVal);
@@ -624,7 +637,9 @@ handle_tick(QName,
                           ok;
                       _ ->
                           ok
-                  end
+                  end,
+                  maybe_apply_policies(Q, Overview),
+                  ok
               catch
                   _:Err ->
                       rabbit_log:debug("~ts: handle tick failed with ~p",
@@ -708,6 +723,21 @@ system_recover(quorum_queues) ->
             ok
     end.
 
+maybe_apply_policies(Q, #{config := CurrentConfig}) ->
+    NewPolicyConfig = gather_policy_config(Q, false),
+
+    RelevantKeys = maps:keys(NewPolicyConfig),
+    CurrentPolicyConfig = maps:with(RelevantKeys, CurrentConfig),
+
+    ShouldUpdate = NewPolicyConfig =/= CurrentPolicyConfig,
+    case ShouldUpdate of
+        true ->
+            rabbit_log:debug("Re-applying policies to ~ts", [rabbit_misc:rs(amqqueue:get_name(Q))]),
+            policy_changed(Q),
+            ok;
+        false -> ok
+    end.
+
 -spec recover(binary(), [amqqueue:amqqueue()]) ->
     {[amqqueue:amqqueue()], [amqqueue:amqqueue()]}.
 recover(_Vhost, Queues) ->
@@ -2064,3 +2094,4 @@ file_handle_other_reservation() ->
 
 file_handle_release_reservation() ->
     ok.
+


Test patch:
diff --git a/deps/rabbit/test/quorum_queue_SUITE.erl b/deps/rabbit/test/quorum_queue_SUITE.erl
index 6f9969e6c879..10c3a8a02ad1 100644
--- a/deps/rabbit/test/quorum_queue_SUITE.erl
+++ b/deps/rabbit/test/quorum_queue_SUITE.erl
@@ -96,6 +96,7 @@ groups() ->
                                             force_shrink_member_to_current_member,
                                             force_all_queues_shrink_member_to_current_member,
                                             force_vhost_queues_shrink_member_to_current_member,
+                                            policy_repair,
                                             gh_12635
                                            ]
                        ++ all_tests()},
@@ -1301,6 +1302,161 @@ force_vhost_queues_shrink_member_to_current_member(Config) ->
         ?assertEqual(3, length(Nodes0))
     end || Q <- QQs, VHost <- VHosts].
 
+
+% Tests that, if the process of a QQ is dead in the moment of declaring a policy
+% that affects such queue, when the process is made available again, the policy
+% will eventually get applied. (https://github.com/rabbitmq/rabbitmq-server/issues/7863)
+policy_repair(Config) ->
+    [Server0, _Server1, _Server2] = Servers =
+        rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server0),
+    #'confirm.select_ok'{} = amqp_channel:call(Ch, #'confirm.select'{}),
+
+    QQ = ?config(queue_name, Config),
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+    RaName = ra_name(QQ),
+    ExpectedMaxLength1 = 10,
+    Priority1 = 1,
+    ok = rabbit_ct_broker_helpers:rpc(
+        Config,
+        0, 
+        rabbit_policy, 
+        set, 
+        [
+            <<"/">>, 
+            <<QQ/binary, "_1">>,
+            QQ, 
+            [{<<"max-length">>, ExpectedMaxLength1}, {<<"overflow">>, <<"reject-publish">>}], 
+            Priority1, 
+            <<"quorum_queues">>, 
+            <<"acting-user">>
+        ]),
+
+    % Wait for the policy to apply
+    QueryFun = fun rabbit_fifo:overview/1,
+    ?awaitMatch({ok, {_, #{config := #{max_length := ExpectedMaxLength1}}}, _},
+                rpc:call(Server0, ra, local_query, [RaName, QueryFun]),
+                ?DEFAULT_AWAIT),
+
+    % Check the policy has been applied
+    %   Insert MaxLength1 + some messages but after consuming all messages only 
+    %   MaxLength1 are retrieved.
+    %   Checking twice to ensure consistency
+    publish_confirm_many(Ch, QQ, ExpectedMaxLength1 + 1),
+    % +1 because QQs let one pass
+    wait_for_messages_ready(Servers, RaName, ExpectedMaxLength1 + 1), 
+    fail = publish_confirm(Ch, QQ),
+    fail = publish_confirm(Ch, QQ),
+    consume_all(Ch, QQ),
+
+    % Set higher priority policy, allowing more messages
+    ExpectedMaxLength2 = 20,
+    Priority2 = 2,
+    ok = rabbit_ct_broker_helpers:rpc(
+        Config,
+        0, 
+        rabbit_policy, 
+        set, 
+        [
+            <<"/">>, 
+            <<QQ/binary, "_2">>,
+            QQ, 
+            [{<<"max-length">>, ExpectedMaxLength2}, {<<"overflow">>, <<"reject-publish">>}], 
+            Priority2,
+            <<"quorum_queues">>, 
+            <<"acting-user">>
+        ]),
+
+    % Wait for the policy to apply
+    ?awaitMatch({ok, {_, #{config := #{max_length := ExpectedMaxLength2}}}, _},
+                rpc:call(Server0, ra, local_query, [RaName, QueryFun]),
+                ?DEFAULT_AWAIT),
+
+    % Check the policy has been applied
+    %   Insert MaxLength2 + some messages but after consuming all messages only 
+    %   MaxLength2 are retrieved.
+    %   Checking twice to ensure consistency.
+    % + 1 because QQs let one pass
+    publish_confirm_many(Ch, QQ, ExpectedMaxLength2 + 1),
+    wait_for_messages_ready(Servers, RaName, ExpectedMaxLength2 + 1),
+    fail = publish_confirm(Ch, QQ),
+    fail = publish_confirm(Ch, QQ),
+    consume_all(Ch, QQ),
+
+    % Ensure the queue process is unavailable
+    lists:foreach(fun(Srv) -> ensure_qq_proc_dead(Config, Srv, RaName) end, Servers),
+
+    % Add policy with higher priority, allowing even more messages.
+    ExpectedMaxLength3 = 30,
+    Priority3 = 3,
+    ok = rabbit_ct_broker_helpers:rpc(
+        Config,
+        0, 
+        rabbit_policy, 
+        set, 
+        [
+            <<"/">>, 
+            <<QQ/binary, "_3">>,
+            QQ, 
+            [{<<"max-length">>, ExpectedMaxLength3}, {<<"overflow">>, <<"reject-publish">>}], 
+            Priority3, 
+            <<"quorum_queues">>, 
+            <<"acting-user">>
+        ]),
+
+    % Restart the queue process.
+    {ok, Queue} = 
+        rabbit_ct_broker_helpers:rpc(
+            Config,
+            0,
+            rabbit_amqqueue,
+            lookup, 
+            [{resource, <<"/">>, queue, QQ}]),
+    lists:foreach(
+        fun(Srv) ->
+            rabbit_ct_broker_helpers:rpc(
+                Config,
+                Srv,
+                rabbit_quorum_queue,
+                recover,
+                [foo, [Queue]]
+            )
+        end,
+        Servers),
+
+    % Wait for the queue to be available again. 
+    lists:foreach(fun(Srv) ->
+        rabbit_ct_helpers:await_condition(
+            fun () ->
+                is_pid( 
+                    rabbit_ct_broker_helpers:rpc(
+                        Config,
+                        Srv,
+                        erlang,
+                        whereis,
+                        [RaName]))
+            end)
+        end,
+        Servers),
+
+    % Wait for the policy to apply
+    ?awaitMatch({ok, {_, #{config := #{max_length := ExpectedMaxLength3}}}, _},
+                rpc:call(Server0, ra, local_query, [RaName, QueryFun]),
+                ?DEFAULT_AWAIT),
+
+    % Check the policy has been applied
+    %   Insert MaxLength3 + some messages but after consuming all messages only 
+    %   MaxLength3 are retrieved.
+    %   Checking twice to ensure consistency.
+    % + 1 because QQs let one pass
+    publish_confirm_many(Ch, QQ, ExpectedMaxLength3 + 1),
+    wait_for_messages_ready(Servers, RaName, ExpectedMaxLength3 + 1),
+    fail = publish_confirm(Ch, QQ),
+    fail = publish_confirm(Ch, QQ),
+    consume_all(Ch, QQ).
+
+
 gh_12635(Config) ->
     % https://github.com/rabbitmq/rabbitmq-server/issues/12635
     [Server0, _Server1, Server2] =
@@ -4395,3 +4551,28 @@ lists_interleave([Item | Items], List)
     {Left, Right} = lists:split(2, List),
     Left ++ [Item | lists_interleave(Items, Right)].
 
+publish_confirm_many(Ch, Queue, Count) ->
+    lists:foreach(fun(_) -> publish_confirm(Ch, Queue) end, lists:seq(1, Count)).
+
+consume_all(Ch, QQ) ->
+    Consume = fun C(Acc) ->
+        case amqp_channel:call(Ch, #'basic.get'{queue = QQ}) of
+            {#'basic.get_ok'{}, Msg} ->
+                C([Msg | Acc]);
+            _ ->
+                Acc
+        end
+    end,
+    Consume([]).
+
+ensure_qq_proc_dead(Config, Server, RaName) ->
+    case rabbit_ct_broker_helpers:rpc(Config, Server, erlang, whereis, [RaName]) of
+        undefined ->
+            ok;
+        Pid ->
+            rabbit_ct_broker_helpers:rpc(Config, Server, erlang, exit, [Pid, kill]),
+            %% Give some time for the supervisor to restart the process
+            timer:sleep(500),
+            ensure_qq_proc_dead(Config, Server, RaName)
+    end.
+