Add username annotation message interceptor for MQTT

Author: getlarge

deps/rabbitmq_mqtt/priv/schema/rabbitmq_mqtt.schema

@@ -332,6 +332,15 @@ end}.
            false ->
              Acc
          end;
+        ({["mqtt", "message_interceptors", "incoming", "set_username_annotation", "enabled"], Enabled}, Acc) ->
+         case Enabled of
+           true ->
+             Mod = rabbit_mqtt_msg_interceptor_username,
+             Cfg = #{},
+             [{Mod, Cfg} | Acc];
+           false ->
+             Acc
+         end;
         (Other, _Acc) ->
          cuttlefish:invalid(io_lib:format("~p is invalid", [Other]))
     end, [], L)

deps/rabbitmq_mqtt/src/rabbit_mqtt_msg_interceptor_username.erl

@@ -0,0 +1,20 @@
+%% 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_mqtt_msg_interceptor_username).
+-behaviour(rabbit_msg_interceptor).
+
+-export([intercept/4]).
+
+-define(KEY, <<"x-opt-mqtt-username">>).
+
+intercept(Msg, #{protocol := Proto, username := Username}, incoming, _Cfg)
+  when Proto =:= mqtt50 orelse
+       Proto =:= mqtt311 orelse
+       Proto =:= mqtt310 ->
+    mc:set_annotation(?KEY, Username, Msg);
+intercept(Msg, _Ctx, _Stage, _Cfg) ->
+    Msg.

deps/rabbitmq_mqtt/test/config_schema_SUITE_data/rabbitmq_mqtt.snippets

@@ -106,11 +106,11 @@
  {ssl_cert_login_other_name_san_type,
   "mqtt.ssl_cert_login_san_type = other_name",
   [{rabbitmq_mqtt,[{ssl_cert_login_san_type,other_name}]}],
-  [rabbitmq_mqtt]},    
+  [rabbitmq_mqtt]},
  {ssl_cert_login_san_index,
   "mqtt.ssl_cert_login_san_index = 0",
   [{rabbitmq_mqtt,[{ssl_cert_login_san_index,0}]}],
-  [rabbitmq_mqtt]},   
+  [rabbitmq_mqtt]},
  {proxy_protocol,
   "listeners.tcp.default = 5672
    mqtt.allow_anonymous  = true
@@ -175,15 +175,18 @@
 
   {message_interceptor_enabled,
    "mqtt.message_interceptors.incoming.set_client_id_annotation.enabled = true",
+   "mqtt.message_interceptors.incoming.set_username_annotation.enabled = true",
    [{rabbitmq_mqtt, [
       {message_interceptors, [
-        {rabbit_mqtt_msg_interceptor_client_id, #{}}
+        {rabbit_mqtt_msg_interceptor_client_id, #{}},
+        {rabbit_mqtt_msg_interceptor_username, #{}},
       ]}
      ]}],
    []},
 
   {message_interceptor_disabled,
    "mqtt.message_interceptors.incoming.set_client_id_annotation.enabled = false",
+   "mqtt.message_interceptors.incoming.set_username_annotation.enabled = false",
    [{rabbitmq_mqtt, [
       {message_interceptors, []}
      ]}],

deps/rabbitmq_mqtt/test/mqtt_shared_SUITE.erl

@@ -1782,6 +1782,7 @@ message_interceptors(Config) ->
              [message_interceptors,
               [
                {rabbit_mqtt_msg_interceptor_client_id, #{}},
+               {rabbit_mqtt_msg_interceptor_username, #{}},
                {rabbit_msg_interceptor_timestamp, #{overwrite => false,
                                                     incoming => true,
                                                     outgoing => true}}
@@ -1821,6 +1822,9 @@ message_interceptors(Config) ->
     ?assertEqual({<<"x-opt-mqtt-client-id">>, longstr, ClientId},
                  lists:keyfind(<<"x-opt-mqtt-client-id">>, 1, Headers)),
 
+    ?assertEqual({<<"x-opt-mqtt-username">>, longstr, <<"guest">>},
+                 lists:keyfind(<<"x-opt-mqtt-username">>, 1, Headers)),
+
     #'basic.qos_ok'{}  = amqp_channel:call(Ch, #'basic.qos'{prefetch_count = 1}),
     CTag = <<"my ctag">>,
     #'basic.consume_ok'{} = amqp_channel:subscribe(