Merge pull request #14195 from zmstone/241106-clientid-override

feat: add clientid override

apps/emqx/src/emqx_channel.erl

@@ -360,6 +360,8 @@ handle_in(?CONNECT_PACKET(ConnPkt) = Packet, Channel) ->
                 fun run_conn_hooks/2,
                 fun check_connect/2,
                 fun enrich_client/2,
+                %% set_log_meta should happen after enrich_client
+                %% because client ID assign and override
                 fun set_log_meta/2,
                 fun check_banned/2,
                 fun count_flapping_event/2
@@ -1661,7 +1663,9 @@ enrich_client(ConnPkt, Channel = #channel{clientinfo = ClientInfo}) ->
             fun maybe_username_as_clientid/2,
             fun maybe_assign_clientid/2,
             %% attr init should happen after clientid and username assign
-            fun maybe_set_client_initial_attrs/2
+            fun maybe_set_client_initial_attrs/2,
+            %% clientid override should happen after client_attrs is initialized
+            fun maybe_override_clientid/2
         ],
         ConnPkt,
         ClientInfo
@@ -1716,11 +1720,16 @@ get_client_attrs_init_config(Zone) ->
     get_mqtt_conf(Zone, client_attrs_init, []).
 
 maybe_set_client_initial_attrs(ConnPkt, #{zone := Zone} = ClientInfo) ->
-    Inits = get_client_attrs_init_config(Zone),
-    UserProperty = get_user_property_as_map(ConnPkt),
-    {ok, initialize_client_attrs(Inits, ClientInfo#{user_property => UserProperty})}.
+    case get_client_attrs_init_config(Zone) of
+        [] ->
+            {ok, ClientInfo};
+        Inits ->
+            UserProperty = get_user_property_as_map(ConnPkt),
+            ClientInfo1 = initialize_client_attrs(Inits, ClientInfo#{user_property => UserProperty}),
+            {ok, maps:remove(user_property, ClientInfo1)}
+    end.
 
-initialize_client_attrs(Inits, ClientInfo) ->
+initialize_client_attrs(Inits, #{clientid := ClientId} = ClientInfo) ->
     lists:foldl(
         fun(#{expression := Variform, set_as_attr := Name}, Acc) ->
             Attrs = maps:get(client_attrs, Acc, #{}),
@@ -1731,6 +1740,8 @@ initialize_client_attrs(Inits, ClientInfo) ->
                         #{
                             msg => "client_attr_rednered_to_empty_string",
                             set_as_attr => Name
+                        }#{
+                            clientid => ClientId
                         }
                     ),
                     Acc;
@@ -1741,7 +1752,8 @@ initialize_client_attrs(Inits, ClientInfo) ->
                             msg => "client_attr_initialized",
                             set_as_attr => Name,
                             attr_value => Value
-                        }
+                        },
+                        #{clientid => ClientId}
                     ),
                     Acc#{client_attrs => Attrs#{Name => Value}};
                 {error, Reason} ->
@@ -1750,7 +1762,8 @@ initialize_client_attrs(Inits, ClientInfo) ->
                         #{
                             msg => "client_attr_initialization_failed",
                             reason => Reason
-                        }
+                        },
+                        #{clientid => ClientId}
                     ),
                     Acc
             end
@@ -1759,6 +1772,41 @@ initialize_client_attrs(Inits, ClientInfo) ->
         Inits
     ).
 
