Merge pull request #9187 from rabbitmq/update-secret

Update secret on the stream protocol

Author: michaelklishin

deps/rabbitmq_stream/src/rabbit_stream_reader.erl

@@ -1461,8 +1461,7 @@ handle_frame_pre_auth(Transport,
                                                                   [Username]),
                                     {C1#stream_connection{connection_step =
                                                               failure},
-                                     {sasl_authenticate,
-                                      ?RESPONSE_SASL_AUTHENTICATION_FAILURE_LOOPBACK,
+                                     {sasl_authenticate, ?RESPONSE_SASL_AUTHENTICATION_FAILURE_LOOPBACK,
                                       <<>>}}
                             end
                     end,
@@ -1643,6 +1642,100 @@ handle_frame_post_auth(Transport,
     rabbit_global_counters:increase_protocol_counter(stream,
                                                      ?PRECONDITION_FAILED, 1),
     {Connection0, State};
+
+handle_frame_post_auth(Transport,
+                       #stream_connection{user = #user{username = Username} = _User,
+                                          socket = S,
+                                          host = Host,
+                                          auth_mechanism = Auth_Mechanism,
+                                          authentication_state = AuthState,
+                                          resource_alarm = false} =
+                           C1,
+                       State,
+                       {request, CorrelationId,
+                        {sasl_authenticate, NewMechanism, NewSaslBin}}) ->
+    rabbit_log:debug("Open frame received sasl_authenticate for username '~ts'", [Username]),
+
+    Connection1 =
+      case Auth_Mechanism of
+        {NewMechanism, AuthMechanism} -> %% Mechanism is the same used during the pre-auth phase
+              {C2, CmdBody} =
+                  case AuthMechanism:handle_response(NewSaslBin, AuthState) of
+                      {refused, NewUsername, Msg, Args} ->
+                          rabbit_core_metrics:auth_attempt_failed(Host,
+                                                                  NewUsername,
+                                                                  stream),
+                          auth_fail(NewUsername, Msg, Args, C1, State),
+                          rabbit_log_connection:warning(Msg, Args),
+                          {C1#stream_connection{connection_step = failure},
+                           {sasl_authenticate,
+                            ?RESPONSE_AUTHENTICATION_FAILURE, <<>>}};
+                      {protocol_error, Msg, Args} ->
+                          rabbit_core_metrics:auth_attempt_failed(Host,
+                                                                  <<>>,
+                                                                  stream),
+                          notify_auth_result(none,
+                                             user_authentication_failure,
+                                             [{error,
+                                               rabbit_misc:format(Msg,
+                                                                  Args)}],
+                                             C1,
+                                             State),
+                          rabbit_log_connection:warning(Msg, Args),
+                          {C1#stream_connection{connection_step = failure},
+                           {sasl_authenticate, ?RESPONSE_SASL_ERROR, <<>>}};
+                      {challenge, Challenge, AuthState1} ->
+                          {C1#stream_connection{authentication_state = AuthState1,
+                                                connection_step = authenticating},
+                           {sasl_authenticate, ?RESPONSE_SASL_CHALLENGE,
+                            Challenge}};
+                      {ok, NewUser = #user{username = NewUsername}} ->
+                          case NewUsername of
+                            Username ->
+                                  rabbit_core_metrics:auth_attempt_succeeded(Host,
+                                                                             Username,
+                                                                             stream),
+                                  notify_auth_result(Username,
+                                                     user_authentication_success,
+                                                     [],
+                                                     C1,
+                                                     State),
+                                  rabbit_log:debug("Successfully updated secret for username '~ts'", [Username]),
+                                  {C1#stream_connection{user = NewUser,
+                                                        authentication_state = done,
+                                                        connection_step = authenticated},
+                                   {sasl_authenticate, ?RESPONSE_CODE_OK,
+                                    <<>>}};
+                              _ ->
+                                  rabbit_core_metrics:auth_attempt_failed(Host,
+                                                                          Username,
+                                                                          stream),
+                                  rabbit_log_connection:warning("Not allowed to change username '~ts'. Only password",
+                                                                [Username]),
+                                  {C1#stream_connection{connection_step =
+                                                            failure},
+                                   {sasl_authenticate,
+                                    ?RESPONSE_SASL_CANNOT_CHANGE_USERNAME,
+                                    <<>>}}
+                          end
+                  end,
+              Frame =
+                  rabbit_stream_core:frame({response, CorrelationId,
+                                            CmdBody}),
+              send(Transport, S, Frame),
+              C2;
+        {OtherMechanism, _} ->
+              rabbit_log_connection:warning("User '~ts' cannot change initial auth mechanism '~ts' for '~ts'",
+                                              [Username, NewMechanism, OtherMechanism]),
+              CmdBody =
+                {sasl_authenticate, ?RESPONSE_SASL_CANNOT_CHANGE_MECHANISM, <<>>},
+              Frame = rabbit_stream_core:frame({response, CorrelationId, CmdBody}),
+              send(Transport, S, Frame),
+              C1#stream_connection{connection_step = failure}
+      end,
+
+    {Connection1, State};
+
 handle_frame_post_auth(Transport,
                        #stream_connection{user = User,
                                           publishers = Publishers0,
@@ -3866,4 +3959,3 @@ stream_from_consumers(SubId, Consumers) ->
         _ ->
             undefined
     end.
-

deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl

@@ -39,6 +39,9 @@ groups() ->
        test_publish_v2,
        test_gc_consumers,
        test_gc_publishers,
+       test_update_secret,
+       cannot_update_username_after_authenticated,
+       cannot_use_another_authmechanism_when_updating_secret,
        unauthenticated_client_rejected_tcp_connected,
        timeout_tcp_connected,
        unauthenticated_client_rejected_peer_properties_exchanged,
@@ -48,7 +51,8 @@ groups() ->
        timeout_close_sent,
        max_segment_size_bytes_validation,
        close_connection_on_consumer_update_timeout,
-       set_filter_size]},
+       set_filter_size
+      ]},
      %% Run `test_global_counters` on its own so the global metrics are
      %% initialised to 0 for each testcase
      {single_node_1, [], [test_global_counters]},
