Merge pull request #14403 from rabbitmq/stream-close-connection-if-unauthorized-vhost-after-secret-update

michaelklishin · web-flow · commit c77751e6200c · 2025-08-19T14:56:15.000-04:00
Close stream connection if unauthorized vhost after secret update
diff --git a/deps/rabbitmq_stream/src/rabbit_stream_reader.erl b/deps/rabbitmq_stream/src/rabbit_stream_reader.erl
@@ -697,6 +697,12 @@ open(info, {OK, S, Data},
             {next_state, close_sent,
              StatemData#statem_data{connection = Connection1,
                                     connection_state = State1}};
+        failure ->
+            _ = demonitor_all_streams(Connection),
+            ?LOG_INFO("Force closing stream connection ~tp because of "
+                      "transition to invalid state",
+                      [self()]),
+            {stop, {shutdown, <<"Invalid state">>}};
         _ ->
             State2 =
                 case Blocked of
@@ -1502,7 +1508,6 @@ handle_frame_pre_auth(Transport,
                   send(Transport, S, F),
                   Connection#stream_connection{connection_step = failure}
         end,
-
     {Connection1, State};
 handle_frame_pre_auth(_Transport, Connection, State, heartbeat) ->
     ?LOG_DEBUG("Received heartbeat frame pre auth"),
@@ -1586,6 +1591,7 @@ handle_frame_post_auth(Transport,
                                                                   stream),
                           auth_fail(NewUsername, Msg, Args, C1, S1),
                           ?LOG_WARNING(Msg, Args),
+                          silent_close_delay(),
                           {C1#stream_connection{connection_step = failure},
                            {sasl_authenticate,
                             ?RESPONSE_AUTHENTICATION_FAILURE, <<>>}};
@@ -1610,27 +1616,15 @@ handle_frame_post_auth(Transport,
                             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,
-                                                     S1),
-                                  ?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,
-                                    <<>>}};
+                              Username ->
+                                  complete_secret_update(NewUser, C1, S1);
                               _ ->
                                   rabbit_core_metrics:auth_attempt_failed(Host,
                                                                           Username,
                                                                           stream),
                                   ?LOG_WARNING("Not allowed to change username '~ts'. Only password",
                                                                 [Username]),
