Improve message interceptors

1. Force the config for timestamp and routing node message interceptors
   to be configured with the overwrite boolean() to avoid defining
   multiple default values throughout the code.
2. Add type specs
3. Extend existing test case for new MQTT client ID interceptor
4. routing node and timestamp should only set the annotation for
   incoming_message_interceptors group
5. Fix `rabbitmq.conf`.
   Prior to this commit there were several issue:
   a.) Setting the right configuration was too user unfriendly, e.g. the user has to set
   ```
   message_interceptor.incoming.rabbit_mqtt_message_interceptor_client_id.annotation_key = x-opt-mqtt-client-id
   ```
   just to enable the MQTT message interceptor.
   b.) The code that parses was too difficult to understand
   c.) MQTT plugin was setting the env for app rabbit, which is an anti-pattern
   d.) disabling a plugin (e.g. MQTT), left its message interceptors still in place
   This is now all fixed, the user sets the rabbitmq.conf as follows:
   ```
   message_interceptors.incoming.set_header_timestamp.overwrite = true
   message_interceptors.incoming.set_header_routing_node.overwrite = false
   mqtt.message_interceptors.incoming.set_client_id_annotation.enabled = true
   ```
   Note that the first two lines use the same format as for RabbitMQ 4.0
   for backwards compatiblity. The last line (MQTT) follows a similar
   pattern.

Author: ansd

deps/rabbit/priv/schema/rabbit.schema

@@ -2667,103 +2667,26 @@ end}.
 {mapping, "message_interceptors.incoming.$interceptor.overwrite", "rabbit.incoming_message_interceptors", [
     {datatype, {enum, [true, false]}}]}.
 
