AMQP 1.0: strictly validate annotations

deps/amqp10_client/src/amqp10_client_session.erl

@@ -1171,16 +1171,13 @@ make_link_ref(Role, Session, Handle) ->
     #link_ref{role = Role, session = Session, link_handle = Handle}.
 
 translate_message_annotations(MA)
-  when is_map(MA) andalso
-       map_size(MA) > 0 ->
-    Content = maps:fold(fun (K, V, Acc) ->
-                                [{sym(K), wrap_map_value(V)} | Acc]
-                        end, [], MA),
-    #'v1_0.message_annotations'{content = Content};
+  when map_size(MA) > 0 ->
+    {map, maps:fold(fun(K, V, Acc) ->
+                            [{sym(K), wrap_map_value(V)} | Acc]
+                    end, [], MA)};
 translate_message_annotations(_MA) ->
     undefined.
 
-
 wrap_map_value(true) ->
     {boolean, true};
 wrap_map_value(false) ->

deps/amqp10_client/test/system_SUITE.erl

@@ -348,8 +348,8 @@ roundtrip(OpenConf, Body) ->
     Msg0 = amqp10_msg:new(<<"my-tag">>, Body, true),
     Msg1 = amqp10_msg:set_application_properties(#{"a_key" => "a_value"}, Msg0),
     Msg2 = amqp10_msg:set_properties(Props, Msg1),
-    Msg = amqp10_msg:set_message_annotations(#{<<"x-key">> => "x-value",
-                                               <<"x_key">> => "x_value"}, Msg2),
+    Msg = amqp10_msg:set_message_annotations(#{<<"x-key 1">> => "value 1",
+                                               <<"x-key 2">> => "value 2"}, Msg2),
     ok = amqp10_client:send_msg(Sender, Msg),
     ok = amqp10_client:detach_link(Sender),
     await_link(Sender, {detached, normal}, link_detach_timeout),
@@ -364,8 +364,8 @@ roundtrip(OpenConf, Body) ->
     % ct:pal(?LOW_IMPORTANCE, "roundtrip message Out: ~tp~nIn: ~tp~n", [OutMsg, Msg]),
     ?assertMatch(Props, amqp10_msg:properties(OutMsg)),
     ?assertEqual(#{<<"a_key">> => <<"a_value">>}, amqp10_msg:application_properties(OutMsg)),
-    ?assertMatch(#{<<"x-key">> := <<"x-value">>,
-                   <<"x_key">> := <<"x_value">>}, amqp10_msg:message_annotations(OutMsg)),
+    ?assertMatch(#{<<"x-key 1">> := <<"value 1">>,
+                   <<"x-key 2">> := <<"value 2">>}, amqp10_msg:message_annotations(OutMsg)),
     ?assertEqual([Body], amqp10_msg:body(OutMsg)),
     ok.
 

deps/amqp10_common/src/amqp10_framing.erl

@@ -122,11 +122,11 @@ decode({described, Descriptor, {map, Fields} = Type}) ->
         #'v1_0.application_properties'{} ->
             #'v1_0.application_properties'{content = decode_map(Fields)};
         #'v1_0.delivery_annotations'{} ->
-            #'v1_0.delivery_annotations'{content = decode_map(Fields)};
+            #'v1_0.delivery_annotations'{content = decode_annotations(Fields)};
         #'v1_0.message_annotations'{} ->
-            #'v1_0.message_annotations'{content = decode_map(Fields)};
+            #'v1_0.message_annotations'{content = decode_annotations(Fields)};
         #'v1_0.footer'{} ->
-            #'v1_0.footer'{content = decode_map(Fields)};
+            #'v1_0.footer'{content = decode_annotations(Fields)};
         #'v1_0.amqp_value'{} ->
             #'v1_0.amqp_value'{content = Type};
         Else ->
