Merge pull request #9904 from lafirest/feat/kick_when_banned

feat(banned): session will be kicked when client is banned by `clientid`

changes/v4.4.15-en.md

@@ -12,6 +12,8 @@
 - Security enhancement for retained messages [#9790](https://github.com/emqx/emqx/pull/9790).
   The retained messages will not be published if the publisher client is banned.
 
+- Now the corresponding session will be kicked when client is banned by `clientid` [#9904](https://github.com/emqx/emqx/pull/9904).
+
 ## Bug fixes
 
 - The returned client lists of HTTP query `GET /api/v4/clients?_page=2&_limit=20` to different nodes might be inconsistent [#9926](https://github.com/emqx/emqx/pull/9926).

changes/v4.4.15-zh.md

@@ -12,6 +12,8 @@
 - 增强 `保留消息` 的安全性 [#9790](https://github.com/emqx/emqx/pull/9790)。
   现在投递保留消息前，会先过滤掉来源客户端被封禁了的那些消息。
 
+- 现在客户端通过 `clientid` 被封禁时将会踢掉对应的会话 [#9904](https://github.com/emqx/emqx/pull/9904)。
+
 ## 修复
 
 - 使用 HTTP API `GET /api/v4/clients?_page=2&_limit=20` 请求客户端列表时，请求发送到不同的 emqx 节点，返回的客户端列表可能不一致 [#9926](https://github.com/emqx/emqx/pull/9926)。

src/emqx_banned.erl

@@ -21,6 +21,7 @@
 -include("emqx.hrl").
 -include("logger.hrl").
 -include("types.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
 
 -logger_header("[Banned]").
 
@@ -95,13 +96,15 @@ create(#{who    := Who,
          reason := Reason,
          at     := At,
          until  := Until}) ->
-    mnesia:dirty_write(?BANNED_TAB, #banned{who = Who,
+    insert_banned(#banned{
+        who = Who,
         by = By,
         reason = Reason,
         at = At,
-                                            until = Until});
+        until = Until
+    });
 create(Banned) when is_record(Banned, banned) ->
-    mnesia:dirty_write(?BANNED_TAB, Banned).
+    insert_banned(Banned).
 
 -spec(delete({clientid, emqx_types:clientid()}
            | {username, emqx_types:username()}
@@ -162,3 +165,21 @@ expire_banned_items(Now) ->
               mnesia:delete_object(?BANNED_TAB, B, sticky_write);
          (_, _Acc) -> ok
       end, ok, ?BANNED_TAB).
+
+insert_banned(Banned) ->
+    mnesia:dirty_write(?BANNED_TAB, Banned),
+    on_banned(Banned).
+
+on_banned(#banned{who = {clientid, ClientId}}) ->
+    %% kick the session if the client is banned by clientid
+    ?tp(
+        warning,
+        kick_session_due_to_banned,
+        #{
+            clientid => ClientId
+        }
+    ),
+    emqx_cm:kick_session(ClientId),
+    ok;
+on_banned(_) ->
+    ok.

test/emqx_banned_SUITE.erl

@@ -21,11 +21,12 @@
 
 -include_lib("emqx/include/emqx.hrl").
 -include_lib("eunit/include/eunit.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
 
 all() -> emqx_ct:all(?MODULE).
 
 init_per_suite(Config) ->
-    application:load(emqx),
+    emqx_ct_helpers:start_apps([]),
     ok = ekka:start(),
     %% for coverage
     ok = emqx_banned:mnesia(copy),
@@ -34,7 +35,8 @@ init_per_suite(Config) ->
 end_per_suite(_Config) ->
     ekka:stop(),
     ekka_mnesia:ensure_stopped(),
-    ekka_mnesia:delete_schema().
+    ekka_mnesia:delete_schema(),
+    emqx_ct_helpers:stop_apps([]).
 
 t_add_delete(_) ->
     Banned = #banned{who = {clientid, <<"TestClient">>},
@@ -84,7 +86,10 @@ t_check(_) ->
     ?assertEqual(0, emqx_banned:info(size)).
 
 t_unused(_) ->
-    {ok, Banned} = emqx_banned:start_link(),
+    Banned = case emqx_banned:start_link() of
+                 {ok, Pid} -> Pid;
+                 {error, {already_started, Pid}} -> Pid
+             end,
     ok = emqx_banned:create(#banned{who = {clientid, <<"BannedClient">>},
                                     until = erlang:system_time(second)}),
     ?assertEqual(ignored, gen_server:call(Banned, unexpected_req)),
@@ -93,3 +98,22 @@ t_unused(_) ->
     timer:sleep(500), %% expiry timer
     ok = emqx_banned:stop().
 
+t_kick(_) ->
+    ClientId = <<"client">>,
+    snabbkaffe:start_trace(),
+
+    Now = erlang:system_time(second),
+    Who = {clientid, ClientId},
+
+    emqx_banned:create(#{
+        who => Who,
+        by => <<"test">>,
+        reason => <<"test">>,
+        at => Now,
+        until => Now + 120
+    }),
+
+    Trace = snabbkaffe:collect_trace(),
+    snabbkaffe:stop(),
+    emqx_banned:delete(Who),
+    ?assertEqual(1, length(?of_kind(kick_session_due_to_banned, Trace))).