-% Pseudo-key to include the interceptor in the list of interceptors.
-%  - If any other configuration is provided for the interceptor this
-%  configuration is not required.
-%  - If no other configuration is provided, this one is required so that the
-%  interceptor gets invoked.
-{mapping, "message_interceptors.incoming.$interceptor.enabled", "rabbit.incoming_message_interceptors", [
-    {datatype, {enum, [true]}}]}.
-
-{mapping, "message_interceptors.outgoing.$interceptor.enabled", "rabbit.outgoing_message_interceptors", [
-    {datatype, {enum, [true]}}]}.
-
-{mapping,
-    "message_interceptors.incoming.set_header_timestamp.overwrite",
-    "rabbit.incoming_message_interceptors",
-    [{datatype, {enum, [true, false]}}]}.
-{mapping,
-    "message_interceptors.incoming.rabbit_message_interceptor_routing_node.overwrite",
-    "rabbit.incoming_message_interceptors",
-    [{datatype, {enum, [true, false]}}]}.
-
-{mapping,
-    "message_interceptors.incoming.set_header_routing_node.overwrite",
-    "rabbit.incoming_message_interceptors",
-    [{datatype, {enum, [true, false]}}]}.
-{mapping,
-    "message_interceptors.incoming.rabbit_message_interceptor_timestamp.overwrite",
-    "rabbit.incoming_message_interceptors",
-    [{datatype, {enum, [true, false]}}]}.
-
 {translation, "rabbit.incoming_message_interceptors",
-    fun(Conf) ->
-        case cuttlefish_variable:filter_by_prefix("message_interceptors.incoming", Conf) of
-            [] ->
-                cuttlefish:unset();
-            L ->
-                InterceptorsConfig = [
-                    {Module0, Config, Value}
-                    || {["message_interceptors", "incoming", Module0, Config], Value} <- L
-                ],
-                {Result, Order0} = lists:foldl(
-                    fun({Interceptor0, Key0, Value}, {Acc, Order}) ->
-                        Interceptor = list_to_atom(Interceptor0),
-                        Key = list_to_atom(Key0),
-                        MapPutFun = fun(Old) -> maps:put(Key, Value, Old) end,
-                        % This Interceptor -> Module alias exists for
-                        % compatibility reasons
-                        Module = case Interceptor  of
-                                     set_header_timestamp ->
-                                         rabbit_message_interceptor_timestamp;
-                                     set_header_routing_node ->
-                                         rabbit_message_interceptor_routing_node;
-                                     _ ->
-                                         Interceptor
-                                 end,
-                        NewAcc = maps:update_with(Module,
-                                                  MapPutFun,
-                                                  #{Key => Value},
-                                                  Acc),
-                        {NewAcc, [Module| Order]}
-                    end,
-                    {#{}, []},
-                    InterceptorsConfig
-                ),
-                Order = lists:uniq(Order0),
-                [{O, maps:without([enabled], maps:get(O, Result))} || O <- Order]
-        end
-    end
-}.
-
-{translation, "rabbit.outgoing_message_interceptors",
-    fun(Conf) ->
-        case cuttlefish_variable:filter_by_prefix("message_interceptors.outgoing", Conf) of
-            [] ->
-                cuttlefish:unset();
-            L ->
-                InterceptorsConfig = [
-                    {Module0, Config, Value}
-                    || {["message_interceptors", "outgoing", Module0, Config], Value} <- L
-                ],
-                {Result, Order0} = lists:foldl(
-                    fun({Interceptor0, Key0, Value}, {Acc, Order}) ->
-                        Module = list_to_atom(Interceptor0),
-                        Key = list_to_atom(Key0),
-                        MapPutFun = fun(Old) -> maps:put(Key, Value, Old) end,
-                        NewAcc = maps:update_with(Module,
-                                                  MapPutFun,
-                                                  #{Key => Value},
-                                                  Acc),
-                        {NewAcc, [Module| Order]}
-                    end,
-                    {#{}, []},
-                    InterceptorsConfig
-                ),
-                Order = lists:uniq(Order0),
-                [{O, maps:without([enabled], maps:get(O, Result))} || O <- Order]
-        end
-    end
+ fun(Conf) ->
+         case cuttlefish_variable:filter_by_prefix("message_interceptors", Conf) of
+             [] ->
+                 cuttlefish:unset();
+             L ->
+                 [begin
+                      Interceptor = list_to_atom(Interceptor0),
+                      Mod = case Interceptor of
+                                set_header_timestamp ->
+                                    rabbit_message_interceptor_timestamp;
+                                set_header_routing_node ->
+                                    rabbit_message_interceptor_routing_node;
+                                _ ->
+                                    cuttlefish:invalid(io_lib:format("~p is invalid", [Interceptor]))
+                            end,
+                      {Mod, #{overwrite => Overwrite}}
+                  end || {["message_interceptors", "incoming", Interceptor0, "overwrite"], Overwrite} <- L]
+         end
+ end
 }.
 
 {mapping, "stream.replication.port_range.min", "osiris.port_range", [

deps/rabbit/src/rabbit.erl

@@ -1655,11 +1655,12 @@ persist_static_configuration() ->
     persist_static_configuration(
       [classic_queue_index_v2_segment_entry_count,
        classic_queue_store_v2_max_cache_size,
-       classic_queue_store_v2_check_crc32,
-       incoming_message_interceptors,
-       outgoing_message_interceptors
+       classic_queue_store_v2_check_crc32
       ]),
 
+    Interceptors = application:get_env(?MODULE, incoming_message_interceptors, []),
+    ok = rabbit_message_interceptor:add(Interceptors, incoming_message_interceptors),
+
     %% Disallow the following two cases:
     %% 1. Negative values
     %% 2. MoreCreditAfter greater than InitialCredit

deps/rabbit/src/rabbit_amqp_session.erl

@@ -284,7 +284,7 @@
           max_incoming_window :: pos_integer(),
           max_link_credit :: pos_integer(),
           max_queue_credit :: pos_integer(),
-          msg_interceptor_ctx :: map()
+          msg_interceptor_ctx :: rabbit_message_interceptor:context()
          }).
 
 -record(state, {
@@ -477,9 +477,9 @@ init({ReaderPid, WriterPid, ChannelNum, MaxFrameSize, User, Vhost, ContainerId,
                            max_link_credit = MaxLinkCredit,
                            max_queue_credit = MaxQueueCredit,
                            msg_interceptor_ctx = #{protocol => ?PROTOCOL,
-                                                   username => User#user.username,
                                                    vhost => Vhost,
-                                                   conn_name => ConnName}
+                                                   username => User#user.username,
+                                                   connection_name => ConnName}
                           }}}.
 
 terminate(_Reason, #state{incoming_links = IncomingLinks,
@@ -2442,12 +2442,12 @@ incoming_link_transfer(
     Mc0 = mc:init(mc_amqp, PayloadBin, #{}),
     case lookup_target(LinkExchange, LinkRKey, Mc0, Vhost, User, PermCache0) of
         {ok, X, RoutingKeys, Mc1, PermCache} ->
+            check_user_id(Mc1, User),
+            TopicPermCache = check_write_permitted_on_topics(
+                               X, User, RoutingKeys, TopicPermCache0),
             Mc2 = rabbit_message_interceptor:intercept(Mc1,
                                                        MsgInterceptorCtx,
                                                        incoming_message_interceptors),
-            check_user_id(Mc2, User),
-            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}},

deps/rabbit/src/rabbit_channel.erl

@@ -111,7 +111,7 @@
           max_consumers,  % taken from rabbit.consumer_max_per_channel
           %% defines how ofter gc will be executed
           writer_gc_threshold,
-          msg_interceptor_ctx
+          msg_interceptor_ctx :: rabbit_message_interceptor:context()
          }).
 
 -record(pending_ack, {
@@ -493,6 +493,10 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
     OptionalVariables = extract_variable_map_from_amqp_params(AmqpParams),
     {ok, GCThreshold} = application:get_env(rabbit, writer_gc_threshold),
     MaxConsumers = application:get_env(rabbit, consumer_max_per_channel, infinity),
+    MsgInterceptorCtx = #{protocol => amqp091,
+                          vhost => VHost,
+                          username => User#user.username,
+                          connection_name => ConnName},
     State = #ch{cfg = #conf{state = starting,
                             protocol = Protocol,
                             channel = Channel,
@@ -511,11 +515,7 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
                             authz_context = OptionalVariables,
                             max_consumers = MaxConsumers,
                             writer_gc_threshold = GCThreshold,
-                            msg_interceptor_ctx = #{protocol => amqp091,
-                                                    username => User#user.username,
-                                                    vhost => VHost,
-                                                    conn_name => ConnName}
-                           },
+                            msg_interceptor_ctx = MsgInterceptorCtx},
                 limiter = Limiter,
                 tx                      = none,
                 next_tag                = 1,
