STOMP: introduce a max frame size limit (backport #8802)

.gitignore

@@ -86,6 +86,7 @@ rabbitmq-server-*.tar.xz
 rabbitmq-server-*.zip
 
 traces*
+deps/rabbitmq_stomp/test/python_SUITE_data/src/deps
 callgrand*
 
 /user.bazelrc

deps/rabbitmq_stomp/include/rabbit_stomp.hrl

@@ -40,3 +40,5 @@
          ssl_hash]).
 
 -define(STOMP_GUIDE_URL, <<"https://rabbitmq.com/stomp.html">>).
+
+-define(DEFAULT_MAX_FRAME_SIZE, 4 * 1024 * 1024).

deps/rabbitmq_stomp/priv/schema/rabbitmq_stomp.schema

@@ -238,3 +238,12 @@ end}.
 
 {mapping, "stomp.default_nack_requeue", "rabbitmq_stomp.default_nack_requeue",
     [{datatype, {enum, [true, false]}}]}.
+
+
+
+%% Maximum frame size.
+%%
+%% Defaults to 192Kb.
+
+{mapping, "stomp.max_frame_size", "rabbitmq_stomp.max_frame_size",
+    [{datatype, integer}, {validators, ["non_negative_integer"]}]}.

deps/rabbitmq_stomp/src/rabbit_stomp_reader.erl

@@ -24,12 +24,24 @@
 -define(OTHER_METRICS, [recv_cnt, send_cnt, send_pend, garbage_collection, state,
                         timeout]).
 
--record(reader_state, {socket, conn_name, parse_state, processor_state, state,
-                       conserve_resources, recv_outstanding, stats_timer,
-                       parent, connection, heartbeat_sup, heartbeat,
-                       timeout_sec %% heartbeat timeout value used, 0 means
-                                   %% heartbeats are disabled
-                      }).
+-record(reader_state, {
+    socket,
+    conn_name,
+    parse_state,
+    processor_state,
+    state,
+    conserve_resources,
+    recv_outstanding,
+    max_frame_size,
+    current_frame_size,
+    stats_timer,
+    parent,
+    connection,
+    heartbeat_sup, heartbeat,
+    %% heartbeat timeout value used, 0 means
+    %% heartbeats are disabled
+    timeout_sec
+}).
 
 %%----------------------------------------------------------------------------
 
@@ -69,6 +81,7 @@ init([SupHelperPid, Ref, Configuration]) ->
             _ = register_resource_alarm(),
 
             LoginTimeout = application:get_env(rabbitmq_stomp, login_timeout, 10_000),
+            MaxFrameSize = application:get_env(rabbitmq_stomp, max_frame_size, ?DEFAULT_MAX_FRAME_SIZE),
             erlang:send_after(LoginTimeout, self(), login_timeout),
 
             gen_server2:enter_loop(?MODULE, [],
@@ -80,6 +93,8 @@ init([SupHelperPid, Ref, Configuration]) ->
                                 processor_state    = ProcState,
                                 heartbeat_sup      = SupHelperPid,
                                 heartbeat          = {none, none},
+                                max_frame_size     = MaxFrameSize,
+                                current_frame_size = 0,
                                 state              = running,
                                 conserve_resources = false,
                                 recv_outstanding   = false})), #reader_state.stats_timer),
@@ -222,23 +237,41 @@ process_received_bytes([], State) ->
     {ok, State};
 process_received_bytes(Bytes,
                        State = #reader_state{
-                         processor_state = ProcState,
-                         parse_state     = ParseState}) ->
+                         max_frame_size = MaxFrameSize,
+                         current_frame_size = FrameLength,
+                         processor_state  = ProcState,
+                         parse_state      = ParseState}) ->
     case rabbit_stomp_frame:parse(Bytes, ParseState) of
         {more, ParseState1} ->
-            {ok, State#reader_state{parse_state = ParseState1}};
+            FrameLength1 = FrameLength + byte_size(Bytes),
+            case FrameLength1 > MaxFrameSize of
+                true ->
+                    log_reason({network_error, {frame_too_big, {FrameLength1, MaxFrameSize}}}, State),
+                    {stop, normal, State};
+                false ->
+                    {ok, State#reader_state{parse_state = ParseState1,
+                                            current_frame_size = FrameLength1}}
+            end;
         {ok, Frame, Rest} ->