+                                  silent_close_delay(),
                                   {C1#stream_connection{connection_step =
                                                             failure},
                                    {sasl_authenticate,
@@ -1652,6 +1646,7 @@ handle_frame_post_auth(Transport,
         {OtherMechanism, _} ->
               ?LOG_WARNING("User '~ts' cannot change initial auth mechanism '~ts' for '~ts'",
                                               [Username, NewMechanism, OtherMechanism]),
+              silent_close_delay(),
               CmdBody =
                 {sasl_authenticate, ?RESPONSE_SASL_CANNOT_CHANGE_MECHANISM, <<>>},
               Frame = rabbit_stream_core:frame({response, CorrelationId, CmdBody}),
@@ -2774,6 +2769,32 @@ handle_frame_post_auth(Transport,
     increase_protocol_counter(?UNKNOWN_FRAME),
     {Connection#stream_connection{connection_step = close_sent}, State}.
 
+complete_secret_update(NewUser = #user{username = Username},
+                       #stream_connection{host = Host,
+                                          socket = S,
+                                          virtual_host = VH} = C1, S1) ->
+    notify_auth_result(Username, user_authentication_success, [], C1, S1),
+    rabbit_core_metrics:auth_attempt_succeeded(Host, Username, stream),
+    ?LOG_DEBUG("Stream connection has successfully checked updated secret (token) for username '~ts'",
+               [Username]),
+    try
+        ?LOG_DEBUG("Stream connection: will verify virtual host access after secret (token) update"),
+        rabbit_access_control:check_vhost_access(NewUser, VH, {socket, S}, #{}),
+        ?LOG_DEBUG("Stream connection: successfully re-verified virtual host access"),
+
+        {C1#stream_connection{user = NewUser,
+                              authentication_state = done,
+                              connection_step = authenticated},
+         {sasl_authenticate, ?RESPONSE_CODE_OK,
+          <<>>}}
+    catch exit:#amqp_error{explanation = Explanation} ->
+              ?LOG_WARNING("Stream connection no longer has the permissions to access its target virtual host ('~ts') after a secret (token) update: ~ts",
+                           [VH, Explanation]),
+              silent_close_delay(),
+              {C1#stream_connection{connection_step = failure},
+               {sasl_authenticate, ?RESPONSE_VHOST_ACCESS_FAILURE, <<>>}}
+    end.
+
 process_client_command_versions(C, []) ->
     C;
 process_client_command_versions(C, [H | T]) ->
diff --git a/deps/rabbitmq_stream/src/rabbit_stream_reader.hrl b/deps/rabbitmq_stream/src/rabbit_stream_reader.hrl
@@ -20,6 +20,10 @@
 -type publisher_reference() :: binary().
 -type subscription_id() :: byte().
 -type internal_id() :: integer().
+-type connection_step() :: tcp_connected | peer_properties_exchanged |
+                           authenticating | authenticated | tuning |
+                           tuned | opened | failure |
+                           closing | close_sent | closing_done.
 
 -record(publisher,
         {publisher_id :: publisher_id(),
@@ -75,8 +79,7 @@
          credits :: atomics:atomics_ref(),
          user :: undefined | #user{},
          virtual_host :: undefined | binary(),
-         connection_step ::
-             atom(), % tcp_connected, peer_properties_exchanged, authenticating, authenticated, tuning, tuned, opened, failure, closing, closing_done
+         connection_step :: connection_step(),
          frame_max :: integer(),
          heartbeat :: undefined | integer(),
          heartbeater :: any(),
diff --git a/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl b/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl
@@ -48,6 +48,8 @@ groups() ->
        test_update_secret,
        cannot_update_username_after_authenticated,
        cannot_use_another_authmechanism_when_updating_secret,
+       update_secret_should_close_connection_if_wrong_secret,
+       update_secret_should_close_connection_if_unauthorized_vhost,
        unauthenticated_client_rejected_tcp_connected,
        timeout_tcp_connected,
        unauthenticated_client_rejected_peer_properties_exchanged,
@@ -165,6 +167,12 @@ 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(update_secret_should_close_connection_if_unauthorized_vhost = TestCase,
+                  Config) ->
+  ok = rabbit_ct_broker_helpers:add_user(Config, <<"other">>),
+  ok = rabbit_ct_broker_helpers:set_full_permissions(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,
@@ -200,6 +208,11 @@ 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(update_secret_should_close_connection_if_unauthorized_vhost = TestCase,
+                 Config) ->
+    ok = rabbit_ct_broker_helpers:delete_user(Config, <<"other">>),
+    rabbit_ct_helpers:testcase_finished(Config, TestCase);
+
 end_per_testcase(close_connection_on_consumer_update_timeout = TestCase, Config) ->
     ok = rabbit_ct_broker_helpers:rpc(Config,
                                       0,
@@ -285,29 +298,53 @@ test_update_secret(Config) ->
     {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">>)),
+           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),
+    _C1 = expect_unsuccessful_authentication(
+            try_authenticate(gen_tcp, S, C0, <<"PLAIN">>, <<"other">>, <<"other">>),
+            ?RESPONSE_SASL_CANNOT_CHANGE_USERNAME),
     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),
+    _C1 = expect_unsuccessful_authentication(
+            try_authenticate(gen_tcp, S, C0, <<"EXTERNAL">>, <<"guest">>, <<"new_password">>),
+            ?RESPONSE_SASL_CANNOT_CHANGE_MECHANISM),
     closed = wait_for_socket_close(gen_tcp, S, 10),
     ok.
 
+update_secret_should_close_connection_if_wrong_secret(Config) ->
+    Transport = gen_tcp,
+    {S, C0} = connect_and_authenticate(Transport, Config),
+    Pwd = rand:bytes(20),
+    _C1 = expect_unsuccessful_authentication(
+            try_authenticate(Transport, S, C0, <<"PLAIN">>, <<"guest">>, Pwd),
+            ?RESPONSE_AUTHENTICATION_FAILURE),
+    closed = wait_for_socket_close(Transport, S, 10),
+    ok.
+
+update_secret_should_close_connection_if_unauthorized_vhost(Config) ->
+    T = gen_tcp,
+    Port = get_port(T, Config),
+    Opts = get_opts(T),
+    {ok, S} = T:connect("localhost", Port, Opts),
+    C0 = rabbit_stream_core:init(0),
+    C1 = test_peer_properties(T, S, C0),
+    Username = <<"other">>,
+    C2 = test_authenticate(T, S, C1, Username),
+    ok = rabbit_ct_broker_helpers:clear_permissions(Config, Username, <<"/">>),
+    _C3 = expect_unsuccessful_authentication(
+            try_authenticate(gen_tcp, S, C2, <<"PLAIN">>, Username, Username),
+            ?RESPONSE_VHOST_ACCESS_FAILURE),
+    closed = wait_for_socket_close(T, S, 10),
+    ok.
+
 test_stream_tls(Config) ->
     Stream = atom_to_binary(?FUNCTION_NAME, utf8),
     test_server(ssl, Stream, Config),