+maybe_override_clientid(_ConnPkt, #{zone := Zone} = ClientInfo) ->
+    Expression = get_mqtt_conf(Zone, clientid_override, disabled),
+    {ok, override_clientid(Expression, ClientInfo)}.
+
+override_clientid(disabled, ClientInfo) ->
+    ClientInfo;
+override_clientid(Expression, #{clientid := OrigClientId} = ClientInfo) ->
+    case emqx_variform:render(Expression, ClientInfo) of
+        {ok, <<>>} ->
+            ?SLOG(
+                warning,
+                #{
+                    msg => "clientid_override_expression_returned_empty_string"
+                },
+                #{clientid => OrigClientId}
+            ),
+            ClientInfo;
+        {ok, ClientId} ->
+            % Must add 'clientid' log meta for trace log filter
+            ?TRACE("MQTT", "clientid_overridden", #{
+                clientid => ClientId, original_clientid => OrigClientId
+            }),
+            ClientInfo#{clientid => ClientId};
+        {error, Reason} ->
+            ?SLOG(
+                warning,
+                #{
+                    msg => "clientid_override_expression_failed",
+                    reason => Reason
+                },
+                #{clientid => OrigClientId}
+            ),
+            ClientInfo
+    end.
+
 get_user_property_as_map(#mqtt_packet_connect{properties = #{'User-Property' := UserProperty}}) when
     is_list(UserProperty)
 ->

apps/emqx/src/emqx_schema.erl

@@ -1819,6 +1819,13 @@ fields("banned") ->
             )}
     ].
 
+compile_variform_allow_disabled(disabled, _Opts) ->
+    disabled;
+compile_variform_allow_disabled(<<"disabled">>, _Opts) ->
+    disabled;
+compile_variform_allow_disabled(Expression, Opts) ->
+    compile_variform(Expression, Opts).
+
 compile_variform(undefined, _Opts) ->
     undefined;
 compile_variform(Expression, #{make_serializable := true}) ->
@@ -3734,6 +3741,15 @@ mqtt_general() ->
                     default => [],
                     desc => ?DESC("client_attrs_init")
                 }
+            )},
+        {"clientid_override",
+            sc(
+                hoconsc:union([disabled, typerefl:alias("string", any())]),
+                #{
+                    default => disabled,
+                    desc => ?DESC("clientid_override"),
+                    converter => fun compile_variform_allow_disabled/2
+                }
             )}
     ].
 %% All session's importance should be lower than general part to organize document.

apps/emqx/test/emqx_client_SUITE.erl

@@ -81,7 +81,10 @@ groups() ->
             t_certcn_as_clientid_default_config_tls,
             t_certcn_as_clientid_tlsv1_3,
             t_certcn_as_clientid_tlsv1_2,
-            t_peercert_preserved_before_connected
+            t_peercert_preserved_before_connected,
+            t_clientid_override,
+            t_clientid_override_fail_with_empty_render_result,
+            t_clientid_override_fail_with_expression_exception
         ]}
     ].
 
@@ -99,7 +102,13 @@ init_per_testcase(_Case, Config) ->
     Config.
 
 end_per_testcase(_Case, _Config) ->
-    emqx_config:put_zone_conf(default, [mqtt, idle_timeout], 15000).
+    %% restore default values
+    emqx_config:put_zone_conf(default, [mqtt, idle_timeout], 15000),
+    emqx_config:put_zone_conf(default, [mqtt, use_username_as_clientid], false),
+    emqx_config:put_zone_conf(default, [mqtt, peer_cert_as_clientid], disabled),
+    emqx_config:put_zone_conf(default, [mqtt, client_attrs_init], []),
+    emqx_config:put_zone_conf(default, [mqtt, clientid_override], disabled),
+    ok.
 
 %%--------------------------------------------------------------------
 %% Test cases for MQTT v3
@@ -444,6 +453,34 @@ t_client_attr_from_user_property(_Config) ->
         emqx_cm:get_chan_info(ClientId)
     ),
     emqtt:disconnect(Client).
