Merge pull request #8492 from rabbitmq/mergify/bp/v3.10.x/pr-8490

Fix SASL external authentication in stream plugin (backport #8488) (backport #8489) (backport #8490)

Author: michaelklishin

deps/rabbitmq_stream/src/rabbit_stream_reader.erl

@@ -59,6 +59,7 @@
          %% client port
          peer_port,
          auth_mechanism,
+         authentication_state :: any(),
          connected_at :: integer(),
          helper_sup :: pid(),
          socket :: rabbit_net:socket(),
@@ -70,7 +71,6 @@
          stream_leaders :: #{stream() => pid()},
          stream_subscriptions :: #{stream() => [subscription_id()]},
          credits :: atomics:atomics_ref(),
-         authentication_state :: atom(),
          user :: undefined | #user{},
          virtual_host :: undefined | binary(),
          connection_step ::
@@ -1188,8 +1188,6 @@ handle_frame_pre_auth(Transport,
                                    ServerProperties}}),
     send(Transport, S, Frame),
     {Connection#stream_connection{client_properties = ClientProperties,
-                                  authentication_state =
-                                      peer_properties_exchanged,
                                   connection_step = peer_properties_exchanged},
      State};
 handle_frame_pre_auth(Transport,
@@ -1252,13 +1250,8 @@ handle_frame_pre_auth(Transport,
                             {C1#stream_connection{connection_step = failure},
                              {sasl_authenticate, ?RESPONSE_SASL_ERROR, <<>>}};
                         {challenge, Challenge, AuthState1} ->
-                            rabbit_core_metrics:auth_attempt_succeeded(RemoteAddress,
-                                                                       <<>>,
-                                                                       stream),
-                            {C1#stream_connection{authentication_state =
-                                                      AuthState1,
-                                                  connection_step =
-                                                      authenticating},
+                            {C1#stream_connection{authentication_state = AuthState1,
+                                                  connection_step = authenticating},
                              {sasl_authenticate, ?RESPONSE_SASL_CHALLENGE,
                               Challenge}};
                         {ok, User = #user{username = Username}} ->
@@ -1275,11 +1268,9 @@ handle_frame_pre_auth(Transport,
                                                        [],
                                                        C1,
                                                        State),
-                                    {C1#stream_connection{authentication_state =
-                                                              done,
-                                                          user = User,
-                                                          connection_step =
-                                                              authenticated},
+                                    {C1#stream_connection{user = User,
+                                                          authentication_state = done,
+                                                          connection_step = authenticated},
                                      {sasl_authenticate, ?RESPONSE_CODE_OK,
                                       <<>>}};
                                 not_allowed ->