-            case rabbit_stomp_processor:process_frame(Frame, ProcState) of
-                {ok, NewProcState, Conn} ->
-                    PS = rabbit_stomp_frame:initial_state(),
-                    NextState = maybe_block(State, Frame),
-                    process_received_bytes(Rest, NextState#reader_state{
-                        processor_state = NewProcState,
-                        parse_state     = PS,
-                        connection      = Conn});
-                {stop, Reason, NewProcState} ->
-                    {stop, Reason,
-                     processor_state(NewProcState, State)}
+            FrameLength1 = FrameLength + byte_size(Bytes) - byte_size(Rest),
+            case FrameLength1 > MaxFrameSize of
+                true ->
+                    log_reason({network_error, {frame_too_big, {FrameLength1, MaxFrameSize}}}, State),
+                    {stop, normal, State};
+                false ->
+                    case rabbit_stomp_processor:process_frame(Frame, ProcState) of
+                        {ok, NewProcState, Conn} ->
+                            PS = rabbit_stomp_frame:initial_state(),
+                            NextState = maybe_block(State, Frame),
+                            process_received_bytes(Rest, NextState#reader_state{
+                                                           current_frame_size = 0,
+                                                           processor_state = NewProcState,
+                                                           parse_state     = PS,
+                                                           connection      = Conn});
+                        {stop, Reason, NewProcState} ->
+                            {stop, Reason,
+                             processor_state(NewProcState, State)}
+                    end
             end;
         {error, Reason} ->
             %% The parser couldn't parse data. We log the reason right

deps/rabbitmq_stomp/test/connections_SUITE.erl

@@ -22,7 +22,9 @@ all() ->
         stats_are_not_leaked,
         stats,
         heartbeat,
-        login_timeout
+        login_timeout,
+        frame_size,
+        frame_size_huge
     ].
 
 merge_app_env(Config) ->
@@ -174,3 +176,36 @@ login_timeout(Config) ->
           Config, 0,
           application, unset_env, [rabbitmq_stomp, login_timeout])
     end.
+
+frame_size(Config) ->
+    rabbit_ct_broker_helpers:rpc(
+      Config, 0,
+      application, set_env, [rabbitmq_stomp, max_frame_size, 80]),
+    StompPort = get_stomp_port(Config),
+    {ok, Client} = rabbit_stomp_client:connect("1.2", "guest", "guest", StompPort,
+                                               [{"heart-beat", "5000,7000"}]),
+    ok = rabbit_stomp_client:send(
+      Client, "SEND", [{"destination", "qwe"}],
+      ["Lorem ipsum dolor sit amet viverra fusce. "
+       "Lorem ipsum dolor sit amet viverra fusce. "
+       "Lorem ipsum dolor sit amet viverra fusce."
+       "Lorem ipsum dolor sit amet viverra fusce."
+       "Lorem ipsum dolor sit amet viverra fusce."]),
+    {S, _} = Client,
+    {error, closed} = gen_tcp:recv(S, 0, 500),
+    ok.
+
+
+frame_size_huge(Config) ->
+    rabbit_ct_broker_helpers:rpc(
+      Config, 0,
+      application, set_env, [rabbitmq_stomp, max_frame_size, 700]),
+    StompPort = get_stomp_port(Config),
+    {ok, Client} = rabbit_stomp_client:connect("1.2", "guest", "guest", StompPort,
+                                               [{"heart-beat", "5000,7000"}]),
+    rabbit_stomp_client:send(
+      Client, "SEND", [{"destination", "qwe"}],
+      [base64:encode(crypto:strong_rand_bytes(100000000))]),
+    {S, _} = Client,
+    {error, closed} = gen_tcp:recv(S, 0, 500),
+    ok.

deps/rabbitmq_stomp/test/python_SUITE.erl

@@ -30,6 +30,17 @@ groups() ->
         ]}
     ].
 