@@ -1213,10 +1213,10 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
             rabbit_misc:precondition_failed("invalid message: ~tp", [Reason]);
         {ok, Message0} ->
             check_write_permitted_on_topics(Exchange, User, Message0, AuthzContext),
+            check_user_id_header(Message0, User),
             Message = rabbit_message_interceptor:intercept(Message0,
                                                            MsgInterceptorCtx,
                                                            incoming_message_interceptors),
-            check_user_id_header(Message, User),
             QNames = rabbit_exchange:route(Exchange, Message, #{return_binding_keys => true}),
             [deliver_reply(RK, Message) || {virtual_reply_queue, RK} <- QNames],
             Queues = rabbit_amqqueue:lookup_many(QNames),

deps/rabbit/src/rabbit_message_interceptor.erl

@@ -1,49 +1,65 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+
 -module(rabbit_message_interceptor).
 
+%% client API
 -export([intercept/3,
-         set_msg_annotation/4]).
+         add/2,
+         remove/2]).
+%% helpers for behaviour implementations
+-export([set_annotation/4]).
 
+%% same protocol names as output by Prometheus endpoint
 -type protocol() :: amqp091 | amqp10 | mqtt310 | mqtt311 | mqtt50.
+-type context() :: #{protocol := protocol(),
+                     vhost := rabbit_types:vhost(),
+                     username := rabbit_types:username(),
+                     connection_name := binary(),
+                     atom() => term()}.
+-type group() :: incoming_message_interceptors |
+                 outgoing_message_interceptors.
+-type config() :: #{atom() => term()}.
+-type interceptor() :: {module(), config()}.
+
+
+-export_type([context/0]).
+
+-callback intercept(mc:state(), context(), group(), config()) ->
+    mc:state().
 
--type msg_interceptor_ctx() :: #{protocol := protocol(),
-                                 vhost := binary(),
-                                 username := binary(),
-                                 conn_name => binary(),
-                                 atom() => term()}.
-
--callback intercept(Msg, MsgInterceptorCtx, Group, Config) -> Msg when
-    Msg :: mc:state(),
-    MsgInterceptorCtx :: msg_interceptor_ctx(),
-    Group :: incoming_message_interceptors | outgoing_message_interceptors,
-    Config :: #{atom() := term()}.
-
--spec intercept(Msg, MsgInterceptorCtx, Group) -> Msg when
-    Msg :: mc:state(),
-    MsgInterceptorCtx :: map(),
-    Group :: incoming_message_interceptors | outgoing_message_interceptors.
-intercept(Msg, MsgInterceptorCtx, Group) ->
-    Interceptors = persistent_term:get(Group, []),
-    lists:foldl(fun({Module, Config}, Msg0) ->
-                        try
-                            Module:intercept(Msg0,
-                                             MsgInterceptorCtx,
-                                             Group,
-                                             Config)
-                        catch
-                            error:undef ->
-                                Msg0
-                        end
-                end, Msg , Interceptors).
-
--spec set_msg_annotation(mc:state(),
-                         mc:ann_key(),
-                         mc:ann_value(),
-                         boolean()
-                        ) -> mc:state().
-set_msg_annotation(Msg, Key, Value, Overwrite) ->
+-spec intercept(mc:state(), context(), group()) ->
+    mc:state().
+intercept(Msg, Ctx, Group) ->
+    Interceptors = persistent_term:get(Group),
+    lists:foldl(fun({Mod, Config}, Msg0) ->
+                        Mod:intercept(Msg0, Ctx, Group, Config)
+                end, Msg, Interceptors).
+
+-spec set_annotation(mc:state(), mc:ann_key(), mc:ann_value(), boolean()) ->
+    mc:state().
+set_annotation(Msg, Key, Value, Overwrite) ->
     case {mc:x_header(Key, Msg), Overwrite} of
         {Val, false} when Val =/= undefined ->
             Msg;
         _ ->
             mc:set_annotation(Key, Value, Msg)
     end.
