Report authentication failures as returned by the broker

deps/amqp_client/include/amqp_client_internal.hrl

@@ -28,8 +28,10 @@
 -define(LOG_WARN(Format, Args), error_logger:warning_msg(Format, Args)).
 -define(LOG_ERR(Format, Args),  error_logger:error_msg(Format, Args)).
 
--define(CLIENT_CAPABILITIES, [{<<"publisher_confirms">>,         bool, true},
+-define(CLIENT_CAPABILITIES,
+    [{<<"publisher_confirms">>,           bool, true},
      {<<"exchange_exchange_bindings">>,   bool, true},
      {<<"basic.nack">>,                   bool, true},
      {<<"consumer_cancel_notify">>,       bool, true},
-                              {<<"connection.blocked">>,         bool, true}]).
+     {<<"connection.blocked">>,           bool, true},
+     {<<"authentication_failure_close">>, bool, true}]).

deps/amqp_client/src/amqp_gen_connection.erl

@@ -181,6 +181,8 @@ handle_call(connect, _From,
     case Mod:connect(AmqpParams, SIF, ChMgr, MState) of
         {ok, Params} ->
             {reply, {ok, self()}, after_connect(Params, State1)};
+        {closing, #amqp_error{name = access_refused} = AmqpError, Error} ->
+            {stop, {shutdown, AmqpError}, Error, State1};
         {closing, Params, #amqp_error{} = AmqpError, Error} ->
             server_misbehaved(self(), AmqpError),
             {reply, Error, after_connect(Params, State1)};

deps/amqp_client/src/amqp_network_connection.erl

@@ -183,7 +183,11 @@ network_handshake(AmqpParams = #amqp_params_network{virtual_host = VHost},
                                 mechanisms = Mechanisms} =
         handshake_recv('connection.start'),
     ok = check_version(Start),
-    Tune = login(AmqpParams, Mechanisms, State0),
+    case login(AmqpParams, Mechanisms, State0) of
+        {closing, #amqp_error{}, _Error} = Err ->
+            do(#'connection.close_ok'{}, State0),
+            Err;
+        Tune ->
             {TuneOk, ChannelMax, State1} = tune(Tune, AmqpParams, SHF, State0),
             do2(TuneOk, State1),
             do2(#'connection.open'{virtual_host = VHost}, State1),
@@ -192,6 +196,7 @@ network_handshake(AmqpParams = #amqp_params_network{virtual_host = VHost},
                 #'connection.open_ok'{}                     -> {ok, Params};
                 {closing, #amqp_error{} = AmqpError, Error} -> {closing, Params,
                                                                 AmqpError, Error}
+            end
     end.
 
 check_version(#'connection.start'{version_major = ?PROTOCOL_VERSION_MAJOR,
@@ -265,7 +270,14 @@ login_loop(Mech, MState0, Params, State) ->
         #'connection.secure'{challenge = Challenge} ->
             {Resp, MState1} = Mech(Challenge, Params, MState0),
             do2(#'connection.secure_ok'{response = Resp}, State),
-            login_loop(Mech, MState1, Params, State)
+            login_loop(Mech, MState1, Params, State);
+        #'connection.close'{reply_code = ?ACCESS_REFUSED,
+                            reply_text = ExplanationBin} ->
+            Explanation = binary_to_list(ExplanationBin),
+            {closing,
+             #amqp_error{name        = access_refused,
+                         explanation = Explanation},
+             {error, {auth_failure, Explanation}}}
     end.
 
 client_properties(UserProperties) ->
@@ -291,6 +303,8 @@ handshake_recv(Expecting) ->
                     Method;
                 {'connection.tune', 'connection.secure'} ->
                     Method;
+                {'connection.tune', 'connection.close'} ->
+                    Method;
                 {'connection.open_ok', _} ->
                     {closing,
                      #amqp_error{name        = command_invalid,
@@ -304,7 +318,7 @@ handshake_recv(Expecting) ->
             end;
         socket_closed ->
             case Expecting of
-                'connection.tune'    -> exit(auth_failure);
+                'connection.tune'    -> exit({auth_failure, "Disconnected"});
                 'connection.open_ok' -> exit(access_refused);
                 _                    -> exit({socket_closed_unexpectedly,
                                               Expecting})

deps/amqp_client/test/negative_test_util.erl

@@ -178,11 +178,11 @@ assert_down_with_error(MonitorRef, CodeAtom) ->
 
 non_existent_user_test() ->
     Params = [{username, test_util:uuid()}, {password, test_util:uuid()}],
-    ?assertMatch({error, auth_failure}, test_util:new_connection(Params)).
+    ?assertMatch({error, {auth_failure, _}}, test_util:new_connection(Params)).
 
 invalid_password_test() ->
     Params = [{username, <<"guest">>}, {password, test_util:uuid()}],
-    ?assertMatch({error, auth_failure}, test_util:new_connection(Params)).
+    ?assertMatch({error, {auth_failure, _}}, test_util:new_connection(Params)).
 
 non_existent_vhost_test() ->
     Params = [{virtual_host, test_util:uuid()}],