Support x-cc message annotation

deps/rabbit/BUILD.bazel

@@ -862,6 +862,12 @@ rabbitmq_integration_suite(
 rabbitmq_integration_suite(
     name = "topic_permission_SUITE",
     size = "medium",
+    additional_beam = [
+        ":test_amqp_utils_beam",
+    ],
+    runtime_deps = [
+        "//deps/rabbitmq_amqp_client:erlang_app",
+    ],
 )
 
 rabbitmq_integration_suite(

deps/rabbit/app.bzl

@@ -1559,7 +1559,7 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         outs = ["test/topic_permission_SUITE.beam"],
         app_name = "rabbit",
         erlc_opts = "//:test_erlc_opts",
-        deps = ["//deps/amqp_client:erlang_app"],
+        deps = ["//deps/amqp10_common:erlang_app", "//deps/amqp_client:erlang_app"],
     )
     erlang_bytecode(
         name = "transactions_SUITE_beam_files",

deps/rabbit/src/mc.erl

@@ -26,6 +26,7 @@
          priority/1,
          set_ttl/2,
          x_header/2,
+         x_headers/1,
          routing_headers/2,
          exchange/1,
          routing_keys/1,
@@ -88,6 +89,7 @@
                         {timestamp, non_neg_integer()} |
                         {list, [tagged_value()]} |
                         {map, [{tagged_value(), tagged_value()}]} |
+                        {array, atom(), [tagged_value()]} |
                         null |
                         undefined.
 
@@ -104,11 +106,16 @@
     {MetadataSize :: non_neg_integer(),
      PayloadSize :: non_neg_integer()}.
 
-%% retrieve and x- header from the protocol data
+%% retrieve an x- header from the protocol data
 %% the return value should be tagged with an AMQP 1.0 type
 -callback x_header(binary(), proto_state()) ->
     tagged_value().
 
+%% retrieve x- headers from the protocol data
+%% the return values should be tagged with an AMQP 1.0 type
+-callback x_headers(proto_state()) ->
+    #{binary() => tagged_value()}.
+
 %% retrieve a property field from the protocol data
 %% e.g. message_id, correlation_id
 -callback property(atom(), proto_state()) ->
@@ -148,7 +155,7 @@ init(Proto, Data, Anns) ->
 -spec init(protocol(), term(), annotations(), environment()) -> state().
 init(Proto, Data, Anns0, Env) ->
     {ProtoData, ProtoAnns} = Proto:init(Data),
-    Anns1 = case map_size(Env) == 0 of
+    Anns1 = case map_size(Env) =:= 0 of
                 true -> Anns0;
                 false -> Anns0#{env => Env}
             end,
@@ -214,6 +221,25 @@ x_header(Key, #?MODULE{protocol = Proto,
 x_header(Key, BasicMsg) ->
     mc_compat:x_header(Key, BasicMsg).
 
+-spec x_headers(state()) ->
+    #{binary() => tagged_value()}.
+x_headers(#?MODULE{protocol = Proto,
+                   annotations = Anns,
+                   data = Data}) ->
+    %% x-headers may be have been added to the annotations map.
+    New = maps:filtermap(
+            fun(Key, Val) ->
+                    case mc_util:is_x_header(Key) of
+                        true ->
+                            {true, mc_util:infer_type(Val)};
+                        false ->
+                            false
+                    end
+            end, Anns),
+    maps:merge(Proto:x_headers(Data), New);
+x_headers(BasicMsg) ->
+    mc_compat:x_headers(BasicMsg).
+
 -spec routing_headers(state(), [x_headers | complex_types]) ->
     #{binary() => property_value()}.
 routing_headers(#?MODULE{protocol = Proto,

deps/rabbit/src/mc_amqp.erl

@@ -8,6 +8,7 @@
          init/1,
          size/1,
          x_header/2,
+         x_headers/1,
          property/2,
          routing_headers/2,
          convert_to/3,
@@ -125,6 +126,9 @@ size(#v1{message_annotations = MA,
 x_header(Key, Msg) ->
     message_annotation(Key, Msg, undefined).
 
+x_headers(Msg) ->
+    #{K => V || {{_T, K}, V} <- message_annotations(Msg)}.
+
 property(_Prop, #msg_body_encoded{properties = undefined}) ->
     undefined;
 property(Prop, #msg_body_encoded{properties = Props}) ->
@@ -618,41 +622,16 @@ encode_deaths(Deaths) ->
               {map, Map}
       end, Deaths).
 
-essential_properties(#msg_body_encoded{message_annotations = MA} = Msg) ->
+essential_properties(Msg) ->
     Durable = get_property(durable, Msg),
     Priority = get_property(priority, Msg),
     Timestamp = get_property(timestamp, Msg),
     Ttl = get_property(ttl, Msg),
-    Anns0 = #{?ANN_DURABLE => Durable},
-    Anns = maps_put_truthy(
-             ?ANN_PRIORITY, Priority,
-             maps_put_truthy(
-               ?ANN_TIMESTAMP, Timestamp,
-               maps_put_truthy(
-                 ttl, Ttl,
-                 Anns0))),
-    case MA of
-        [] ->
-            Anns;
-        _ ->
-            lists:foldl(
-              fun ({{symbol, <<"x-routing-key">>},
-                    {utf8, Key}}, Acc) ->
-                      maps:update_with(?ANN_ROUTING_KEYS,
-                                       fun(L) -> [Key | L] end,
-                                       [Key],
-                                       Acc);
-                  ({{symbol, <<"x-cc">>},
-                    {list, CCs0}}, Acc) ->
-                      CCs = [CC || {_T, CC} <- CCs0],
-                      maps:update_with(?ANN_ROUTING_KEYS,
-                                       fun(L) -> L ++ CCs end,
-                                       CCs,
-                                       Acc);
-                  ({{symbol, <<"x-exchange">>},
-                    {utf8, Exchange}}, Acc) ->
-                      Acc#{?ANN_EXCHANGE => Exchange};
-                  (_, Acc) ->
-                      Acc
-              end, Anns, MA)
-    end.
+    Anns = #{?ANN_DURABLE => Durable},
+    maps_put_truthy(
+      ?ANN_PRIORITY, Priority,
+      maps_put_truthy(
+        ?ANN_TIMESTAMP, Timestamp,
+        maps_put_truthy(
+          ttl, Ttl,
+          Anns))).

deps/rabbit/src/mc_amqpl.erl

@@ -11,6 +11,7 @@
          init/1,
          size/1,
          x_header/2,
+         x_headers/1,
          routing_headers/2,
          convert_to/3,
          convert_from/3,
@@ -273,6 +274,23 @@ x_header(Key, #content{properties = none} = Content0) ->
     Content = rabbit_binary_parser:ensure_content_decoded(Content0),
     x_header(Key, Content).
 
+x_headers(#content{properties = #'P_basic'{headers = undefined}}) ->
+    #{};
+x_headers(#content{properties = #'P_basic'{headers = Headers}}) ->
+    L = lists:filtermap(
+          fun({Name, Type, Val}) ->
+                  case mc_util:is_x_header(Name) of
+                      true ->
+                          {true, {Name, from_091(Type, Val)}};
+                      false ->
+                          false
+                  end
+          end, Headers),
+    maps:from_list(L);
+x_headers(#content{properties = none} = Content0) ->
+    Content = rabbit_binary_parser:ensure_content_decoded(Content0),
+    x_headers(Content).
+
 property(Prop, Content) ->
     mc_util:infer_type(mc_compat:get_property(Prop, Content)).
 
@@ -707,7 +725,6 @@ supported_header_value_type(table) ->
 supported_header_value_type(_) ->
     true.
 
-
 amqp10_map_get(_K, []) ->
     undefined;
 amqp10_map_get(K, Tuples) ->

deps/rabbit/src/mc_compat.erl

@@ -20,6 +20,7 @@
          priority/1,
          set_ttl/2,
          x_header/2,
+         x_headers/1,
          routing_headers/2,
          %%%
          convert_to/2,
@@ -138,6 +139,9 @@ set_ttl(Value, #basic_message{content = Content0} = Msg) ->
 x_header(Key,#basic_message{content = Content}) ->
     mc_amqpl:x_header(Key, Content).
 
+x_headers(#basic_message{content = Content}) ->
+    mc_amqpl:x_headers(Content).
+
 routing_headers(#basic_message{content = Content}, Opts) ->
     mc_amqpl:routing_headers(Content, Opts).
 

deps/rabbit/src/mc_util.erl

@@ -61,7 +61,7 @@ utf8_string_is_ascii(UTF8String) ->
 amqp_map_get(Key, {map, List}, Default) ->
     amqp_map_get(Key, List, Default);
 amqp_map_get(Key, List, Default) when is_list(List) ->
-    case lists:search(fun ({{_, K}, _}) -> K == Key end, List) of
+    case lists:search(fun ({{_, K}, _}) -> K =:= Key end, List) of
         {value, {_K, V}} ->
             V;
         false ->

deps/rabbit/src/rabbit_amqp_session.erl

@@ -154,6 +154,7 @@
           %% The routing key is either defined in the ATTACH frame and static for
           %% the life time of the link or dynamically provided in each message's
           %% "to" field (address v2) or "subject" field (address v1).
+          %% (A publisher can set additional routing keys via the x-cc message annotation.)
           routing_key :: rabbit_types:routing_key() | to | subject,
           %% queue_name_bin is only set if the link target address refers to a queue.
           queue_name_bin :: undefined | rabbit_misc:resource_name(),
@@ -2369,11 +2370,11 @@ incoming_link_transfer(
 
     Mc0 = mc:init(mc_amqp, PayloadBin, #{}),
     case lookup_target(LinkExchange, LinkRKey, Mc0, Vhost, User, PermCache0) of
-        {ok, X, RoutingKey, Mc1, PermCache} ->
+        {ok, X, RoutingKeys, Mc1, PermCache} ->
             Mc2 = rabbit_message_interceptor:intercept(Mc1),
             check_user_id(Mc2, User),
-            TopicPermCache = check_write_permitted_on_topic(
-                               X, User, RoutingKey, TopicPermCache0),
+            TopicPermCache = check_write_permitted_on_topics(
+                               X, User, RoutingKeys, TopicPermCache0),
             QNames = rabbit_exchange:route(X, Mc2, #{return_binding_keys => true}),
             rabbit_trace:tap_in(Mc2, QNames, ConnName, ChannelNum, Username, Trace),
             Opts = #{correlation => {HandleInt, DeliveryId}},
@@ -2408,14 +2409,14 @@ incoming_link_transfer(
                                    "delivery_tag=~p, delivery_id=~p, reason=~p",
                                    [DeliveryTag, DeliveryId, Reason])
             end;
-        {error, #'v1_0.error'{} = Err} ->
+        {error, {anonymous_terminus, false}, #'v1_0.error'{} = Err} ->
             Disposition = case Settled of
                               true -> [];
                               false -> [released(DeliveryId)]
                           end,
             Detach = [detach(HandleInt, Link0, Err)],
             {error, Disposition ++ Detach};
-        {error, anonymous_terminus, #'v1_0.error'{} = Err} ->
+        {error, {anonymous_terminus, true}, #'v1_0.error'{} = Err} ->
             %% https://docs.oasis-open.org/amqp/anonterm/v1.0/cs01/anonterm-v1.0-cs01.html#doc-routingerrors
             case Settled of
                 true ->
@@ -2440,13 +2441,13 @@ incoming_link_transfer(
     end.
 
 lookup_target(#exchange{} = X, LinkRKey, Mc, _, _, PermCache) ->
-    lookup_routing_key(X, LinkRKey, Mc, PermCache);
+    lookup_routing_key(X, LinkRKey, Mc, false, PermCache);
 lookup_target(#resource{} = XName, LinkRKey, Mc, _, _, PermCache) ->
     case rabbit_exchange:lookup(XName) of
         {ok, X} ->
-            lookup_routing_key(X, LinkRKey, Mc, PermCache);
+            lookup_routing_key(X, LinkRKey, Mc, false, PermCache);
         {error, not_found} ->
-            {error, error_not_found(XName)}
+            {error, {anonymous_terminus, false}, error_not_found(XName)}
     end;
 lookup_target(to, to, Mc, Vhost, User, PermCache0) ->
     case mc:property(to, Mc) of
@@ -2458,25 +2459,26 @@ lookup_target(to, to, Mc, Vhost, User, PermCache0) ->
                     case rabbit_exchange:lookup(XName) of
                         {ok, X} ->
                             check_internal_exchange(X),
-                            lookup_routing_key(X, RKey, Mc, PermCache);
+                            lookup_routing_key(X, RKey, Mc, true, PermCache);
                         {error, not_found} ->
-                            {error, anonymous_terminus, error_not_found(XName)}
+                            {error, {anonymous_terminus, true}, error_not_found(XName)}
                     end;
                 {error, bad_address} ->
-                    {error, anonymous_terminus,
+                    {error, {anonymous_terminus, true},
                      #'v1_0.error'{
                         condition = ?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
                         description = {utf8, <<"bad 'to' address string: ", String/binary>>}}}
             end;
         undefined ->
-            {error, anonymous_terminus,
+            {error, {anonymous_terminus, true},
              #'v1_0.error'{
                 condition = ?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
                 description = {utf8, <<"anonymous terminus requires 'to' address to be set">>}}}
     end.
 
 lookup_routing_key(X = #exchange{name = #resource{name = XNameBin}},
-                   RKey0, Mc0, PermCache) ->
+                   RKey0, Mc0, AnonTerm, PermCache) ->
+    Mc1 = mc:set_annotation(?ANN_EXCHANGE, XNameBin, Mc0),
     RKey = case RKey0 of
                subject ->
                    case mc:property(subject, Mc0) of
@@ -2488,9 +2490,31 @@ lookup_routing_key(X = #exchange{name = #resource{name = XNameBin}},
                _ when is_binary(RKey0) ->
                    RKey0
            end,
-    Mc1 = mc:set_annotation(?ANN_EXCHANGE, XNameBin, Mc0),
-    Mc = mc:set_annotation(?ANN_ROUTING_KEYS, [RKey], Mc1),
-    {ok, X, RKey, Mc, PermCache}.
+    case mc:x_header(<<"x-cc">>, Mc0) of
+        undefined ->
+            RKeys = [RKey],
+            Mc = mc:set_annotation(?ANN_ROUTING_KEYS, RKeys, Mc1),
+            {ok, X, RKeys, Mc, PermCache};
+        {list, CCs0} = L ->
+            try lists:map(fun({utf8, CC}) -> CC end, CCs0) of
+                CCs ->
+                    RKeys = [RKey | CCs],
+                    Mc = mc:set_annotation(?ANN_ROUTING_KEYS, RKeys, Mc1),
+                    {ok, X, RKeys, Mc, PermCache}
+            catch error:function_clause ->
+                      {error, {anonymous_terminus, AnonTerm}, bad_x_cc(L)}
+            end;
+        BadValue ->
+            {error, {anonymous_terminus, AnonTerm}, bad_x_cc(BadValue)}
+    end.
+
+bad_x_cc(Value) ->
+    Desc = unicode:characters_to_binary(
+             lists:flatten(
+               io_lib:format(
+                 "bad value for 'x-cc' message-annotation: ~tp", [Value]))),
+    #'v1_0.error'{condition = ?V_1_0_AMQP_ERROR_INVALID_FIELD,
+                  description = {utf8, Desc}}.
 
 process_routing_confirm([], _SenderSettles = true, _, U) ->
     rabbit_global_counters:messages_unroutable_dropped(?PROTOCOL, 1),
@@ -3445,14 +3469,20 @@ check_resource_access(Resource, Perm, User, Cache) ->
             end
     end.
 
--spec check_write_permitted_on_topic(
+-spec check_write_permitted_on_topics(
         rabbit_types:exchange(),
         rabbit_types:user(),
-        rabbit_types:routing_key(),
+        [rabbit_types:routing_key(),...],
         topic_permission_cache()) ->
     topic_permission_cache().
-check_write_permitted_on_topic(Resource, User, RoutingKey, TopicPermCache) ->
-    check_topic_authorisation(Resource, User, RoutingKey, write, TopicPermCache).
+check_write_permitted_on_topics(#exchange{type = topic} = Resource,
+                                User, RoutingKeys, TopicPermCache) ->
+    lists:foldl(
+      fun(RoutingKey, Cache) ->
+              check_topic_authorisation(Resource, User, RoutingKey, write, Cache)
+      end, TopicPermCache, RoutingKeys);
+check_write_permitted_on_topics(_, _, _, TopicPermCache) ->
+    TopicPermCache.
 
 -spec check_read_permitted_on_topic(
         rabbit_types:exchange(),