+
+-spec add([interceptor()], group()) -> ok.
+add(Interceptors, Group) ->
+    %% validation
+    lists:foreach(fun({Mod, #{}}) ->
+                          case erlang:function_exported(Mod, intercept, 4) of
+                              true -> ok;
+                              false -> error(Mod)
+                          end
+                  end, Interceptors),
+    persistent_term:put(Group, persistent_term:get(Group, []) ++ Interceptors).
+
+-spec remove([interceptor()], group()) -> ok.
+remove(Interceptors, Group) ->
+    persistent_term:put(Group, persistent_term:get(Group, []) -- Interceptors).

deps/rabbit/src/rabbit_message_interceptor_routing_node.erl

@@ -1,14 +1,22 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+
 -module(rabbit_message_interceptor_routing_node).
 -behaviour(rabbit_message_interceptor).
 
 -define(HEADER_ROUTING_NODE, <<"x-routed-by">>).
 
 -export([intercept/4]).
 
-intercept(Msg, _MsgInterceptorCtx, _Group, Config) ->
+intercept(Msg, _Ctx, incoming_message_interceptors, Config) ->
     Node = atom_to_binary(node()),
-    Overwrite = maps:get(overwrite, Config, false),
-    rabbit_message_interceptor:set_msg_annotation(Msg,
-                                                  ?HEADER_ROUTING_NODE,
-                                                  Node,
-                                                  Overwrite).
+    Overwrite = maps:get(overwrite, Config),
+    rabbit_message_interceptor:set_annotation(Msg,
+                                              ?HEADER_ROUTING_NODE,
+                                              Node,
+                                              Overwrite);
+intercept(Msg, _Ctx, _Group, _Config) ->
+    Msg.

deps/rabbit/src/rabbit_message_interceptor_timestamp.erl

@@ -1,3 +1,9 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+
 -module(rabbit_message_interceptor_timestamp).
 -behaviour(rabbit_message_interceptor).
 
@@ -7,20 +13,21 @@
 
 -export([intercept/4]).
 
-intercept(Msg0, _MsgInterceptorCtx, _Group, Config) ->
+intercept(Msg0, _Ctx, incoming_message_interceptors, Config) ->
     Ts = mc:get_annotation(?ANN_RECEIVED_AT_TIMESTAMP, Msg0),
-    Overwrite = maps:get(overwrite, Config, false),
-    Msg = rabbit_message_interceptor:set_msg_annotation(
-            Msg0,
-            ?HEADER_TIMESTAMP,
-            Ts,
-            Overwrite),
-    set_msg_timestamp(Msg, Ts, Overwrite).
+    Overwrite = maps:get(overwrite, Config),
+    Msg = rabbit_message_interceptor:set_annotation(Msg0,
+                                                    ?HEADER_TIMESTAMP,
+                                                    Ts,
+                                                    Overwrite),
+    set_timestamp(Msg, Ts, Overwrite);
+intercept(Msg, _MsgInterceptorCtx, _Group, _Config) ->
+    Msg.
 
-set_msg_timestamp(Msg, Timestamp, Overwrite) ->
+set_timestamp(Msg, Ts, Overwrite) ->
     case {mc:timestamp(Msg), Overwrite} of
-        {Ts, false} when is_integer(Ts) ->
+        {ExistingTs, false} when is_integer(ExistingTs) ->
             Msg;
         _ ->
-            mc:set_annotation(?ANN_TIMESTAMP, Timestamp, Msg)
+            mc:set_annotation(?ANN_TIMESTAMP, Ts, Msg)
     end.