Merge pull request #7531 from zmstone/merge-v43-to-v44

Merge v43 to v44

CHANGES-4.3.md

@@ -10,6 +10,29 @@ File format:
 - One list item per change topic
   Change log ends with a list of github PRs
 
+## v4.3.14
+
+### Enhancements
+
+* In order to fix the execution order of exhook, e.g. before/after other plugins/modules,
+  ExHook now supports user customizing emqx_hook execute priority.
+* add api: PUT /rules/{id}/reset_metrics.
+  This api reset the metrics of the rule engine of a rule, and reset the metrics of the action related to this rule. [#7474]
+* Enhanced rule engine error handling when json parsing error.
+* Add support for `RSA-PSK-AES256-GCM-SHA384`, `RSA-PSK-AES256-CBC-SHA384`,
+ `RSA-PSK-AES128-GCM-SHA256`, `RSA-PSK-AES128-CBC-SHA256` PSK ciphers, and remove `PSK-3DES-EDE-CBC-SHA`,
+ `PSK-RC4-SHA` from the default configuration. [#7427]
+
+### Bug fixes
+
+* Prohibit empty topics in strict mode
+* Make sure ehttpc delete useless pool always succeed.
+* Update mongodb driver to fix potential process leak.
+* Dashboard admin password persists after leaving/joining the cluster
+* Silence grep/sed warnings in docker-entrypoint.sh. [#7520]
+* Generate `loaded_modules` and `loaded_plugins` files with default
+  values when no such files exists. [#7520]
+
 ## v4.3.13
 
 ### Important changes

Makefile

@@ -101,6 +101,7 @@ $(PROFILES:%=clean-%):
 clean-all:
 	@rm -f rebar.lock
 	@rm -rf _build
+	@rm -f rebar.lock
 
 .PHONY: deps-all
 deps-all: $(REBAR) $(PROFILES:%=deps-%)

apps/emqx_auth_jwt/src/emqx_auth_jwt.app.src

@@ -1,6 +1,6 @@
 {application, emqx_auth_jwt,
  [{description, "EMQ X Authentication with JWT"},
-  {vsn, "4.4.0"}, % strict semver, bump manually!
+  {vsn, "4.4.1"}, % strict semver, bump manually!
   {modules, []},
   {registered, [emqx_auth_jwt_sup]},
   {applications, [kernel,stdlib,jose]},

apps/emqx_auth_jwt/src/emqx_auth_jwt.appup.src

@@ -1,13 +1,13 @@
 %% -*-: erlang -*-
 {VSN,
  [
-    {"4.3.0", [
+    {<<"4\\.4\\.0">>, [
       {load_module, emqx_auth_jwt_svr, brutal_purge, soft_purge, []}
     ]},
     {<<".*">>, []}
  ],
  [
-    {"4.3.0", [
+    {<<"4\\.4\\.0">>, [
       {load_module, emqx_auth_jwt_svr, brutal_purge, soft_purge, []}
     ]},
     {<<".*">>, []}

apps/emqx_auth_jwt/src/emqx_auth_jwt_svr.erl

@@ -91,7 +91,7 @@ do_init_jwks(Options) ->
                                   [K, V, Reason]),
                              undefined;
                          J -> J
-                     catch T:R:_ ->
+                     catch T:R ->
                          ?LOG(warning, "Build ~p JWK ~p failed: {~p, ~p}~n",
                               [K, V, T, R]),
                          undefined

apps/emqx_auth_mongo/rebar.config

@@ -1,7 +1,6 @@
 {deps,
   %% NOTE: mind poolboy version when updating mongodb-erlang version
- [{mongodb, {git,"https://github.com/emqx/mongodb-erlang", {tag, "v3.0.12"}}},
+ [%% mongodb-erlang uses a special fork https://github.com/comtihon/poolboy.git
-  %% mongodb-erlang uses a special fork https://github.com/comtihon/poolboy.git
   %% (which has overflow_ttl feature added).
   %% However, it references `{branch, "master}` (commit 9c06a9a on 2021-04-07).
   %% By accident, We have always been using the upstream fork due to
@@ -29,4 +28,3 @@
 {cover_enabled, true}.
 {cover_opts, [verbose]}.
 {cover_export_enabled, true}.
-

apps/emqx_bridge_mqtt/README.md

@@ -227,7 +227,7 @@ topic:   topic2/#
 Add a forwarding topic for the specified bridge
 
 ``` bash
-$ ./bin/emqx_ctl bridges add-forwards emqx topic3/#
+$ ./bin/emqx_ctl bridges add-forward emqx topic3/#
 Add-forward topic successfully.
 ```
 
@@ -235,7 +235,7 @@ Delete the forwarding topic for the specified bridge
 
 
 ``` bash
-$ ./bin/emqx_ctl bridges del-forwards emqx topic3/#
+$ ./bin/emqx_ctl bridges del-forward emqx topic3/#
 Del-forward topic successfully.
 ```
 

apps/emqx_bridge_mqtt/docs/guide.rst

@@ -250,14 +250,14 @@ Add a forwarding topic for the specified bridge
 
 .. code-block:: bash
 
-   $ ./bin/emqx_ctl bridges add-forwards emqx topic3/#
+   $ ./bin/emqx_ctl bridges add-forward emqx topic3/#
    Add-forward topic successfully.
 
 Delete the forwarding topic for the specified bridge
 
 .. code-block:: bash
 
-   $ ./bin/emqx_ctl bridges del-forwards emqx topic3/#
+   $ ./bin/emqx_ctl bridges del-forward emqx topic3/#
    Del-forward topic successfully.
 
 List subscriptions for the specified bridge

apps/emqx_bridge_mqtt/etc/emqx_bridge_mqtt.conf

@@ -129,7 +129,7 @@ bridge.mqtt.aws.ciphers = TLS_AES_256_GCM_SHA384,TLS_AES_128_GCM_SHA256,TLS_CHAC
 ## Note that 'bridge.${BridgeName}.ciphers' and 'bridge.${BridgeName}.psk_ciphers' cannot
 ## be configured at the same time.
 ## See 'https://tools.ietf.org/html/rfc4279#section-2'.
-#bridge.mqtt.aws.psk_ciphers = PSK-AES128-CBC-SHA,PSK-AES256-CBC-SHA,PSK-3DES-EDE-CBC-SHA,PSK-RC4-SHA
+#bridge.mqtt.aws.psk_ciphers = RSA-PSK-AES256-GCM-SHA384,RSA-PSK-AES256-CBC-SHA384,RSA-PSK-AES128-GCM-SHA256,RSA-PSK-AES128-CBC-SHA256,RSA-PSK-AES256-CBC-SHA,RSA-PSK-AES128-CBC-SHA
 
 ## Ping interval of a down bridge.
 ##

apps/emqx_bridge_mqtt/priv/emqx_bridge_mqtt.schema

@@ -134,13 +134,29 @@
 ]}.
 
 {translation, "emqx_bridge_mqtt.bridges", fun(Conf) ->
-
+    AvaiableCiphers = ["RSA-PSK-AES256-GCM-SHA384","RSA-PSK-AES256-CBC-SHA384",
+                       "RSA-PSK-AES128-GCM-SHA256","RSA-PSK-AES128-CBC-SHA256",
+                       "RSA-PSK-AES256-CBC-SHA","RSA-PSK-AES128-CBC-SHA"
+                      ],
+    %% Compatible with legacy PSK Cipher strings
+    PskMapping = fun("PSK-AES128-CBC-SHA") -> {true, "RSA-PSK-AES128-CBC-SHA"};
+                    ("PSK-AES256-CBC-SHA") -> {true, "RSA-PSK-AES256-CBC-SHA"};
+                    ("PSK-3DES-EDE-CBC-SHA") -> {true, "PSK-3DES-EDE-CBC-SHA"};
+                    ("PSK-RC4-SHA") -> {true, "PSK-RC4-SHA"};
+                    (C) -> case lists:member(C, AvaiableCiphers) of
+                               true -> {true, C};
+                               false -> false
+                           end
+                 end,
     MapPSKCiphers = fun(PSKCiphers) ->
-                        lists:map(
+                      lists:filtermap(fun(C0) ->
-                          fun("PSK-AES128-CBC-SHA") -> {psk, aes_128_cbc, sha};
+                        case PskMapping(C0) of
-                             ("PSK-AES256-CBC-SHA") -> {psk, aes_256_cbc, sha};
+                            false ->
-                             ("PSK-3DES-EDE-CBC-SHA") -> {psk, '3des_ede_cbc', sha};
+                              cuttlefish:invalid(
-                             ("PSK-RC4-SHA") -> {psk, rc4_128, sha}
+                                io_lib:format("psk_ciphers: not support ~s", [C0]));
+                            {true, C} ->
+                                {true, C}
+                        end
                       end, PSKCiphers)
                     end,
 

apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_mqtt,
  [{description, "EMQ X Bridge to MQTT Broker"},
-  {vsn, "4.3.4"}, % strict semver, bump manually!
+  {vsn, "4.3.5"}, % strict semver, bump manually!
   {modules, []},
   {registered, []},
   {applications, [kernel,stdlib,replayq,emqtt]},

apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt.appup.src

@@ -1,6 +1,10 @@
 %% -*-: erlang -*-
 {VSN,
   [
+    {"4.3.4", [
+      %% There are only changes to the schema file, so we don't need
+      %% any commands here.
+    ]},
     {"4.3.3", [
       {load_module, emqx_bridge_mqtt, brutal_purge, soft_purge, []}
     ]},
@@ -16,6 +20,8 @@
     {<<".*">>, []}
   ],
   [
+    {"4.3.4", [
+    ]},
     {"4.3.3", [
       {load_module, emqx_bridge_mqtt, brutal_purge, soft_purge, []}
     ]},

apps/emqx_exhook/etc/emqx_exhook.conf

@@ -30,6 +30,18 @@
 ## Value: Integer
 #exhook.pool_size = 16
 
+## The exhook execution priority on the Chain of the emqx hooks.
+##
+## Modify the field to fix the exhook execute order before/after other plugins/modules.
+## By default, most hooks registered by plugins or modules have a priority of 0.
+##
+## With the same priority of 0, the execute order depends on hookpoints mount order.
+## Scilicet is the loaded order of plugins/ modules.
+##
+## Default: 0
+## Value: Integer
+#exhook.hook_priority = 0
+
 ##--------------------------------------------------------------------
 ## The Hook callback servers
 

apps/emqx_exhook/include/emqx_exhook.hrl

@@ -41,4 +41,6 @@
       , {'message.dropped',     {emqx_exhook_handler, on_message_dropped,      []}}
       ]).
 
+-define(DEFAULT_HOOK_PRIORITY, 0).
+
 -endif.

apps/emqx_exhook/priv/emqx_exhook.schema

@@ -15,6 +15,11 @@
   {datatype, string}
 ]}.
 
+{mapping, "exhook.hook_priority", "emqx_exhook.hook_priority", [
+  {default, 0},
+  {datatype, integer}
+]}.
+
 {translation, "emqx_exhook.auto_reconnect", fun(Conf) ->
   case cuttlefish:conf_get("exhook.auto_reconnect", Conf) of
       "false" -> false;

apps/emqx_exhook/src/emqx_exhook.app.src

@@ -1,6 +1,6 @@
 {application, emqx_exhook,
  [{description, "EMQ X Extension for Hook"},
-  {vsn, "4.4.0"},
+  {vsn, "4.4.1"},
   {modules, []},
   {registered, []},
   {mod, {emqx_exhook_app, []}},

apps/emqx_exhook/src/emqx_exhook.appup.src

@@ -1,7 +1,13 @@
-%% -*-: erlang -*-
+%% -*- mode: erlang -*-
+%% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
- [{<<".*">>, []}
+  [{"4.4.0",
- ],
+    [{load_module,emqx_exhook_sup,brutal_purge,soft_purge,[]},
- [{<<".*">>, []}
+     {load_module,emqx_exhook_server,brutal_purge,soft_purge,[]},
- ]
+     {update, emqx_exhook_mngr, {advanced, ["4.4.0"]}}]},
-}.
+   {<<".*">>,[]}],
+  [{"4.4.0",
+    [{load_module,emqx_exhook_sup,brutal_purge,soft_purge,[]},
+     {load_module,emqx_exhook_server,brutal_purge,soft_purge,[]},
+     {update, emqx_exhook_mngr, {advanced, ["4.4.0"]}}]},
+   {<<".*">>,[]}]}.

apps/emqx_exhook/src/emqx_exhook_mngr.erl

@@ -23,7 +23,7 @@
 -include_lib("emqx/include/logger.hrl").
 
 %% APIs
--export([start_link/3]).
+-export([start_link/4]).
 
 %% Mgmt API
 -export([ enable/2
@@ -61,9 +61,14 @@
           %% Request options
           request_options :: grpc_client:options(),
           %% Timer references
-          trefs :: map()
+          trefs :: map(),
+          %% Hooks execute options
+          hooks_options :: hooks_options()
          }).
 
+-export_type([ server_options/0
+             , hooks_options/0]).
+
 -type servers() :: [{Name :: atom(), server_options()}].
 
 -type server_options() :: [ {scheme, http | https}
@@ -71,6 +76,10 @@
                           | {port, inet:port_number()}
                           ].
 
+-type hooks_options() :: #{hook_priority => integer()}.
+
+-define(DEFAULT_HOOK_OPTS, #{hook_priority => ?DEFAULT_HOOK_PRIORITY}).
+
 -define(DEFAULT_TIMEOUT, 60000).
 
 -define(CNTER, emqx_exhook_counter).
@@ -79,12 +88,12 @@
 %% APIs
 %%--------------------------------------------------------------------
 
--spec start_link(servers(), false | non_neg_integer(), grpc_client:options())
+-spec start_link(servers(), false | non_neg_integer(), grpc_client:options(), hooks_options())
     ->ignore
      | {ok, pid()}
      | {error, any()}.
-start_link(Servers, AutoReconnect, ReqOpts) ->
+start_link(Servers, AutoReconnect, ReqOpts, HooksOpts) ->
-    gen_server:start_link(?MODULE, [Servers, AutoReconnect, ReqOpts], []).
+    gen_server:start_link(?MODULE, [Servers, AutoReconnect, ReqOpts, HooksOpts], []).
 
 -spec enable(pid(), atom() | string()) -> ok | {error, term()}.
 enable(Pid, Name) ->
@@ -104,7 +113,7 @@ call(Pid, Req) ->
 %% gen_server callbacks
 %%--------------------------------------------------------------------
 
-init([Servers, AutoReconnect, ReqOpts0]) ->
+init([Servers, AutoReconnect, ReqOpts0, HooksOpts]) ->
     process_flag(trap_exit, true),
     %% XXX: Due to the ExHook Module in the enterprise,
     %% this process may start multiple times and they will share this table
@@ -125,32 +134,34 @@ init([Servers, AutoReconnect, ReqOpts0]) ->
 
     %% Load the hook servers
     ReqOpts = maps:without([request_failed_action], ReqOpts0),
-    {Waiting, Running} = load_all_servers(Servers, ReqOpts),
+    {Waiting, Running} = load_all_servers(Servers, ReqOpts, HooksOpts),
     {ok, ensure_reload_timer(
            #state{waiting = Waiting,
                   running = Running,
                   stopped = #{},
                   request_options = ReqOpts,
                   auto_reconnect = AutoReconnect,
-                  trefs = #{}
+                  trefs = #{},
+                  hooks_options = HooksOpts
                  }
           )}.
 
 %% @private
-load_all_servers(Servers, ReqOpts) ->
+load_all_servers(Servers, ReqOpts, HooksOpts) ->
-    load_all_servers(Servers, ReqOpts, #{}, #{}).
+    load_all_servers(Servers, ReqOpts, HooksOpts, #{}, #{}).
-load_all_servers([], _Request, Waiting, Running) ->
+
+load_all_servers([], _Request, _HooksOpts, Waiting, Running) ->
     {Waiting, Running};
-load_all_servers([{Name, Options} | More], ReqOpts, Waiting, Running) ->
+load_all_servers([{Name, Options} | More], ReqOpts, HooksOpts, Waiting, Running) ->
     {NWaiting, NRunning} =
-        case emqx_exhook_server:load(Name, Options, ReqOpts) of
+        case emqx_exhook_server:load(Name, Options, ReqOpts, HooksOpts) of
             {ok, ServerState} ->
                 save(Name, ServerState),
                 {Waiting, Running#{Name => Options}};
             {error, _} ->
                 {Waiting#{Name => Options}, Running}
         end,
-    load_all_servers(More, ReqOpts, NWaiting, NRunning).
+    load_all_servers(More, ReqOpts, HooksOpts, NWaiting, NRunning).
 
 handle_call({load, Name}, _From, State) ->
     {Result, NState} = do_load_server(Name, State),
@@ -204,8 +215,27 @@ terminate(_Reason, State = #state{running = Running}) ->
     _ = unload_exhooks(),
     ok.
 
-code_change(_OldVsn, State, _Extra) ->
+%% in the emqx_exhook:v4.3.5, we have added one new field in the state last:
-    {ok, State}.
+%%  - hooks_options :: map()
+code_change({down, _Vsn}, State, [ToVsn]) ->
+    case re:run(ToVsn, "4\\.4\\.0") of
+        {match, _} ->
+            NState = list_to_tuple(
+                       lists:droplast(
+                         tuple_to_list(State))),
+            {ok, NState};
+        _ ->
+            {ok, State}
+    end;
+code_change(_Vsn, State, [FromVsn]) ->
+    case re:run(FromVsn, "4\\.4\\.0") of
+        {match, _} ->
+            NState = list_to_tuple(
+                       tuple_to_list(State) ++ [?DEFAULT_HOOK_OPTS]),
+            {ok, NState};
+        _ ->
+            {ok, State}
+    end.
 
 %%--------------------------------------------------------------------
 %% Internal funcs
@@ -219,7 +249,8 @@ do_load_server(Name, State0 = #state{
                                  waiting = Waiting,
                                  running = Running,
                                  stopped = Stopped,
-                                 request_options = ReqOpts}) ->
+                                 request_options = ReqOpts,
+                                 hooks_options = HooksOpts}) ->
     State = clean_reload_timer(Name, State0),
     case maps:get(Name, Running, undefined) of
         undefined ->
@@ -228,7 +259,7 @@ do_load_server(Name, State0 = #state{
                 undefined ->
                     {{error, not_found}, State};
                 Options ->
-                    case emqx_exhook_server:load(Name, Options, ReqOpts) of
+                    case emqx_exhook_server:load(Name, Options, ReqOpts, HooksOpts) of
                         {ok, ServerState} ->
                             save(Name, ServerState),
                             ?LOG(info, "Load exhook callback server "

apps/emqx_exhook/src/emqx_exhook_server.erl

@@ -25,7 +25,7 @@
 -define(PB_CLIENT_MOD, emqx_exhook_v_1_hook_provider_client).
 
 %% Load/Unload
--export([ load/3
+-export([ load/4
         , unload/1
         ]).
 
@@ -83,8 +83,9 @@
 %% Load/Unload APIs
 %%--------------------------------------------------------------------
 
--spec load(atom(), list(), map()) -> {ok, server()} | {error, term()} .
+-spec load(atom(), emqx_exhook_mngr:server_options(), grpc_client:options(), emqx_exhook_mngr:hooks_options())
-load(Name0, Opts0, ReqOpts) ->
+          -> {ok, server()} | {error, term()} .
+load(Name0, Opts0, ReqOpts, HooksOpts) ->
     Name = to_list(Name0),
     {SvrAddr, ClientOpts} = channel_opts(Opts0),
     case emqx_exhook_sup:start_grpc_client_channel(
@@ -99,7 +100,7 @@ load(Name0, Opts0, ReqOpts) ->
                                io_lib:format("exhook.~s.", [Name])),
                     ensure_metrics(Prefix, HookSpecs),
                     %% Ensure hooks
-                    ensure_hooks(HookSpecs),
+                    ensure_hooks(HookSpecs, maps:get(hook_priority, HooksOpts, ?DEFAULT_HOOK_PRIORITY)),
                     {ok, #server{name = Name,
                                  options = ReqOpts,
                                  channel = _ChannPoolPid,
@@ -181,18 +182,16 @@ resovle_hookspec(HookSpecs) when is_list(HookSpecs) ->
         case maps:get(name, HookSpec, undefined) of
             undefined -> Acc;
             Name0 ->
-                Name = try
+                Name = try binary_to_existing_atom(Name0, utf8) catch T:R -> {T,R} end,
-                           binary_to_existing_atom(Name0, utf8)
+                case lists:member(Name, AvailableHooks) of
-                       catch T:R -> {T,R}
+                    true ->
-                       end,
+                        case lists:member(Name, MessageHooks) of
-                case {lists:member(Name, AvailableHooks),
+                            true ->
-                      lists:member(Name, MessageHooks)} of
+                                Acc#{Name => #{topics => maps:get(topics, HookSpec, [])}};
-                    {false, _} ->
+                            _ ->
-                        error({unknown_hookpoint, Name});
+                                Acc#{Name => #{}}
-                    {true, false} ->
+                        end;
-                        Acc#{Name => #{}};
+                    _ -> error({unknown_hookpoint, Name})
-                    {true, true} ->
-                        Acc#{Name => #{topics => maps:get(topics, HookSpec, [])}}
                 end
         end
     end, #{}, HookSpecs).
@@ -202,13 +201,13 @@ ensure_metrics(Prefix, HookSpecs) ->
             || Hookpoint <- maps:keys(HookSpecs)],
     lists:foreach(fun emqx_metrics:ensure/1, Keys).
 
-ensure_hooks(HookSpecs) ->
+ensure_hooks(HookSpecs, Priority) ->
     lists:foreach(fun(Hookpoint) ->
         case lists:keyfind(Hookpoint, 1, ?ENABLED_HOOKS) of
             false ->
                 ?LOG(error, "Unknown name ~s to hook, skip it!", [Hookpoint]);
             {Hookpoint, {M, F, A}} ->
-                emqx_hooks:put(Hookpoint, {M, F, A}),
+                emqx_hooks:put(Hookpoint, {M, F, A}, Priority),
                 ets:update_counter(?CNTER, Hookpoint, {2, 1}, {Hookpoint, 0})
         end
     end, maps:keys(HookSpecs)).

apps/emqx_exhook/src/emqx_exhook_sup.erl

@@ -18,6 +18,8 @@
 
 -behaviour(supervisor).
 
+-include("emqx_exhook.hrl").
+
 -export([ start_link/0
         , init/1
         ]).
@@ -43,7 +45,7 @@ start_link() ->
 
 init([]) ->
     Mngr = ?CHILD(emqx_exhook_mngr, worker,
-                  [servers(), auto_reconnect(), request_options()]),
+                  [servers(), auto_reconnect(), request_options(), hooks_options()]),
     {ok, {{one_for_one, 10, 100}, [Mngr]}}.
 
 servers() ->
@@ -58,6 +60,10 @@ request_options() ->
       pool_size => env(pool_size, erlang:system_info(schedulers))
      }.
 
+hooks_options() ->
+    #{hook_priority => env(hook_priority, ?DEFAULT_HOOK_PRIORITY)
+     }.
+
 env(Key, Def) ->
     application:get_env(emqx_exhook, Key, Def).
 

apps/emqx_exhook/test/emqx_exhook_SUITE.erl

@@ -98,10 +98,31 @@ t_cli_stats(_) ->
     _ = emqx_exhook_cli:cli(x),
     unmeck_print().
 
+t_priority(_) ->
+    restart_exhook_with_envs([{emqx_exhook, hook_priority, 1}]),
+
+    emqx_exhook:disable(default),
+    ok = emqx_exhook:enable(default),
+    [Callback | _] = emqx_hooks:lookup('client.connected'),
+    1 = emqx_hooks:callback_priority(Callback).
+
 %%--------------------------------------------------------------------
 %% Utils
 %%--------------------------------------------------------------------
 
+%% TODO: make it more general and move to `emqx_ct_helpers`
+restart_exhook_with_envs(Envs) ->
+    emqx_ct_helpers:stop_apps([emqx_exhook]),
+    SetPriorityFun
+        = fun(emqx) ->
+                  set_special_cfgs(emqx);
+             (emqx_exhook) ->
+                  lists:foreach(fun({App, Key, Val}) ->
+                                        application:set_env(App, Key, Val)
+                                end, Envs)
+          end,
+    emqx_ct_helpers:start_apps([emqx_exhook], SetPriorityFun).
+
 meck_print() ->
     meck:new(emqx_ctl, [passthrough, no_history, no_link]),
     meck:expect(emqx_ctl, print, fun(_) -> ok end),

apps/emqx_exproto/etc/emqx_exproto.conf

@@ -224,7 +224,7 @@ exproto.listener.protoname.reuseaddr = true
 ## Note that 'listener.ssl.external.ciphers' and 'listener.ssl.external.psk_ciphers' cannot
 ## be configured at the same time.
 ## See 'https://tools.ietf.org/html/rfc4279#section-2'.
-#exproto.listener.protoname.psk_ciphers = PSK-AES128-CBC-SHA,PSK-AES256-CBC-SHA,PSK-3DES-EDE-CBC-SHA,PSK-RC4-SHA
+#exproto.listener.protoname.psk_ciphers = RSA-PSK-AES256-GCM-SHA384,RSA-PSK-AES256-CBC-SHA384,RSA-PSK-AES128-GCM-SHA256,RSA-PSK-AES128-CBC-SHA256,RSA-PSK-AES256-CBC-SHA,RSA-PSK-AES128-CBC-SHA
 
 ## SSL parameter renegotiation is a feature that allows a client and a server
 ## to renegotiate the parameters of the SSL connection on the fly.

apps/emqx_exproto/priv/emqx_exproto.schema

@@ -274,12 +274,29 @@ end}.
                           {reuseaddr, cuttlefish:conf_get(Prefix ++ ".reuseaddr", Conf, undefined)}])
               end,
     SplitFun = fun(undefined) -> undefined; (S) -> string:tokens(S, ",") end,
+    AvaiableCiphers = ["RSA-PSK-AES256-GCM-SHA384","RSA-PSK-AES256-CBC-SHA384",
+                       "RSA-PSK-AES128-GCM-SHA256","RSA-PSK-AES128-CBC-SHA256",
+                       "RSA-PSK-AES256-CBC-SHA","RSA-PSK-AES128-CBC-SHA"
+                      ],
+    %% Compatible with legacy PSK Cipher strings
+    PskMapping = fun("PSK-AES128-CBC-SHA") -> {true, "RSA-PSK-AES128-CBC-SHA"};
+                    ("PSK-AES256-CBC-SHA") -> {true, "RSA-PSK-AES256-CBC-SHA"};
+                    ("PSK-3DES-EDE-CBC-SHA") -> {true, "PSK-3DES-EDE-CBC-SHA"};
+                    ("PSK-RC4-SHA") -> {true, "PSK-RC4-SHA"};
+                    (C) -> case lists:member(C, AvaiableCiphers) of
+                               true -> {true, C};
+                               false -> false
+                           end
+                 end,
     MapPSKCiphers = fun(PSKCiphers) ->
-                      lists:map(
+                      lists:filtermap(fun(C0) ->
-                          fun("PSK-AES128-CBC-SHA") -> {psk, aes_128_cbc, sha};
+                        case PskMapping(C0) of
-                             ("PSK-AES256-CBC-SHA") -> {psk, aes_256_cbc, sha};
+                            false ->
-                             ("PSK-3DES-EDE-CBC-SHA") -> {psk, '3des_ede_cbc', sha};
+                              cuttlefish:invalid(
-                             ("PSK-RC4-SHA") -> {psk, rc4_128, sha}
+                                io_lib:format("psk_ciphers: not support ~s", [C0]));
+                            {true, C} ->
+                                {true, C}
+                        end
                       end, PSKCiphers)
                     end,
     SslOpts = fun(Prefix) ->

apps/emqx_exproto/src/emqx_exproto.app.src

@@ -1,6 +1,6 @@
 {application, emqx_exproto,
  [{description, "EMQ X Extension for Protocol"},
-  {vsn, "4.3.6"}, %% 4.3.3 is used by ee
+  {vsn, "4.3.7"}, %% 4.3.3 is used by ee
   {modules, []},
   {registered, []},
   {mod, {emqx_exproto_app, []}},

apps/emqx_exproto/src/emqx_exproto.appup.src

@@ -1,6 +1,11 @@
 %% -*- mode: erlang -*-
 {VSN,
-  [{<<"4\\.3\\.[4-5]">>,
+  [
+   {"4.3.6",
+    [ %% There are only changes to the schema file, so we don't need any
+      %% commands here
+    ]},
+   {<<"4\\.3\\.[4-5]">>,
     [{load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
      {load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]},
    {<<"4\\.3\\.[2-3]">>,
@@ -12,7 +17,8 @@
      {load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
      {load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
-  [{<<"4\\.3\\.[4-5]">>,
+  [{"4.3.6", []},
+   {<<"4\\.3\\.[4-5]">>,
     [{load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
      {load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]},
    {<<"4\\.3\\.[2-3]">>,

apps/emqx_lwm2m/priv/emqx_lwm2m.schema

@@ -190,21 +190,41 @@ end}.
       case cuttlefish:conf_get("lwm2m.dtls.ciphers", Conf, undefined) of
           undefined ->
               [];
-          C ->
+          Ciphers0 ->
-              [{ciphers, SplitFun(C)}]
+              [{ciphers, SplitFun(Ciphers0)}]
       end,
   PskCiphers =
       case cuttlefish:conf_get("lwm2m.dtls.psk_ciphers", Conf, undefined) of
           undefined ->
               [];
           C2 ->
-              Psk = lists:map(fun("PSK-AES128-CBC-SHA") -> "RSA-PSK-AES128-CBC-SHA";
+              AvaiableCiphers = ["RSA-PSK-AES256-GCM-SHA384","RSA-PSK-AES256-CBC-SHA384",
-                                 ("PSK-AES256-CBC-SHA") -> "RSA-PSK-AES256-CBC-SHA";
+                                 "RSA-PSK-AES128-GCM-SHA256","RSA-PSK-AES128-CBC-SHA256",
-                                 ("PSK-3DES-EDE-CBC-SHA") -> "RSA-PSK-3DES-EDE-CBC-SHA";
+                                 "RSA-PSK-AES256-CBC-SHA","RSA-PSK-AES128-CBC-SHA"
-                                 ("PSK-RC4-SHA") -> "RSA-PSK-RC4-SHA";
+                                ],
-                                 (Suite) -> Suite
+              %% Compatible with legacy PSK Cipher strings
-                              end, SplitFun(C2)),
+              PskMapping = fun("PSK-AES128-CBC-SHA") -> {true, "RSA-PSK-AES128-CBC-SHA"};
-              [{ciphers, Psk}, {user_lookup_fun, {fun emqx_psk:lookup/3, <<>>}}]
+                              ("PSK-AES256-CBC-SHA") -> {true, "RSA-PSK-AES256-CBC-SHA"};
+                              ("PSK-3DES-EDE-CBC-SHA") -> {true, "PSK-3DES-EDE-CBC-SHA"};
+                              ("PSK-RC4-SHA") -> {true, "PSK-RC4-SHA"};
+                              (C) -> case lists:member(C, AvaiableCiphers) of
+                                         true -> {true, C};
+                                         false -> false
+                                     end
+                           end,
+              MapPSKCiphers = fun(PSKCiphers) ->
+                                lists:filtermap(fun(C0) ->
+                                  case PskMapping(C0) of
+                                      false ->
+                                        cuttlefish:invalid(
+                                          io_lib:format("psk_ciphers: not support ~s", [C0]));
+                                      {true, C} ->
+                                          {true, C}
+                                  end
+                                end, PSKCiphers)
+                              end,
+              [{ciphers, MapPSKCiphers(SplitFun(C2))},
+               {user_lookup_fun, {fun emqx_psk:lookup/3, <<>>}}]
       end,
   Ciphers /= []
   andalso PskCiphers /= []

apps/emqx_lwm2m/src/emqx_lwm2m.app.src

@@ -1,6 +1,6 @@
 {application,emqx_lwm2m,
  [{description,"EMQ X LwM2M Gateway"},
-  {vsn, "4.3.6"}, % strict semver, bump manually!
+  {vsn, "4.3.7"}, % strict semver, bump manually!
   {modules,[]},
   {registered,[emqx_lwm2m_sup]},
   {applications,[kernel,stdlib,lwm2m_coap]},

apps/emqx_lwm2m/src/emqx_lwm2m.appup.src

@@ -10,7 +10,11 @@
     [{load_module,emqx_lwm2m_protocol,brutal_purge,soft_purge,[]},
      {load_module,emqx_lwm2m_api,brutal_purge,soft_purge,[]}]},
    {"4.3.5",
-    [{load_module,emqx_lwm2m_api,brutal_purge,soft_purge,[]}]}],
+    [{load_module,emqx_lwm2m_api,brutal_purge,soft_purge,[]}]},
+   {"4.3.6",
+    [ %% There are only changes to the schema file, so we don't need any
+      %% commands here
+    ]}],
   [{<<"4\\.3\\.[0-1]">>,
     [{restart_application,emqx_lwm2m}]},
    {"4.3.2",
@@ -21,4 +25,5 @@
     [{load_module,emqx_lwm2m_protocol,brutal_purge,soft_purge,[]},
      {load_module,emqx_lwm2m_api,brutal_purge,soft_purge,[]}]},
    {"4.3.5",
-    [{load_module,emqx_lwm2m_api,brutal_purge,soft_purge,[]}]}]}.
+    [{load_module,emqx_lwm2m_api,brutal_purge,soft_purge,[]}]},
+   {"4.3.6", []}]}.

apps/emqx_plugin_libs/src/emqx_plugin_libs.app.src

@@ -1,6 +1,6 @@
 {application, emqx_plugin_libs,
  [{description, "EMQ X Plugin utility libs"},
-  {vsn, "4.4.2"},
+  {vsn, "4.4.3"},
   {modules, []},
   {applications, [kernel,stdlib]},
   {env, []}

apps/emqx_plugin_libs/src/emqx_plugin_libs.appup.src

@@ -1,25 +1,27 @@
 %% -*- mode: erlang -*-
+%% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
-  [{"4.4.1",
+  [{"4.4.2",[{load_module,emqx_plugin_libs_ssl,brutal_purge,soft_purge,[]}]},
-    [ {load_module,emqx_trace,brutal_purge,soft_purge,[]}
+   {"4.4.1",
-    , {load_module,emqx_trace_api,brutal_purge,soft_purge,[]}
+    [{load_module,emqx_plugin_libs_ssl,brutal_purge,soft_purge,[]},
-    ]},
+     {load_module,emqx_trace,brutal_purge,soft_purge,[]},
+     {load_module,emqx_trace_api,brutal_purge,soft_purge,[]}]},
    {"4.4.0",
-    [ {load_module,emqx_trace,brutal_purge,soft_purge,[]}
+    [{load_module,emqx_plugin_libs_ssl,brutal_purge,soft_purge,[]},
-    , {load_module,emqx_trace_api,brutal_purge,soft_purge,[]}
+     {load_module,emqx_trace,brutal_purge,soft_purge,[]},
-    , {update, emqx_slow_subs, {advanced, ["4.4.0"]}}
+     {load_module,emqx_trace_api,brutal_purge,soft_purge,[]},
-    , {load_module,emqx_slow_subs_api,brutal_purge,soft_purge,[]}
+     {update,emqx_slow_subs,{advanced,["4.4.0"]}},
-    ]},
+     {load_module,emqx_slow_subs_api,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
- [{"4.4.1",
+  [{"4.4.2",[{load_module,emqx_plugin_libs_ssl,brutal_purge,soft_purge,[]}]},
-    [ {load_module,emqx_trace,brutal_purge,soft_purge,[]}
+   {"4.4.1",
-    , {load_module,emqx_trace_api,brutal_purge,soft_purge,[]}
+    [{load_module,emqx_plugin_libs_ssl,brutal_purge,soft_purge,[]},
-    ]},
+     {load_module,emqx_trace,brutal_purge,soft_purge,[]},
+     {load_module,emqx_trace_api,brutal_purge,soft_purge,[]}]},
    {"4.4.0",
-   [ {load_module,emqx_trace,brutal_purge,soft_purge,[]}
+    [{load_module,emqx_plugin_libs_ssl,brutal_purge,soft_purge,[]},
-   , {load_module,emqx_trace_api,brutal_purge,soft_purge,[]}
+     {load_module,emqx_trace,brutal_purge,soft_purge,[]},
-   , {update, emqx_slow_subs, {advanced, ["4.4.0"]}}
+     {load_module,emqx_trace_api,brutal_purge,soft_purge,[]},
-   , {load_module,emqx_slow_subs_api,brutal_purge,soft_purge,[]}
+     {update,emqx_slow_subs,{advanced,["4.4.0"]}},
-   ]},
+     {load_module,emqx_slow_subs_api,brutal_purge,soft_purge,[]}]},
-  {<<".*">>,[]}]
+   {<<".*">>,[]}]}.
-}.

apps/emqx_plugin_libs/src/emqx_plugin_libs_ssl.erl

@@ -58,9 +58,9 @@ save_files_return_opts(Options, Dir) ->
     KeyFile = Get(<<"keyfile">>),
     CertFile = Get(<<"certfile">>),
     CAFile = GetD(<<"cacertfile">>, Get(<<"cafile">>)),
-    Key = do_save_file(KeyFile, Dir),
+    Key = maybe_save_file(KeyFile, Dir),
-    Cert = do_save_file(CertFile, Dir),
+    Cert = maybe_save_file(CertFile, Dir),
-    CA = do_save_file(CAFile, Dir),
+    CA = maybe_save_file(CAFile, Dir),
     Verify = case GetD(<<"verify">>, false) of
                   false -> verify_none;
                   _ -> verify_peer
@@ -80,25 +80,47 @@ save_files_return_opts(Options, Dir) ->
 -spec save_file(file_input(), atom() | string() | binary()) -> string().
 save_file(Param, SubDir) ->
    Dir = filename:join([emqx:get_env(data_dir), SubDir]),
-   do_save_file( Param, Dir).
+   maybe_save_file(Param, Dir).
 
 filter([]) -> [];
 filter([{_, ""} | T]) -> filter(T);
 filter([{_, undefined} | T]) -> filter(T);
 filter([H | T]) -> [H | filter(T)].
 
-do_save_file(#{<<"filename">> := FileName, <<"file">> := Content}, Dir)
+maybe_save_file(#{<<"filename">> := FileName, <<"file">> := Content}, Dir)
   when FileName =/= undefined andalso Content =/= undefined ->
-    do_save_file(ensure_str(FileName), iolist_to_binary(Content), Dir);
+    maybe_save_file(ensure_str(FileName), iolist_to_binary(Content), Dir);
-do_save_file(FilePath, _) when is_binary(FilePath) ->
+maybe_save_file(FilePath, _) when is_binary(FilePath) ->
     ensure_str(FilePath);
-do_save_file(FilePath, _) when is_list(FilePath) ->
+maybe_save_file(FilePath, _) when is_list(FilePath) ->
     FilePath;
-do_save_file(_, _) -> "".
+maybe_save_file(_, _) -> "".
 
-do_save_file("", _, _Dir) -> ""; %% ignore
+maybe_save_file("", _, _Dir) -> ""; %% no filename, ignore
-do_save_file(_, <<>>, _Dir) -> ""; %% ignore
+maybe_save_file(FileName, <<>>, Dir) ->  %% no content, see if file exists
-do_save_file(FileName, Content, Dir) ->
+    {ok, Cwd} = file:get_cwd(),
+    %% NOTE: when FileName is an absolute path, filename:join has no effect
+    CwdFile = ensure_str(filename:join([Cwd, FileName])),
+    DataDirFile = ensure_str(filename:join([Dir, FileName])),
+    Possibles0 = case CwdFile =:= DataDirFile of
+                    true -> [CwdFile];
+                    false -> [CwdFile, DataDirFile]
+                end,
+    Possibles = Possibles0 ++
+                 case FileName of
+                     "etc/certs/" ++ Path ->
+                         %% this is the dir hard-coded in rule-engine resources as
+                         %% default, unfortunatly we cannot change the deaults
+                         %% due to compatibilty reasons, so we have to make a guess
+                         ["/etc/emqx/certs/" ++ Path];
+                     _ ->
+                         []
+                 end,
+    case find_exist_file(FileName, Possibles) of
+        false -> erlang:throw({bad_cert_file, Possibles});
+        Found -> Found
+    end;
+maybe_save_file(FileName, Content, Dir) ->
      FullFilename = filename:join([Dir, FileName]),
      ok = filelib:ensure_dir(FullFilename),
      case file:write_file(FullFilename, Content) of
@@ -112,3 +134,9 @@ do_save_file(FileName, Content, Dir) ->
 ensure_str(L) when is_list(L) -> L;
 ensure_str(B) when is_binary(B) -> unicode:characters_to_list(B, utf8).
 
+find_exist_file(_Name, []) -> false;
+find_exist_file(Name, [F | Rest]) ->
+    case filelib:is_regular(F) of
+        true -> F;
+        false -> find_exist_file(Name, Rest)
+    end.

apps/emqx_plugin_libs/test/emqx_plugin_libs_ssl_tests.erl

@@ -42,7 +42,8 @@ prop_file_or_content() ->
                         {prop_cert_file_name(), proper_types:binary()}]).
 
 prop_cert_file_name() ->
-    proper_types:oneof(["certname1", <<"certname2">>, "", <<>>, undefined]).
+    File = code:which(?MODULE), %% existing
+    proper_types:oneof(["", <<>>, undefined, File]).
 
 prop_tls_versions() ->
     proper_types:oneof(["tlsv1.3",
@@ -76,3 +77,10 @@ file_or_content({Name, Content}) ->
     #{<<"file">> => Content, <<"filename">> => Name};
 file_or_content(Name) ->
     Name.
+
+bad_cert_file_test() ->
+    Input = #{<<"keyfile">> =>
+                #{<<"filename">> => "notafile",
+                  <<"file">> => ""}},
+    ?assertThrow({bad_cert_file, _},
+                 emqx_plugin_libs_ssl:save_files_return_opts(Input, "test-data")).

apps/emqx_psk_file/test/emqx_psk_file_SUITE.erl

@@ -15,10 +15,85 @@
 %%--------------------------------------------------------------------
 
 -module(emqx_psk_file_SUITE).
+
 -compile(nowarn_export_all).
 -compile(export_all).
 
-all() -> [].
+-include_lib("eunit/include/eunit.hrl").
 
-groups() ->
+all() ->
-    [].
+    emqx_ct:all(?MODULE).
+
+init_per_suite(Config) ->
+    emqx_ct_helpers:start_apps([emqx_psk_file], fun set_special_confs/1),
+    Config.
+
+end_per_suite(_) ->
+    emqx_ct_helpers:stop_apps([emqx_psk_file]).
+
+set_special_confs(emqx) ->
+    emqx_ct_helpers:change_emqx_opts(
+      ssl_twoway, [{ssl_options,
+                    [{versions, ['tlsv1.2','tlsv1.1', tlsv1]},
+                     {ciphers, psk_ciphers()},
+                     {user_lookup_fun,{fun emqx_psk:lookup/3,<<>>}}
+                    ]
+                   }]),
+    application:set_env(emqx, plugins_loaded_file,
+                        emqx_ct_helpers:deps_path(emqx, "test/emqx_SUITE_data/loaded_plugins"));
+set_special_confs(emqx_psk_file) ->
+    Path = emqx_ct_helpers:deps_path(emqx_psk_file, "etc/psk.txt"),
+    application:set_env(emqx_psk_file, path, Path),
+    application:set_env(emqx_psk_file, delimiter, ":");
+set_special_confs(_App) ->
+    ok.
+
+%%--------------------------------------------------------------------
+%% cases
+
+t_psk_loaded(_) ->
+    ?assertEqual({stop, <<"1234">>},
+                  emqx_psk_file:on_psk_lookup(<<"client1">>, undefined)).
+
+t_psk_ciphers(_) ->
+    lists:foreach(fun(Cipher) ->
+        {ok, C} = do_emqtt_connect(Cipher),
+        emqtt:disconnect(C)
+    end, psk_ciphers()).
+
+do_emqtt_connect(Cipher) ->
+    {ok, C} = emqtt:start_link(
+                [{proto_ver, v5},
+                 {port, 8883},
+                 {ssl, true},
+                 {ssl_opts, ssl_opts(Cipher)}
+                ]),
+    {ok, _} = emqtt:connect(C),
+    {ok, C}.
+
+psk_ciphers() ->
+    ["RSA-PSK-AES256-GCM-SHA384","RSA-PSK-AES256-CBC-SHA384",
+     "RSA-PSK-AES128-GCM-SHA256","RSA-PSK-AES128-CBC-SHA256",
+     "RSA-PSK-AES256-CBC-SHA","RSA-PSK-AES128-CBC-SHA"].
+
+ssl_opts(Cipher) ->
+    TlsFile = fun(Name) ->
+        emqx_ct_helpers:app_path(
+          emqx,
+          filename:join(["etc", "certs", Name]))
+    end,
+    [{cacertfile, TlsFile("cacert.pem")},
+     {certfile, TlsFile("client-cert.pem")},
+     {keyfile, TlsFile("client-key.pem")},
+     {verify, verify_peer},
+     {server_name_indication, disable},
+     {protocol, tls},
+     {versions, ['tlsv1.2', 'tlsv1.1']},
+     {psk_identity, "client1"},
+     {user_lookup_fun, {fun ?MODULE:on_psk_client_lookup/3, #{}}},
+     {ciphers, [Cipher]}
+    ].
+
+on_psk_client_lookup(psk, _PSKId, _UserState) ->
+    {stop, Psk} = emqx_psk_file:on_psk_lookup(<<"client1">>, undefined),
+    {ok, Psk}.

apps/emqx_rule_engine/src/emqx_rule_engine.app.src

@@ -1,6 +1,6 @@
 {application, emqx_rule_engine,
  [{description, "EMQ X Rule Engine"},
-  {vsn, "4.4.2"}, % strict semver, bump manually!
+  {vsn, "4.4.3"}, % strict semver, bump manually!
   {modules, []},
   {registered, [emqx_rule_engine_sup, emqx_rule_registry]},
   {applications, [kernel,stdlib,rulesql,getopt]},

apps/emqx_rule_engine/src/emqx_rule_engine.appup.src

@@ -1,15 +1,28 @@
 %% -*- mode: erlang -*-
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
-  [{"4.4.1",
+  [{"4.4.2",
-    [{load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]}]},
+   {"4.4.1",
+    [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
    {"4.4.0",
-    [{load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
      {update,emqx_rule_metrics,{advanced,["4.4.0"]}},
@@ -18,15 +31,28 @@
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
-  [{"4.4.1",
+  [{"4.4.2",
-    [{load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]}]},
+   {"4.4.1",
+    [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
    {"4.4.0",
-    [{load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
      {update,emqx_rule_metrics,{advanced,["4.4.0"]}},

apps/emqx_rule_engine/src/emqx_rule_engine.erl

@@ -37,6 +37,7 @@
         , test_resource/1
         , start_resource/1
         , get_resource_status/1
+        , is_source_alive/1
         , get_resource_params/1
         , delete_resource/1
         , update_resource/2
@@ -231,7 +232,10 @@ delete_rule(RuleId) ->
     end.
 
 -spec(create_resource(#{type := _, config := _, _ => _}) -> {ok, resource()} | {error, Reason :: term()}).
-create_resource(#{type := Type, config := Config0} = Params) ->
+create_resource(Params) ->
+    create_resource(Params, with_retry).
+
+create_resource(#{type := Type, config := Config0} = Params, Retry) ->
     case emqx_rule_registry:find_resource_type(Type) of
         {ok, #resource_type{on_create = {M, F}, params_spec = ParamSpec}} ->
             Config = emqx_rule_validator:validate_params(Config0, ParamSpec),
@@ -243,10 +247,20 @@ create_resource(#{type := Type, config := Config0} = Params) ->
                                  created_at = erlang:system_time(millisecond)
                                 },
             ok = emqx_rule_registry:add_resource(Resource),
+            case Retry of
+                with_retry ->
                     %% Note that we will return OK in case of resource creation failure,
                     %% A timer is started to re-start the resource later.
-            catch _ = ?CLUSTER_CALL(init_resource, [M, F, ResId, Config]),
+                    _ = (catch (?CLUSTER_CALL(init_resource, [M, F, ResId, Config]))),
                     {ok, Resource};
+                no_retry ->
+                    try
+                        _ = ?CLUSTER_CALL(init_resource, [M, F, ResId, Config]),
+                        {ok, Resource}
+                    catch throw : Reason ->
+                        {error, Reason}
+                    end
+            end;
         not_found ->
             {error, {resource_type_not_found, Type}}
     end.
@@ -314,24 +328,47 @@ start_resource(ResId) ->
     end.
 
 -spec(test_resource(#{type := _, config := _, _ => _}) -> ok | {error, Reason :: term()}).
-test_resource(#{type := Type, config := Config0}) ->
+test_resource(#{type := Type} = Params) ->
     case emqx_rule_registry:find_resource_type(Type) of
-        {ok, #resource_type{on_create = {ModC, Create},
+        {ok, #resource_type{}} ->
-                            on_destroy = {ModD, Destroy},
+            ResId = maps:get(id, Params, resource_id()),
-                            params_spec = ParamSpec}} ->
-            Config = emqx_rule_validator:validate_params(Config0, ParamSpec),
-            ResId = resource_id(),
             try
-                _ = ?CLUSTER_CALL(init_resource, [ModC, Create, ResId, Config]),
+                case create_resource(maps:put(id, ResId, Params), no_retry) of
-                _ = ?CLUSTER_CALL(clear_resource, [ModD, Destroy, ResId]),
+                    {ok, _} ->
-                ok
+                        case is_source_alive(ResId) of
-            catch
+                            true ->
-                throw:Reason -> {error, Reason}
+                                ok;
+                            false ->
+                                %% in is_source_alive, the cluster-call RPC logs errors
+                                %% so we do not log anything here
+                                {error, {resource_down, ResId}}
+                        end;
+                    {error, Reason} ->
+                        {error, Reason}
+                end
+            catch E:R:S ->
+                ?LOG(warning, "test resource failed, ~0p:~0p ~0p", [E, R, S]),
+                {error, R}
+            after
+                _ = ?CLUSTER_CALL(delete_resource, [ResId])
             end;
         not_found ->
             {error, {resource_type_not_found, Type}}
     end.
 
+is_source_alive(ResId) ->
+    case rpc:multicall(ekka_mnesia:running_nodes(), ?MODULE, get_resource_status, [ResId], 5000) of
+        {ResL, []} ->
+            is_source_alive_(ResL);
+        {_, _Errors} ->
+            false
+    end.
+
+is_source_alive_([]) -> true;
+is_source_alive_([{ok, #{is_alive := true}} | ResL]) -> is_source_alive_(ResL);
+is_source_alive_([{ok, #{is_alive := false}} | _ResL]) -> false;
+is_source_alive_([_Error | _ResL]) -> false.
+
 -spec(get_resource_status(resource_id()) -> {ok, resource_status()} | {error, Reason :: term()}).
 get_resource_status(ResId) ->
     case emqx_rule_registry:find_resource_params(ResId) of

apps/emqx_rule_engine/src/emqx_rule_engine_api.erl

@@ -58,6 +58,13 @@
             descr  => "Delete a rule"
            }).
 
+-rest_api(#{name   => reset_metrics,
+            method => 'PUT',
+            path   => "/rules/:bin:id/reset_metrics",
+            func   => reset_metrics,
+            descr  => "reset a rule metrics"
+           }).
+
 -rest_api(#{name   => list_actions,
             method => 'GET',
             path   => "/actions/",
@@ -154,6 +161,7 @@
         , list_rules/2
         , show_rule/2
         , delete_rule/2
+        , reset_metrics/2
         ]).
 
 -export([ list_actions/2
@@ -252,6 +260,10 @@ delete_rule(#{id := Id}, _Params) ->
     ok = emqx_rule_engine:delete_rule(Id),
     return(ok).
 
+reset_metrics(#{id := Id}, _Params) ->
+    ok = emqx_rule_metrics:reset_metrics(Id),
+    return(ok).
+
 %%------------------------------------------------------------------------------
 %% Actions API
 %%------------------------------------------------------------------------------
@@ -296,7 +308,7 @@ do_create_resource(Create, ParsedParams) ->
 list_resources(#{}, _Params) ->
     Data0 = lists:foldr(fun maybe_record_to_map/2, [], emqx_rule_registry:get_resources()),
     Data = lists:map(fun(Res = #{id := ResId}) ->
-               Status = get_aggregated_status(ResId),
+               Status = emqx_rule_engine:is_source_alive(ResId),
                maps:put(status, Status, Res)
            end, Data0),
     return({ok, Data}).
@@ -304,14 +316,6 @@ list_resources(#{}, _Params) ->
 list_resources_by_type(#{type := Type}, _Params) ->
     return_all(emqx_rule_registry:get_resources_by_type(Type)).
 
-get_aggregated_status(ResId) ->
-    lists:all(fun(Node) ->
-        case rpc:call(Node, emqx_rule_engine, get_resource_status, [ResId]) of
-            {ok, #{is_alive := true}} -> true;
-            _ -> false
-        end
-    end, ekka_mnesia:running_nodes()).
-
 show_resource(#{id := Id}, _Params) ->
     case emqx_rule_registry:find_resource(Id) of
         {ok, R} ->

apps/emqx_rule_engine/src/emqx_rule_events.erl

@@ -188,18 +188,18 @@ eventmsg_connected(_ClientInfo = #{
                     is_bridge := IsBridge,
                     mountpoint := Mountpoint
                    },
-                   _ConnInfo = #{
+                   ConnInfo = #{
                     peername := PeerName,
                     sockname := SockName,
                     clean_start := CleanStart,
                     proto_name := ProtoName,
                     proto_ver := ProtoVer,
-                    keepalive := Keepalive,
                     connected_at := ConnectedAt,
-                    conn_props := ConnProps,
+                    receive_maximum := RcvMax
-                    receive_maximum := RcvMax,
-                    expiry_interval := ExpiryInterval
                    }) ->
+    Keepalive = maps:get(keepalive, ConnInfo, 0),
+    ConnProps = maps:get(conn_props, ConnInfo, #{}),
+    ExpiryInterval = maps:get(expiry_interval, ConnInfo, 0),
     with_basic_columns('client.connected',
         #{clientid => ClientId,
           username => Username,

apps/emqx_rule_engine/src/emqx_rule_funcs.erl

@@ -200,6 +200,9 @@
         , rfc3339_to_unix_ts/2
         , now_timestamp/0
         , now_timestamp/1
+        , mongo_date/0
+        , mongo_date/1
+        , mongo_date/2
         ]).
 
 %% Proc Dict Func
@@ -906,6 +909,24 @@ time_unit(<<"millisecond">>) -> millisecond;
 time_unit(<<"microsecond">>) -> microsecond;
 time_unit(<<"nanosecond">>) -> nanosecond.
 
+mongo_date() ->
+    erlang:timestamp().
+
+mongo_date(MillisecondsTimestamp) ->
+    convert_timestamp(MillisecondsTimestamp).
+
+mongo_date(Timestamp, Unit) ->
+    InsertedTimeUnit = time_unit(Unit),
+    ScaledEpoch = erlang:convert_time_unit(Timestamp, InsertedTimeUnit, millisecond),
+    convert_timestamp(ScaledEpoch).
+
+convert_timestamp(MillisecondsTimestamp) ->
+    MicroTimestamp = MillisecondsTimestamp * 1000,
+    MegaSecs = MicroTimestamp div 1000_000_000_000,
+    Secs = MicroTimestamp div 1000_000 - MegaSecs*1000_000,
+    MicroSecs = MicroTimestamp rem 1000_000,
+    {MegaSecs, Secs, MicroSecs}.
+
 %% @doc This is for sql funcs that should be handled in the specific modules.
 %% Here the emqx_rule_funcs module acts as a proxy, forwarding
 %% the function handling to the worker module.

apps/emqx_rule_engine/src/emqx_rule_metrics.erl

@@ -63,6 +63,7 @@
         , create_metrics/1
         , clear_rule_metrics/1
         , clear_metrics/1
+        , reset_metrics/1
         ]).
 
 -export([ get_rule_metrics/1
@@ -129,6 +130,45 @@ clear_rule_metrics(Id) ->
 clear_metrics(Id) ->
     gen_server:call(?MODULE, {delete_metrics, Id}).
 
+-spec(reset_metrics(rule_id()) -> ok).
+reset_metrics(Id) ->
+    reset_speeds(Id),
+    reset_metrics(Id, rule_metrics()),
+    case emqx_rule_registry:get_rule(Id) of
+        not_found -> ok;
+        {ok, #rule{actions = Actions}} ->
+            [ reset_metrics(ActionId, action_metrics())
+              || #action_instance{ id = ActionId} <- Actions],
+            ok
+    end.
+
+reset_metrics(Id, Metrics) ->
+    case couters_ref(Id) of
+        not_found -> ok;
+        Ref -> [counters:put(Ref, metrics_idx(Idx), 0)
+                || Idx <- Metrics],
+               ok
+    end.
+
+reset_speeds(Id) ->
+    gen_server:call(?MODULE, {reset_speeds, Id}).
+
+rule_metrics() ->
+    [ 'rules.matched'
+    , 'rules.failed'
+    , 'rules.passed'
+    , 'rules.exception'
+    , 'rules.no_result'
+    ].
+
+action_metrics() ->
+    [ 'actions.success'
+    , 'actions.error'
+    , 'actions.taken'
+    , 'actions.exception'
+    , 'actions.retry'
+    ].
+
 -spec(get(rule_id(), atom()) -> number()).
 get(Id, Metric) ->
     case couters_ref(Id) of
@@ -290,6 +330,9 @@ handle_call({create_rule_metrics, Id}, _From,
                                     _ -> RuleSpeeds#{Id => #rule_speed{}}
                                 end}};
 
+handle_call({reset_speeds, Id}, _From, State = #state{rule_speeds = RuleSpeedMap}) ->
+    {reply, ok, State#state{rule_speeds = maps:put(Id, #rule_speed{}, RuleSpeedMap)}};
+
 handle_call({delete_metrics, Id}, _From,
             State = #state{metric_ids = MIDs, rule_speeds = undefined}) ->
     {reply, delete_counters(Id), State#state{metric_ids = sets:del_element(Id, MIDs)}};

apps/emqx_rule_engine/src/emqx_rule_runtime.erl

@@ -50,31 +50,8 @@ apply_rules([], _Input) ->
     ok;
 apply_rules([#rule{enabled = false}|More], Input) ->
     apply_rules(More, Input);
-apply_rules([Rule = #rule{id = RuleID}|More], Input) ->
+apply_rules([Rule|More], Input) ->
-    try apply_rule_discard_result(Rule, Input)
+    apply_rule_discard_result(Rule, Input),
-    catch
-        %% ignore the errors if select or match failed
-        _:{select_and_transform_error, Error} ->
-            emqx_rule_metrics:inc_rules_exception(RuleID),
-            ?LOG(warning, "SELECT clause exception for ~s failed: ~p",
-                 [RuleID, Error]);
-        _:{match_conditions_error, Error} ->
-            emqx_rule_metrics:inc_rules_exception(RuleID),
-            ?LOG(warning, "WHERE clause exception for ~s failed: ~p",
-                 [RuleID, Error]);
-        _:{select_and_collect_error, Error} ->
-            emqx_rule_metrics:inc_rules_exception(RuleID),
-            ?LOG(warning, "FOREACH clause exception for ~s failed: ~p",
-                 [RuleID, Error]);
-        _:{match_incase_error, Error} ->
-            emqx_rule_metrics:inc_rules_exception(RuleID),
-            ?LOG(warning, "INCASE clause exception for ~s failed: ~p",
-                 [RuleID, Error]);
-        _:Error:StkTrace ->
-            emqx_rule_metrics:inc_rules_exception(RuleID),
-            ?LOG(error, "Apply rule ~s failed: ~p. Stacktrace:~n~p",
-                 [RuleID, Error, StkTrace])
-    end,
     apply_rules(More, Input).
 
 apply_rule_discard_result(Rule, Input) ->
@@ -84,7 +61,35 @@ apply_rule_discard_result(Rule, Input) ->
 apply_rule(Rule = #rule{id = RuleID}, Input) ->
     clear_rule_payload(),
     ok = emqx_rule_metrics:inc_rules_matched(RuleID),
-    do_apply_rule(Rule, add_metadata(Input, #{rule_id => RuleID})).
+    try do_apply_rule(Rule, add_metadata(Input, #{rule_id => RuleID}))
+    catch
+        %% ignore the errors if select or match failed
+        _:Reason = {select_and_transform_error, Error} ->
+            emqx_rule_metrics:inc_rules_exception(RuleID),
+            ?LOG(warning, "SELECT clause exception for ~s failed: ~p",
+                 [RuleID, Error]),
+            {error, Reason};
+        _:Reason = {match_conditions_error, Error} ->
+            emqx_rule_metrics:inc_rules_exception(RuleID),
+            ?LOG(warning, "WHERE clause exception for ~s failed: ~p",
+                 [RuleID, Error]),
+            {error, Reason};
+        _:Reason = {select_and_collect_error, Error} ->
+            emqx_rule_metrics:inc_rules_exception(RuleID),
+            ?LOG(warning, "FOREACH clause exception for ~s failed: ~p",
+                 [RuleID, Error]),
+            {error, Reason};
+        _:Reason = {match_incase_error, Error} ->
+            emqx_rule_metrics:inc_rules_exception(RuleID),
+            ?LOG(warning, "INCASE clause exception for ~s failed: ~p",
+                 [RuleID, Error]),
+            {error, Reason};
+        _:Error:StkTrace ->
+            emqx_rule_metrics:inc_rules_exception(RuleID),
+            ?LOG(error, "Apply rule ~s failed: ~p. Stacktrace:~n~p",
+                 [RuleID, Error, StkTrace]),
+            {error, {Error, StkTrace}}
+    end.
 
 do_apply_rule(#rule{id = RuleId,
                     is_foreach = true,
@@ -452,7 +457,8 @@ cache_payload(DecodedP) ->
 
 safe_decode_and_cache(MaybeJson) ->
     try cache_payload(emqx_json:decode(MaybeJson, [return_maps]))
-    catch _:_ -> #{}
+    catch
+        _:_:_-> error({decode_json_failed, MaybeJson})
     end.
 
 ensure_list(List) when is_list(List) -> List;

apps/emqx_rule_engine/src/emqx_rule_sqltester.erl

@@ -77,7 +77,7 @@ test_rule(Sql, Select, Context, EventTopics) ->
         R
     of
         {ok, Data} -> {ok, flatten(Data)};
-        {error, nomatch} -> {error, nomatch}
+        {error, Reason} -> {error, Reason}
     after
         ok = emqx_rule_registry:remove_action_instance_params(ActInstId)
     end.

apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl

@@ -49,6 +49,7 @@ groups() ->
       [t_register_provider,
        t_unregister_provider,
        t_create_rule,
+       t_reset_metrics,
        t_create_resource
       ]},
      {actions, [],
@@ -124,7 +125,8 @@ groups() ->
        t_sqlparse_array_range_1,
        t_sqlparse_array_range_2,
        t_sqlparse_true_false,
-       t_sqlparse_new_map
+       t_sqlparse_new_map,
+       t_sqlparse_invalid_json
       ]},
      {rule_metrics, [],
       [t_metrics,
@@ -353,6 +355,43 @@ t_inspect_action(_Config) ->
     emqx_rule_registry:remove_resource(ResId),
     ok.
 
+t_reset_metrics(_Config) ->
+    ok = emqx_rule_engine:load_providers(),
+    {ok, #resource{id = ResId}} = emqx_rule_engine:create_resource(
+                                    #{type => built_in,
+                                      config => #{},
+                                      description => <<"debug resource">>}),
+    {ok, #rule{id = Id}} = emqx_rule_engine:create_rule(
+                             #{rawsql => "select clientid as c, username as u "
+                               "from \"t1\" ",
+                               actions => [#{name => 'inspect',
+                                             args => #{'$resource' => ResId, a=>1, b=>2}}],
+                               type => built_in,
+                               description => <<"Inspect rule">>
+                              }),
+    {ok, Client} = emqtt:start_link([{username, <<"emqx">>}]),
+    {ok, _} = emqtt:connect(Client),
+    [ begin
+          emqtt:publish(Client, <<"t1">>, <<"{\"id\": 1, \"name\": \"ha\"}">>, 0),
+          timer:sleep(100)
+      end
+      || _ <- lists:seq(1,10)],
+    emqx_rule_metrics:reset_metrics(Id),
+    Expected = #{exception => 0, failed => 0, matched => 0, no_result => 0,
+                 passed => 0, speed => 0.0, speed_last5m => 0.0, speed_max => 0},
+    Got = emqx_rule_metrics:get_rule_metrics(Id),
+    %% use == instead of =:=, so that 0 and 0.0 are compared equal
+    case Expected == Got of
+        true -> ok;
+        false -> ?assertEqual(Expected, Got)
+    end,
+    ?assertEqual(#{failed => 0, success => 0, taken => 0},
+                   emqx_rule_metrics:get_action_metrics(ResId)),
+    emqtt:stop(Client),
+    emqx_rule_registry:remove_rule(Id),
+    emqx_rule_registry:remove_resource(ResId),
+    ok.
+
 t_republish_action(_Config) ->
     Qos0Received = emqx_metrics:val('messages.qos0.received'),
     Received = emqx_metrics:val('messages.received'),
@@ -2278,12 +2317,13 @@ t_sqlparse_array_range_1(_Config) ->
     Sql02 = "select "
            "  payload.a[1..4] as c "
            "from \"t/#\" ",
-    ?assertThrow({select_and_transform_error, {error,{range_get,non_list_data},_}},
+    ?assertMatch({error, {select_and_transform_error, {error,{range_get,non_list_data},_}}},
         emqx_rule_sqltester:test(
             #{<<"rawsql">> => Sql02,
                 <<"ctx">> =>
                     #{<<"payload">> => <<"{\"x\":[0,1,2,3,4,5]}">>,
                       <<"topic">> => <<"t/a">>}})),
+
     %% construct a range:
     Sql1 = "select "
            "  [1..4] as c, "
@@ -2410,6 +2450,29 @@ t_sqlparse_nested_get(_Config) ->
               <<"payload">> => <<"{\"a\": {\"b\": 0}}">>
           }})).
 
+t_sqlparse_invalid_json(_Config) ->
+    Sql02 = "select "
+           "  payload.a[1..4] as c "
+           "from \"t/#\" ",
+    ?assertMatch({error, {select_and_transform_error, {error,{decode_json_failed,_},_}}},
+        emqx_rule_sqltester:test(
+            #{<<"rawsql">> => Sql02,
+                <<"ctx">> =>
+                    #{<<"payload">> => <<"{\"x\":[0,1,2,3,}">>,
+                      <<"topic">> => <<"t/a">>}})),
+
+
+    Sql2 = "foreach payload.sensors "
+        "do item.cmd as msg_type "
+        "from \"t/#\" ",
+    ?assertMatch({error, {select_and_collect_error, {error,{decode_json_failed,_},_}}},
+                 emqx_rule_sqltester:test(
+                   #{<<"rawsql">> => Sql2,
+                     <<"ctx">> =>
+                         #{<<"payload">> =>
+                               <<"{\"sensors\": [{\"cmd\":\"1\"} {\"cmd\":}]}">>,
+                           <<"topic">> => <<"t/a">>}})).
+
 %%------------------------------------------------------------------------------
 %% Internal helpers
 %%------------------------------------------------------------------------------

apps/emqx_rule_engine/test/emqx_rule_metrics_SUITE.erl

@@ -44,6 +44,7 @@ groups() ->
 
 init_per_suite(Config) ->
     emqx_ct_helpers:start_apps([emqx]),
+    catch emqx_rule_metrics:stop(),
     {ok, _} = emqx_rule_metrics:start_link(),
     Config.
 

bin/install_upgrade.escript

@@ -262,6 +262,7 @@ make_symlink_or_copy(Filename, ReleaseLink) ->
 
 unpack_zipballs(RelNameStr, Version) ->
     {ok, Cwd} = file:get_cwd(),
+    try
         GzFile = filename:absname(filename:join(["releases", RelNameStr ++ "-" ++ Version ++ ".tar.gz"])),
         ZipFiles = filelib:wildcard(filename:join(["releases", RelNameStr ++ "-*" ++ Version ++ "*.zip"])),
         ?INFO("unzip ~p", [ZipFiles]),
@@ -273,8 +274,11 @@ unpack_zipballs(RelNameStr, Version) ->
             {ok, _FileList} = zip:unzip("emqx.zip"),
             ok = file:set_cwd(filename:join([TmdTarD, "emqx"])),
             ok = erl_tar:create(GzFile, filelib:wildcard("*"), [compressed])
-     end || Zip <- ZipFiles],
+        end || Zip <- ZipFiles]
-    file:set_cwd(Cwd).
+    after
+        % restore cwd
+        file:set_cwd(Cwd)
+    end.
 
 first_value(_Fun, []) -> no_value;
 first_value(Fun, [Value | Rest]) ->

deploy/docker/docker-entrypoint.sh

@@ -105,13 +105,13 @@ fill_tuples() {
     local file=$1
     local elements=${*:2}
     for var in $elements; do
-        if grep -qE "\{\s*$var\s*,\s*(true|false)\s*\}\s*\." "$file"; then
+        if grep -qE "\{\s*$var\s*,\s*(true|false)\s*\}\s*\." "$file" 2>/dev/null; then
-            sed -r "s/\{\s*($var)\s*,\s*(true|false)\s*\}\s*\./{\1, true}./1" "$file" > tmpfile && cat tmpfile > "$file" 
+            sed -r "s/\{\s*($var)\s*,\s*(true|false)\s*\}\s*\./{\1, true}./1" "$file" 2>/dev/null > tmpfile && cat tmpfile > "$file"
-        elif grep -q "$var\s*\." "$file"; then
+        elif grep -q "$var\s*\." "$file" 2>/dev/null; then
             # backward compatible.
-            sed -r "s/($var)\s*\./{\1, true}./1" "$file" > tmpfile && cat tmpfile > "$file"
+            sed -r "s/($var)\s*\./{\1, true}./1" "$file" > tmpfile 2>/dev/null && cat tmpfile > "$file"
         else
-            sed '$a'\\ "$file" > tmpfile && cat tmpfile > "$file"
+            sed '$a'\\ "$file" 2>/dev/null > tmpfile && cat tmpfile > "$file"
             echo "{$var, true}." >> "$file"
         fi
     done

etc/emqx.conf

@@ -1473,7 +1473,7 @@ listener.ssl.external.ciphers = TLS_AES_256_GCM_SHA384,TLS_AES_128_GCM_SHA256,TL
 ## Note that 'listener.ssl.external.ciphers' and 'listener.ssl.external.psk_ciphers' cannot
 ## be configured at the same time.
 ## See 'https://tools.ietf.org/html/rfc4279#section-2'.
-#listener.ssl.external.psk_ciphers = PSK-AES128-CBC-SHA,PSK-AES256-CBC-SHA,PSK-3DES-EDE-CBC-SHA,PSK-RC4-SHA
+#listener.ssl.external.psk_ciphers = RSA-PSK-AES256-GCM-SHA384,RSA-PSK-AES256-CBC-SHA384,RSA-PSK-AES128-GCM-SHA256,RSA-PSK-AES128-CBC-SHA256,RSA-PSK-AES256-CBC-SHA,RSA-PSK-AES128-CBC-SHA
 
 ## SSL parameter renegotiation is a feature that allows a client and a server
 ## to renegotiate the parameters of the SSL connection on the fly.
@@ -1998,7 +1998,7 @@ listener.wss.external.ciphers = TLS_AES_256_GCM_SHA384,TLS_AES_128_GCM_SHA256,TL
 ## Note that 'listener.wss.external.ciphers' and 'listener.wss.external.psk_ciphers' cannot
 ## be configured at the same time.
 ## See 'https://tools.ietf.org/html/rfc4279#section-2'.
-## listener.wss.external.psk_ciphers = PSK-AES128-CBC-SHA,PSK-AES256-CBC-SHA,PSK-3DES-EDE-CBC-SHA,PSK-RC4-SHA
+## listener.wss.external.psk_ciphers = RSA-PSK-AES256-GCM-SHA384,RSA-PSK-AES256-CBC-SHA384,RSA-PSK-AES128-GCM-SHA256,RSA-PSK-AES128-CBC-SHA256,RSA-PSK-AES256-CBC-SHA,RSA-PSK-AES128-CBC-SHA
 
 ## See: listener.ssl.$name.secure_renegotiate
 ##

lib-ce/emqx_dashboard/src/emqx_dashboard.app.src

@@ -1,6 +1,6 @@
 {application, emqx_dashboard,
  [{description, "EMQ X Web Dashboard"},
-  {vsn, "4.4.3"}, % strict semver, bump manually!
+  {vsn, "4.4.4"}, % strict semver, bump manually!
   {modules, []},
   {registered, [emqx_dashboard_sup]},
   {applications, [kernel,stdlib,mnesia,minirest]},

lib-ce/emqx_dashboard/src/emqx_dashboard_admin.erl

@@ -156,7 +156,16 @@ update_pwd(Username, Fun) ->
 
 
 -spec(lookup_user(binary()) -> [mqtt_admin()]).
-lookup_user(Username) when is_binary(Username) -> mnesia:dirty_read(mqtt_admin, Username).
+lookup_user(Username) when is_binary(Username) ->
+    IsDefaultUser = binenv(default_user_username) =:= Username,
+    case mnesia:dirty_read(mqtt_admin, Username) of
+        [] when IsDefaultUser ->
+            _ = ensure_default_user_in_db(Username),
+            %% try to read again
+            mnesia:dirty_read(mqtt_admin, Username);
+        Res ->
+            Res
+    end.
 
 -spec(all_users() -> [#mqtt_admin{}]).
 all_users() -> ets:tab2list(mqtt_admin).
@@ -172,8 +181,8 @@ check(_, undefined) ->
     {error, <<"Password undefined">>};
 check(Username, Password) ->
     case lookup_user(Username) of
-        [#mqtt_admin{password = <<Salt:4/binary, Hash/binary>>}] ->
+        [#mqtt_admin{password = PwdHash}] ->
-            case Hash =:= md5_hash(Salt, Password) of
+            case is_valid_pwd(PwdHash, Password) of
                 true  -> ok;
                 false -> {error, <<"Username/Password error">>}
             end;
@@ -181,13 +190,19 @@ check(Username, Password) ->
             {error, <<"Username/Password error">>}
     end.
 
+is_valid_pwd(<<Salt:4/binary, Hash/binary>>, Password) ->
+    Hash =:= md5_hash(Salt, Password).
+
 %%--------------------------------------------------------------------
 %% gen_server callbacks
 %%--------------------------------------------------------------------
 
 init([]) ->
     %% Add default admin user
-    _ = add_default_user(binenv(default_user_username), binenv(default_user_passwd)),
+    {ok, _} = mnesia:subscribe({table, mqtt_admin, simple}),
+    PasswordHash = ensure_default_user_in_db(binenv(default_user_username)),
+    ok = ensure_default_user_passwd_hashed_in_pt(PasswordHash),
+    ok = maybe_warn_default_pwd(),
     {ok, state}.
 
 handle_call(_Req, _From, State) ->
@@ -196,6 +211,17 @@ handle_call(_Req, _From, State) ->
 handle_cast(_Msg, State) ->
     {noreply, State}.
 
+handle_info({mnesia_table_event, {write, Admin, _}}, State) ->
+    %% the password is chagned from another node, sync it to persistent_term
+    #mqtt_admin{username = Username, password = HashedPassword} = Admin,
+    case binenv(default_user_username) of
+        Username ->
+            ok = ensure_default_user_passwd_hashed_in_pt(HashedPassword);
+        _ ->
+            ignore
+    end,
+    {noreply, State};
+
 handle_info(_Msg, State) ->
     {noreply, State}.
 
@@ -222,37 +248,59 @@ salt() ->
     <<Salt:32>>.
 
 binenv(Key) ->
-    iolist_to_binary(application:get_env(emqx_dashboard, Key, "")).
+    iolist_to_binary(application:get_env(emqx_dashboard, Key, <<>>)).
 
-add_default_user(Username, Password) when ?EMPTY_KEY(Username) orelse ?EMPTY_KEY(Password) ->
+ensure_default_user_in_db(Username) ->
-    ignore;
+    F =
-
+        fun() ->
-add_default_user(Username, Password) ->
+                case mnesia:wread({mqtt_admin, Username}) of
-    case lookup_user(Username) of
+                    [] ->
-        [] -> add_user(Username, Password, <<"administrator">>);
+                        PasswordHash = initial_default_user_passwd_hashed(),
-        _  ->
+                        Admin = #mqtt_admin{username = Username,
-            case check(Username, Password) of
+                                            password = PasswordHash,
-                ok ->
+                                            tags = <<"administrator">>},
-                    ?LOG(warning,
+                        ok = mnesia:write(Admin),
-                        "[Dashboard] The initial default password for dashboard 'admin' user in emqx_dashboard.conf\n"
+                        PasswordHash;
-                        "For safety, it should be changed as soon as possible.\n"
+                    [#mqtt_admin{password = PasswordHash}] ->
-                        "Please use the './bin/emqx_ctl admins' CLI to change it.\n"
+                        PasswordHash
-                        "Then remove `dashboard.default_user.login/password` from emqx_dashboard.conf"
-                    );
-                {error, _} ->
-                    %% We can't force add default,
-                    %% otherwise passwords that have been updated via HTTP API will be reset after reboot.
-                    ?LOG(warning,
-                        "[Dashboard] dashboard.default_user.password in the plugins/emqx_dashboard.conf\n"
-                        "does not match the password in the database(mnesia).\n"
-                        "1. If you have already changed the password via the HTTP API or `./bin/emqx_ctl admins`,"
-                        "this warning has no effect.\n"
-                        "You should remove the `dashboard.default_user.login/password` from emqx_dashboard.conf "
-                        "to resolve this warning.\n"
-                        "2. If you just want to update the password by manually changing the configuration file,\n"
-                        "you need to delete the old user and password using `emqx_ctl admins del ~s` first\n"
-                        "the new password in emqx_dashboard.conf can take effect after reboot.",
-                        [])
                 end
         end,
-    ok.
+    {atomic, PwdHash} = mnesia:transaction(F),
+    PwdHash.
+
+initial_default_user_passwd_hashed() ->
+    case get_default_user_passwd_hashed_from_pt() of
+        Empty when ?EMPTY_KEY(Empty) ->
+            %% in case it's not set yet
+            case binenv(default_user_passwd) of
+                Empty when ?EMPTY_KEY(Empty) ->
+                    error({missing_configuration, default_user_passwd});
+                Pwd ->
+                    hash(Pwd)
+            end;
+        PwdHash ->
+            PwdHash
+    end.
+
+%% use this persistent_term for a copy of the value in mnesia database
+%% so that after the node leaves a cluster, db gets purged,
+%% we can still find the changed password back from PT
+ensure_default_user_passwd_hashed_in_pt(Hashed) ->
+    ok = persistent_term:put({?MODULE, default_user_passwd_hashed}, Hashed).
+
+get_default_user_passwd_hashed_from_pt() ->
+    persistent_term:get({?MODULE, default_user_passwd_hashed}, <<>>).
+
+maybe_warn_default_pwd() ->
+    case is_valid_pwd(initial_default_user_passwd_hashed(), <<"public">>) of
+        true ->
+            ?LOG(warning,
+                 "[Dashboard] Using default password for dashboard 'admin' user. "
+                 "Please use './bin/emqx_ctl admins' command to change it. "
+                 "NOTE: the default password in config file is only "
+                 "used to initialise the database record, changing the config "
+                 "file after database is initialised has no effect."
+                );
+        false ->
+            ok
+    end.

lib-ce/emqx_dashboard/test/emqx_dashboard_SUITE.erl

@@ -29,6 +29,8 @@
 
 -include_lib("emqx/include/emqx.hrl").
 
+-include("emqx_dashboard.hrl").
+
 -define(CONTENT_TYPE, "application/x-www-form-urlencoded").
 
 -define(HOST, "http://127.0.0.1:18083/").
@@ -40,31 +42,45 @@
 -define(OVERVIEWS, ['alarms/activated', 'alarms/deactivated', banned, brokers, stats, metrics, listeners, clients, subscriptions, routes, plugins]).
 
 all() ->
-    [{group, overview},
+    [
+     {group, overview},
      {group, admins},
      {group, rest},
      {group, cli}
      ].
 
 groups() ->
-    [{overview, [sequence], [t_overview]},
+    [
-     {admins, [sequence], [t_admins_add_delete]},
+     {overview, [sequence], [t_overview]},
-     {rest, [sequence], [t_rest_api, t_auth_exhaustive_attack]},
+     {admins, [sequence], [t_admins_add_delete, t_admins_persist_default_password, t_default_password_persists_after_leaving_cluster]},
+     {rest, [sequence], [t_rest_api]},
      {cli, [sequence], [t_cli]}
     ].
 
 init_per_suite(Config) ->
     application:load(emqx_plugin_libs),
-    emqx_ct_helpers:start_apps([emqx_modules, emqx_management, emqx_dashboard]),
+    ok = emqx_ct_helpers:start_apps([emqx_modules, emqx_management, emqx_dashboard]),
     Config.
 
 end_per_suite(_Config) ->
     emqx_ct_helpers:stop_apps([emqx_dashboard, emqx_management, emqx_modules]),
     ekka_mnesia:ensure_stopped().
 
+init_per_testcase(Case, Config) ->
+    ?MODULE:Case({init, Config}).
+
+end_per_testcase(Case, Config) ->
+    %% revert to default password
+    emqx_dashboard_admin:change_password(<<"admin">>, <<"public">>),
+    ?MODULE:Case({'end', Config}).
+
+t_overview({init, Config}) -> Config;
+t_overview({'end', _Config}) -> ok;
 t_overview(_) ->
     [?assert(request_dashboard(get, api_path(erlang:atom_to_list(Overview)), auth_header_()))|| Overview <- ?OVERVIEWS].
 
+t_admins_add_delete({init, Config}) -> Config;
+t_admins_add_delete({'end', _Config}) -> ok;
 t_admins_add_delete(_) ->
     ok = emqx_dashboard_admin:add_user(<<"username">>, <<"password">>, <<"tag">>),
     ok = emqx_dashboard_admin:add_user(<<"username1">>, <<"password1">>, <<"tag1">>),
@@ -85,9 +101,100 @@ t_admins_add_delete(_) ->
     ?assertNotEqual(true, request_dashboard(get, api_path("brokers"),
         auth_header_("username", "pwd"))).
 
+t_admins_persist_default_password({init, Config}) -> Config;
+t_admins_persist_default_password({'end', _Config}) -> ok;
+t_admins_persist_default_password(_) ->
+    emqx_dashboard_admin:change_password(<<"admin">>, <<"new_password">>),
+    ct:sleep(100),
+    [#mqtt_admin{password=Password, tags= <<"administrator">>}] = emqx_dashboard_admin:lookup_user(<<"admin">>),
+
+    %% To ensure that state persists even if the process dies
+    application:stop(emqx_dashboard),
+    application:start(emqx_dashboard),
+
+    ct:sleep(100),
+
+    %% It gets restarted by the app automatically
+    [#mqtt_admin{password=PasswordAfterRestart}] = emqx_dashboard_admin:lookup_user(<<"admin">>),
+    ?assertEqual(Password, PasswordAfterRestart).
+
+debug(Label, Slave) ->
+    ct:print(
+      "[~p]~nusers local ~p~nusers remote: ~p~nenv local: ~p~nenv remote: ~p",
+      [
+       Label,
+       ets:tab2list(mqtt_admin),
+       rpc:call(Slave, ets, tab2list, [mqtt_admin]),
+       application:get_all_env(emqx_dashboard),
+       rpc:call(Slave, application, get_all_env, [emqx_dashboard])
+      ]).
+
+
+t_default_password_persists_after_leaving_cluster({init, Config}) ->
+    Slave = start_slave('test1', [emqx_modules, emqx_management, emqx_dashboard]),
+    [{slave, Slave} | Config];
+t_default_password_persists_after_leaving_cluster({'end', Config}) ->
+    Slave = proplists:get_value(slave, Config),
+    {ok, _} = stop_slave(Slave, [emqx_dashboard, emqx_management, emqx_modules]),
+    ok;
+t_default_password_persists_after_leaving_cluster(Config) ->
+    Slave = proplists:get_value(slave, Config),
+    [#mqtt_admin{password=InitialPassword}] = emqx_dashboard_admin:lookup_user(<<"admin">>),
+
+    ct:print("Cluster status: ~p", [ekka_cluster:info()]),
+    ct:print("Table nodes: ~p", [mnesia:table_info(mqtt_admin, active_replicas)]),
+
+
+    %% To make sure that subscription is not lost during reconnection
+    rpc:call(Slave, ekka, leave, []),
+    ct:sleep(100), %% To ensure that leave gets processed
+    rpc:call(Slave, ekka, join, [node()]),
+    ct:sleep(100), %% To ensure that join gets processed
+
+    ct:print("Cluster status: ~p", [ekka_cluster:info()]),
+    ct:print("Table nodes: ~p", [mnesia:table_info(mqtt_admin, active_replicas)]),
+
+    ct:print("Apps: ~p", [
+                          rpc:call(Slave, application, which_applications, [])
+                         ]),
+
+    debug(0, Slave),
+
+    emqx_dashboard_admin:change_password(<<"admin">>, <<"new_password">>),
+    ct:sleep(100), %% To ensure that event gets processed
+
+    debug(1, Slave),
+
+    [#mqtt_admin{password=Password}] = rpc:call(Slave, emqx_dashboard_admin, lookup_user, [<<"admin">>]),
+    ?assertNotEqual(InitialPassword, Password),
+
+    rpc:call(Slave, ekka, leave, []),
+
+    debug(2, Slave),
+
+    rpc:call(Slave, application, stop, [emqx_dashboard]),
+
+    debug(3, Slave),
+
+    rpc:call(Slave, application, start, [emqx_dashboard]),
+
+    debug(4, Slave),
+
+    ?assertEqual(
+       ok,
+       rpc:call(Slave, emqx_dashboard_admin, check, [<<"admin">>, <<"new_password">>])),
+
+    ?assertMatch(
+       {error, _},
+       rpc:call(Slave, emqx_dashboard_admin, check, [<<"admin">>, <<"password">>])),
+    ok.
+
+t_rest_api({init, Config}) -> Config;
+t_rest_api({'end', _Config}) -> ok;
 t_rest_api(_Config) ->
     {ok, Res0} = http_get("users"),
     Users = get_http_data(Res0),
+    ct:pal("~p", [emqx_dashboard_admin:all_users()]),
     ?assert(lists:member(#{<<"username">> => <<"admin">>, <<"tags">> => <<"administrator">>},
         Users)),
 
@@ -104,11 +211,15 @@ t_rest_api(_Config) ->
              ]],
     ok.
 
+t_auth_exhaustive_attack({init, Config}) -> Config;
+t_auth_exhaustive_attack({'end', _Config}) -> ok;
 t_auth_exhaustive_attack(_Config) ->
     {ok, Res0} = http_post("auth", #{<<"username">> => <<"invalid_login">>, <<"password">> => <<"newpwd">>}),
     {ok, Res1} = http_post("auth", #{<<"username">> => <<"admin">>, <<"password">> => <<"invalid_password">>}),
     ?assertEqual(Res0, Res1).
 
+t_cli({init, Config}) -> Config;
+t_cli({'end', _Config}) -> ok;
 t_cli(_Config) ->
     [mnesia:dirty_delete({mqtt_admin, Admin}) ||  Admin <- mnesia:dirty_all_keys(mqtt_admin)],
     emqx_dashboard_cli:admins(["add", "username", "password"]),
@@ -171,3 +282,53 @@ api_path(Path) ->
 
 json(Data) ->
     {ok, Jsx} = emqx_json:safe_decode(Data, [return_maps]), Jsx.
+
+start_slave(Name, Apps) ->
+    {ok, Node} = ct_slave:start(list_to_atom(atom_to_list(Name) ++ "@" ++ host()),
+                                [{kill_if_fail, true},
+                                 {monitor_master, true},
+                                 {init_timeout, 10000},
+                                 {startup_timeout, 10000},
+                                 {erl_flags, ebin_path()}]),
+
+    pong = net_adm:ping(Node),
+    setup_node(Node, Apps),
+    Node.
+
+stop_slave(Node, Apps) ->
+    [ok = Res || Res <- rpc:call(Node, emqx_ct_helpers, stop_apps, [Apps])],
+    rpc:call(Node, ekka, leave, []),
+    ct_slave:stop(Node).
+
+host() ->
+    [_, Host] = string:tokens(atom_to_list(node()), "@"), Host.
+
+ebin_path() ->
+    string:join(["-pa" | lists:filter(fun is_lib/1, code:get_path())], " ").
+
+is_lib(Path) ->
+    string:prefix(Path, code:lib_dir()) =:= nomatch.
+
+setup_node(Node, Apps) ->
+    EnvHandler =
+        fun(emqx) ->
+                application:set_env(emqx, listeners, []),
+                application:set_env(gen_rpc, port_discovery, manual),
+                ok;
+           (emqx_management) ->
+                application:set_env(emqx_management, listeners, []),
+                ok;
+           (emqx_dashboard) ->
+                application:set_env(emqx_dashboard, listeners, []),
+                ok;
+           (_) ->
+                ok
+        end,
+
+    [ok = rpc:call(Node, application, load, [App]) || App <- [gen_rpc, emqx | Apps]],
+    ok = rpc:call(Node, emqx_ct_helpers, start_apps, [Apps, EnvHandler]),
+
+    rpc:call(Node, ekka, join, [node()]),
+    rpc:call(Node, application, stop, [emqx_dashboard]),
+    rpc:call(Node, application, start, [emqx_dashboard]),
+    ok.

lib-ce/emqx_modules/src/emqx_modules.app.src

@@ -1,6 +1,6 @@
 {application, emqx_modules,
  [{description, "EMQ X Module Management"},
-  {vsn, "4.4.2"},
+  {vsn, "4.4.3"},
   {modules, []},
   {applications, [kernel,stdlib]},
   {mod, {emqx_modules_app, []}},

lib-ce/emqx_modules/src/emqx_modules.appup.src

@@ -1,21 +1,27 @@
 %% -*- mode: erlang -*-
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
-  [{"4.4.1",
+  [{"4.4.2", [{load_module,emqx_modules,brutal_purge,soft_purge,[]}]},
-    [{load_module,emqx_mod_subscription,brutal_purge,soft_purge,[]},
+   {"4.4.1",
+    [{load_module,emqx_modules,brutal_purge,soft_purge,[]},
+     {load_module,emqx_mod_subscription,brutal_purge,soft_purge,[]},
      {load_module,emqx_mod_delayed,brutal_purge,soft_purge,[]}]},
    {"4.4.0",
-    [{load_module,emqx_mod_subscription,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_modules,brutal_purge,soft_purge,[]},
+     {load_module,emqx_mod_subscription,brutal_purge,soft_purge,[]},
      {load_module,emqx_mod_delayed,brutal_purge,soft_purge,[]},
      {load_module,emqx_mod_presence,brutal_purge,soft_purge,[]},
      {load_module,emqx_mod_sup,brutal_purge,soft_purge,[]},
      {load_module,emqx_mod_trace_api,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
-  [{"4.4.1",
+  [{"4.4.2", [{load_module,emqx_modules,brutal_purge,soft_purge,[]}]},
-    [{load_module,emqx_mod_subscription,brutal_purge,soft_purge,[]},
+   {"4.4.1",
+    [{load_module,emqx_modules,brutal_purge,soft_purge,[]},
+     {load_module,emqx_mod_subscription,brutal_purge,soft_purge,[]},
      {load_module,emqx_mod_delayed,brutal_purge,soft_purge,[]}]},
    {"4.4.0",
-    [{load_module,emqx_mod_subscription,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_modules,brutal_purge,soft_purge,[]},
+     {load_module,emqx_mod_subscription,brutal_purge,soft_purge,[]},
      {load_module,emqx_mod_delayed,brutal_purge,soft_purge,[]},
      {load_module,emqx_mod_presence,brutal_purge,soft_purge,[]},
      {load_module,emqx_mod_sup,brutal_purge,soft_purge,[]},

lib-ce/emqx_modules/src/emqx_modules.erl

@@ -43,6 +43,7 @@ load() ->
     case emqx:get_env(modules_loaded_file) of
         undefined -> ok;
         File ->
+            ensure_loaded_modules_file(File),
             load_modules(File)
     end.
 
@@ -58,6 +59,31 @@ load(ModuleName) ->
             emqx_modules:load_module(ModuleName, true)
     end.
 
+%% @doc Creates a `loaded_modules' file with default values if one
+%% doesn't exist.
+-spec ensure_loaded_modules_file(file:filename()) -> ok.
+ensure_loaded_modules_file(Filepath) ->
+    case filelib:is_regular(Filepath) of
+        true ->
+            ok;
+        false ->
+            do_ensure_loaded_modules_file(Filepath)
+    end.
+
+do_ensure_loaded_modules_file(Filepath) ->
+    DefaultModules = [emqx_mod_acl_internal, emqx_mod_presence],
+    Res = file:write_file(Filepath,
+                          [io_lib:format("{~p, true}.~n", [Mod])
+                           || Mod <- DefaultModules]),
+    case Res of
+        ok ->
+            ok;
+        {error, Reason} ->
+            ?LOG(error, "Could not write default loaded_modules file ~p ; Error: ~p",
+                 [Filepath, Reason]),
+            ok
+    end.
+
 %% @doc Unload all the extended modules.
 -spec(unload() -> ok).
 unload() ->
@@ -175,6 +201,8 @@ write_loaded(false) -> ok.
 
 %%--------------------------------------------------------------------
 %% @doc Modules Command
+%%--------------------------------------------------------------------
+
 cli(["list"]) ->
     lists:foreach(fun({Name, Active}) ->
                     emqx_ctl:print("Module(~s, description=~s, active=~s)~n",

lib-ce/emqx_modules/test/emqx_modules_SUITE.erl

@@ -20,6 +20,7 @@
 -compile(nowarn_export_all).
 
 -include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
 
 -define(CONTENT_TYPE, "application/x-www-form-urlencoded").
 
@@ -44,6 +45,32 @@ end_per_suite(_Config) ->
     emqx_ct_http:delete_default_app(),
     emqx_ct_helpers:stop_apps([emqx_modules, emqx_management]).
 
+init_per_testcase(t_ensure_default_loaded_modules_file, Config) ->
+    LoadedModulesFilepath = application:get_env(emqx, modules_loaded_file),
+    ok = application:stop(emqx_modules),
+    TmpFilepath = filename:join(["/", "tmp", "loaded_modules_tmp"]),
+    case file:delete(TmpFilepath) of
+        ok -> ok;
+        {error, enoent} -> ok
+    end,
+    application:set_env(emqx, modules_loaded_file, TmpFilepath),
+    [ {loaded_modules_filepath, LoadedModulesFilepath}
+    , {tmp_filepath, TmpFilepath}
+    | Config];
+init_per_testcase(_TestCase, Config) ->
+    Config.
+
+end_per_testcase(t_ensure_default_loaded_modules_file, Config) ->
+    LoadedModulesFilepath = ?config(loaded_modules_filepath, Config),
+    TmpFilepath = ?config(tmp_filepath, Config),
+    file:delete(TmpFilepath),
+    ok = application:stop(emqx_modules),
+    application:set_env(emqx, modules_loaded_file, LoadedModulesFilepath),
+    ok = application:start(emqx_modules),
+    ok;
+end_per_testcase(_TestCase, _Config) ->
+    ok.
+
 t_load(_) ->
     ?assertEqual(ok, emqx_modules:unload()),
     ?assertEqual(ok, emqx_modules:load()),
@@ -52,6 +79,19 @@ t_load(_) ->
     ?assertEqual(ignore, emqx_modules:reload(emqx_mod_rewrite)),
     ?assertEqual(ok, emqx_modules:reload(emqx_mod_acl_internal)).
 
+t_ensure_default_loaded_modules_file(_Config) ->
+    ok = application:start(emqx_modules),
+    ?assertEqual(
+       [ {emqx_mod_acl_internal,true}
+       , {emqx_mod_delayed,false}
+       , {emqx_mod_presence,true}
+       , {emqx_mod_rewrite,false}
+       , {emqx_mod_subscription,false}
+       , {emqx_mod_topic_metrics,false}
+       ],
+       lists:sort(emqx_modules:list())),
+    ok.
+
 t_list(_) ->
     ?assertMatch([{_, _} | _ ], emqx_modules:list()).
 

priv/emqx.schema

@@ -2102,12 +2102,30 @@ end}.
                           {reuseaddr, cuttlefish:conf_get(Prefix ++ ".reuseaddr", Conf, undefined)}])
               end,
     SplitFun = fun(undefined) -> undefined; (S) -> string:tokens(S, ",") end,
+    %% In erlang, we only support the following PSK ciphers (ssl_cipher:psk_suites(3))
+    AvaiableCiphers = ["RSA-PSK-AES256-GCM-SHA384","RSA-PSK-AES256-CBC-SHA384",
+                       "RSA-PSK-AES128-GCM-SHA256","RSA-PSK-AES128-CBC-SHA256",
+                       "RSA-PSK-AES256-CBC-SHA","RSA-PSK-AES128-CBC-SHA"
+                      ],
+    %% Compatible with legacy PSK Cipher strings
+    PskMapping = fun("PSK-AES128-CBC-SHA") -> {true, "RSA-PSK-AES128-CBC-SHA"};
+                    ("PSK-AES256-CBC-SHA") -> {true, "RSA-PSK-AES256-CBC-SHA"};
+                    ("PSK-3DES-EDE-CBC-SHA") -> {true, "PSK-3DES-EDE-CBC-SHA"};
+                    ("PSK-RC4-SHA") -> {true, "PSK-RC4-SHA"};
+                    (C) -> case lists:member(C, AvaiableCiphers) of
+                               true -> {true, C};
+                               false -> false
+                           end
+                 end,
     MapPSKCiphers = fun(PSKCiphers) ->
-                      lists:map(
+                      lists:filtermap(fun(C0) ->
-                          fun("PSK-AES128-CBC-SHA") -> {psk, aes_128_cbc, sha};
+                        case PskMapping(C0) of
-                             ("PSK-AES256-CBC-SHA") -> {psk, aes_256_cbc, sha};
+                            false ->
-                             ("PSK-3DES-EDE-CBC-SHA") -> {psk, '3des_ede_cbc', sha};
+                              cuttlefish:invalid(
-                             ("PSK-RC4-SHA") -> {psk, rc4_128, sha}
+                                io_lib:format("psk_ciphers: not support ~s", [C0]));
+                            {true, C} ->
+                                {true, C}
+                        end
                       end, PSKCiphers)
                     end,
     SslOpts = fun(Prefix) ->
@@ -2124,7 +2142,14 @@ end}.
                       {TLSCiphers, undefined} ->
                         SplitFun(TLSCiphers);
                       {undefined, PSKCiphers} ->
-                        MapPSKCiphers(SplitFun(PSKCiphers));
+                        case Versions == undefined orelse lists:member('tlsv1.3', Versions) of
+                            true ->
+                                cuttlefish:invalid(
+                                  Prefix++".tls_versions cannot contain tlsv1.3 "
+                                  "if "++Prefix++".psk_ciphers is configured");
+                            _ ->
+                                MapPSKCiphers(SplitFun(PSKCiphers))
+                        end;
                       {_TLSCiphers, _PSKCiphers} ->
                         cuttlefish:invalid(Prefix++".ciphers and "++Prefix++".psk_ciphers cannot be configured at the same time")
                     end,

rebar.config

@@ -6,6 +6,7 @@
 %% with rebar.config.erl module. Final result is written to
 %% rebar.config.rendered if environment DEBUG is set.
 
+{minimum_otp_vsn, "23"}.
 {edoc_opts, [{preprocess,true}]}.
 {erl_opts, [warn_unused_vars,warn_shadow_vars,warn_unused_import,
             warn_obsolete_guard,compressed,
@@ -37,7 +38,7 @@
 
 {deps,
     [ {gpb, "4.11.2"} %% gpb only used to build, but not for release, pin it here to avoid fetching a wrong version due to rebar plugins scattered in all the deps
-    , {ehttpc, {git, "https://github.com/emqx/ehttpc", {tag, "0.1.14"}}}
+    , {ehttpc, {git, "https://github.com/emqx/ehttpc", {tag, "0.1.15"}}}
     , {eredis_cluster, {git, "https://github.com/emqx/eredis_cluster", {tag, "0.6.5"}}}
     , {gproc, {git, "https://github.com/uwiger/gproc", {tag, "0.8.0"}}}
     , {jiffy, {git, "https://github.com/emqx/jiffy", {tag, "1.0.5"}}}
@@ -56,7 +57,8 @@
     , {observer_cli, "1.6.1"} % NOTE: depends on recon 2.5.1
     , {getopt, "1.0.1"}
     , {snabbkaffe, {git, "https://github.com/kafka4beam/snabbkaffe.git", {tag, "0.15.0"}}}
-    , {lc, {git, "https://github.com/emqx/lc.git", {tag, "0.2.0"}}}
+    , {lc, {git, "https://github.com/emqx/lc.git", {tag, "0.2.1"}}}
+    , {mongodb, {git,"https://github.com/emqx/mongodb-erlang", {tag, "v3.0.13"}}}
     ]}.
 
 {xref_ignores,

scripts/update-appup.sh

@@ -55,6 +55,10 @@ while [ "$#" -gt 0 ]; do
         SKIP_BUILD='yes'
         shift
         ;;
+    --skip-build-base)
+        SKIP_BUILD_BASE='yes'
+        shift
+        ;;
     --check)
         # hijack the --check option
         IS_CHECK='yes'
@@ -88,16 +92,20 @@ else
     NEW_COPY='no'
 fi
 
-pushd "${PREV_DIR_BASE}/${PREV_TAG}"
+if [ "${SKIP_BUILD_BASE:-no}" = 'yes' ]; then
-if [ "$NEW_COPY" = 'no' ]; then
+    echo "not building relup base ${PREV_DIR_BASE}/${PREV_TAG}"
+else
+    pushd "${PREV_DIR_BASE}/${PREV_TAG}"
+    if [ "$NEW_COPY" = 'no' ]; then
         REMOTE="$(git remote -v | grep "${GIT_REPO}" | head -1 | awk '{print $1}')"
         git fetch "$REMOTE"
+    fi
+    git reset --hard
+    git clean -fdx
+    git checkout "${PREV_TAG}"
+    make "$PROFILE"
+    popd
 fi
-git reset --hard
-git clean -fdx
-git checkout "${PREV_TAG}"
-make "$PROFILE"
-popd
 
 PREV_REL_DIR="${PREV_DIR_BASE}/${PREV_TAG}/_build/${PROFILE}/lib"
 

src/emqx.appup.src

@@ -2,10 +2,16 @@
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
   [{"4.4.2",
-    [{load_module,emqx_app,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_plugins,brutal_purge,soft_purge,[]},
+     {load_module,emqx_frame,brutal_purge,soft_purge,[]},
+     {load_module,emqx,brutal_purge,soft_purge,[]},
+     {load_module,emqx_hooks,brutal_purge,soft_purge,[]},
+     {load_module,emqx_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_relup}]},
    {"4.4.1",
-    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
+    [{load_module,emqx,brutal_purge,soft_purge,[]},
+     {load_module,emqx_hooks,brutal_purge,soft_purge,[]},
+     {load_module,emqx_listeners,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]},
      {load_module,emqx_flapping,brutal_purge,soft_purge,[]},
@@ -26,7 +32,8 @@
      {load_module,emqx_connection,brutal_purge,soft_purge,[]},
      {add_module,emqx_relup}]},
    {"4.4.0",
-    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_hooks,brutal_purge,soft_purge,[]},
+     {load_module,emqx_listeners,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]},
      {load_module,emqx_flapping,brutal_purge,soft_purge,[]},
      {load_module,emqx_pmon,brutal_purge,soft_purge,[]},
@@ -55,10 +62,16 @@
      {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
   [{"4.4.2",
-    [{load_module,emqx_app,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_plugins,brutal_purge,soft_purge,[]},
+     {load_module,emqx_frame,brutal_purge,soft_purge,[]},
+     {load_module,emqx,brutal_purge,soft_purge,[]},
+     {load_module,emqx_hooks,brutal_purge,soft_purge,[]},
+     {load_module,emqx_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_relup}]},
    {"4.4.1",
-    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
+    [{load_module,emqx,brutal_purge,soft_purge,[]},
+     {load_module,emqx_hooks,brutal_purge,soft_purge,[]},
+     {load_module,emqx_listeners,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]},
      {load_module,emqx_flapping,brutal_purge,soft_purge,[]},
@@ -79,7 +92,8 @@
      {load_module,emqx_connection,brutal_purge,soft_purge,[]},
      {delete_module,emqx_relup}]},
    {"4.4.0",
-    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_hooks,brutal_purge,soft_purge,[]},
+     {load_module,emqx_listeners,brutal_purge,soft_purge,[]},
      {load_module,emqx_cm,brutal_purge,soft_purge,[]},
      {load_module,emqx_flapping,brutal_purge,soft_purge,[]},
      {load_module,emqx_pmon,brutal_purge,soft_purge,[]},

src/emqx.erl

@@ -234,7 +234,19 @@ shutdown(Reason) ->
                  ).
 
 reboot() ->
-    lists:foreach(fun application:start/1 , default_started_applications()).
+    case is_application_running(emqx_dashboard) of
+        true ->
+            application:stop(emqx_dashboard), %% dashboard must be started after mnesia
+            lists:foreach(fun application:start/1 , default_started_applications()),
+            application:start(emqx_dashboard);
+
+        false ->
+            lists:foreach(fun application:start/1 , default_started_applications())
+    end.
+
+is_application_running(App) ->
+    StartedApps = proplists:get_value(started, application:info()),
+    proplists:is_defined(App, StartedApps).
 
 -ifdef(EMQX_ENTERPRISE).
 default_started_applications() ->

src/emqx_frame.erl

@@ -265,7 +265,7 @@ parse_packet(#mqtt_packet_header{type = ?CONNACK}, <<AckFlags:8, ReasonCode:8, R
 
 parse_packet(#mqtt_packet_header{type = ?PUBLISH, qos = QoS}, Bin,
              #{strict_mode := StrictMode, version := Ver}) ->
-    {TopicName, Rest} = parse_utf8_string(Bin, StrictMode),
+    {TopicName, Rest} = parse_topic_name(Bin, StrictMode),
     {PacketId, Rest1} = case QoS of
                             ?QOS_0 -> {undefined, Rest};
                             _ -> parse_packet_id(Rest)
@@ -357,7 +357,7 @@ parse_will_message(Packet = #mqtt_packet_connect{will_flag = true,
                                                  proto_ver = Ver},
                    Bin, StrictMode) ->
     {Props, Rest} = parse_properties(Bin, Ver, StrictMode),
-    {Topic, Rest1} = parse_utf8_string(Rest, StrictMode),
+    {Topic, Rest1} = parse_topic_name(Rest, StrictMode),
     {Payload, Rest2} = parse_binary_data(Rest1),
     {Packet#mqtt_packet_connect{will_props   = Props,
                                 will_topic   = Topic,
@@ -524,6 +524,14 @@ parse_binary_data(Bin)
   when 2 > byte_size(Bin) ->
     error(malformed_binary_data_length).
 
+parse_topic_name(Bin, false) ->
+    parse_utf8_string(Bin, false);
+parse_topic_name(Bin, true) ->
+    case parse_utf8_string(Bin, true) of
+        {<<>>, _Rest} -> error(empty_topic_name);
+        Result -> Result
+    end.
+
 %%--------------------------------------------------------------------
 %% Serialize MQTT Packet
 %%--------------------------------------------------------------------

src/emqx_hooks.erl

@@ -32,6 +32,8 @@
         , add/3
         , add/4
         , put/2
+        , put/3
+        , put/4
         , del/2
         , run/2
         , run_fold/3
@@ -75,6 +77,8 @@
           priority :: integer()
          }).
 
+-type callback() :: #callback{}.
+
 -record(hook, {
           name :: hookpoint(),
           callbacks :: list(#callback{})
@@ -110,7 +114,7 @@ callback_priority(#callback{priority= P}) -> P.
 %%--------------------------------------------------------------------
 
 %% @doc Register a callback
--spec(add(hookpoint(), action() | #callback{}) -> ok_or_error(already_exists)).
+-spec(add(hookpoint(), action() | callback()) -> ok_or_error(already_exists)).
 add(HookPoint, Callback) when is_record(Callback, callback) ->
     gen_server:call(?SERVER, {add, HookPoint, Callback}, infinity);
 add(HookPoint, Action) when is_function(Action); is_tuple(Action) ->
@@ -131,12 +135,24 @@ add(HookPoint, Action, Filter, Priority) when is_integer(Priority) ->
     add(HookPoint, #callback{action = Action, filter = Filter, priority = Priority}).
 
 %% @doc Like add/2, it register a callback, discard 'already_exists' error.
--spec(put(hookpoint(), action() | #callback{}) -> ok).
+-spec put(hookpoint(), action() | callback()) -> ok.
-put(HookPoint, Callback) ->
+put(HookPoint, Callback) when is_record(Callback, callback) ->
     case add(HookPoint, Callback) of
         ok -> ok;
         {error, already_exists} -> ok
-    end.
+    end;
+put(HookPoint, Action) when is_function(Action); is_tuple(Action) ->
+    ?MODULE:put(HookPoint, #callback{action = Action, priority = 0}).
+
+-spec put(hookpoint(), action(), filter() | integer() | list()) -> ok.
+put(HookPoint, Action, {_M, _F, _A} = Filter) ->
+    ?MODULE:put(HookPoint, #callback{action = Action, filter = Filter, priority = 0});
+put(HookPoint, Action, Priority) when is_integer(Priority) ->
+    ?MODULE:put(HookPoint, #callback{action = Action, priority = Priority}).
+
+-spec put(hookpoint(), action(), filter(), integer()) -> ok.
+put(HookPoint, Action, Filter, Priority) when is_integer(Priority) ->
+    ?MODULE:put(HookPoint, #callback{action = Action, filter = Filter, priority = Priority}).
 
 %% @doc Unregister a callback.
 -spec(del(hookpoint(), action() | {module(), atom()}) -> ok).
@@ -205,7 +221,7 @@ execute({M, F, A}, Args) ->
     erlang:apply(M, F, Args ++ A).
 
 %% @doc Lookup callbacks.
--spec(lookup(hookpoint()) -> [#callback{}]).
+-spec(lookup(hookpoint()) -> [callback()]).
 lookup(HookPoint) ->
     case ets:lookup(?TAB, HookPoint) of
         [#hook{callbacks = Callbacks}] ->
@@ -288,4 +304,3 @@ del_callback(Func, [#callback{action = {Func, _A}} | Callbacks], Acc) ->
     del_callback(Func, Callbacks, Acc);
 del_callback(Action, [Callback | Callbacks], Acc) ->
     del_callback(Action, Callbacks, [Callback | Acc]).
-

src/emqx_limiter.erl

@@ -159,7 +159,7 @@ update_overall_limiter(Zone, Capacity, Interval) ->
     try
         esockd_limiter:update({Zone, overall_messages_routing}, Capacity, Interval),
         true
-    catch _:_:_ ->
+    catch _:_ ->
         false
     end.
 
@@ -167,6 +167,6 @@ delete_overall_limiter(Zone) ->
     try
         esockd_limiter:delete({Zone, overall_messages_routing}),
         true
-    catch _:_:_ ->
+    catch _:_ ->
         false
     end.

src/emqx_plugins.erl

@@ -215,7 +215,21 @@ load_plugin_conf(AppName, PluginDir) ->
     end, AppsEnv).
 
 ensure_file(File) ->
-    case filelib:is_file(File) of false -> write_loaded([]); true -> ok end.
+    case filelib:is_file(File) of
+        false ->
+            DefaultPlugins = [ {emqx_management, true}
+                             , {emqx_dashboard, true}
+                             , {emqx_modules, false}
+                             , {emqx_recon, true}
+                             , {emqx_retainer, true}
+                             , {emqx_telemetry, true}
+                             , {emqx_rule_engine, true}
+                             , {emqx_bridge_mqtt, false}
+                             ],
+            write_loaded(DefaultPlugins);
+        true ->
+            ok
+    end.
 
 with_loaded_file(File, SuccFun) ->
     case read_loaded(File) of

test/emqx_frame_SUITE.erl

@@ -162,6 +162,14 @@ t_parse_malformed_utf8_string(_) ->
     ParseState = emqx_frame:initial_parse_state(#{strict_mode => true}),
     ?catch_error(utf8_string_invalid, emqx_frame:parse(MalformedPacket, ParseState)).
 
+t_parse_empty_topic_name(_) ->
+    Packet = <<48, 4, 0, 0, 0, 1>>,
+    NormalState = emqx_frame:initial_parse_state(#{strict_mode => false}),
+    ?assertMatch({_, _}, emqx_frame:parse(Packet, NormalState)),
+
+    StrictState = emqx_frame:initial_parse_state(#{strict_mode => true}),
+    ?catch_error(empty_topic_name, emqx_frame:parse(Packet, StrictState)).
+
 t_parse_frame_proxy_protocol(_) ->
     BinList = [ <<"PROXY TCP4 ">>, <<"PROXY TCP6 ">>, <<"PROXY UNKNOWN">>
               , <<"\r\n\r\n\0\r\nQUIT\n">>],

test/emqx_plugins_SUITE.erl

@@ -21,11 +21,11 @@
 
 -include_lib("emqx/include/emqx.hrl").
 -include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
 
 all() -> emqx_ct:all(?MODULE).
 
 init_per_suite(Config) ->
-
     %% Compile extra plugin code
 
     DataPath = proplists:get_value(data_dir, Config),
@@ -47,7 +47,35 @@ set_special_cfg(PluginsDir) ->
     ok.
 
 end_per_suite(_Config) ->
-    emqx_ct_helpers:stop_apps([]).
+    emqx_ct_helpers:stop_apps([]),
+    file:delete(get(loaded_file)).
+
+init_per_testcase(t_ensure_default_loaded_plugins_file, Config) ->
+    {ok, LoadedPluginsFilepath} = application:get_env(emqx, plugins_loaded_file),
+    TmpFilepath = filename:join(["/", "tmp", "loaded_plugins_tmp"]),
+    case file:delete(TmpFilepath) of
+        ok -> ok;
+        {error, enoent} -> ok
+    end,
+    application:set_env(emqx, plugins_loaded_file, TmpFilepath),
+    [ {loaded_plugins_filepath, LoadedPluginsFilepath}
+    , {tmp_filepath, TmpFilepath}
+    | Config];
+init_per_testcase(_TestCase, Config) ->
+    Config.
+
+end_per_testcase(t_ensure_default_loaded_plugins_file, Config) ->
+    LoadedPluginsFilepath = ?config(loaded_plugins_filepath, Config),
+    TmpFilepath = ?config(tmp_filepath, Config),
+    file:delete(TmpFilepath),
+    emqx_plugins:unload(),
+    application:set_env(emqx, plugins_loaded_file, LoadedPluginsFilepath),
+    %% need to purge the plugin to avoid inter-testcase dependencies.
+    code:purge(emqx_mini_plugin_app),
+    ok;
+end_per_testcase(_TestCase, _Config) ->
+    emqx_plugins:unload(),
+    ok.
 
 t_load(_) ->
     ?assertEqual(ok, emqx_plugins:load()),
@@ -61,6 +89,25 @@ t_load(_) ->
     ?assertEqual(ignore, emqx_plugins:load()),
     ?assertEqual(ignore, emqx_plugins:unload()).
 
+t_ensure_default_loaded_plugins_file(Config) ->
+    %% this will trigger it to write the default plugins to the
+    %% inexistent file; but it won't truly load them in this test
+    %% because there are no config files in `expand_plugins_dir'.
+    TmpFilepath = ?config(tmp_filepath, Config),
+    ok = emqx_plugins:load(),
+    {ok, Contents} = file:consult(TmpFilepath),
+    ?assertEqual(
+       [ {emqx_bridge_mqtt, false}
+       , {emqx_dashboard, true}
+       , {emqx_management, true}
+       , {emqx_modules, false}
+       , {emqx_recon, true}
+       , {emqx_retainer, true}
+       , {emqx_rule_engine, true}
+       , {emqx_telemetry, true}
+       ],
+       lists:sort(Contents)),
+    ok.
 
 t_init_config(_) ->
     ConfFile = "emqx_mini_plugin.config",