Merge pull request #4237 from emqx/fix-merge-conflict-master-to-5.0
Auto-pull-request-on-2021-02-24
This commit is contained in:
commit
52777efc8e
|
@ -47,6 +47,7 @@ jobs:
|
|||
printenv > .env
|
||||
docker exec -i erlang bash -c "make xref"
|
||||
docker exec --env-file .env -i erlang bash -c "make ct"
|
||||
docker exec --env-file .env -i erlang bash -c "make eunit"
|
||||
docker exec -i erlang bash -c "make cover"
|
||||
docker exec -i erlang bash -c "make coveralls"
|
||||
- uses: actions/upload-artifact@v1
|
||||
|
|
|
@ -1,4 +1,5 @@
|
|||
.eunit
|
||||
test-data/
|
||||
deps
|
||||
!deps/.placeholder
|
||||
*.o
|
||||
|
|
7
Makefile
7
Makefile
|
@ -2,6 +2,7 @@ REBAR_VERSION = 3.14.3-emqx-4
|
|||
DASHBOARD_VERSION = v4.3.0
|
||||
REBAR = $(CURDIR)/rebar3
|
||||
BUILD = $(CURDIR)/build
|
||||
SCRIPTS = $(CURDIR)/scripts
|
||||
export EMQX_ENTERPRISE=false
|
||||
export PKG_VSN ?= $(shell $(CURDIR)/pkg-vsn.sh)
|
||||
|
||||
|
@ -20,17 +21,17 @@ all: $(REBAR) $(PROFILES)
|
|||
|
||||
.PHONY: ensure-rebar3
|
||||
ensure-rebar3:
|
||||
$(CURDIR)/ensure-rebar3.sh $(REBAR_VERSION)
|
||||
$(SCRIPTS)/ensure-rebar3.sh $(REBAR_VERSION)
|
||||
|
||||
$(REBAR): ensure-rebar3
|
||||
|
||||
.PHONY: get-dashboard
|
||||
get-dashboard:
|
||||
$(CURDIR)/get-dashboard.sh $(DASHBOARD_VERSION)
|
||||
$(SCRIPTS)/get-dashboard.sh $(DASHBOARD_VERSION)
|
||||
|
||||
.PHONY: eunit
|
||||
eunit: $(REBAR)
|
||||
$(REBAR) eunit
|
||||
$(REBAR) eunit -v -c
|
||||
|
||||
.PHONY: proper
|
||||
proper: $(REBAR)
|
||||
|
|
|
@ -109,13 +109,13 @@ load_hooks() ->
|
|||
{ok, _} = ehttpc_sup:start_pool(PoolName, PoolOpts),
|
||||
case application:get_env(?APP, super_req) of
|
||||
undefined ->
|
||||
emqx:hook('client.authenticate', {emqx_auth_http, check, [#{auth => maps:from_list(AuthReq),
|
||||
emqx_hooks:put('client.authenticate', {emqx_auth_http, check, [#{auth => maps:from_list(AuthReq),
|
||||
super => undefined}]});
|
||||
{ok, SuperReq} ->
|
||||
PoolOpts1 = proplists:get_value(pool_opts, SuperReq),
|
||||
PoolName1 = proplists:get_value(pool_name, SuperReq),
|
||||
{ok, _} = ehttpc_sup:start_pool(PoolName1, PoolOpts1),
|
||||
emqx:hook('client.authenticate', {emqx_auth_http, check, [#{auth => maps:from_list(AuthReq),
|
||||
emqx_hooks:put('client.authenticate', {emqx_auth_http, check, [#{auth => maps:from_list(AuthReq),
|
||||
super => maps:from_list(SuperReq)}]})
|
||||
end
|
||||
end,
|
||||
|
@ -126,7 +126,7 @@ load_hooks() ->
|
|||
PoolOpts2 = proplists:get_value(pool_opts, ACLReq),
|
||||
PoolName2 = proplists:get_value(pool_name, ACLReq),
|
||||
{ok, _} = ehttpc_sup:start_pool(PoolName2, PoolOpts2),
|
||||
emqx:hook('client.check_acl', {emqx_acl_http, check_acl, [#{acl => maps:from_list(ACLReq)}]})
|
||||
emqx_hooks:put('client.check_acl', {emqx_acl_http, check_acl, [#{acl => maps:from_list(ACLReq)}]})
|
||||
end,
|
||||
ok.
|
||||
|
||||
|
|
|
@ -0,0 +1,10 @@
|
|||
%% -*-: erlang -*-
|
||||
|
||||
{VSN,
|
||||
[
|
||||
{<<".*">>, []}
|
||||
],
|
||||
[
|
||||
{<<".*">>, []}
|
||||
]
|
||||
}.
|
|
@ -195,4 +195,3 @@ feedvar(max_inflight, 0, _) ->
|
|||
|
||||
feedvar(max_inflight, Size, _) ->
|
||||
Size.
|
||||
|
||||
|
|
|
@ -185,18 +185,16 @@
|
|||
},
|
||||
ssl => #{
|
||||
order => 14,
|
||||
type => string,
|
||||
required => false,
|
||||
default => <<"off">>,
|
||||
enum => [<<"on">>, <<"off">>],
|
||||
title => #{en => <<"Bridge SSL">>,
|
||||
zh => <<"Bridge SSL"/utf8>>},
|
||||
description => #{en => <<"Switch which used to enable ssl connection of the bridge">>,
|
||||
zh => <<"是否启用 Bridge SSL 连接"/utf8>>}
|
||||
type => boolean,
|
||||
default => false,
|
||||
title => #{en => <<"Enable SSL">>,
|
||||
zh => <<"开启SSL链接"/utf8>>},
|
||||
description => #{en => <<"Enable SSL or not">>,
|
||||
zh => <<"是否开启 SSL"/utf8>>}
|
||||
},
|
||||
cacertfile => #{
|
||||
order => 15,
|
||||
type => string,
|
||||
type => file,
|
||||
required => false,
|
||||
default => <<"etc/certs/cacert.pem">>,
|
||||
title => #{en => <<"CA certificates">>,
|
||||
|
@ -206,7 +204,7 @@
|
|||
},
|
||||
certfile => #{
|
||||
order => 16,
|
||||
type => string,
|
||||
type => file,
|
||||
required => false,
|
||||
default => <<"etc/certs/client-cert.pem">>,
|
||||
title => #{en => <<"SSL Certfile">>,
|
||||
|
@ -216,7 +214,7 @@
|
|||
},
|
||||
keyfile => #{
|
||||
order => 17,
|
||||
type => string,
|
||||
type => file,
|
||||
required => false,
|
||||
default => <<"etc/certs/client-key.pem">>,
|
||||
title => #{en => <<"SSL Keyfile">>,
|
||||
|
@ -246,7 +244,6 @@
|
|||
}
|
||||
}).
|
||||
|
||||
|
||||
-define(RESOURCE_CONFIG_SPEC_MQTT_SUB, #{
|
||||
address => #{
|
||||
order => 1,
|
||||
|
@ -424,7 +421,6 @@
|
|||
}
|
||||
}).
|
||||
|
||||
|
||||
-define(RESOURCE_CONFIG_SPEC_RPC, #{
|
||||
address => #{
|
||||
order => 1,
|
||||
|
@ -573,7 +569,7 @@ on_resource_create(ResId, Params) ->
|
|||
?LOG(info, "Initiating Resource ~p, ResId: ~p", [?RESOURCE_TYPE_MQTT, ResId]),
|
||||
{ok, _} = application:ensure_all_started(ecpool),
|
||||
PoolName = pool_name(ResId),
|
||||
Options = options(Params, PoolName),
|
||||
Options = options(Params, PoolName, ResId),
|
||||
start_resource(ResId, PoolName, Options),
|
||||
case test_resource_status(PoolName) of
|
||||
true -> ok;
|
||||
|
@ -719,7 +715,7 @@ name(Pool, Id) ->
|
|||
pool_name(ResId) ->
|
||||
list_to_atom("bridge_mqtt:" ++ str(ResId)).
|
||||
|
||||
options(Options, PoolName) ->
|
||||
options(Options, PoolName, ResId) ->
|
||||
GetD = fun(Key, Default) -> maps:get(Key, Options, Default) end,
|
||||
Get = fun(Key) -> GetD(Key, undefined) end,
|
||||
Address = Get(<<"address">>),
|
||||
|
@ -743,8 +739,6 @@ options(Options, PoolName) ->
|
|||
Topic ->
|
||||
[{subscriptions, [{Topic, Get(<<"qos">>)}]} | Subscriptions]
|
||||
end,
|
||||
%% TODO check why only ciphers are configurable but not versions
|
||||
TlsVersions = emqx_tls_lib:default_versions(),
|
||||
[{address, binary_to_list(Address)},
|
||||
{bridge_mode, GetD(<<"bridge_mode">>, true)},
|
||||
{clean_start, true},
|
||||
|
@ -755,17 +749,16 @@ options(Options, PoolName) ->
|
|||
{username, str(Get(<<"username">>))},
|
||||
{password, str(Get(<<"password">>))},
|
||||
{proto_ver, mqtt_ver(Get(<<"proto_ver">>))},
|
||||
{retry_interval, cuttlefish_duration:parse(str(GetD(<<"retry_interval">>, "30s")), s)},
|
||||
{ssl, cuttlefish_flag:parse(str(Get(<<"ssl">>)))},
|
||||
{ssl_opts, [ {keyfile, str(Get(<<"keyfile">>))}
|
||||
, {certfile, str(Get(<<"certfile">>))}
|
||||
, {cacertfile, str(Get(<<"cacertfile">>))}
|
||||
, {versions, TlsVersions}
|
||||
, {ciphers, emqx_tls_lib:integral_ciphers(TlsVersions, Get(<<"ciphers">>))}
|
||||
]}
|
||||
{retry_interval, cuttlefish_duration:parse(str(GetD(<<"retry_interval">>, "30s")), s)}
|
||||
| maybe_ssl(Options, cuttlefish_flag:parse(str(Get(<<"ssl">>))), ResId)
|
||||
] ++ Subscriptions1
|
||||
end.
|
||||
|
||||
maybe_ssl(_Options, false, _ResId) ->
|
||||
[];
|
||||
maybe_ssl(Options, true, ResId) ->
|
||||
Dir = filename:join([emqx:get_env(data_dir), "rule", ResId]),
|
||||
[{ssl, true}, {ssl_opts, emqx_plugin_libs_ssl:save_files_return_opts(Options, Dir)}].
|
||||
|
||||
mqtt_ver(ProtoVer) ->
|
||||
case ProtoVer of
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
{application, emqx_exhook,
|
||||
[{description, "EMQ X Extension for Hook"},
|
||||
{vsn, "git"},
|
||||
{vsn, "4.3.0"},
|
||||
{modules, []},
|
||||
{registered, []},
|
||||
{mod, {emqx_exhook_app, []}},
|
||||
|
|
|
@ -565,15 +565,14 @@ enrich_clientinfo(InClientInfo = #{proto_name := ProtoName}, ClientInfo) ->
|
|||
NClientInfo#{protocol => ProtoName}.
|
||||
|
||||
default_conninfo(ConnInfo) ->
|
||||
ConnInfo#{proto_name => undefined,
|
||||
proto_ver => undefined,
|
||||
clean_start => true,
|
||||
ConnInfo#{clean_start => true,
|
||||
clientid => undefined,
|
||||
username => undefined,
|
||||
conn_props => [],
|
||||
conn_mod => undefined,
|
||||
conn_props => #{},
|
||||
connected => true,
|
||||
connected_at => erlang:system_time(millisecond),
|
||||
keepalive => undefined,
|
||||
keepalive => 0,
|
||||
receive_maximum => 0,
|
||||
expiry_interval => 0}.
|
||||
|
||||
|
|
|
@ -264,7 +264,7 @@ init_state(WrappedSock, Peername, Options) ->
|
|||
run_loop(Parent, State = #state{socket = Socket,
|
||||
peername = Peername}) ->
|
||||
emqx_logger:set_metadata_peername(esockd:format(Peername)),
|
||||
emqx_misc:tune_heap_size(?DEFAULT_OOM_POLICY),
|
||||
_ = emqx_misc:tune_heap_size(?DEFAULT_OOM_POLICY),
|
||||
case activate_socket(State) of
|
||||
{ok, NState} ->
|
||||
hibernate(Parent, NState);
|
||||
|
@ -273,6 +273,7 @@ run_loop(Parent, State = #state{socket = Socket,
|
|||
exit_on_sock_error(Reason)
|
||||
end.
|
||||
|
||||
-spec exit_on_sock_error(atom()) -> no_return().
|
||||
exit_on_sock_error(Reason) when Reason =:= einval;
|
||||
Reason =:= enotconn;
|
||||
Reason =:= closed ->
|
||||
|
@ -449,6 +450,7 @@ handle_msg(Msg, State) ->
|
|||
%%--------------------------------------------------------------------
|
||||
%% Terminate
|
||||
|
||||
-spec terminate(atom(), state()) -> no_return().
|
||||
terminate(Reason, State = #state{channel = Channel}) ->
|
||||
?LOG(debug, "Terminated due to ~p", [Reason]),
|
||||
_ = emqx_exproto_channel:terminate(Reason, Channel),
|
||||
|
|
|
@ -44,11 +44,11 @@ lwm2m.topics.update = "up/resp"
|
|||
# When publish the update message.
|
||||
#
|
||||
# Can be one of:
|
||||
# - object_list_changed: only if the object list is changed
|
||||
# - contains_object_list: only if the update message contains object list
|
||||
# - always: always publish the update message
|
||||
#
|
||||
# Defaults to object_list_changed
|
||||
#lwm2m.publish_update_when = object_list_changed
|
||||
# Defaults to contains_object_list
|
||||
#lwm2m.update_msg_publish_condition = contains_object_list
|
||||
|
||||
# Dir where the object definition files can be found
|
||||
lwm2m.xml_dir = "{{ platform_etc_dir }}/lwm2m_xml"
|
||||
|
|
|
@ -112,9 +112,9 @@ end}.
|
|||
{default, "lwm2m/%e/up/resp"}
|
||||
]}.
|
||||
|
||||
{mapping, "lwm2m.publish_update_when", "emqx_lwm2m.publish_update_when", [
|
||||
{datatype, {enum, [object_list_changed, always]}},
|
||||
{default, object_list_changed}
|
||||
{mapping, "lwm2m.update_msg_publish_condition", "emqx_lwm2m.update_msg_publish_condition", [
|
||||
{datatype, {enum, [contains_object_list, always]}},
|
||||
{default, contains_object_list}
|
||||
]}.
|
||||
|
||||
{translation, "emqx_lwm2m.topics", fun(Conf) ->
|
||||
|
|
|
@ -101,12 +101,12 @@ get_lwm2m_opts(Envs) ->
|
|||
AutoObserve = proplists:get_value(auto_observe, Envs, []),
|
||||
QmodeTimeWindow = proplists:get_value(qmode_time_window, Envs, []),
|
||||
Topics = proplists:get_value(topics, Envs, []),
|
||||
PublishUpdateWhen = proplists:get_value(publish_update_when, Envs, object_list_changed),
|
||||
PublishCondition = proplists:get_value(update_msg_publish_condition, Envs, contains_object_list),
|
||||
[{lifetime_max, LifetimeMax},
|
||||
{lifetime_min, LifetimeMin},
|
||||
{mountpoint, list_to_binary(Mountpoint)},
|
||||
{port, Sockport},
|
||||
{auto_observe, AutoObserve},
|
||||
{qmode_time_window, QmodeTimeWindow},
|
||||
{publish_update_when, PublishUpdateWhen},
|
||||
{update_msg_publish_condition, PublishCondition},
|
||||
{topics, Topics}].
|
||||
|
|
|
@ -121,11 +121,11 @@ update_reg_info(NewRegInfo, Lwm2mState = #lwm2m_state{
|
|||
|
||||
UpdatedRegInfo = maps:merge(RegInfo, NewRegInfo),
|
||||
|
||||
case proplists:get_value(publish_update_when,
|
||||
lwm2m_coap_responder:options(), object_list_changed) of
|
||||
case proplists:get_value(update_msg_publish_condition,
|
||||
lwm2m_coap_responder:options(), contains_object_list) of
|
||||
always ->
|
||||
send_to_broker(<<"update">>, #{<<"data">> => UpdatedRegInfo}, Lwm2mState);
|
||||
object_list_changed ->
|
||||
contains_object_list ->
|
||||
%% - report the registration info update, but only when objectList is updated.
|
||||
case NewRegInfo of
|
||||
#{<<"objectList">> := _} ->
|
||||
|
|
|
@ -0,0 +1,7 @@
|
|||
{application, emqx_plugin_libs,
|
||||
[{description, "EMQ X Plugin utility libs"},
|
||||
{vsn, "4.3.0"},
|
||||
{modules, []},
|
||||
{applications, [kernel,stdlib]},
|
||||
{env, []}
|
||||
]}.
|
|
@ -0,0 +1,18 @@
|
|||
%%--------------------------------------------------------------------
|
||||
%% Copyright (c) 2021 EMQ Technologies Co., Ltd. All Rights Reserved.
|
||||
%%
|
||||
%% Licensed under the Apache License, Version 2.0 (the "License");
|
||||
%% you may not use this file except in compliance with the License.
|
||||
%% You may obtain a copy of the License at
|
||||
%%
|
||||
%% http://www.apache.org/licenses/LICENSE-2.0
|
||||
%%
|
||||
%% Unless required by applicable law or agreed to in writing, software
|
||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
%% See the License for the specific language governing permissions and
|
||||
%% limitations under the License.
|
||||
%%--------------------------------------------------------------------
|
||||
|
||||
-module(emqx_plugin_libs).
|
||||
|
|
@ -0,0 +1,89 @@
|
|||
%%--------------------------------------------------------------------
|
||||
%% Copyright (c) 2021 EMQ Technologies Co., Ltd. All Rights Reserved.
|
||||
%%
|
||||
%% Licensed under the Apache License, Version 2.0 (the "License");
|
||||
%% you may not use this file except in compliance with the License.
|
||||
%% You may obtain a copy of the License at
|
||||
%%
|
||||
%% http://www.apache.org/licenses/LICENSE-2.0
|
||||
%%
|
||||
%% Unless required by applicable law or agreed to in writing, software
|
||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
%% See the License for the specific language governing permissions and
|
||||
%% limitations under the License.
|
||||
%%--------------------------------------------------------------------
|
||||
|
||||
-module(emqx_plugin_libs_ssl).
|
||||
|
||||
-export([save_files_return_opts/2]).
|
||||
|
||||
-type file_input_key() :: binary(). %% <<"file">> | <<"filename">>
|
||||
-type file_input() :: #{file_input_key() => binary()}.
|
||||
|
||||
%% options are below paris
|
||||
%% <<"keyfile">> => file_input()
|
||||
%% <<"certfile">> => file_input()
|
||||
%% <<"cafile">> => file_input() %% backward compatible
|
||||
%% <<"cacertfile">> => file_input()
|
||||
%% <<"verify">> => boolean()
|
||||
%% <<"tls_versions">> => binary()
|
||||
%% <<"ciphers">> => binary()
|
||||
-type opts_key() :: binary().
|
||||
-type opts_input() :: #{opts_key() => file_input() | boolean() | binary()}.
|
||||
|
||||
-type opt_key() :: keyfile | certfile | cacertfile | verify | versions | ciphers.
|
||||
-type opt_value() :: term().
|
||||
-type opts() :: [{opt_key(), opt_value()}].
|
||||
|
||||
%% @doc Parse ssl options input.
|
||||
%% If the input contains file content, save the files in the given dir.
|
||||
%% Returns ssl options for Erlang's ssl application.
|
||||
-spec save_files_return_opts(opts_input(), file:name_all()) -> opts().
|
||||
save_files_return_opts(Options, Dir) ->
|
||||
GetD = fun(Key, Default) -> maps:get(Key, Options, Default) end,
|
||||
Get = fun(Key) -> GetD(Key, undefined) end,
|
||||
KeyFile = Get(<<"keyfile">>),
|
||||
CertFile = Get(<<"certfile">>),
|
||||
CAFile = GetD(<<"cacertfile">>, Get(<<"cafile">>)),
|
||||
Key = save_file(KeyFile, Dir),
|
||||
Cert = save_file(CertFile, Dir),
|
||||
CA = save_file(CAFile, Dir),
|
||||
Verify = case GetD(<<"verify">>, false) of
|
||||
false -> verify_none;
|
||||
_ -> verify_peer
|
||||
end,
|
||||
Versions = emqx_tls_lib:integral_versions(Get(<<"tls_versions">>)),
|
||||
Ciphers = emqx_tls_lib:integral_ciphers(Versions, Get(<<"ciphers">>)),
|
||||
filter([{keyfile, Key}, {certfile, Cert}, {cacertfile, CA},
|
||||
{verify, Verify}, {versions, Versions}, {ciphers, Ciphers}]).
|
||||
|
||||
filter([]) -> [];
|
||||
filter([{_, ""} | T]) -> filter(T);
|
||||
filter([H | T]) -> [H | filter(T)].
|
||||
|
||||
save_file(#{<<"filename">> := FileName, <<"file">> := Content}, Dir)
|
||||
when FileName =/= undefined andalso Content =/= undefined ->
|
||||
save_file(ensure_str(FileName), iolist_to_binary(Content), Dir);
|
||||
save_file(FilePath, _) when is_binary(FilePath) ->
|
||||
ensure_str(FilePath);
|
||||
save_file(FilePath, _) when is_list(FilePath) ->
|
||||
FilePath;
|
||||
save_file(_, _) -> "".
|
||||
|
||||
save_file("", _, _Dir) -> ""; %% ignore
|
||||
save_file(_, <<>>, _Dir) -> ""; %% ignore
|
||||
save_file(FileName, Content, Dir) ->
|
||||
FullFilename = filename:join([Dir, FileName]),
|
||||
ok = filelib:ensure_dir(FullFilename),
|
||||
case file:write_file(FullFilename, Content) of
|
||||
ok ->
|
||||
ensure_str(FullFilename);
|
||||
{error, Reason} ->
|
||||
logger:error("failed_to_save_ssl_file ~s: ~0p", [FullFilename, Reason]),
|
||||
error({"failed_to_save_ssl_file", FullFilename, Reason})
|
||||
end.
|
||||
|
||||
ensure_str(L) when is_list(L) -> L;
|
||||
ensure_str(B) when is_binary(B) -> unicode:characters_to_list(B, utf8).
|
||||
|
|
@ -0,0 +1,78 @@
|
|||
%%--------------------------------------------------------------------
|
||||
%% Copyright (c) 2021 EMQ Technologies Co., Ltd. All Rights Reserved.
|
||||
%%
|
||||
%% Licensed under the Apache License, Version 2.0 (the "License");
|
||||
%% you may not use this file except in compliance with the License.
|
||||
%% You may obtain a copy of the License at
|
||||
%%
|
||||
%% http://www.apache.org/licenses/LICENSE-2.0
|
||||
%%
|
||||
%% Unless required by applicable law or agreed to in writing, software
|
||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
%% See the License for the specific language governing permissions and
|
||||
%% limitations under the License.
|
||||
%%--------------------------------------------------------------------
|
||||
|
||||
-module(emqx_plugin_libs_ssl_tests).
|
||||
|
||||
-include_lib("proper/include/proper.hrl").
|
||||
-include_lib("eunit/include/eunit.hrl").
|
||||
|
||||
no_crash_test_() ->
|
||||
Opts = [{numtests, 1000}, {to_file, user}],
|
||||
{timeout, 60,
|
||||
fun() -> ?assert(proper:quickcheck(prop_run(), Opts)) end}.
|
||||
|
||||
prop_run() ->
|
||||
?FORALL(Generated, prop_opts_input(), test_opts_input(Generated)).
|
||||
|
||||
%% proper type to generate input value.
|
||||
prop_opts_input() ->
|
||||
[{keyfile, prop_file_or_content()},
|
||||
{certfile, prop_file_or_content()},
|
||||
{cacertfile, prop_file_or_content()},
|
||||
{verify, proper_types:boolean()},
|
||||
{versions, prop_tls_versions()},
|
||||
{ciphers, prop_tls_ciphers()},
|
||||
{other, proper_types:binary()}].
|
||||
|
||||
prop_file_or_content() ->
|
||||
proper_types:oneof([prop_cert_file_name(),
|
||||
{prop_cert_file_name(), proper_types:binary()}]).
|
||||
|
||||
prop_cert_file_name() ->
|
||||
proper_types:oneof(["certname1", <<"certname2">>, "", <<>>, undefined]).
|
||||
|
||||
prop_tls_versions() ->
|
||||
proper_types:oneof(["tlsv1.3",
|
||||
<<"tlsv1.3,tlsv1.2">>,
|
||||
"tlsv1.2 , tlsv1.1",
|
||||
"1.2",
|
||||
"v1.3",
|
||||
"",
|
||||
<<>>,
|
||||
undefined]).
|
||||
|
||||
prop_tls_ciphers() ->
|
||||
proper_types:oneof(["TLS_AES_256_GCM_SHA384,TLS_AES_128_GCM_SHA256",
|
||||
<<>>,
|
||||
"",
|
||||
undefined]).
|
||||
|
||||
test_opts_input(Inputs) ->
|
||||
KF = fun(K) -> {_, V} = lists:keyfind(K, 1, Inputs), V end,
|
||||
Generated = #{<<"keyfile">> => file_or_content(KF(keyfile)),
|
||||
<<"certfile">> => file_or_content(KF(certfile)),
|
||||
<<"cafile">> => file_or_content(KF(cacertfile)),
|
||||
<<"verify">> => file_or_content(KF(verify)),
|
||||
<<"tls_versions">> => KF(versions),
|
||||
<<"ciphers">> => KF(ciphers),
|
||||
<<"other">> => KF(other)},
|
||||
_ = emqx_plugin_libs_ssl:save_files_return_opts(Generated, "test-data"),
|
||||
true.
|
||||
|
||||
file_or_content({Name, Content}) ->
|
||||
#{<<"file">> => Content, <<"filename">> => Name};
|
||||
file_or_content(Name) ->
|
||||
Name.
|
|
@ -269,16 +269,10 @@ do_update_resource_check(Id, NewParams) ->
|
|||
config = OldConfig,
|
||||
description = OldDescription} = _OldResource} ->
|
||||
try
|
||||
do_update_resource(#{id => Id,
|
||||
config => case maps:find(<<"config">>, NewParams) of
|
||||
{ok, NewConfig} -> NewConfig;
|
||||
error -> OldConfig
|
||||
end,
|
||||
type => Type,
|
||||
description => case maps:find(<<"description">>, NewParams) of
|
||||
{ok, NewDescription} -> NewDescription;
|
||||
error -> OldDescription
|
||||
end}),
|
||||
Conifg = maps:get(<<"config">>, NewParams, OldConfig),
|
||||
Descr = maps:get(<<"description">>, NewParams, OldDescription),
|
||||
do_update_resource(#{id => Id, config => Conifg, type => Type,
|
||||
description => Descr}),
|
||||
ok
|
||||
catch _ : Reason ->
|
||||
{error, Reason}
|
||||
|
@ -294,11 +288,13 @@ do_update_resource(#{id := Id, type := Type, description := NewDescription, conf
|
|||
Config = emqx_rule_validator:validate_params(NewConfig, ParamSpec),
|
||||
case test_resource(#{type => Type, config => NewConfig}) of
|
||||
ok ->
|
||||
Resource = #resource{id = Id,
|
||||
Resource = #resource{
|
||||
id = Id,
|
||||
type = Type,
|
||||
config = Config,
|
||||
description = NewDescription,
|
||||
created_at = erlang:system_time(millisecond)},
|
||||
created_at = erlang:system_time(millisecond)
|
||||
},
|
||||
cluster_call(init_resource, [Module, Create, Id, Config]),
|
||||
emqx_rule_registry:add_resource(Resource);
|
||||
{error, Reason} ->
|
||||
|
@ -468,18 +464,19 @@ may_update_rule_params(Rule, Params = #{rawsql := SQL}) ->
|
|||
maps:remove(rawsql, Params));
|
||||
Reason -> throw(Reason)
|
||||
end;
|
||||
may_update_rule_params(Rule = #rule{enabled = OldE, actions = Actions},
|
||||
Params = #{enabled := ToE}) ->
|
||||
case {OldE, ToE} of
|
||||
may_update_rule_params(Rule = #rule{enabled = OldEnb, actions = Actions},
|
||||
Params = #{enabled := NewEnb}) ->
|
||||
case {OldEnb, NewEnb} of
|
||||
{false, true} -> refresh_rule(Rule);
|
||||
{true, false} -> clear_actions(Actions);
|
||||
_ -> ok
|
||||
end,
|
||||
may_update_rule_params(Rule#rule{enabled = ToE}, maps:remove(enabled, Params));
|
||||
may_update_rule_params(Rule#rule{enabled = NewEnb}, maps:remove(enabled, Params));
|
||||
may_update_rule_params(Rule, Params = #{description := Descr}) ->
|
||||
may_update_rule_params(Rule#rule{description = Descr}, maps:remove(description, Params));
|
||||
may_update_rule_params(Rule, Params = #{on_action_failed := OnFailed}) ->
|
||||
may_update_rule_params(Rule#rule{on_action_failed = OnFailed}, maps:remove(on_action_failed, Params));
|
||||
may_update_rule_params(Rule#rule{on_action_failed = OnFailed},
|
||||
maps:remove(on_action_failed, Params));
|
||||
may_update_rule_params(Rule = #rule{actions = OldActions}, Params = #{actions := Actions}) ->
|
||||
%% prepare new actions before removing old ones
|
||||
NewActions = prepare_actions(Actions, maps:get(enabled, Params, true)),
|
||||
|
|
|
@ -11,7 +11,7 @@ Please see: [EMQ X - WebHook](https://docs.emqx.io/broker/latest/en/advanced/web
|
|||
## The web services URL for Hook request
|
||||
##
|
||||
## Value: String
|
||||
web.hook.api.url = http://127.0.0.1:8080
|
||||
web.hook.url = http://127.0.0.1:8080
|
||||
|
||||
## Encode message payload field
|
||||
##
|
||||
|
|
|
@ -15,14 +15,17 @@
|
|||
]}.
|
||||
|
||||
{mapping, "web.hook.ssl.cacertfile", "emqx_web_hook.cacertfile", [
|
||||
{default, ""},
|
||||
{datatype, string}
|
||||
]}.
|
||||
|
||||
{mapping, "web.hook.ssl.certfile", "emqx_web_hook.certfile", [
|
||||
{default, ""},
|
||||
{datatype, string}
|
||||
]}.
|
||||
|
||||
{mapping, "web.hook.ssl.keyfile", "emqx_web_hook.keyfile", [
|
||||
{default, ""},
|
||||
{datatype, string}
|
||||
]}.
|
||||
|
||||
|
|
|
@ -18,13 +18,3 @@
|
|||
{cover_enabled, true}.
|
||||
{cover_opts, [verbose]}.
|
||||
{cover_export_enabled, true}.
|
||||
|
||||
{profiles,
|
||||
[{test,
|
||||
[{erl_opts, [export_all, nowarn_export_all]},
|
||||
{deps,
|
||||
[
|
||||
{emqtt, {git, "https://github.com/emqx/emqtt", {tag, "1.2.3"}}}
|
||||
]}
|
||||
]}
|
||||
]}.
|
||||
|
|
|
@ -0,0 +1,10 @@
|
|||
%% -*-: erlang -*-
|
||||
|
||||
{VSN,
|
||||
[
|
||||
{<<".*">>, []}
|
||||
],
|
||||
[
|
||||
{<<".*">>, []}
|
||||
]
|
||||
}.
|
|
@ -17,85 +17,85 @@
|
|||
%% Define the default actions.
|
||||
-module(emqx_web_hook_actions).
|
||||
|
||||
-export([ on_resource_create/2
|
||||
, on_get_resource_status/2
|
||||
, on_resource_destroy/2
|
||||
]).
|
||||
|
||||
-export([ on_action_create_data_to_webserver/2
|
||||
, on_action_data_to_webserver/2
|
||||
]).
|
||||
|
||||
-export_type([action_fun/0]).
|
||||
|
||||
-include_lib("emqx/include/emqx.hrl").
|
||||
-include_lib("emqx/include/logger.hrl").
|
||||
-include_lib("emqx_rule_engine/include/rule_actions.hrl").
|
||||
-include("emqx_web_hook.hrl").
|
||||
|
||||
-type(action_fun() :: fun((Data :: map(), Envs :: map()) -> Result :: any())).
|
||||
|
||||
-type(url() :: binary()).
|
||||
|
||||
-define(RESOURCE_TYPE_WEBHOOK, 'web_hook').
|
||||
-define(RESOURCE_CONFIG_SPEC, #{
|
||||
url => #{
|
||||
order => 1,
|
||||
url => #{order => 1,
|
||||
type => string,
|
||||
format => url,
|
||||
required => true,
|
||||
title => #{en => <<"URL">>,
|
||||
zh => <<"URL"/utf8>>},
|
||||
title => #{en => <<"Request URL">>,
|
||||
zh => <<"请求 URL"/utf8>>},
|
||||
description => #{en => <<"The URL of the server that will receive the Webhook requests.">>,
|
||||
zh => <<"用于接收 Webhook 请求的服务器的 URL。"/utf8>>}
|
||||
},
|
||||
connect_timeout => #{
|
||||
order => 2,
|
||||
type => number,
|
||||
default => 5,
|
||||
zh => <<"用于接收 Webhook 请求的服务器的 URL。"/utf8>>}},
|
||||
connect_timeout => #{order => 2,
|
||||
type => string,
|
||||
default => <<"5s">>,
|
||||
title => #{en => <<"Connect Timeout">>,
|
||||
zh => <<"连接超时时间"/utf8>>},
|
||||
description => #{en => <<"Connect timeout in seconds">>,
|
||||
zh => <<"连接超时时间,单位秒"/utf8>>}},
|
||||
request_timeout => #{
|
||||
order => 3,
|
||||
type => number,
|
||||
default => 5,
|
||||
description => #{en => <<"Connect Timeout In Seconds">>,
|
||||
zh => <<"连接超时时间"/utf8>>}},
|
||||
request_timeout => #{order => 3,
|
||||
type => string,
|
||||
default => <<"5s">>,
|
||||
title => #{en => <<"Request Timeout">>,
|
||||
zh => <<"请求超时时间时间"/utf8>>},
|
||||
description => #{en => <<"Request timeout in seconds">>,
|
||||
zh => <<"请求超时时间,单位秒"/utf8>>}},
|
||||
cacertfile => #{
|
||||
order => 4,
|
||||
description => #{en => <<"Request Timeout In Seconds">>,
|
||||
zh => <<"请求超时时间"/utf8>>}},
|
||||
pool_size => #{order => 4,
|
||||
type => number,
|
||||
default => 8,
|
||||
title => #{en => <<"Pool Size">>, zh => <<"连接池大小"/utf8>>},
|
||||
description => #{en => <<"Connection Pool">>,
|
||||
zh => <<"连接池大小"/utf8>>}
|
||||
},
|
||||
cacertfile => #{order => 5,
|
||||
type => file,
|
||||
default => <<>>,
|
||||
default => <<"">>,
|
||||
title => #{en => <<"CA Certificate File">>,
|
||||
zh => <<"CA 证书文件"/utf8>>},
|
||||
description => #{en => <<"CA certificate file.">>,
|
||||
zh => <<"CA 证书文件。"/utf8>>}
|
||||
},
|
||||
certfile => #{
|
||||
order => 5,
|
||||
description => #{en => <<"CA Certificate file">>,
|
||||
zh => <<"CA 证书文件"/utf8>>}},
|
||||
keyfile => #{order => 6,
|
||||
type => file,
|
||||
default => <<>>,
|
||||
title => #{en => <<"Certificate File">>,
|
||||
zh => <<"证书文件"/utf8>>},
|
||||
description => #{en => <<"Certificate file.">>,
|
||||
zh => <<"证书文件。"/utf8>>}
|
||||
},
|
||||
keyfile => #{
|
||||
order => 6,
|
||||
default => <<"">>,
|
||||
title =>#{en => <<"SSL Key">>,
|
||||
zh => <<"SSL Key"/utf8>>},
|
||||
description => #{en => <<"Your ssl keyfile">>,
|
||||
zh => <<"SSL 私钥"/utf8>>}},
|
||||
certfile => #{order => 7,
|
||||
type => file,
|
||||
default => <<>>,
|
||||
title => #{en => <<"Private Key File">>,
|
||||
zh => <<"私钥文件"/utf8>>},
|
||||
description => #{en => <<"Private key file.">>,
|
||||
zh => <<"私钥文件。"/utf8>>}
|
||||
},
|
||||
verify => #{
|
||||
order => 7,
|
||||
default => <<"">>,
|
||||
title =>#{en => <<"SSL Cert">>,
|
||||
zh => <<"SSL Cert"/utf8>>},
|
||||
description => #{en => <<"Your ssl certfile">>,
|
||||
zh => <<"SSL 证书"/utf8>>}},
|
||||
verify => #{order => 8,
|
||||
type => boolean,
|
||||
default => false,
|
||||
title => #{en => <<"Verify">>,
|
||||
zh => <<"Verify"/utf8>>},
|
||||
description => #{en => <<"Turn on peer certificate verification.">>,
|
||||
zh => <<"是否开启对端证书验证。"/utf8>>}
|
||||
},
|
||||
pool_size => #{
|
||||
order => 8,
|
||||
type => number,
|
||||
default => 32,
|
||||
title => #{en => <<"Pool Size">>,
|
||||
zh => <<"连接池大小"/utf8>>},
|
||||
description => #{en => <<"Pool size for HTTP server.">>,
|
||||
zh => <<"HTTP server 连接池大小。"/utf8>>}
|
||||
}
|
||||
}).
|
||||
title =>#{en => <<"Verify Server Certfile">>,
|
||||
zh => <<"校验服务器证书"/utf8>>},
|
||||
description => #{en => <<"Whether to verify the server certificate. By default, the client will not verify the server's certificate. If verification is required, please set it to true.">>,
|
||||
zh => <<"是否校验服务器证书。 默认客户端不会去校验服务器的证书,如果需要校验,请设置成true。"/utf8>>}}
|
||||
}).
|
||||
|
||||
-define(ACTION_PARAM_RESOURCE, #{
|
||||
order => 0,
|
||||
|
@ -103,9 +103,9 @@
|
|||
required => true,
|
||||
title => #{en => <<"Resource ID">>,
|
||||
zh => <<"资源 ID"/utf8>>},
|
||||
description => #{en => <<"Bind a resource to this action.">>,
|
||||
zh => <<"给动作绑定一个资源。"/utf8>>}
|
||||
}).
|
||||
description => #{en => <<"Bind a resource to this action">>,
|
||||
zh => <<"给动作绑定一个资源"/utf8>>}
|
||||
}).
|
||||
|
||||
-define(ACTION_DATA_SPEC, #{
|
||||
'$resource' => ?ACTION_PARAM_RESOURCE,
|
||||
|
@ -140,7 +140,7 @@
|
|||
description => #{en => <<"HTTP headers.">>,
|
||||
zh => <<"HTTP headers。"/utf8>>}},
|
||||
body => #{
|
||||
order => 5,
|
||||
order => 4,
|
||||
type => string,
|
||||
input => textarea,
|
||||
required => false,
|
||||
|
@ -153,7 +153,8 @@
|
|||
"默认 HTTP 请求体的内容为规则输出的所有字段的键和值构成的 JSON 字符串。"/utf8>>}}
|
||||
}).
|
||||
|
||||
-resource_type(#{name => ?RESOURCE_TYPE_WEBHOOK,
|
||||
-resource_type(
|
||||
#{name => ?RESOURCE_TYPE_WEBHOOK,
|
||||
create => on_resource_create,
|
||||
status => on_get_resource_status,
|
||||
destroy => on_resource_destroy,
|
||||
|
@ -162,7 +163,7 @@
|
|||
zh => <<"WebHook"/utf8>>},
|
||||
description => #{en => <<"WebHook">>,
|
||||
zh => <<"WebHook"/utf8>>}
|
||||
}).
|
||||
}).
|
||||
|
||||
-rule_action(#{name => data_to_webserver,
|
||||
category => data_forward,
|
||||
|
@ -174,18 +175,7 @@
|
|||
zh => <<"发送数据到 Web 服务"/utf8>>},
|
||||
description => #{en => <<"Forward Messages to Web Server">>,
|
||||
zh => <<"将数据转发给 Web 服务"/utf8>>}
|
||||
}).
|
||||
|
||||
-type(url() :: binary()).
|
||||
|
||||
-export([ on_resource_create/2
|
||||
, on_get_resource_status/2
|
||||
, on_resource_destroy/2
|
||||
]).
|
||||
|
||||
-export([ on_action_create_data_to_webserver/2
|
||||
, on_action_data_to_webserver/2
|
||||
]).
|
||||
}).
|
||||
|
||||
%%------------------------------------------------------------------------------
|
||||
%% Actions for web hook
|
||||
|
@ -194,7 +184,7 @@
|
|||
-spec(on_resource_create(binary(), map()) -> map()).
|
||||
on_resource_create(ResId, Conf) ->
|
||||
{ok, _} = application:ensure_all_started(ehttpc),
|
||||
Options = pool_opts(Conf),
|
||||
Options = pool_opts(Conf, ResId),
|
||||
PoolName = pool_name(ResId),
|
||||
case test_http_connect(Conf) of
|
||||
true -> ok;
|
||||
|
@ -299,7 +289,7 @@ parse_action_params(Params = #{<<"url">> := URL}) ->
|
|||
path => path(filename:join(CommonPath, maps:get(<<"path">>, Params, <<>>))),
|
||||
headers => NHeaders,
|
||||
body => maps:get(<<"body">>, Params, <<>>),
|
||||
request_timeout => timer:seconds(maps:get(<<"request_timeout">>, Params, 5)),
|
||||
request_timeout => cuttlefish_duration:parse(str(maps:get(<<"request_timeout">>, Params, <<"5s">>))),
|
||||
pool => maps:get(<<"pool">>, Params)}
|
||||
catch _:_ ->
|
||||
throw({invalid_params, Params})
|
||||
|
@ -328,38 +318,33 @@ str(Str) when is_list(Str) -> Str;
|
|||
str(Atom) when is_atom(Atom) -> atom_to_list(Atom);
|
||||
str(Bin) when is_binary(Bin) -> binary_to_list(Bin).
|
||||
|
||||
pool_opts(Params = #{<<"url">> := URL}) ->
|
||||
#{host := Host0,
|
||||
scheme := Scheme} = URIMap = uri_string:parse(binary_to_list(URL)),
|
||||
Port = maps:get(port, URIMap, case Scheme of
|
||||
"https" -> 443;
|
||||
_ -> 80
|
||||
end),
|
||||
PoolSize = maps:get(<<"pool_size">>, Params, 32),
|
||||
ConnectTimeout = timer:seconds(maps:get(<<"connect_timeout">>, Params, 5)),
|
||||
{Inet, Host} = parse_host(Host0),
|
||||
MoreOpts = case Scheme of
|
||||
"http" ->
|
||||
[{transport_opts, [Inet]}];
|
||||
"https" ->
|
||||
KeyFile = maps:get(<<"keyfile">>, Params),
|
||||
CertFile = maps:get(<<"certfile">>, Params),
|
||||
CACertFile = maps:get(<<"cacertfile">>, Params),
|
||||
VerifyType = case maps:get(<<"verify">>, Params) of
|
||||
true -> verify_peer;
|
||||
false -> verify_none
|
||||
add_default_scheme(<<"http://", _/binary>> = URL) ->
|
||||
URL;
|
||||
add_default_scheme(<<"https://", _/binary>> = URL) ->
|
||||
URL;
|
||||
add_default_scheme(URL) ->
|
||||
<<"http://", URL/binary>>.
|
||||
|
||||
pool_opts(Params = #{<<"url">> := URL}, ResId) ->
|
||||
#{host := Host0, scheme := Scheme} = URIMap =
|
||||
uri_string:parse(binary_to_list(add_default_scheme(URL))),
|
||||
DefaultPort = case is_https(Scheme) of
|
||||
true -> 443;
|
||||
false -> 80
|
||||
end,
|
||||
TLSOpts = lists:filter(fun({_K, V}) when V =:= <<>> ->
|
||||
false;
|
||||
(_) ->
|
||||
true
|
||||
end, [{keyfile, KeyFile}, {certfile, CertFile}, {cacertfile, CACertFile}]),
|
||||
NTLSOpts = [ {verify, VerifyType}
|
||||
, {versions, emqx_tls_lib:default_versions()}
|
||||
, {ciphers, emqx_tls_lib:default_ciphers()}
|
||||
| TLSOpts
|
||||
],
|
||||
[{transport, ssl}, {transport_opts, [Inet | NTLSOpts]}]
|
||||
Port = maps:get(port, URIMap, DefaultPort),
|
||||
PoolSize = maps:get(<<"pool_size">>, Params, 32),
|
||||
ConnectTimeout =
|
||||
cuttlefish_duration:parse(str(maps:get(<<"connect_timeout">>, Params, <<"5s">>))),
|
||||
{Inet, Host} = parse_host(Host0),
|
||||
TransportOpts =
|
||||
case is_https(Scheme) of
|
||||
true -> [Inet | get_ssl_opts(Params, ResId)];
|
||||
false -> [Inet]
|
||||
end,
|
||||
Opts = case is_https(Scheme) of
|
||||
true -> [{transport_opts, TransportOpts}, {transport, ssl}];
|
||||
false -> [{transport_opts, TransportOpts}]
|
||||
end,
|
||||
[{host, Host},
|
||||
{port, Port},
|
||||
|
@ -367,11 +352,19 @@ pool_opts(Params = #{<<"url">> := URL}) ->
|
|||
{pool_type, hash},
|
||||
{connect_timeout, ConnectTimeout},
|
||||
{retry, 5},
|
||||
{retry_timeout, 1000}] ++ MoreOpts.
|
||||
{retry_timeout, 1000} | Opts].
|
||||
|
||||
pool_name(ResId) ->
|
||||
list_to_atom("webhook:" ++ str(ResId)).
|
||||
|
||||
is_https(Scheme) when is_list(Scheme) -> is_https(list_to_binary(Scheme));
|
||||
is_https(<<"https", _/binary>>) -> true;
|
||||
is_https(_) -> false.
|
||||
|
||||
get_ssl_opts(Opts, ResId) ->
|
||||
Dir = filename:join([emqx:get_env(data_dir), "rule", ResId]),
|
||||
[{ssl, true}, {ssl_opts, emqx_plugin_libs_ssl:save_files_return_opts(Opts, Dir)}].
|
||||
|
||||
parse_host(Host) ->
|
||||
case inet:parse_address(Host) of
|
||||
{ok, Addr} when size(Addr) =:= 4 -> {inet, Addr};
|
||||
|
|
|
@ -34,8 +34,9 @@
|
|||
|
||||
prop_confs() ->
|
||||
Schema = cuttlefish_schema:files(filelib:wildcard(code:priv_dir(emqx_web_hook) ++ "/*.schema")),
|
||||
?ALL(Confs, confs(),
|
||||
?ALL({Url, Confs0}, {url(), confs()},
|
||||
begin
|
||||
Confs = [{"web.hook.url", Url}|Confs0],
|
||||
Envs = cuttlefish_generator:map(Schema, cuttlefish_conf_file(Confs)),
|
||||
|
||||
assert_confs(Confs, Envs),
|
||||
|
@ -65,7 +66,7 @@ set_special_cfgs(_) ->
|
|||
application:set_env(emqx, modules_loaded_file, undefined),
|
||||
ok.
|
||||
|
||||
assert_confs([{"web.hook.api.url", Url}|More], Envs) ->
|
||||
assert_confs([{"web.hook.url", Url}|More], Envs) ->
|
||||
%% Assert!
|
||||
Url = deep_get_env("emqx_web_hook.url", Envs),
|
||||
assert_confs(More, Envs);
|
||||
|
@ -112,8 +113,7 @@ cuttlefish_conf_option(K, V)
|
|||
%%--------------------------------------------------------------------
|
||||
|
||||
confs() ->
|
||||
nof([{"web.hook.api.url", url()},
|
||||
{"web.hook.encode_payload", oneof(["base64", "base62"])},
|
||||
nof([{"web.hook.encode_payload", oneof(["base64", "base62"])},
|
||||
{"web.hook.rule.client.connect.1", rule_spec()},
|
||||
{"web.hook.rule.client.connack.1", rule_spec()},
|
||||
{"web.hook.rule.client.connected.1", rule_spec()},
|
||||
|
|
|
@ -105,37 +105,33 @@ validate_params(Params) ->
|
|||
{error, ?ERROR8, Msg}
|
||||
end.
|
||||
|
||||
%% TODO who and reason is undefined - causing dialyzer errors. fix later
|
||||
-dialyzer({nowarn_function,pack_banned/1}).
|
||||
pack_banned(Params) ->
|
||||
Now = erlang:system_time(second),
|
||||
do_pack_banned(Params, #banned{by = <<"user">>,
|
||||
at = Now,
|
||||
until = Now + 300}).
|
||||
do_pack_banned(Params, #{by => <<"user">>, at => Now, until => Now + 300}).
|
||||
|
||||
do_pack_banned([], Banned) ->
|
||||
{ok, Banned};
|
||||
do_pack_banned([], #{who := Who, by := By, reason := Reason, at := At, until := Until}) ->
|
||||
{ok, #banned{who = Who, by = By, reason = Reason, at = At, until = Until}};
|
||||
do_pack_banned([{<<"who">>, Who} | Params], Banned) ->
|
||||
case lists:keytake(<<"as">>, 1, Params) of
|
||||
{value, {<<"as">>, <<"peerhost">>}, Params2} ->
|
||||
{ok, IPAddress} = inet:parse_address(str(Who)),
|
||||
do_pack_banned(Params2, Banned#banned{who = {peerhost, IPAddress}});
|
||||
do_pack_banned(Params2, Banned#{who => {peerhost, IPAddress}});
|
||||
{value, {<<"as">>, <<"clientid">>}, Params2} ->
|
||||
do_pack_banned(Params2, Banned#banned{who = {clientid, Who}});
|
||||
do_pack_banned(Params2, Banned#{who => {clientid, Who}});
|
||||
{value, {<<"as">>, <<"username">>}, Params2} ->
|
||||
do_pack_banned(Params2, Banned#banned{who = {username, Who}})
|
||||
do_pack_banned(Params2, Banned#{who => {username, Who}})
|
||||
end;
|
||||
do_pack_banned([P1 = {<<"as">>, _}, P2 | Params], Banned) ->
|
||||
do_pack_banned([P2, P1 | Params], Banned);
|
||||
do_pack_banned([{<<"by">>, By} | Params], Banned) ->
|
||||
do_pack_banned(Params, Banned#banned{by = By});
|
||||
do_pack_banned(Params, Banned#{by => By});
|
||||
do_pack_banned([{<<"reason">>, Reason} | Params], Banned) ->
|
||||
do_pack_banned(Params, Banned#banned{reason = Reason});
|
||||
do_pack_banned(Params, Banned#{reason => Reason});
|
||||
do_pack_banned([{<<"at">>, At} | Params], Banned) ->
|
||||
do_pack_banned(Params, Banned#banned{at = At});
|
||||
do_pack_banned(Params, Banned#{at => At});
|
||||
do_pack_banned([{<<"until">>, Until} | Params], Banned) ->
|
||||
do_pack_banned(Params, Banned#banned{until = Until});
|
||||
do_pack_banned([_P | Params], Banned) -> %% ingore other params
|
||||
do_pack_banned(Params, Banned#{until => Until});
|
||||
do_pack_banned([_P | Params], Banned) -> %% ignore other params
|
||||
do_pack_banned(Params, Banned).
|
||||
|
||||
do_delete(<<"peerhost">>, Who) ->
|
||||
|
|
|
@ -65,7 +65,7 @@ restart(#{identifier := Identifier}, _Params) ->
|
|||
Results = [{Node, emqx_mgmt:restart_listener(Node, Identifier)} || {Node, _Info} <- emqx_mgmt:list_nodes()],
|
||||
case lists:filter(fun({_, Result}) -> Result =/= ok end, Results) of
|
||||
[] -> return(ok);
|
||||
Errors -> return({error, Errors})
|
||||
Errors -> return({error, {restart, Errors}})
|
||||
end.
|
||||
|
||||
format(Listeners) when is_list(Listeners) ->
|
||||
|
|
|
@ -6,13 +6,10 @@
|
|||
%% with rebar.config.erl module. Final result is written to
|
||||
%% rebar.config.rendered if environment DEBUG is set.
|
||||
|
||||
{minimum_otp_vsn, "21.3"}.
|
||||
{edoc_opts, [{preprocess,true}]}.
|
||||
{erl_opts, [warn_unused_vars,warn_shadow_vars,warn_unused_import,
|
||||
warn_obsolete_guard,compressed]}.
|
||||
|
||||
{overrides,[{add,[{extra_src_dirs, [{"etc", [{recursive,true}]}]}]}
|
||||
]}.
|
||||
{extra_src_dirs, [{"etc", [{recursive,true}]}]}.
|
||||
|
||||
{xref_checks,[undefined_function_calls,undefined_functions,locals_not_used,
|
||||
|
|
|
@ -6,7 +6,7 @@ do(Dir, CONFIG) ->
|
|||
ok = compile_and_load_pase_transforms(Dir),
|
||||
C1 = deps(CONFIG),
|
||||
Config = dialyzer(C1),
|
||||
dump(Config ++ coveralls() ++ config()).
|
||||
dump(Config ++ [{overrides, overrides()}] ++ coveralls() ++ config()).
|
||||
|
||||
bcrypt() ->
|
||||
{bcrypt, {git, "https://github.com/emqx/erlang-bcrypt.git", {branch, "0.6.0"}}}.
|
||||
|
@ -19,6 +19,14 @@ deps(Config) ->
|
|||
end,
|
||||
lists:keystore(deps, 1, Config, {deps, OldDpes ++ MoreDeps}).
|
||||
|
||||
overrides() ->
|
||||
[ {add, [ {extra_src_dirs, [{"etc", [{recursive,true}]}]}
|
||||
, {erl_opts, [ deterministic
|
||||
, {compile_info, [{emqx_vsn, get_vsn()}]}
|
||||
]}
|
||||
]}
|
||||
].
|
||||
|
||||
config() ->
|
||||
[ {plugins, plugins()}
|
||||
, {profiles, profiles()}
|
||||
|
@ -51,27 +59,42 @@ test_deps() ->
|
|||
, meck
|
||||
].
|
||||
|
||||
default_compile_opts() ->
|
||||
[compressed, deterministic, no_debug_info, warnings_as_errors, {parse_transform, mod_vsn}].
|
||||
common_compile_opts() ->
|
||||
[ deterministic
|
||||
, {compile_info, [{emqx_vsn, get_vsn()}]}
|
||||
].
|
||||
|
||||
prod_compile_opts() ->
|
||||
[ compressed
|
||||
, no_debug_info
|
||||
, warnings_as_errors
|
||||
| common_compile_opts()
|
||||
].
|
||||
|
||||
test_compile_opts() ->
|
||||
[ debug_info
|
||||
| common_compile_opts()
|
||||
].
|
||||
|
||||
profiles() ->
|
||||
[ {'emqx', [ {erl_opts, default_compile_opts()}
|
||||
[ {'emqx', [ {erl_opts, prod_compile_opts()}
|
||||
, {relx, relx('emqx')}
|
||||
]}
|
||||
, {'emqx-pkg', [ {erl_opts, default_compile_opts()}
|
||||
, {'emqx-pkg', [ {erl_opts, prod_compile_opts()}
|
||||
, {relx, relx('emqx-pkg')}
|
||||
]}
|
||||
, {'emqx-edge', [ {erl_opts, default_compile_opts()}
|
||||
, {'emqx-edge', [ {erl_opts, prod_compile_opts()}
|
||||
, {relx, relx('emqx-edge')}
|
||||
]}
|
||||
, {'emqx-edge-pkg', [ {erl_opts, default_compile_opts()}
|
||||
, {'emqx-edge-pkg', [ {erl_opts, prod_compile_opts()}
|
||||
, {relx, relx('emqx-edge-pkg')}
|
||||
]}
|
||||
, {check, [ {erl_opts, [debug_info, warnings_as_errors, {parse_transform, mod_vsn}]}
|
||||
, {check, [ {erl_opts, test_compile_opts()}
|
||||
]}
|
||||
, {test, [ {deps, test_deps()}
|
||||
, {plugins, test_plugins()}
|
||||
, {erl_opts, [debug_info, {parse_transform, mod_vsn}] ++ erl_opts_i()}
|
||||
, {erl_opts, test_compile_opts() ++ erl_opts_i()}
|
||||
, {extra_src_dirs, [{"test", [{recursive,true}]}]}
|
||||
]}
|
||||
].
|
||||
|
||||
|
@ -125,6 +148,7 @@ relx_apps(ReleaseType) ->
|
|||
, emqx
|
||||
, {mnesia, load}
|
||||
, {ekka, load}
|
||||
, {emqx_plugin_libs, load}
|
||||
]
|
||||
++ [bcrypt || provide_bcrypt_release(ReleaseType)]
|
||||
++ relx_apps_per_rel(ReleaseType)
|
||||
|
|
|
@ -1,12 +1,11 @@
|
|||
#!/bin/sh
|
||||
#!/bin/bash
|
||||
|
||||
#set -euo pipefail
|
||||
set -eu
|
||||
set -euo pipefail
|
||||
|
||||
VERSION="$1"
|
||||
|
||||
# ensure dir
|
||||
cd -P -- "$(dirname -- "$0")"
|
||||
cd -P -- "$(dirname -- "${BASH_SOURCE[0]}")/.."
|
||||
|
||||
DOWNLOAD_URL='https://github.com/emqx/rebar3/releases/download'
|
||||
|
|
@ -1,15 +1,19 @@
|
|||
#!/bin/bash
|
||||
|
||||
set -eu
|
||||
|
||||
VERSION="$1"
|
||||
set -euo pipefail
|
||||
|
||||
# ensure dir
|
||||
cd -P -- "$(dirname -- "$0")"
|
||||
cd -P -- "$(dirname -- "${BASH_SOURCE[0]}")/.."
|
||||
|
||||
DOWNLOAD_URL='https://github.com/emqx/emqx-dashboard-frontend/releases/download'
|
||||
if [[ "$1" == https://* ]]; then
|
||||
VERSION='*' # alwyas download
|
||||
DOWNLOAD_URL="$1"
|
||||
else
|
||||
VERSION="$1"
|
||||
DOWNLOAD_URL="https://github.com/emqx/emqx-dashboard-frontend/releases/download/${VERSION}/emqx-dashboard.zip"
|
||||
fi
|
||||
|
||||
if [ "$EMQX_ENTERPRISE" = 'true' ] || [ "$EMQX_ENTERPRISE" == '1' ]; then
|
||||
if [ "${EMQX_ENTERPRISE:-}" = 'true' ] || [ "${EMQX_ENTERPRISE:-}" == '1' ]; then
|
||||
DASHBOARD_PATH='lib-ee/emqx_dashboard/priv'
|
||||
else
|
||||
DASHBOARD_PATH='lib-ce/emqx_dashboard/priv'
|
||||
|
@ -28,7 +32,7 @@ if [ -d "$DASHBOARD_PATH/www" ] && [ "$(version)" = "$VERSION" ]; then
|
|||
exit 0
|
||||
fi
|
||||
|
||||
curl -f -L "${DOWNLOAD_URL}/${VERSION}/emqx-dashboard.zip" -o ./emqx-dashboard.zip
|
||||
curl -f -L "${DOWNLOAD_URL}" -o ./emqx-dashboard.zip
|
||||
unzip -q ./emqx-dashboard.zip -d "$DASHBOARD_PATH"
|
||||
rm -rf "$DASHBOARD_PATH/www"
|
||||
mv "$DASHBOARD_PATH/dist" "$DASHBOARD_PATH/www"
|
|
@ -99,7 +99,7 @@ ensure_all_started([L | Rest], Results) ->
|
|||
ensure_all_started(Rest, NewResults).
|
||||
|
||||
%% @doc Format address:port for logging.
|
||||
-spec(format_listen_on(esockd:listen_on()) -> binary()).
|
||||
-spec(format_listen_on(esockd:listen_on()) -> [char()]).
|
||||
format_listen_on(ListenOn) -> format(ListenOn).
|
||||
|
||||
-spec(start_listener(listener()) -> ok).
|
||||
|
@ -197,9 +197,8 @@ restart_listener(Proto, ListenOn, Options) when Proto == https; Proto == wss ->
|
|||
restart_listener(Proto, ListenOn, _Opts) ->
|
||||
esockd:reopen(Proto, ListenOn).
|
||||
|
||||
ok(ok) -> ok;
|
||||
ok({ok, _}) -> ok;
|
||||
ok(Error) -> Error.
|
||||
ok(Other) -> Other.
|
||||
|
||||
%% @doc Stop all listeners.
|
||||
-spec(stop() -> ok).
|
||||
|
|
|
@ -23,7 +23,10 @@
|
|||
, integral_ciphers/2
|
||||
]).
|
||||
|
||||
-define(IS_STRING_LIST(L), (is_list(L) andalso L =/= [] andalso is_list(hd(L)))).
|
||||
%% non-empty string
|
||||
-define(IS_STRING(L), (is_list(L) andalso L =/= [] andalso is_integer(hd(L)))).
|
||||
%% non-empty list of strings
|
||||
-define(IS_STRING_LIST(L), (is_list(L) andalso L =/= [] andalso ?IS_STRING(hd(L)))).
|
||||
|
||||
%% @doc Returns the default supported tls versions.
|
||||
-spec default_versions() -> [atom()].
|
||||
|
@ -33,7 +36,19 @@ default_versions() ->
|
|||
|
||||
%% @doc Validate a given list of desired tls versions.
|
||||
%% raise an error exception if non of them are available.
|
||||
-spec integral_versions([ssl:tls_version()]) -> [ssl:tls_version()].
|
||||
%% The input list can be a string/binary of comma separated versions.
|
||||
-spec integral_versions(undefined | string() | binary() | [ssl:tls_version()]) ->
|
||||
[ssl:tls_version()].
|
||||
integral_versions(undefined) ->
|
||||
integral_versions(default_versions());
|
||||
integral_versions([]) ->
|
||||
integral_versions(default_versions());
|
||||
integral_versions(<<>>) ->
|
||||
integral_versions(default_versions());
|
||||
integral_versions(Desired) when ?IS_STRING(Desired) ->
|
||||
integral_versions(iolist_to_binary(Desired));
|
||||
integral_versions(Desired) when is_binary(Desired) ->
|
||||
integral_versions(parse_versions(Desired));
|
||||
integral_versions(Desired) ->
|
||||
{_, Available} = lists:keyfind(available, 1, ssl:versions()),
|
||||
case lists:filter(fun(V) -> lists:member(V, Available) end, Desired) of
|
||||
|
@ -96,3 +111,32 @@ default_versions(_) ->
|
|||
%% Deduplicate a list without re-ordering the elements.
|
||||
dedup([]) -> [];
|
||||
dedup([H | T]) -> [H | dedup([I || I <- T, I =/= H])].
|
||||
|
||||
%% parse comma separated tls version strings
|
||||
parse_versions(Versions) ->
|
||||
do_parse_versions(split_by_comma(Versions), []).
|
||||
|
||||
do_parse_versions([], Acc) -> lists:reverse(Acc);
|
||||
do_parse_versions([V | More], Acc) ->
|
||||
case parse_version(V) of
|
||||
unknown ->
|
||||
emqx_logger:warning("unknown_tls_version_discarded: ~p", [V]),
|
||||
do_parse_versions(More, Acc);
|
||||
Parsed ->
|
||||
do_parse_versions(More, [Parsed | Acc])
|
||||
end.
|
||||
|
||||
parse_version(<<"tlsv", Vsn/binary>>) -> parse_version(Vsn);
|
||||
parse_version(<<"v", Vsn/binary>>) -> parse_version(Vsn);
|
||||
parse_version(<<"1.3">>) -> 'tlsv1.3';
|
||||
parse_version(<<"1.2">>) -> 'tlsv1.2';
|
||||
parse_version(<<"1.1">>) -> 'tlsv1.1';
|
||||
parse_version(<<"1">>) -> 'tlsv1';
|
||||
parse_version(_) -> unknown.
|
||||
|
||||
split_by_comma(Bin) ->
|
||||
[trim_space(I) || I <- binary:split(Bin, <<",">>, [global])].
|
||||
|
||||
%% trim spaces
|
||||
trim_space(Bin) ->
|
||||
hd([I || I <- binary:split(Bin, <<" ">>), I =/= <<>>]).
|
||||
|
|
103
sync-apps.sh
103
sync-apps.sh
|
@ -1,103 +0,0 @@
|
|||
#!/bin/bash
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
force="${1:-no}"
|
||||
|
||||
apps=(
|
||||
# "emqx_auth_http" # permanently diverged
|
||||
# "emqx_web_hook" # permanently diverged
|
||||
"emqx_auth_jwt"
|
||||
"emqx_auth_ldap"
|
||||
"emqx_auth_mongo"
|
||||
"emqx_auth_mysql"
|
||||
"emqx_auth_pgsql"
|
||||
"emqx_auth_redis"
|
||||
"emqx_bridge_mqtt"
|
||||
"emqx_coap"
|
||||
# "emqx_dashboard" # moved to lib-ce
|
||||
"emqx_exhook"
|
||||
"emqx_exproto"
|
||||
"emqx_lua_hook"
|
||||
"emqx_lwm2m"
|
||||
# "emqx_management" # moved to lib-ce
|
||||
"emqx_prometheus"
|
||||
"emqx_psk_file"
|
||||
"emqx_recon"
|
||||
"emqx_retainer"
|
||||
"emqx_rule_engine"
|
||||
"emqx_sasl"
|
||||
"emqx_sn"
|
||||
"emqx_stomp"
|
||||
"emqx_telemetry"
|
||||
)
|
||||
|
||||
if git status --porcelain | grep -qE 'apps/'; then
|
||||
echo 'apps dir is not git-clear, refuse to sync'
|
||||
# exit 1
|
||||
fi
|
||||
|
||||
mkdir -p tmp/
|
||||
|
||||
download_zip() {
|
||||
local app="$1"
|
||||
local ref="$2"
|
||||
local vsn
|
||||
vsn="$(echo "$ref" | tr '/' '-')"
|
||||
local file="tmp/${app}-${vsn}.zip"
|
||||
if [ -f "$file" ] && [ "$force" != "force" ]; then
|
||||
return 0
|
||||
fi
|
||||
local repo
|
||||
repo=${app//_/-}
|
||||
local url="https://github.com/emqx/$repo/archive/$ref.zip"
|
||||
echo "downloading ${url}"
|
||||
curl -fLsS -o "$file" "$url"
|
||||
}
|
||||
|
||||
default_vsn="dev/v4.3.0"
|
||||
download_zip "emqx_auth_mnesia" "e4.2.3"
|
||||
for app in "${apps[@]}"; do
|
||||
download_zip "$app" "$default_vsn"
|
||||
done
|
||||
|
||||
extract_zip(){
|
||||
local app="$1"
|
||||
local ref="$2"
|
||||
local vsn_arg="${3:-}"
|
||||
local vsn_dft
|
||||
vsn_dft="$(echo "$ref" | tr '/' '-')"
|
||||
local vsn
|
||||
if [ -n "$vsn_arg" ]; then
|
||||
vsn="$vsn_arg"
|
||||
else
|
||||
vsn="$vsn_dft"
|
||||
fi
|
||||
local file="tmp/${app}-${vsn_dft}.zip"
|
||||
local repo
|
||||
repo=${app//_/-}
|
||||
rm -rf "apps/${app}/"
|
||||
unzip "$file" -d apps/
|
||||
mv "apps/${repo}-${vsn}/" "apps/$app/"
|
||||
}
|
||||
|
||||
extract_zip "emqx_auth_mnesia" "e4.2.3" "e4.2.3"
|
||||
for app in "${apps[@]}"; do
|
||||
extract_zip "$app" "$default_vsn"
|
||||
done
|
||||
|
||||
cleanup_app(){
|
||||
local app="$1"
|
||||
pushd "apps/$app"
|
||||
rm -f Makefile rebar.config.script LICENSE src/*.app.src.script src/*.appup.src
|
||||
rm -rf ".github" ".ci"
|
||||
# restore rebar.config and app.src
|
||||
git checkout rebar.config
|
||||
git checkout src/*.app.src
|
||||
popd
|
||||
}
|
||||
|
||||
apps+=( "emqx_auth_mnesia" )
|
||||
for app in "${apps[@]}"; do
|
||||
cleanup_app "$app"
|
||||
done
|
|
@ -53,8 +53,12 @@ tls_versions_test() ->
|
|||
?assert(lists:member('tlsv1.3', emqx_tls_lib:default_versions())).
|
||||
|
||||
tls_version_unknown_test() ->
|
||||
?assertError(#{reason := no_available_tls_version},
|
||||
?assertEqual(emqx_tls_lib:default_versions(),
|
||||
emqx_tls_lib:integral_versions([])),
|
||||
?assertEqual(emqx_tls_lib:default_versions(),
|
||||
emqx_tls_lib:integral_versions(<<>>)),
|
||||
?assertEqual(emqx_tls_lib:default_versions(),
|
||||
emqx_tls_lib:integral_versions("foo")),
|
||||
?assertError(#{reason := no_available_tls_version},
|
||||
emqx_tls_lib:integral_versions([foo])).
|
||||
|
||||
|
|
Loading…
Reference in New Issue