+
+t_clientid_override(_) ->
+    emqx_logger:set_log_level(debug),
+    ClientId = <<"original-clientid-0">>,
+    Username = <<"username1">>,
+    Override = <<"username">>,
+    {ok, Rule1} = emqx_variform:compile(Override),
+    emqx_config:put_zone_conf(default, [mqtt, clientid_override], Rule1),
+    {ok, Client} = emqtt:start_link([{clientid, ClientId}, {port, 1883}, {username, Username}]),
+    {ok, _} = emqtt:connect(Client),
+    ?assertMatch(#{clientid := Username}, maps:get(clientinfo, emqx_cm:get_chan_info(Username))),
+    ?assertMatch(undefined, emqx_cm:get_chan_info(ClientId)),
+    emqtt:disconnect(Client).
+
+t_clientid_override_fail_with_empty_render_result(_) ->
+    test_clientid_override_fail(<<"original-clientid-1">>, <<"undefined_var">>).
+
+t_clientid_override_fail_with_expression_exception(_) ->
+    test_clientid_override_fail(<<"original-clientid-2">>, <<"nth(1,undefined_var)">>).
+
+test_clientid_override_fail(ClientId, Expr) ->
+    {ok, Rule1} = emqx_variform:compile(Expr),
+    emqx_config:put_zone_conf(default, [mqtt, clientid_override], Rule1),
+    {ok, Client} = emqtt:start_link([{clientid, ClientId}, {port, 1883}]),
+    {ok, _} = emqtt:connect(Client),
+    ?assertMatch(#{clientid := ClientId}, maps:get(clientinfo, emqx_cm:get_chan_info(ClientId))),
+    emqtt:disconnect(Client).
+
 t_certcn_as_clientid_default_config_tls(_) ->
     tls_certcn_as_clientid(default).
 
@@ -480,7 +517,8 @@ t_peercert_preserved_before_connected(_) ->
     ?assertMatch(
         #{conninfo := ConnInfo} when not is_map_key(peercert, ConnInfo),
         emqx_connection:info(ConnPid)
-    ).
+    ),
+    emqtt:disconnect(Client).
 
 on_hook(ConnInfo, _, 'client.connect' = HP, Pid) ->
     _ = Pid ! {HP, ConnInfo},

apps/emqx/test/emqx_config_SUITE.erl

@@ -456,7 +456,8 @@ zone_global_defaults() ->
                 upgrade_qos => false,
                 use_username_as_clientid => false,
                 wildcard_subscription => true,
-                client_attrs_init => []
+                client_attrs_init => [],
+                clientid_override => disabled
             },
         overload_protection =>
             #{

changes/ce/feat-14195.en.md

@@ -0,0 +1,4 @@
+Support clientid override.
+
+EMQX now allows custom client ID overrides with `mqtt.clientid_override={Expression}`, offering more flexibility.
+This deprecates the `use_userid_as_clientid` and `peer_cert_as_clientid` options, which remain available for compatibility until version 6.0.

rel/i18n/emqx_schema.hocon

@@ -1715,6 +1715,24 @@ client_attrs_init_set_as_attr {
     The extracted attribute will be stored in the `client_attrs` property with this name."""
 }
 
+clientid_override {
+  label: "Client ID Override Expression"
+  desc: """~
+    A one line expression to evaluate a set of predefined string functions (like in the rule engine SQL statements).
+    The expression can be a function call with nested calls as its arguments, or direct variable reference.
+    So far, it does not provide user-defined variable binding (like `var a=1`) or user-defined functions.
+    As an example, to extract the prefix of client ID delimited by a dot: `nth(1, tokens(username, '.'))`.
+
+    The variables pre-bound variables are:
+    - `cn`: Client's TLS certificate common name.
+    - `dn`: Client's TLS certificate distinguished name (the subject).
+    - `clientid`: The original MQTT Client ID.
+    - `username`: MQTT Client's username.
+    - `client_attrs.{NAME}`: Client attributes initialized by per config `client_attrs_init`.
+
+    You can read more about variform expressions in EMQX docs."""
+}
+
 banned_bootstrap_file.desc:
 """The bootstrap file is a CSV file used to batch loading banned data when initializing a single node or cluster, in other words, the import operation is performed only if there is no data in the database.