diff --git a/CHANGES-4.3.md b/CHANGES-4.3.md index 4d420a1d4..81f91450e 100644 --- a/CHANGES-4.3.md +++ b/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 diff --git a/Makefile b/Makefile index 5cfccf879..0ba09b68c 100644 --- a/Makefile +++ b/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-%) diff --git a/apps/emqx_auth_jwt/src/emqx_auth_jwt.app.src b/apps/emqx_auth_jwt/src/emqx_auth_jwt.app.src index 7d784e3b2..e88f6daf1 100644 --- a/apps/emqx_auth_jwt/src/emqx_auth_jwt.app.src +++ b/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]}, diff --git a/apps/emqx_auth_jwt/src/emqx_auth_jwt.appup.src b/apps/emqx_auth_jwt/src/emqx_auth_jwt.appup.src index b9831bb6f..dca3ef2c1 100644 --- a/apps/emqx_auth_jwt/src/emqx_auth_jwt.appup.src +++ b/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, []} ]}, {<<".*">>, []} diff --git a/apps/emqx_auth_jwt/src/emqx_auth_jwt_svr.erl b/apps/emqx_auth_jwt/src/emqx_auth_jwt_svr.erl index b9d19bf57..f34cde783 100644 --- a/apps/emqx_auth_jwt/src/emqx_auth_jwt_svr.erl +++ b/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 diff --git a/apps/emqx_auth_mongo/rebar.config b/apps/emqx_auth_mongo/rebar.config index c89c15d3c..8db0176b0 100644 --- a/apps/emqx_auth_mongo/rebar.config +++ b/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}. - diff --git a/apps/emqx_bridge_mqtt/README.md b/apps/emqx_bridge_mqtt/README.md index fea8f46b3..0052f8408 100644 --- a/apps/emqx_bridge_mqtt/README.md +++ b/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. ``` diff --git a/apps/emqx_bridge_mqtt/docs/guide.rst b/apps/emqx_bridge_mqtt/docs/guide.rst index bb0808367..3c2caeec7 100644 --- a/apps/emqx_bridge_mqtt/docs/guide.rst +++ b/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 diff --git a/apps/emqx_bridge_mqtt/etc/emqx_bridge_mqtt.conf b/apps/emqx_bridge_mqtt/etc/emqx_bridge_mqtt.conf index 1192863c0..faf2fd39c 100644 --- a/apps/emqx_bridge_mqtt/etc/emqx_bridge_mqtt.conf +++ b/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. ## diff --git a/apps/emqx_bridge_mqtt/priv/emqx_bridge_mqtt.schema b/apps/emqx_bridge_mqtt/priv/emqx_bridge_mqtt.schema index 12a571f45..30120a7d5 100644 --- a/apps/emqx_bridge_mqtt/priv/emqx_bridge_mqtt.schema +++ b/apps/emqx_bridge_mqtt/priv/emqx_bridge_mqtt.schema @@ -134,14 +134,30 @@ ]}. {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( - fun("PSK-AES128-CBC-SHA") -> {psk, aes_128_cbc, sha}; - ("PSK-AES256-CBC-SHA") -> {psk, aes_256_cbc, sha}; - ("PSK-3DES-EDE-CBC-SHA") -> {psk, '3des_ede_cbc', sha}; - ("PSK-RC4-SHA") -> {psk, rc4_128, sha} - end, 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, Split = fun(undefined) -> undefined; (S) -> string:tokens(S, ",") end, diff --git a/apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt.app.src b/apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt.app.src index 339c69eba..31c795ff5 100644 --- a/apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt.app.src +++ b/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]}, diff --git a/apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt.appup.src b/apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt.appup.src index 5aa48cea7..b1fa8b85d 100644 --- a/apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt.appup.src +++ b/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, []} ]}, diff --git a/apps/emqx_exhook/etc/emqx_exhook.conf b/apps/emqx_exhook/etc/emqx_exhook.conf index 0608fdadc..70dba4e06 100644 --- a/apps/emqx_exhook/etc/emqx_exhook.conf +++ b/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 diff --git a/apps/emqx_exhook/include/emqx_exhook.hrl b/apps/emqx_exhook/include/emqx_exhook.hrl index 7301fdcbb..58deb707a 100644 --- a/apps/emqx_exhook/include/emqx_exhook.hrl +++ b/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. diff --git a/apps/emqx_exhook/priv/emqx_exhook.schema b/apps/emqx_exhook/priv/emqx_exhook.schema index f55913d72..277e05f0c 100644 --- a/apps/emqx_exhook/priv/emqx_exhook.schema +++ b/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; diff --git a/apps/emqx_exhook/src/emqx_exhook.app.src b/apps/emqx_exhook/src/emqx_exhook.app.src index 715060df4..500f0ef05 100644 --- a/apps/emqx_exhook/src/emqx_exhook.app.src +++ b/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, []}}, diff --git a/apps/emqx_exhook/src/emqx_exhook.appup.src b/apps/emqx_exhook/src/emqx_exhook.appup.src index 0fe0ea78f..05590198e 100644 --- a/apps/emqx_exhook/src/emqx_exhook.appup.src +++ b/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"]}}]}, + {<<".*">>,[]}]}. diff --git a/apps/emqx_exhook/src/emqx_exhook_mngr.erl b/apps/emqx_exhook/src/emqx_exhook_mngr.erl index d4c493cb8..b3d7ea089 100644 --- a/apps/emqx_exhook/src/emqx_exhook_mngr.erl +++ b/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) -> - gen_server:start_link(?MODULE, [Servers, AutoReconnect, ReqOpts], []). +start_link(Servers, AutoReconnect, ReqOpts, HooksOpts) -> + 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, #{}, #{}). -load_all_servers([], _Request, Waiting, Running) -> +load_all_servers(Servers, ReqOpts, HooksOpts) -> + load_all_servers(Servers, ReqOpts, HooksOpts, #{}, #{}). + +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) -> - {ok, State}. +%% in the emqx_exhook:v4.3.5, we have added one new field in the state last: +%% - 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 " diff --git a/apps/emqx_exhook/src/emqx_exhook_server.erl b/apps/emqx_exhook/src/emqx_exhook_server.erl index 276f5a638..0f6d7b87d 100644 --- a/apps/emqx_exhook/src/emqx_exhook_server.erl +++ b/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()} . -load(Name0, Opts0, ReqOpts) -> +-spec load(atom(), emqx_exhook_mngr:server_options(), grpc_client:options(), emqx_exhook_mngr:hooks_options()) + -> {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 - binary_to_existing_atom(Name0, utf8) - catch T:R -> {T,R} - end, - case {lists:member(Name, AvailableHooks), - lists:member(Name, MessageHooks)} of - {false, _} -> - error({unknown_hookpoint, Name}); - {true, false} -> - Acc#{Name => #{}}; - {true, true} -> - Acc#{Name => #{topics => maps:get(topics, HookSpec, [])}} + Name = try binary_to_existing_atom(Name0, utf8) catch T:R -> {T,R} end, + case lists:member(Name, AvailableHooks) of + true -> + case lists:member(Name, MessageHooks) of + true -> + Acc#{Name => #{topics => maps:get(topics, HookSpec, [])}}; + _ -> + Acc#{Name => #{}} + end; + _ -> error({unknown_hookpoint, Name}) 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)). diff --git a/apps/emqx_exhook/src/emqx_exhook_sup.erl b/apps/emqx_exhook/src/emqx_exhook_sup.erl index c92fd6ca4..1b60392dc 100644 --- a/apps/emqx_exhook/src/emqx_exhook_sup.erl +++ b/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). diff --git a/apps/emqx_exhook/test/emqx_exhook_SUITE.erl b/apps/emqx_exhook/test/emqx_exhook_SUITE.erl index b4c58fe62..5aed2f2b8 100644 --- a/apps/emqx_exhook/test/emqx_exhook_SUITE.erl +++ b/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), diff --git a/apps/emqx_exproto/etc/emqx_exproto.conf b/apps/emqx_exproto/etc/emqx_exproto.conf index ae79e1a42..712442eb9 100644 --- a/apps/emqx_exproto/etc/emqx_exproto.conf +++ b/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. diff --git a/apps/emqx_exproto/priv/emqx_exproto.schema b/apps/emqx_exproto/priv/emqx_exproto.schema index fb114dc77..e8a210c2d 100644 --- a/apps/emqx_exproto/priv/emqx_exproto.schema +++ b/apps/emqx_exproto/priv/emqx_exproto.schema @@ -274,13 +274,30 @@ 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( - fun("PSK-AES128-CBC-SHA") -> {psk, aes_128_cbc, sha}; - ("PSK-AES256-CBC-SHA") -> {psk, aes_256_cbc, sha}; - ("PSK-3DES-EDE-CBC-SHA") -> {psk, '3des_ede_cbc', sha}; - ("PSK-RC4-SHA") -> {psk, rc4_128, sha} - end, 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, SslOpts = fun(Prefix) -> Versions = case SplitFun(cuttlefish:conf_get(Prefix ++ ".tls_versions", Conf, undefined)) of diff --git a/apps/emqx_exproto/src/emqx_exproto.app.src b/apps/emqx_exproto/src/emqx_exproto.app.src index 4ce137fc5..fe566ae52 100644 --- a/apps/emqx_exproto/src/emqx_exproto.app.src +++ b/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, []}}, diff --git a/apps/emqx_exproto/src/emqx_exproto.appup.src b/apps/emqx_exproto/src/emqx_exproto.appup.src index 4b58335b4..0da87e289 100644 --- a/apps/emqx_exproto/src/emqx_exproto.appup.src +++ b/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]">>, diff --git a/apps/emqx_lwm2m/priv/emqx_lwm2m.schema b/apps/emqx_lwm2m/priv/emqx_lwm2m.schema index ded81df05..d459d74b8 100644 --- a/apps/emqx_lwm2m/priv/emqx_lwm2m.schema +++ b/apps/emqx_lwm2m/priv/emqx_lwm2m.schema @@ -190,21 +190,41 @@ end}. case cuttlefish:conf_get("lwm2m.dtls.ciphers", Conf, undefined) of undefined -> []; - C -> - [{ciphers, SplitFun(C)}] + Ciphers0 -> + [{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"; - ("PSK-AES256-CBC-SHA") -> "RSA-PSK-AES256-CBC-SHA"; - ("PSK-3DES-EDE-CBC-SHA") -> "RSA-PSK-3DES-EDE-CBC-SHA"; - ("PSK-RC4-SHA") -> "RSA-PSK-RC4-SHA"; - (Suite) -> Suite - end, SplitFun(C2)), - [{ciphers, Psk}, {user_lookup_fun, {fun emqx_psk:lookup/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: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 /= [] diff --git a/apps/emqx_lwm2m/src/emqx_lwm2m.app.src b/apps/emqx_lwm2m/src/emqx_lwm2m.app.src index 717433c96..18c85faad 100644 --- a/apps/emqx_lwm2m/src/emqx_lwm2m.app.src +++ b/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]}, diff --git a/apps/emqx_lwm2m/src/emqx_lwm2m.appup.src b/apps/emqx_lwm2m/src/emqx_lwm2m.appup.src index a25d6cae1..1d2dd58fc 100644 --- a/apps/emqx_lwm2m/src/emqx_lwm2m.appup.src +++ b/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", []}]}. diff --git a/apps/emqx_plugin_libs/src/emqx_plugin_libs.app.src b/apps/emqx_plugin_libs/src/emqx_plugin_libs.app.src index fe4fff0b5..5fbd21fab 100644 --- a/apps/emqx_plugin_libs/src/emqx_plugin_libs.app.src +++ b/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, []} diff --git a/apps/emqx_plugin_libs/src/emqx_plugin_libs.appup.src b/apps/emqx_plugin_libs/src/emqx_plugin_libs.appup.src index 46e76ed82..b183c5e0a 100644 --- a/apps/emqx_plugin_libs/src/emqx_plugin_libs.appup.src +++ b/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", - [ {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_trace_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.2",[{load_module,emqx_plugin_libs_ssl,brutal_purge,soft_purge,[]}]}, + {"4.4.1", + [{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_plugin_libs_ssl,brutal_purge,soft_purge,[]}, + {load_module,emqx_trace,brutal_purge,soft_purge,[]}, + {load_module,emqx_trace_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", - [ {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_trace_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.2",[{load_module,emqx_plugin_libs_ssl,brutal_purge,soft_purge,[]}]}, + {"4.4.1", + [{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_plugin_libs_ssl,brutal_purge,soft_purge,[]}, + {load_module,emqx_trace,brutal_purge,soft_purge,[]}, + {load_module,emqx_trace_api,brutal_purge,soft_purge,[]}, + {update,emqx_slow_subs,{advanced,["4.4.0"]}}, + {load_module,emqx_slow_subs_api,brutal_purge,soft_purge,[]}]}, + {<<".*">>,[]}]}. diff --git a/apps/emqx_plugin_libs/src/emqx_plugin_libs_ssl.erl b/apps/emqx_plugin_libs/src/emqx_plugin_libs_ssl.erl index 2c80dfcfb..8d5e2daa5 100644 --- a/apps/emqx_plugin_libs/src/emqx_plugin_libs_ssl.erl +++ b/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), - Cert = do_save_file(CertFile, Dir), - CA = do_save_file(CAFile, Dir), + Key = maybe_save_file(KeyFile, Dir), + Cert = maybe_save_file(CertFile, 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); -do_save_file(FilePath, _) when is_binary(FilePath) -> + maybe_save_file(ensure_str(FileName), iolist_to_binary(Content), Dir); +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 -do_save_file(_, <<>>, _Dir) -> ""; %% ignore -do_save_file(FileName, Content, Dir) -> +maybe_save_file("", _, _Dir) -> ""; %% no filename, ignore +maybe_save_file(FileName, <<>>, Dir) -> %% no content, see if file exists + {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. diff --git a/apps/emqx_plugin_libs/test/emqx_plugin_libs_ssl_tests.erl b/apps/emqx_plugin_libs/test/emqx_plugin_libs_ssl_tests.erl index d989b9711..54ba48e18 100644 --- a/apps/emqx_plugin_libs/test/emqx_plugin_libs_ssl_tests.erl +++ b/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")). diff --git a/apps/emqx_psk_file/test/emqx_psk_file_SUITE.erl b/apps/emqx_psk_file/test/emqx_psk_file_SUITE.erl index d0083247d..5c4448c67 100644 --- a/apps/emqx_psk_file/test/emqx_psk_file_SUITE.erl +++ b/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}. diff --git a/apps/emqx_rule_engine/src/emqx_rule_engine.app.src b/apps/emqx_rule_engine/src/emqx_rule_engine.app.src index bcf2ee469..d7b77f41d 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_engine.app.src +++ b/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]}, diff --git a/apps/emqx_rule_engine/src/emqx_rule_engine.appup.src b/apps/emqx_rule_engine/src/emqx_rule_engine.appup.src index d5b2c6319..2c01ba70d 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_engine.appup.src +++ b/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", - [{load_module,emqx_rule_events,brutal_purge,soft_purge,[]}, + [{"4.4.2", + [{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", - [{load_module,emqx_rule_events,brutal_purge,soft_purge,[]}, + [{"4.4.2", + [{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"]}}, diff --git a/apps/emqx_rule_engine/src/emqx_rule_engine.erl b/apps/emqx_rule_engine/src/emqx_rule_engine.erl index c8e69a17f..068edc571 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_engine.erl +++ b/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), - %% 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]), - {ok, 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]))), + {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}, - on_destroy = {ModD, Destroy}, - params_spec = ParamSpec}} -> - Config = emqx_rule_validator:validate_params(Config0, ParamSpec), - ResId = resource_id(), + {ok, #resource_type{}} -> + ResId = maps:get(id, Params, resource_id()), try - _ = ?CLUSTER_CALL(init_resource, [ModC, Create, ResId, Config]), - _ = ?CLUSTER_CALL(clear_resource, [ModD, Destroy, ResId]), - ok - catch - throw:Reason -> {error, Reason} + case create_resource(maps:put(id, ResId, Params), no_retry) of + {ok, _} -> + case is_source_alive(ResId) of + true -> + 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 diff --git a/apps/emqx_rule_engine/src/emqx_rule_engine_api.erl b/apps/emqx_rule_engine/src/emqx_rule_engine_api.erl index 39ac1e9c2..b2766e0e0 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_engine_api.erl +++ b/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} -> diff --git a/apps/emqx_rule_engine/src/emqx_rule_events.erl b/apps/emqx_rule_engine/src/emqx_rule_events.erl index 1cef282c8..57aed3868 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_events.erl +++ b/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, - expiry_interval := ExpiryInterval + receive_maximum := RcvMax }) -> + 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, diff --git a/apps/emqx_rule_engine/src/emqx_rule_funcs.erl b/apps/emqx_rule_engine/src/emqx_rule_funcs.erl index 5181731d1..7c972323a 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_funcs.erl +++ b/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. diff --git a/apps/emqx_rule_engine/src/emqx_rule_metrics.erl b/apps/emqx_rule_engine/src/emqx_rule_metrics.erl index 0dfec930e..a725ca35d 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_metrics.erl +++ b/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)}}; diff --git a/apps/emqx_rule_engine/src/emqx_rule_runtime.erl b/apps/emqx_rule_engine/src/emqx_rule_runtime.erl index 35121c046..2912bd7b3 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_runtime.erl +++ b/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) -> - try 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([Rule|More], Input) -> + apply_rule_discard_result(Rule, Input), 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; diff --git a/apps/emqx_rule_engine/src/emqx_rule_sqltester.erl b/apps/emqx_rule_engine/src/emqx_rule_sqltester.erl index 747de87d7..cd00c4384 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_sqltester.erl +++ b/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. diff --git a/apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl b/apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl index 2a0498d2f..562254285 100644 --- a/apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl +++ b/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 %%------------------------------------------------------------------------------ diff --git a/apps/emqx_rule_engine/test/emqx_rule_metrics_SUITE.erl b/apps/emqx_rule_engine/test/emqx_rule_metrics_SUITE.erl index 5a8d99d0e..265072b97 100644 --- a/apps/emqx_rule_engine/test/emqx_rule_metrics_SUITE.erl +++ b/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. diff --git a/bin/install_upgrade.escript b/bin/install_upgrade.escript index 2489ceed5..1af1df11a 100755 --- a/bin/install_upgrade.escript +++ b/bin/install_upgrade.escript @@ -262,19 +262,23 @@ make_symlink_or_copy(Filename, ReleaseLink) -> unpack_zipballs(RelNameStr, Version) -> {ok, Cwd} = file:get_cwd(), - GzFile = filename:absname(filename:join(["releases", RelNameStr ++ "-" ++ Version ++ ".tar.gz"])), - ZipFiles = filelib:wildcard(filename:join(["releases", RelNameStr ++ "-*" ++ Version ++ "*.zip"])), - ?INFO("unzip ~p", [ZipFiles]), - [begin - TmdTarD="/tmp/emqx_untar_" ++ integer_to_list(erlang:system_time()), - ok = filelib:ensure_dir(filename:join([TmdTarD, "dummy"])), - {ok, _} = file:copy(Zip, filename:join([TmdTarD, "emqx.zip"])), - ok = file:set_cwd(filename:join([TmdTarD])), - {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], - file:set_cwd(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]), + [begin + TmdTarD="/tmp/emqx_untar_" ++ integer_to_list(erlang:system_time()), + ok = filelib:ensure_dir(filename:join([TmdTarD, "dummy"])), + {ok, _} = file:copy(Zip, filename:join([TmdTarD, "emqx.zip"])), + ok = file:set_cwd(filename:join([TmdTarD])), + {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] + after + % restore cwd + file:set_cwd(Cwd) + end. first_value(_Fun, []) -> no_value; first_value(Fun, [Value | Rest]) -> diff --git a/deploy/docker/docker-entrypoint.sh b/deploy/docker/docker-entrypoint.sh index ecbec57df..8d82318d7 100755 --- a/deploy/docker/docker-entrypoint.sh +++ b/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 - sed -r "s/\{\s*($var)\s*,\s*(true|false)\s*\}\s*\./{\1, true}./1" "$file" > tmpfile && cat tmpfile > "$file" - elif grep -q "$var\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" 2>/dev/null > tmpfile && cat tmpfile > "$file" + 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 diff --git a/etc/emqx.conf b/etc/emqx.conf index 3b4cca025..2be03e193 100644 --- a/etc/emqx.conf +++ b/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 ## diff --git a/lib-ce/emqx_dashboard/src/emqx_dashboard.app.src b/lib-ce/emqx_dashboard/src/emqx_dashboard.app.src index 0156ea865..b02625117 100644 --- a/lib-ce/emqx_dashboard/src/emqx_dashboard.app.src +++ b/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]}, diff --git a/lib-ce/emqx_dashboard/src/emqx_dashboard_admin.erl b/lib-ce/emqx_dashboard/src/emqx_dashboard_admin.erl index 420380f88..8b35d167d 100644 --- a/lib-ce/emqx_dashboard/src/emqx_dashboard_admin.erl +++ b/lib-ce/emqx_dashboard/src/emqx_dashboard_admin.erl @@ -146,9 +146,9 @@ update_pwd(Username, Fun) -> Trans = fun() -> User = case lookup_user(Username) of - [Admin] -> Admin; - [] -> - mnesia:abort(<<"Username Not Found">>) + [Admin] -> Admin; + [] -> + mnesia:abort(<<"Username Not Found">>) end, mnesia:write(Fun(User)) end, @@ -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 = <>}] -> - case Hash =:= md5_hash(Salt, Password) of + [#mqtt_admin{password = PwdHash}] -> + 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(<>, 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() -> <>. 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) -> - ignore; +ensure_default_user_in_db(Username) -> + F = + fun() -> + case mnesia:wread({mqtt_admin, Username}) of + [] -> + PasswordHash = initial_default_user_passwd_hashed(), + Admin = #mqtt_admin{username = Username, + password = PasswordHash, + tags = <<"administrator">>}, + ok = mnesia:write(Admin), + PasswordHash; + [#mqtt_admin{password = PasswordHash}] -> + PasswordHash + end + end, + {atomic, PwdHash} = mnesia:transaction(F), + PwdHash. -add_default_user(Username, Password) -> - case lookup_user(Username) of - [] -> add_user(Username, Password, <<"administrator">>); - _ -> - case check(Username, Password) of - ok -> - ?LOG(warning, - "[Dashboard] The initial default password for dashboard 'admin' user in emqx_dashboard.conf\n" - "For safety, it should be changed as soon as possible.\n" - "Please use the './bin/emqx_ctl admins' CLI to change it.\n" - "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. +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. diff --git a/lib-ce/emqx_dashboard/test/emqx_dashboard_SUITE.erl b/lib-ce/emqx_dashboard/test/emqx_dashboard_SUITE.erl index 8a5a47a24..c4b47d4f9 100644 --- a/lib-ce/emqx_dashboard/test/emqx_dashboard_SUITE.erl +++ b/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]}, - {rest, [sequence], [t_rest_api, t_auth_exhaustive_attack]}, + [ + {overview, [sequence], [t_overview]}, + {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. diff --git a/lib-ce/emqx_modules/src/emqx_modules.app.src b/lib-ce/emqx_modules/src/emqx_modules.app.src index fbcc6fc68..1ff39b0f8 100644 --- a/lib-ce/emqx_modules/src/emqx_modules.app.src +++ b/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, []}}, diff --git a/lib-ce/emqx_modules/src/emqx_modules.appup.src b/lib-ce/emqx_modules/src/emqx_modules.appup.src index 7e79063f8..51382cdfa 100644 --- a/lib-ce/emqx_modules/src/emqx_modules.appup.src +++ b/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", - [{load_module,emqx_mod_subscription,brutal_purge,soft_purge,[]}, + [{"4.4.2", [{load_module,emqx_modules,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", - [{load_module,emqx_mod_subscription,brutal_purge,soft_purge,[]}, + [{"4.4.2", [{load_module,emqx_modules,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,[]}, diff --git a/lib-ce/emqx_modules/src/emqx_modules.erl b/lib-ce/emqx_modules/src/emqx_modules.erl index 3de9c6ba3..c5565d7d6 100644 --- a/lib-ce/emqx_modules/src/emqx_modules.erl +++ b/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,8 +201,10 @@ write_loaded(false) -> ok. %%-------------------------------------------------------------------- %% @doc Modules Command +%%-------------------------------------------------------------------- + cli(["list"]) -> - lists:foreach(fun({Name, Active}) -> + lists:foreach(fun({Name, Active}) -> emqx_ctl:print("Module(~s, description=~s, active=~s)~n", [Name, Name:description(), Active]) end, emqx_modules:list()); diff --git a/lib-ce/emqx_modules/test/emqx_modules_SUITE.erl b/lib-ce/emqx_modules/test/emqx_modules_SUITE.erl index 175b24bba..a149d9991 100644 --- a/lib-ce/emqx_modules/test/emqx_modules_SUITE.erl +++ b/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()). diff --git a/priv/emqx.schema b/priv/emqx.schema index de742f761..2359461ee 100644 --- a/priv/emqx.schema +++ b/priv/emqx.schema @@ -2102,13 +2102,31 @@ 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( - fun("PSK-AES128-CBC-SHA") -> {psk, aes_128_cbc, sha}; - ("PSK-AES256-CBC-SHA") -> {psk, aes_256_cbc, sha}; - ("PSK-3DES-EDE-CBC-SHA") -> {psk, '3des_ede_cbc', sha}; - ("PSK-RC4-SHA") -> {psk, rc4_128, sha} - end, 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, SslOpts = fun(Prefix) -> Versions = case SplitFun(cuttlefish:conf_get(Prefix ++ ".tls_versions", Conf, undefined)) of @@ -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, diff --git a/rebar.config b/rebar.config index 10ba754e9..50a60fe89 100644 --- a/rebar.config +++ b/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, diff --git a/scripts/update-appup.sh b/scripts/update-appup.sh index d62faee74..55bcc0122 100755 --- a/scripts/update-appup.sh +++ b/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 [ "$NEW_COPY" = 'no' ]; then - REMOTE="$(git remote -v | grep "${GIT_REPO}" | head -1 | awk '{print $1}')" - git fetch "$REMOTE" +if [ "${SKIP_BUILD_BASE:-no}" = 'yes' ]; 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" diff --git a/src/emqx.appup.src b/src/emqx.appup.src index fa24eedd2..6c25d9f5c 100644 --- a/src/emqx.appup.src +++ b/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,[]}, diff --git a/src/emqx.erl b/src/emqx.erl index 5c90cf953..7fb3c734f 100644 --- a/src/emqx.erl +++ b/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() -> diff --git a/src/emqx_frame.erl b/src/emqx_frame.erl index 48e8b71fb..aa71b16ae 100644 --- a/src/emqx_frame.erl +++ b/src/emqx_frame.erl @@ -265,7 +265,7 @@ parse_packet(#mqtt_packet_header{type = ?CONNACK}, < - {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 %%-------------------------------------------------------------------- diff --git a/src/emqx_hooks.erl b/src/emqx_hooks.erl index 95ef33fdd..bd795ee0a 100644 --- a/src/emqx_hooks.erl +++ b/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). -put(HookPoint, Callback) -> +-spec put(hookpoint(), action() | callback()) -> ok. +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]). - diff --git a/src/emqx_limiter.erl b/src/emqx_limiter.erl index 1cca7140b..9a13e78f4 100644 --- a/src/emqx_limiter.erl +++ b/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. diff --git a/src/emqx_plugins.erl b/src/emqx_plugins.erl index 7c4633e38..3aa0287da 100644 --- a/src/emqx_plugins.erl +++ b/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 diff --git a/test/emqx_frame_SUITE.erl b/test/emqx_frame_SUITE.erl index d98786e99..81c861bdb 100644 --- a/test/emqx_frame_SUITE.erl +++ b/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">>], diff --git a/test/emqx_plugins_SUITE.erl b/test/emqx_plugins_SUITE.erl index 6d8847f43..395318d43 100644 --- a/test/emqx_plugins_SUITE.erl +++ b/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",