@@ -149,6 +149,16 @@ decode(Other) ->
 decode_map(Fields) ->
     [{decode(K), decode(V)} || {K, V} <- Fields].
 
+%% "The annotations type is a map where the keys are restricted to be of type symbol
+%% or of type ulong. All ulong keys, and all symbolic keys except those beginning
+%% with "x-" are reserved." [3.2.10]
+%% Since we already parse annotations here and neither the client nor server uses
+%% reserved keys, we perform strict validation and crash if any reserved keys are used.
+decode_annotations(Fields) ->
+    lists:map(fun({{symbol, <<"x-", _/binary>>} = K, V}) ->
+                      {K, decode(V)}
+              end, Fields).
+
 -spec encode_described(list | map | binary | annotations | '*',
                        non_neg_integer(),
                        amqp10_frame()) ->
@@ -216,7 +226,7 @@ pprint(Other) -> Other.
 -include_lib("eunit/include/eunit.hrl").
 
 encode_decode_test_() ->
-    Data = [{{utf8, <<"k">>}, {binary, <<"v">>}}],
+    Data = [{{symbol, <<"x-my key">>}, {binary, <<"my value">>}}],
     Test = fun(M) -> [M] = decode_bin(iolist_to_binary(encode_bin(M))) end,
     [
      fun() -> Test(#'v1_0.application_properties'{content = Data}) end,

deps/amqp10_common/test/prop_SUITE.erl

@@ -412,14 +412,21 @@ footer_section() ->
 
 annotations() ->
     ?LET(KvList,
-         list({oneof([amqp_symbol(),
-                      amqp_ulong()]),
+         list({non_reserved_annotation_key(),
                prefer_simple_type()}),
          begin
              KvList1 = lists:uniq(fun({K, _V}) -> K end, KvList),
              lists:filter(fun({_K, V}) -> V =/= null end, KvList1)
          end).
 
+non_reserved_annotation_key() ->
+    {symbol, ?LET(L,
+                  ?SIZED(Size, resize(Size * 10, list(ascii_char()))),
+                  begin
+                      Bin = list_to_binary(L) ,
+                      <<"x-", Bin/binary>>
+                  end)}.
+
 sequence_no() ->
     amqp_uint().
 

deps/rabbit/src/rabbit_amqp_session.erl

@@ -1912,13 +1912,15 @@ settle_op_from_outcome(#'v1_0.modified'{delivery_failed = DelFailed,
                                         undeliverable_here = UndelHere,
                                         message_annotations = Anns0}) ->
     Anns = case Anns0 of
-               #'v1_0.message_annotations'{content = C} ->
-                   Anns1 = lists:map(fun({{symbol, K}, V}) ->
-                                             {K, unwrap(V)}
-                                     end, C),
-                   maps:from_list(Anns1);
-               _ ->
-                   #{}
+               undefined ->
+                   #{};
+               {map, KVList} ->
+                   Anns1 = lists:map(
+                             %% "all symbolic keys except those beginning with "x-" are reserved." [3.2.10]
+                             fun({{symbol, <<"x-", _/binary>> = K}, V}) ->
+                                     {K, unwrap(V)}
+                             end, KVList),
+                   maps:from_list(Anns1)
            end,
     {modify,
      default(DelFailed, false),

deps/rabbit/test/amqp_client_SUITE.erl

@@ -355,7 +355,7 @@ reliable_send_receive_with_outcomes(QType, Config) ->
     Outcomes = [
                 accepted,
                 modified,
-                {modified, true, false, #{<<"fruit">> => <<"banana">>}},
+                {modified, true, false, #{<<"x-fruit">> => <<"banana">>}},
                 {modified, false, true, #{}},
                 rejected,
                 released
@@ -1124,7 +1124,7 @@ amqp_amqpl(QType, Config) ->
                #{"my int" => -2},
                amqp10_msg:new(<<>>, Body1, true)))),
     %% Send with footer
-    Footer = #'v1_0.footer'{content = [{{symbol, <<"my footer">>}, {ubyte, 255}}]},
+    Footer = #'v1_0.footer'{content = [{{symbol, <<"x-my footer">>}, {ubyte, 255}}]},
     ok = amqp10_client:send_msg(
            Sender,
            amqp10_msg:from_amqp_records(
@@ -5155,7 +5155,7 @@ footer_checksum(FooterOpt, Config) ->
              priority => 7,
              ttl => 100_000},
            amqp10_msg:set_delivery_annotations(
-             #{"a" => "b"},
+             #{"x-a" => "b"},
              amqp10_msg:set_message_annotations(
                #{"x-string" => "string-value",
                  "x-int" => 3,

deps/rabbit/test/amqp_system_SUITE_data/fsharp-tests/Program.fs

@@ -298,10 +298,10 @@ module Test =
         use c = connectAnon uri
         let sender, receiver = senderReceiver c "test" "/queues/message_annotations"
         let ann = MessageAnnotations()
-        let k1 = Symbol "key1"
-        let k2 = Symbol "key2"
-        ann.[Symbol "key1"] <- "value1"
-        ann.[Symbol "key2"] <- "value2"
+        let k1 = Symbol "x-key1"
+        let k2 = Symbol "x-key2"
+        ann.[Symbol "x-key1"] <- "value1"
+        ann.[Symbol "x-key2"] <- "value2"
         let m = new Message("testing annotations", MessageAnnotations = ann)
         sender.Send m
         let m' = receive receiver

deps/rabbit/test/mc_unit_SUITE.erl

@@ -524,8 +524,6 @@ amqp_amqpl(_Config) ->
                        durable = true},
     MAC = [
            {{symbol, <<"x-stream-filter">>}, {utf8, <<"apple">>}},
-           thead2(list, [utf8(<<"l">>)]),
-           thead2(map, [{utf8(<<"k">>), utf8(<<"v">>)}]),
            thead2('x-list', list, [utf8(<<"l">>)]),
            thead2('x-map', map, [{utf8(<<"k">>), utf8(<<"v">>)}])
           ],
@@ -591,9 +589,6 @@ amqp_amqpl(_Config) ->
     ?assertMatch(#'P_basic'{expiration = <<"20000">>}, Props),
 
     ?assertMatch({_, longstr, <<"apple">>}, header(<<"x-stream-filter">>, HL)),
-    %% these are not coverted as not x- headers
-    ?assertEqual(undefined, header(<<"list">>, HL)),
-    ?assertEqual(undefined, header(<<"map">>, HL)),
     ?assertMatch({_ ,array, [{longstr,<<"l">>}]}, header(<<"x-list">>, HL)),
     ?assertMatch({_, table, [{<<"k">>,longstr,<<"v">>}]}, header(<<"x-map">>, HL)),
 

deps/rabbitmq_consistent_hash_exchange/test/rabbit_exchange_type_consistent_hash_SUITE.erl

@@ -244,21 +244,21 @@ amqp_dead_letter(Config) ->
          Msg1 = case Seq rem 2 of
                     0 ->
                         amqp10_msg:set_message_annotations(
-                          #{<<"k1">> => Seq}, Msg0);
+                          #{<<"x-k1">> => Seq}, Msg0);
                     1 ->
                         Msg0
                 end,
          Msg2 = case Seq rem 3 of
                     0 ->
                         amqp10_msg:set_application_properties(
-                          #{<<"k2">> => Seq}, Msg1);
+                          #{<<"x-k2">> => Seq}, Msg1);
                     _ ->
                         Msg1
                 end,
          Msg = case Seq rem 4 of
                    0 ->
                        amqp10_msg:set_delivery_annotations(
-                         #{<<"k3">> => Seq}, Msg2);
+                         #{<<"x-k3">> => Seq}, Msg2);
                    _ ->
                        Msg2
                end,

deps/rabbitmq_mqtt/test/mc_mqtt_SUITE.erl

@@ -265,7 +265,7 @@ amqp_to_mqtt_reply_to(_Config) ->
 
 amqp_to_mqtt_footer(_Config) ->
     Body = <<"hey">>,
-    Footer = #'v1_0.footer'{content = [{{symbol, <<"key">>}, {utf8, <<"value">>}}]},
+    Footer = #'v1_0.footer'{content = [{{symbol, <<"x-key">>}, {utf8, <<"value">>}}]},
     %% We can translate, but lose the footer.
     #mqtt_msg{payload = Payload} = amqp_to_mqtt([#'v1_0.data'{content = Body}, Footer]),
     ?assertEqual(<<"hey">>, iolist_to_binary(Payload)).
@@ -404,8 +404,6 @@ amqp_mqtt(_Config) ->
                        durable = true},
     MAC = [
            {{symbol, <<"x-stream-filter">>}, {utf8, <<"apple">>}},
-           thead2(list, [utf8(<<"l">>)]),
-           thead2(map, [{utf8(<<"k">>), utf8(<<"v">>)}]),
            thead2('x-list', list, [utf8(<<"l">>)]),
            thead2('x-map', map, [{utf8(<<"k">>), utf8(<<"v">>)}])
           ],

deps/rabbitmq_shovel/test/amqp10_dynamic_SUITE.erl

@@ -121,12 +121,12 @@ test_amqp10_destination(Config, Src, Dest, Sess, Protocol, ProtocolSrc) ->
                                   end},
                                  {<<"dest-message-annotations">>,
                                   case MapConfig of
-                                     true ->
-                                         #{<<"message-ann-key">> =>
-                                               <<"message-ann-value">>};
-                                     _ ->
-                                         [{<<"message-ann-key">>,
-                                           <<"message-ann-value">>}]
+                                      true ->
+                                          #{<<"x-message-ann-key">> =>
+                                            <<"message-ann-value">>};
+                                      _ ->
+                                          [{<<"x-message-ann-key">>,
+                                            <<"message-ann-value">>}]
                                   end}]),
     Msg = publish_expect(Sess, Src, Dest, <<"tag1">>, <<"hello">>),
     AppProps = amqp10_msg:application_properties(Msg),
@@ -138,7 +138,7 @@ test_amqp10_destination(Config, Src, Dest, Sess, Protocol, ProtocolSrc) ->
                     <<"app-prop-key">> := <<"app-prop-value">>}),
                  (AppProps)),
     ?assertEqual(undefined, maps:get(<<"delivery_mode">>, AppProps, undefined)),
-    ?assertMatch((#{<<"message-ann-key">> := <<"message-ann-value">>}),
+    ?assertMatch((#{<<"x-message-ann-key">> := <<"message-ann-value">>}),
                  (amqp10_msg:message_annotations(Msg))).
 
 simple_amqp10_src(Config) ->

release-notes/4.0.0.md

@@ -117,6 +117,13 @@ RabbitMQ 3.13 `rabbitmq.conf` setting `rabbitmq_amqp1_0.default_vhost` is unsupp
 Instead `default_vhost` will be used to determine the default vhost an AMQP 1.0 client connects to(i.e. when the AMQP 1.0 client
 does not define the vhost in the `hostname` field of the `open` frame).
 
+Starting with RabbitMQ 4.0, RabbitMQ strictly validates that
+[delivery annotations](https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-delivery-annotations),
+[message annotations](https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-message-annotations), and
+[footer](https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-footer) contain only
+[non-reserved annotation keys](https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-annotations).
+As a result, clients can only send symbolic keys that begin with `x-`.
+
 ### MQTT
 
 RabbitMQ 3.13 [rabbitmq.conf](https://www.rabbitmq.com/docs/configure#config-file) settings `mqtt.default_user`, `mqtt.default_password`,