Merge pull request #8304 from thalesmg/fix-client-sub-hook-50

fix(hooks): allow `client.subscribe` hook to reject subscriptions

apps/emqx/src/emqx.appup.src

@@ -2,12 +2,14 @@
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
   [{"5.0.0",
-    [{load_module,emqx_schema,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_channel,brutal_purge,soft_purge,[]},
+     {load_module,emqx_schema,brutal_purge,soft_purge,[]},
      {load_module,emqx_release,brutal_purge,soft_purge,[]},
      {load_module,emqx_relup}]},
    {<<".*">>,[]}],
   [{"5.0.0",
-    [{load_module,emqx_schema,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_channel,brutal_purge,soft_purge,[]},
+     {load_module,emqx_schema,brutal_purge,soft_purge,[]},
      {load_module,emqx_release,brutal_purge,soft_purge,[]},
      {load_module,emqx_relup}]},
    {<<".*">>,[]}]}.

apps/emqx/src/emqx_channel.erl

@@ -513,12 +513,6 @@ handle_in(
                 true ->
                     handle_out(disconnect, ?RC_NOT_AUTHORIZED, Channel);
                 false ->
-                    Replace = fun
-                        _Fun(TupleList, [Tuple = {Key, _Value} | More]) ->
-                            _Fun(lists:keyreplace(Key, 1, TupleList, Tuple), More);
-                        _Fun(TupleList, []) ->
-                            TupleList
-                    end,
                     TopicFilters2 = [TopicFilter || {TopicFilter, 0} <- TupleTopicFilters0],
                     TopicFilters3 = run_hooks(
                         'client.subscribe',
@@ -530,7 +524,18 @@ handle_in(
                         Properties,
                         Channel
                     ),
-                    TupleTopicFilters2 = Replace(TupleTopicFilters0, TupleTopicFilters1),
+                    TupleTopicFilters2 =
+                        lists:foldl(
+                            fun
+                                ({{Topic, Opts = #{delete := true}}, _QoS}, Acc) ->
+                                    Key = {Topic, maps:without([delete], Opts)},
+                                    lists:keydelete(Key, 1, Acc);
+                                (Tuple = {Key, _Value}, Acc) ->
+                                    lists:keyreplace(Key, 1, Acc, Tuple)
+                            end,
+                            TupleTopicFilters0,
+                            TupleTopicFilters1
+                        ),
                     ReasonCodes2 = [
                         ReasonCode
                      || {_TopicFilter, ReasonCode} <- TupleTopicFilters2

apps/emqx/test/emqx_broker_SUITE.erl

@@ -715,6 +715,24 @@ t_connack_auth_error(Config) when is_list(Config) ->
     ?assertEqual(2, emqx_metrics:val('packets.connack.auth_error')),
     ok.
 
+t_handle_in_empty_client_subscribe_hook({init, Config}) ->
+    Hook = {?MODULE, client_subscribe_delete_all_hook, []},
+    ok = emqx_hooks:put('client.subscribe', Hook, _Priority = 100),
+    Config;
+t_handle_in_empty_client_subscribe_hook({'end', _Config}) ->
+    emqx_hooks:del('client.subscribe', {?MODULE, client_subscribe_delete_all_hook}),
+    ok;
+t_handle_in_empty_client_subscribe_hook(Config) when is_list(Config) ->
+    {ok, C} = emqtt:start_link(),
+    {ok, _} = emqtt:connect(C),
+    try
+        {ok, _, RCs} = emqtt:subscribe(C, <<"t">>),
+        ?assertEqual([], RCs),
+        ok
+    after
+        emqtt:disconnect(C)
+    end.
+
 wait_for_events(Action, Kinds) ->
     wait_for_events(Action, Kinds, 500).
 
@@ -771,3 +789,7 @@ recv_msgs(Count, Msgs) ->
     after 100 ->
         Msgs
     end.
+
+client_subscribe_delete_all_hook(_ClientInfo, _Username, TopicFilter) ->
+    EmptyFilters = [{T, Opts#{delete => true}} || {T, Opts} <- TopicFilter],
+    {stop, EmptyFilters}.