Intercept message directly before routing

Intercept message directly before routing and have this consistent
across all protocols since it's easier to reason about.

Author: ansd

deps/rabbit/src/rabbit_amqp_session.erl

@@ -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

@@ -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/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -1637,11 +1637,11 @@ publish_to_queues(
     Anns = #{?ANN_EXCHANGE => ExchangeNameBin,
              ?ANN_ROUTING_KEYS => [mqtt_to_amqp(Topic)]},
     Msg0 = mc:init(mc_mqtt, MqttMsg, Anns, mc_env()),
-    Msg = rabbit_message_interceptor:intercept(Msg0,
-                                               msg_interceptor_ctx(State),
-                                               incoming_message_interceptors),
     case rabbit_exchange:lookup(ExchangeName) of
         {ok, Exchange} ->
+            Msg = rabbit_message_interceptor:intercept(Msg0,
+                                                       msg_interceptor_ctx(State),
+                                                       incoming_message_interceptors),
             QNames0 = rabbit_exchange:route(Exchange, Msg, #{return_binding_keys => true}),
             QNames = drop_local(QNames0, State),
             rabbit_trace:tap_in(Msg, QNames, ConnName, Username, TraceState),