Merge pull request #16081 from rabbitmq/mergify/bp/v4.2.x/pr-16079

AMQP 0-9-1: refactor command assembler (backport #16076) (backport #16079)

Author: michaelklishin

deps/rabbit_common/src/rabbit_command_assembler.erl

@@ -35,7 +35,7 @@
 -type state() ::
         {'method',         protocol()} |
         {'content_header', method(), class_id(), protocol()} |
-        {'content_body',   method(), body_size(), class_id(), protocol()}.
+        {'content_body',   method(), body_size(), content(), protocol()}.
 
 -spec analyze_frame(frame_type(), binary(), protocol()) ->
           frame() | 'heartbeat' | 'error'.
@@ -85,9 +85,17 @@ process({content_header, ClassId, 0, 0, PropertiesBin},
     Content = empty_content(ClassId, PropertiesBin, Protocol),
     {ok, Method, Content, {method, Protocol}};
 process({content_header, ClassId, 0, BodySize, PropertiesBin},
-        {content_header, Method, ClassId, Protocol}) ->
+        {content_header, Method, ClassId, Protocol})
+  when BodySize =< ?MAX_MSG_SIZE ->
     Content = empty_content(ClassId, PropertiesBin, Protocol),
     {ok, {content_body, Method, BodySize, Content, Protocol}};
+process({content_header, ClassId, 0, BodySize, _PropertiesBin},
+        {content_header, Method, ClassId, _Protocol}) when BodySize > ?MAX_MSG_SIZE ->
+    {error, rabbit_misc:amqp_error(
+              frame_error,
+              "content body size ~B exceeds maximum allowed size ~B",
+              [BodySize, ?MAX_MSG_SIZE],
+              rabbit_misc:method_record_type(Method))};
 process({content_header, HeaderClassId, 0, _BodySize, _PropertiesBin},
         {content_header, Method, ClassId, _Protocol}) ->
     unexpected_frame("expected content header for class ~w, "
@@ -98,13 +106,21 @@ process(_Frame, {content_header, Method, ClassId, _Protocol}) ->
                      "got non content header frame instead", [ClassId], Method);
 process({content_body, FragmentBin},
         {content_body, Method, RemainingSize,
-         Content = #content{payload_fragments_rev = Fragments}, Protocol}) ->
+         Content = #content{payload_fragments_rev = Fragments}, Protocol})
+  when byte_size(FragmentBin) =< RemainingSize ->
     NewContent = Content#content{
                    payload_fragments_rev = [FragmentBin | Fragments]},
-    case RemainingSize - size(FragmentBin) of
+    case RemainingSize - byte_size(FragmentBin) of
         0  -> {ok, Method, NewContent, {method, Protocol}};
         Sz -> {ok, {content_body, Method, Sz, NewContent, Protocol}}
     end;
+process({content_body, _FragmentBin},
+        {content_body, Method, _RemainingSize, _Content, _Protocol}) ->
+    {error, rabbit_misc:amqp_error(
+              frame_error,
+              "content body frame exceeds remaining content size",
+              [],
+              rabbit_misc:method_record_type(Method))};
 process(_Frame, {content_body, Method, _RemainingSize, _Content, _Protocol}) ->
     unexpected_frame("expected content body, "
                      "got non content body frame instead", [], Method).