+init_per_suite(Config) ->
+    DataDir = ?config(data_dir, Config),
+    {ok, _} = rabbit_ct_helpers:exec(["pip", "install", "-r", requirements_path(Config),
+                                                        "--target", deps_path(Config)]),
+    Config.
+
+end_per_suite(Config) ->
+    DataDir = ?config(data_dir, Config),
+    ok = file:del_dir_r(deps_path(Config)),
+    Config.
+
 init_per_group(_, Config) ->
     Config1 = rabbit_ct_helpers:set_config(Config,
                                            [
@@ -54,6 +65,9 @@ end_per_testcase(Test, Config) ->
 
 
 main(Config) ->
+    rabbit_ct_broker_helpers:rpc(
+      Config, 0,
+      application, set_env, [rabbitmq_stomp, max_frame_size, 17 * 1024 * 1024]),
     run(Config, filename:join("src", "main_runner.py")).
 
 implicit_connect(Config) ->
@@ -64,7 +78,6 @@ tls_connections(Config) ->
 
 
 run(Config, Test) ->
-    DataDir = ?config(data_dir, Config),
     CertsDir = rabbit_ct_helpers:get_config(Config, rmq_certsdir),
     StompPort = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_stomp),
     StompPortTls = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_stomp_tls),
@@ -75,8 +88,26 @@ run(Config, Test) ->
     os:putenv("STOMP_PORT_TLS", integer_to_list(StompPortTls)),
     os:putenv("RABBITMQ_NODENAME", atom_to_list(NodeName)),
     os:putenv("SSL_CERTS_PATH", CertsDir),
-    {ok, _} = rabbit_ct_helpers:exec([filename:join(DataDir, Test)]).
+    run_python(Config, Test).
+
+run_python(Config, What) ->
+    DataDir = ?config(data_dir, Config),
+    os:putenv("PYTHONPATH", python_path(Config)),
+    {ok, _} = rabbit_ct_helpers:exec([filename:join(DataDir, What)]).
+
+deps_path(Config) ->
+    DataDir = ?config(data_dir, Config),
+    filename:join([DataDir, "src", "deps"]).
+
+requirements_path(Config) ->
+    DataDir = ?config(data_dir, Config),
+    filename:join([DataDir, "src", "requirements.txt"]).
 
+python_path(Config) ->
+    case os:getenv("PYTHONPATH") of
+        false -> deps_path(Config);
+        P -> deps_path(Config) ++ ":" ++ P
+    end.
 
 cur_dir() ->
     {ok, Src} = filelib:find_source(?MODULE),

deps/rabbitmq_stomp/test/python_SUITE_data/src/main_runner.py

@@ -1,14 +1,5 @@
 #!/usr/bin/env python3
 
-import sys
-import subprocess
-
-# implement pip as a subprocess:
-subprocess.check_call([sys.executable, '-m', 'pip', 'install',
-                       'stomp.py==8.1.0'])
-subprocess.check_call([sys.executable, '-m', 'pip', 'install',
-                       'pika==1.1.0'])
-
 import test_runner
 
 if __name__ == '__main__':

deps/rabbitmq_stomp/test/python_SUITE_data/src/requirements.txt

@@ -0,0 +1,3 @@
+stomp.py==8.1.0
+pika==1.1.0
+

deps/rabbitmq_stomp/test/python_SUITE_data/src/test_runner.py

@@ -7,15 +7,6 @@
 ## Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 ##
 
-import sys
-import subprocess
-
-# implement pip as a subprocess:
-subprocess.check_call([sys.executable, '-m', 'pip', 'install',
-                       'stomp.py==8.1.0'])
-subprocess.check_call([sys.executable, '-m', 'pip', 'install',
-                       'pika==1.1.0'])
-
 import unittest
 import sys
 import os

deps/rabbitmq_stomp/test/python_SUITE_data/src/tls_runner.py

@@ -7,15 +7,6 @@
 ## Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 ##
 
-import sys
-import subprocess
-
-# implement pip as a subprocess:
-subprocess.check_call([sys.executable, '-m', 'pip', 'install',
-                       'stomp.py==8.1.0'])
-subprocess.check_call([sys.executable, '-m', 'pip', 'install',
-                       'pika==1.1.0'])
-
 import test_runner
 import test_util