@@ -132,6 +136,13 @@ end_per_group(_, Config) ->
     rabbit_ct_helpers:run_steps(Config,
                                 rabbit_ct_broker_helpers:teardown_steps()).
 
+init_per_testcase(test_update_secret = TestCase, Config) ->
+    rabbit_ct_helpers:testcase_started(Config, TestCase);
+
+init_per_testcase(cannot_update_username_after_authenticated = TestCase, Config) ->
+  ok = rabbit_ct_broker_helpers:add_user(Config, <<"other">>),
+  rabbit_ct_helpers:testcase_started(Config, TestCase);
+
 init_per_testcase(close_connection_on_consumer_update_timeout = TestCase, Config) ->
     ok = rabbit_ct_broker_helpers:rpc(Config,
                                       0,
@@ -142,6 +153,14 @@ init_per_testcase(close_connection_on_consumer_update_timeout = TestCase, Config
 init_per_testcase(TestCase, Config) ->
     rabbit_ct_helpers:testcase_started(Config, TestCase).
 
+end_per_testcase(test_update_secret = TestCase, Config) ->
+    ok = rabbit_ct_broker_helpers:change_password(Config, <<"guest">>, <<"guest">>),
+    rabbit_ct_helpers:testcase_finished(Config, TestCase);
+
+end_per_testcase(cannot_update_username_after_authenticated = TestCase, Config) ->
+    ok = rabbit_ct_broker_helpers:delete_user(Config, <<"other">>),
+    rabbit_ct_helpers:testcase_finished(Config, TestCase);
+
 end_per_testcase(filtering_ff = TestCase, Config) ->
     _ = rabbit_ct_broker_helpers:rpc(Config,
                                      0,
@@ -221,6 +240,34 @@ test_stream(Config) ->
     test_server(gen_tcp, Stream, Config),
     ok.
 
+test_update_secret(Config) ->
+    Transport = gen_tcp,
+    {S, C0} = connect_and_authenticate(Transport, Config),
+    rabbit_ct_broker_helpers:change_password(Config, <<"guest">>, <<"password">>),
+    C1 = expect_successful_authentication(
+      try_authenticate(Transport, S, C0, <<"PLAIN">>, <<"guest">>, <<"password">>)),
+    _C2 = test_close(Transport, S, C1),
+    closed = wait_for_socket_close(Transport, S, 10),
+    ok.
+
+cannot_update_username_after_authenticated(Config) ->
+    {S, C0} = connect_and_authenticate(gen_tcp, Config),
+    C1 = expect_unsuccessful_authentication(
+      try_authenticate(gen_tcp, S, C0, <<"PLAIN">>, <<"other">>, <<"other">>),
+        ?RESPONSE_SASL_CANNOT_CHANGE_USERNAME),
+    _C2 = test_close(gen_tcp, S, C1),
+    closed = wait_for_socket_close(gen_tcp, S, 10),
+    ok.
+
+cannot_use_another_authmechanism_when_updating_secret(Config) ->
+    {S, C0} = connect_and_authenticate(gen_tcp, Config),
+    C1 = expect_unsuccessful_authentication(
+      try_authenticate(gen_tcp, S, C0, <<"EXTERNAL">>, <<"guest">>, <<"new_password">>),
+        ?RESPONSE_SASL_CANNOT_CHANGE_MECHANISM),
+    _C2 = test_close(gen_tcp, S, C1),
+    closed = wait_for_socket_close(gen_tcp, S, 10),
+    ok.
+
 test_stream_tls(Config) ->
     Stream = atom_to_binary(?FUNCTION_NAME, utf8),
     test_server(ssl, Stream, Config),
@@ -577,23 +624,43 @@ get_node_name(Config) ->
 get_node_name(Config, Node) ->
     rabbit_ct_broker_helpers:get_node_config(Config, Node, nodename).
 
+get_port(Transport, Config) ->
+  case Transport of
+      gen_tcp ->
+          get_stream_port(Config);
+      ssl ->
+          application:ensure_all_started(ssl),
+          get_stream_port_tls(Config)
+  end.
+get_opts(Transport) ->
+  case Transport of
+      gen_tcp ->
+          [{active, false}, {mode, binary}];
+      ssl ->
+          [{active, false}, {mode, binary}, {verify, verify_none}]
+  end.
+
+connect_and_authenticate(Transport, Config) ->
+  Port = get_port(Transport, Config),
+  Opts = get_opts(Transport),
+  {ok, S} = Transport:connect("localhost", Port, Opts),
+  C0 = rabbit_stream_core:init(0),
+  C1 = test_peer_properties(Transport, S, C0),
+  {S, test_authenticate(Transport, S, C1)}.
+
+try_authenticate(Transport, S, C, AuthMethod, Username, Password) ->
+  case AuthMethod of
+    <<"PLAIN">> ->
+      plain_sasl_authenticate(Transport, S, C, Username, Password);
+    _ ->
+      Null = 0,
+      sasl_authenticate(Transport, S, C, AuthMethod, <<Null:8, Username/binary, Null:8, Password/binary>>)
+  end.
+
 test_server(Transport, Stream, Config) ->
     QName = rabbit_misc:r(<<"/">>, queue, Stream),
-    Port =
-        case Transport of
-            gen_tcp ->
-                get_stream_port(Config);
-            ssl ->
-                application:ensure_all_started(ssl),
-                get_stream_port_tls(Config)
-        end,
-    Opts =
-        case Transport of
-            gen_tcp ->
-                [{active, false}, {mode, binary}];
-            ssl ->
-                [{active, false}, {mode, binary}, {verify, verify_none}]
-        end,
+    Port = get_port(Transport, Config),
+    Opts = get_opts(Transport),
     {ok, S} =
         Transport:connect("localhost", Port, Opts),
     C0 = rabbit_stream_core:init(0),
@@ -652,6 +719,9 @@ test_peer_properties(Transport, S, C0) ->
     C.
 
 test_authenticate(Transport, S, C0) ->
+  tune(Transport, S, test_plain_sasl_authenticate(Transport, S, sasl_handshake(Transport, S, C0))).
+
+sasl_handshake(Transport, S, C0) ->
     SaslHandshakeFrame =
         rabbit_stream_core:frame({request, 1, sasl_handshake}),
     ok = Transport:send(S, SaslHandshakeFrame),
@@ -664,18 +734,33 @@ test_authenticate(Transport, S, C0) ->
         _ ->
             ct:fail("invalid cmd ~tp", [Cmd])
     end,
+    C1.
+
+test_plain_sasl_authenticate(Transport, S, C1) ->
+  expect_successful_authentication(plain_sasl_authenticate(Transport, S, C1)).
 
-    Username = <<"guest">>,
-    Password = <<"guest">>,
+plain_sasl_authenticate(Transport, S, C1) ->
+    plain_sasl_authenticate(Transport, S, C1, <<"guest">>, <<"guest">>).
+
+plain_sasl_authenticate(Transport, S, C1, Username, Password) ->
     Null = 0,
-    PlainSasl = <<Null:8, Username/binary, Null:8, Password/binary>>,
+    sasl_authenticate(Transport, S, C1, <<"PLAIN">>, <<Null:8, Username/binary, Null:8, Password/binary>>).
+
+expect_successful_authentication({SaslAuth, C2} = _SaslReponse) ->
+  {response, 2, {sasl_authenticate, ?RESPONSE_CODE_OK}} = SaslAuth,
+  C2.
+expect_unsuccessful_authentication({SaslAuth, C2} = _SaslReponse, ExpectedError) ->
+  {response, 2, {sasl_authenticate, ExpectedError}} = SaslAuth,
+  C2.
 
+sasl_authenticate(Transport, S, C1, AuthMethod, AuthBody) ->
     SaslAuthenticateFrame =
         rabbit_stream_core:frame({request, 2,
-                                  {sasl_authenticate, Plain, PlainSasl}}),
+                                  {sasl_authenticate, AuthMethod, AuthBody}}),
     ok = Transport:send(S, SaslAuthenticateFrame),
-    {SaslAuth, C2} = receive_commands(Transport, S, C1),
-    {response, 2, {sasl_authenticate, ?RESPONSE_CODE_OK}} = SaslAuth,
+    receive_commands(Transport, S, C1).
+
+tune(Transport, S, C2) ->
     {Tune, C3} = receive_commands(Transport, S, C2),
     {tune, ?DEFAULT_FRAME_MAX, ?DEFAULT_HEARTBEAT} = Tune,
 
@@ -816,9 +901,9 @@ test_unsubscribe(Transport, Socket, SubscriptionId, C0) ->
     C.
 
 test_deliver(Transport, S, SubscriptionId, COffset, Body, C0) ->
-    ct:pal("test_deliver ", []),
     {{deliver, SubscriptionId, Chunk}, C} =
         receive_commands(Transport, S, C0),
+    ct:pal("test_deliver ~p", [Chunk]),
     <<5:4/unsigned,
       0:4/unsigned,
       0:8,
@@ -838,9 +923,9 @@ test_deliver(Transport, S, SubscriptionId, COffset, Body, C0) ->
     C.
 
 test_deliver_v2(Transport, S, SubscriptionId, COffset, Body, C0) ->
-    ct:pal("test_deliver ", []),
     {{deliver_v2, SubscriptionId, _CommittedOffset, Chunk}, C} =
         receive_commands(Transport, S, C0),
+    ct:pal("test_deliver_v2 ~p", [Chunk]),
     <<5:4/unsigned,
       0:4/unsigned,
       0:8,

deps/rabbitmq_stream_common/include/rabbit_stream.hrl

@@ -52,7 +52,8 @@
 -define(RESPONSE_CODE_PRECONDITION_FAILED, 17).
 -define(RESPONSE_CODE_PUBLISHER_DOES_NOT_EXIST, 18).
 -define(RESPONSE_CODE_NO_OFFSET, 19).
-
+-define(RESPONSE_SASL_CANNOT_CHANGE_MECHANISM, 20).
+-define(RESPONSE_SASL_CANNOT_CHANGE_USERNAME, 21).
 
 -define(OFFSET_TYPE_NONE, 0).
 -define(OFFSET_TYPE_FIRST, 1).

deps/rabbitmq_stream_common/src/rabbit_stream_core.erl

@@ -58,7 +58,9 @@
     ?RESPONSE_CODE_ACCESS_REFUSED |
     ?RESPONSE_CODE_PRECONDITION_FAILED |
     ?RESPONSE_CODE_PUBLISHER_DOES_NOT_EXIST |
-    ?RESPONSE_CODE_NO_OFFSET.
+    ?RESPONSE_CODE_NO_OFFSET |
+    ?RESPONSE_SASL_CANNOT_CHANGE_MECHANISM |
+    ?RESPONSE_SASL_CANNOT_CHANGE_USERNAME .
 -type error_code() :: response_code().
 -type sequence() :: non_neg_integer().
 -type credit() :: non_neg_integer().