Add test for AMQP 1.0 clients using OAuth token

(cherry picked from commit 0397035)

Author: ansd

deps/rabbitmq_auth_backend_oauth2/BUILD.bazel

@@ -151,6 +151,7 @@ rabbitmq_integration_suite(
     ],
     runtime_deps = [
         "//deps/oauth2_client:erlang_app",
+        "//deps/rabbitmq_amqp_client:erlang_app",
         "@emqtt//:erlang_app",
     ],
 )

deps/rabbitmq_auth_backend_oauth2/Makefile

@@ -8,7 +8,7 @@ export BUILD_WITHOUT_QUIC
 LOCAL_DEPS = inets public_key
 BUILD_DEPS = rabbit_common
 DEPS = rabbit cowlib jose base64url oauth2_client
-TEST_DEPS = cowboy rabbitmq_web_dispatch rabbitmq_ct_helpers rabbitmq_ct_client_helpers amqp_client rabbitmq_web_mqtt emqtt
+TEST_DEPS = cowboy rabbitmq_web_dispatch rabbitmq_ct_helpers rabbitmq_ct_client_helpers amqp_client rabbitmq_web_mqtt emqtt rabbitmq_amqp_client
 
 PLT_APPS += rabbitmqctl
 

deps/rabbitmq_auth_backend_oauth2/test/system_SUITE.erl

@@ -44,9 +44,10 @@ groups() ->
                        test_failed_connection_with_a_token_with_insufficient_vhost_permission,
                        test_failed_connection_with_a_token_with_insufficient_resource_permission,
                        more_than_one_resource_server_id_not_allowed_in_one_token,
+                       mqtt_expired_token,
                        mqtt_expirable_token,
                        web_mqtt_expirable_token,
-                       mqtt_expired_token
+                       amqp_expirable_token
                       ]},
 
      {token_refresh, [], [
@@ -433,6 +434,18 @@ mqtt(Config) ->
     ok = emqtt:disconnect(Sub),
     ok = emqtt:disconnect(Pub).
 
+mqtt_expired_token(Config) ->
+    {_Algo, Token} = generate_expired_token(Config),
+    Opts = [{port, rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_mqtt)},
+            {proto_ver, v5},
+            {username, <<"">>},
+            {password, Token}],
+    ClientId = atom_to_binary(?FUNCTION_NAME),
+    {ok, C} = emqtt:start_link([{clientid, ClientId} | Opts]),
+    true = unlink(C),
+    ?assertMatch({error, {bad_username_or_password, _}},
+                 emqtt:connect(C)).
+
 mqtt_expirable_token(Config) ->
     mqtt_expirable_token0(tcp_port_mqtt,
                           [],
@@ -487,17 +500,49 @@ mqtt_expirable_token0(Port, AdditionalOpts, Connect, Config) ->
     after Millis * 2 -> ct:fail("missing DISCONNECT packet from server")
     end.
 
-mqtt_expired_token(Config) ->
-    {_Algo, Token} = generate_expired_token(Config),
-    Opts = [{port, rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_mqtt)},
-            {proto_ver, v5},
-            {username, <<"">>},
-            {password, Token}],
-    ClientId = atom_to_binary(?FUNCTION_NAME),
-    {ok, C} = emqtt:start_link([{clientid, ClientId} | Opts]),
-    true = unlink(C),
-    ?assertMatch({error, {bad_username_or_password, _}},
-                 emqtt:connect(C)).
+amqp_expirable_token(Config) ->
+    {ok, _} = application:ensure_all_started(rabbitmq_amqp_client),
+
+    Seconds = 4,
+    Millis = Seconds * 1000,
+    {_Algo, Token} = generate_expirable_token(Config,
+                                              [<<"rabbitmq.configure:*/*">>,
+                                               <<"rabbitmq.write:*/*">>,
+                                               <<"rabbitmq.read:*/*">>],
+                                              Seconds),
+
+    %% Send and receive a message via AMQP 1.0.
+    QName = atom_to_binary(?FUNCTION_NAME),
+    Address = rabbitmq_amqp_address:queue(QName),
+    Host = ?config(rmq_hostname, Config),
+    Port = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp),
+    OpnConf = #{address => Host,
+                port => Port,
+                container_id => <<"my container">>,
+                sasl => {plain, <<"">>, Token}},
+    {ok, Connection} = amqp10_client:open_connection(OpnConf),
+    {ok, Session} = amqp10_client:begin_session_sync(Connection),
+    {ok, LinkPair} = rabbitmq_amqp_client:attach_management_link_pair_sync(Session, <<"my link pair">>),
+    {ok, _} = rabbitmq_amqp_client:declare_queue(LinkPair, QName, #{}),
+    {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"my sender">>, Address),
+    receive {amqp10_event, {link, Sender, credited}} -> ok
+    after 5000 -> ct:fail({missing_event, ?LINE})
+    end,
+    Body = <<"hey">>,
+    Msg0 = amqp10_msg:new(<<"tag">>, Body),
+    ok = amqp10_client:send_msg(Sender, Msg0),
+    {ok, Receiver} = amqp10_client:attach_receiver_link(Session, <<"my receiver">>, Address),
+    {ok, Msg} = amqp10_client:get_msg(Receiver),
+    ?assertEqual([Body], amqp10_msg:body(Msg)),
+
+    %% In 4 seconds from now, we expect that RabbitMQ disconnects us because our token expired.
+    receive {amqp10_event,
+             {connection, Connection,
+              {closed, {unauthorized_access, <<"credential expired">>}}}} ->
+                ok
+    after Millis * 2 ->
+              ct:fail("server did not close our connection")
+    end.
 
 test_successful_connection_with_complex_claim_as_a_map(Config) ->
     {_Algo, Token} = generate_valid_token_with_extra_fields(