Close stream connection with delay in case of authentication failure

For consistency with other protocols (to protect from potential DoS
attacks). Wrong credentials and virtual host access errors trigger
the delay.

Author: acogoluegnes

deps/rabbitmq_stream/src/rabbit_stream_reader.erl

@@ -79,6 +79,7 @@
          peer_cert_subject,
          peer_cert_validity]).
 -define(UNKNOWN_FIELD, unknown_field).
+-define(SILENT_CLOSE_DELAY, 3_000).
 
 %% client API
 -export([start_link/4,
@@ -1325,6 +1326,7 @@ handle_frame_pre_auth(Transport,
                                                                     stream),
                             auth_fail(Username, Msg, Args, C1, State),
                             rabbit_log_connection:warning(Msg, Args),
+                            silent_close_delay(),
                             {C1#stream_connection{connection_step = failure},
                              {sasl_authenticate,
                               ?RESPONSE_AUTHENTICATION_FAILURE, <<>>}};
@@ -1490,6 +1492,7 @@ handle_frame_pre_auth(Transport,
             Conn
         catch exit:#amqp_error{explanation = Explanation} ->
                   rabbit_log:warning("Opening connection failed: ~ts", [Explanation]),
+                  silent_close_delay(),
                   F = rabbit_stream_core:frame({response, CorrelationId,
                                                 {open,
                                                  ?RESPONSE_VHOST_ACCESS_FAILURE,
@@ -4041,3 +4044,8 @@ stream_from_consumers(SubId, Consumers) ->
         _ ->
             undefined
     end.
+
+%% We don't trust the client at this point - force them to wait
+%% for a bit so they can't DOS us with repeated failed logins etc.
+silent_close_delay() ->
+    timer:sleep(?SILENT_CLOSE_DELAY).

deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl

@@ -61,7 +61,9 @@ groups() ->
        should_receive_metadata_update_after_update_secret,
        store_offset_requires_read_access,
        offset_lag_calculation,
-       test_super_stream_duplicate_partitions
+       test_super_stream_duplicate_partitions,
+       authentication_error_should_close_with_delay,
+       unauthorized_vhost_access_should_close_with_delay
       ]},
      %% Run `test_global_counters` on its own so the global metrics are
      %% initialised to 0 for each testcase
@@ -173,6 +175,10 @@ init_per_testcase(store_offset_requires_read_access = TestCase, Config) ->
   ok = rabbit_ct_broker_helpers:add_user(Config, <<"test">>),
   rabbit_ct_helpers:testcase_started(Config, TestCase);
 
+init_per_testcase(unauthorized_vhost_access_should_close_with_delay = TestCase, Config) ->
+  ok = rabbit_ct_broker_helpers:add_user(Config, <<"other">>),
+  rabbit_ct_helpers:testcase_started(Config, TestCase);
+
 init_per_testcase(TestCase, Config) ->
     rabbit_ct_helpers:testcase_started(Config, TestCase).
 
@@ -201,6 +207,9 @@ end_per_testcase(vhost_queue_limit = TestCase, Config) ->
 end_per_testcase(store_offset_requires_read_access = TestCase, Config) ->
     ok = rabbit_ct_broker_helpers:delete_user(Config, <<"test">>),
     rabbit_ct_helpers:testcase_finished(Config, TestCase);
+end_per_testcase(unauthorized_vhost_access_should_close_with_delay = TestCase, Config) ->
+    ok = rabbit_ct_broker_helpers:delete_user(Config, <<"other">>),
+    rabbit_ct_helpers:testcase_finished(Config, TestCase);
 end_per_testcase(TestCase, Config) ->
     rabbit_ct_helpers:testcase_finished(Config, TestCase).
 
@@ -890,6 +899,41 @@ offset_lag_calculation(Config) ->
 
     ok.
 
+authentication_error_should_close_with_delay(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),
+    Start = erlang:monotonic_time(millisecond),
+    _ = expect_unsuccessful_authentication(
+      try_authenticate(T, S, C1, <<"PLAIN">>, <<"guest">>, <<"wrong password">>),
+        ?RESPONSE_AUTHENTICATION_FAILURE),
+    End = erlang:monotonic_time(millisecond),
+    %% the stream reader module defines the delay (3 seconds)
+    ?assert(End - Start > 2_000),
+    closed = wait_for_socket_close(T, S, 10),
+    ok.
+
+unauthorized_vhost_access_should_close_with_delay(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),
+    User = <<"other">>,
+    C2 = test_plain_sasl_authenticate(T, S, sasl_handshake(T, S, C1), User),
+    Start = erlang:monotonic_time(millisecond),
+    R = do_tune(T, S, C2),
+    ?assertMatch({{response,_,{open,12}}, _}, R),
+    End = erlang:monotonic_time(millisecond),
+    %% the stream reader module defines the delay (3 seconds)
+    ?assert(End - Start > 2_000),
+    closed = wait_for_socket_close(T, S, 10),
+    ok.
+
 consumer_offset_info(Config, ConnectionName) ->
     [[{offset, Offset},
       {offset_lag, Lag}]] = rpc(Config, 0, ?MODULE,
@@ -1093,12 +1137,15 @@ test_peer_properties(Transport, S, Properties, C0) ->
     C.
 
 test_authenticate(Transport, S, C0) ->
-  tune(Transport, S,
-       test_plain_sasl_authenticate(Transport, S, sasl_handshake(Transport, S, C0), <<"guest">>)).
+    tune(Transport, S,
+         test_plain_sasl_authenticate(Transport, S, sasl_handshake(Transport, S, C0), <<"guest">>)).
 
 test_authenticate(Transport, S, C0, Username) ->
-  tune(Transport, S,
-       test_plain_sasl_authenticate(Transport, S, sasl_handshake(Transport, S, C0), Username)).
+    test_authenticate(Transport, S, C0, Username, Username).
+
+test_authenticate(Transport, S, C0, Username, Password) ->
+    tune(Transport, S,
+         test_plain_sasl_authenticate(Transport, S, sasl_handshake(Transport, S, C0), Username, Password)).
 
 sasl_handshake(Transport, S, C0) ->
     SaslHandshakeFrame = request(sasl_handshake),
@@ -1115,7 +1162,10 @@ sasl_handshake(Transport, S, C0) ->
     C1.
 
 test_plain_sasl_authenticate(Transport, S, C1, Username) ->
-  expect_successful_authentication(plain_sasl_authenticate(Transport, S, C1, Username, Username)).
+    test_plain_sasl_authenticate(Transport, S, C1, Username, Username).
+
+test_plain_sasl_authenticate(Transport, S, C1, Username, Password) ->
+    expect_successful_authentication(plain_sasl_authenticate(Transport, S, C1, Username, Password)).
 
 plain_sasl_authenticate(Transport, S, C1, Username, Password) ->
     Null = 0,
@@ -1136,6 +1186,10 @@ sasl_authenticate(Transport, S, C1, AuthMethod, AuthBody) ->
     receive_commands(Transport, S, C1).
 
 tune(Transport, S, C2) ->
+    {{response, _, {open, ?RESPONSE_CODE_OK, _}}, C3} = do_tune(Transport, S, C2),
+    C3.
+
+do_tune(Transport, S, C2) ->
     {Tune, C3} = receive_commands(Transport, S, C2),
     {tune, ?DEFAULT_FRAME_MAX, ?DEFAULT_HEARTBEAT} = Tune,
 
@@ -1147,10 +1201,7 @@ tune(Transport, S, C2) ->
     VirtualHost = <<"/">>,
     OpenFrame = request(3, {open, VirtualHost}),
     ok = Transport:send(S, OpenFrame),
-    {{response, 3, {open, ?RESPONSE_CODE_OK, _ConnectionProperties}},
-     C4} =
-        receive_commands(Transport, S, C3),
-    C4.
+    receive_commands(Transport, S, C3).
 
 test_create_stream(Transport, S, Stream, C0) ->
     CreateStreamFrame = request({create_stream, Stream, #{}}),