mc: increase utf8 scanning limit for longstr conversions.

The AMQP 0.9.1 longstr type is problematic as it can contain arbitrary
binary data but is typically used for utf8 by users.

The current conversion into AMQP avoids scanning arbitrarily large
longstr to see if they only contain valid utf8 by treating all
longstr data longer than 255 bytes as binary. This is in hindsight
too strict and thus this commit increases the scanning limit to
4096 bytes - enough to cover the vast majority of AMQP 0.9.1 header
values.

This change also conversts the AMQP binary types into longstr to
ensure that existing data (held in streams for example) is converted
to an AMQP 0.9.1 type most likely what the user intended.

(cherry picked from commit 131379a)

Author: kjnilsson

deps/rabbit/src/mc_amqpl.erl

@@ -40,6 +40,7 @@
 -define(AMQP10_MESSAGE_ANNOTATIONS_HEADER, <<"x-amqp-1.0-message-annotations">>).
 -define(PROTOMOD, rabbit_framing_amqp_0_9_1).
 -define(CLASS_ID, 60).
+-define(LONGSTR_UTF8_LIMIT, 4096).
 
 -opaque state() :: #content{}.
 
@@ -534,16 +535,19 @@ wrap(_Type, undefined) ->
 wrap(Type, Val) ->
     {Type, Val}.
 
-from_091(longstr, V) ->
-    case mc_util:is_valid_shortstr(V) of
+from_091(longstr, V)
+  when is_binary(V) andalso
+       byte_size(V) =< ?LONGSTR_UTF8_LIMIT ->
+    %% if a longstr is longer than 4096 bytes we just assume it is binary
+    %% it _may_ still be valid utf8 but checking this for every longstr header
+    %% value is going to be excessively slow
+    case mc_util:is_utf8_no_null(V) of
         true ->
             {utf8, V};
         false ->
-            %%  if a string is longer than 255 bytes we just assume it is binary
-            %%  it _may_ still be valid utf8 but checking this is going to be
-            %%  excessively slow
             {binary, V}
     end;
+from_091(longstr, V) -> {binary, V};
 from_091(long, V) -> {long, V};
 from_091(unsignedbyte, V) -> {ubyte, V};
 from_091(short, V) -> {short, V};
@@ -614,7 +618,7 @@ to_091(Key, {int, V}) -> {Key, signedint, V};
 to_091(Key, {double, V}) -> {Key, double, V};
 to_091(Key, {float, V}) -> {Key, float, V};
 to_091(Key, {timestamp, V}) -> {Key, timestamp, V div 1000};
-to_091(Key, {binary, V}) -> {Key, binary, V};
+to_091(Key, {binary, V}) -> {Key, longstr, V};
 to_091(Key, {boolean, V}) -> {Key, bool, V};
 to_091(Key, true) -> {Key, bool, true};
 to_091(Key, false) -> {Key, bool, false};
@@ -637,7 +641,7 @@ to_091({int, V}) -> {signedint, V};
 to_091({double, V}) -> {double, V};
 to_091({float, V}) -> {float, V};
 to_091({timestamp, V}) -> {timestamp, V div 1000};
-to_091({binary, V}) -> {binary, V};
+to_091({binary, V}) -> {longstr, V};
 to_091({boolean, V}) -> {bool, V};
 to_091(true) -> {bool, true};
 to_091(false) -> {bool, false};

deps/rabbit/test/mc_unit_SUITE.erl

@@ -663,7 +663,7 @@ amqp_amqpl(_Config) ->
     ?assertMatch({_, long, 5}, header(<<"long">>, HL)),
     ?assertMatch({_, long, 5}, header(<<"ulong">>, HL)),
     ?assertMatch({_, longstr, <<"a-string">>}, header(<<"utf8">>, HL)),
-    ?assertMatch({_, binary, <<"data">>}, header(<<"binary">>, HL)),
+    ?assertMatch({_, longstr, <<"data">>}, header(<<"binary">>, HL)),
     ?assertMatch({_, longstr, <<"symbol">>}, header(<<"symbol">>, HL)),
     ?assertMatch({_, unsignedbyte, 1}, header(<<"ubyte">>, HL)),
     ?assertMatch({_, short, 2}, header(<<"short">>, HL)),