yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #4067 from HJianBo/fix/webhook_certs 

fix(webhook): fix bad https confs
This commit is contained in:
JianBo He 2021-01-27 10:44:28 +08:00 committed by GitHub
parent 068c351936 8d56371465
commit 7919f08e03
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 123 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
apps/emqx_web_hook/src/emqx_web_hook_app.erl
View File

@ -72,10 +72,8 @@ translate_env() ->
true -> verify_peer; true -> verify_peer;
false -> verify_none false -> verify_none
end, end,
TLSOpts = lists:filter(fun({_K, V}) when V =:= <<>> -> TLSOpts = lists:filter(fun({_K, V}) ->
false; V /= <<>> andalso V /= undefined andalso V /= "" andalso true
(_) ->
true
end, [{keyfile, KeyFile}, {certfile, CertFile}, {cacertfile, CACertFile}]), end, [{keyfile, KeyFile}, {certfile, CertFile}, {cacertfile, CACertFile}]),
TlsVers = ['tlsv1.2','tlsv1.1',tlsv1], TlsVers = ['tlsv1.2','tlsv1.1',tlsv1],
NTLSOpts = [{verify, VerifyType}, NTLSOpts = [{verify, VerifyType},

97
apps/emqx_web_hook/test/emqx_web_hook_SUITE.erl
View File

@ -47,49 +47,38 @@ init_per_group(Name, Config) ->
set_special_cfgs(), set_special_cfgs(),
case Name of case Name of
http -> http ->
http_server:start_http(),
emqx_ct_helpers:start_apps([emqx_web_hook], fun set_special_configs_http/1); emqx_ct_helpers:start_apps([emqx_web_hook], fun set_special_configs_http/1);
https -> https ->
http_server:start_https(),
emqx_ct_helpers:start_apps([emqx_web_hook], fun set_special_configs_https/1); emqx_ct_helpers:start_apps([emqx_web_hook], fun set_special_configs_https/1);
ipv6http -> ipv6http ->
http_server:start_http(),
emqx_ct_helpers:start_apps([emqx_web_hook], fun set_special_configs_ipv6_http/1); emqx_ct_helpers:start_apps([emqx_web_hook], fun set_special_configs_ipv6_http/1);
ipv6https -> ipv6https ->
http_server:start_https(),
emqx_ct_helpers:start_apps([emqx_web_hook], fun set_special_configs_ipv6_https/1) emqx_ct_helpers:start_apps([emqx_web_hook], fun set_special_configs_ipv6_https/1)
end, end,
Config. Config.
end_per_group(Name, Config) -> end_per_group(_Name, Config) ->
case lists:member(Name,[http, ipv6http]) of
true ->
http_server:stop_http();
_ ->
http_server:stop_https()
end,
emqx_ct_helpers:stop_apps([emqx_web_hook]), emqx_ct_helpers:stop_apps([emqx_web_hook]),
Config. Config.
set_special_configs_http(_) -> set_special_configs_http(_) ->
ok. application:set_env(emqx_web_hook, url, "http://127.0.0.1:9999").
set_special_configs_https(_) -> set_special_configs_https(_) ->
Path = emqx_ct_helpers:deps_path(emqx_web_hook, "test/emqx_web_hook_SUITE_data/"), Path = emqx_ct_helpers:deps_path(emqx_web_hook, "test/emqx_web_hook_SUITE_data/"),
SslOpts = [{keyfile, Path ++ "/client-key.pem"}, SslOpts = [{keyfile, Path ++ "/client-key.pem"},
{certfile, Path ++ "/client-cert.pem"}, {certfile, Path ++ "/client-cert.pem"},
{cacertfile, Path ++ "/ca.pem"}], {cafile, Path ++ "/ca.pem"}],
application:set_env(emqx_web_hook, ssl, true), application:set_env(emqx_web_hook, ssl, true),
application:set_env(emqx_web_hook, ssloptions, SslOpts), application:set_env(emqx_web_hook, ssloptions, SslOpts),
application:set_env(emqx_web_hook, url, "https://127.0.0.1:8081"). application:set_env(emqx_web_hook, url, "https://127.0.0.1:8888").
set_special_configs_ipv6_http(N) -> set_special_configs_ipv6_http(_) ->
set_special_configs_http(N), application:set_env(emqx_web_hook, url, "http://[::1]:9999").
application:set_env(emqx_web_hook, url, "http://[::1]:8080").
set_special_configs_ipv6_https(N) -> set_special_configs_ipv6_https(N) ->
set_special_configs_https(N), set_special_configs_https(N),
application:set_env(emqx_web_hook, url, "https://[::1]:8081"). application:set_env(emqx_web_hook, url, "https://[::1]:8888").
set_special_cfgs() -> set_special_cfgs() ->
AllRules = [{"message.acked", "{\"action\": \"on_message_acked\"}"}, AllRules = [{"message.acked", "{\"action\": \"on_message_acked\"}"},
@ -109,6 +98,27 @@ set_special_cfgs() ->
%% Test cases %% Test cases
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
t_valid(Config) ->
{ok, ServerPid} = http_server:start_link(),
application:set_env(emqx_web_hook, headers, [{"k1","K1"}, {"k2", "K2"}]),
{ok, C} = emqtt:start_link([ {clientid, <<"simpleClient">>}
, {proto_ver, v5}
, {keepalive, 60}
]),
{ok, _} = emqtt:connect(C),
emqtt:subscribe(C, <<"TopicA">>, qos2),
emqtt:publish(C, <<"TopicA">>, <<"Payload...">>, qos2),
emqtt:unsubscribe(C, <<"TopicA">>),
emqtt:disconnect(C),
[begin
Maps = emqx_json:decode(P, [return_maps]),
validate_hook_resp(Maps),
validate_hook_headers(Headers)
end
|| {{P, _Bool}, Headers} <- http_server:get_received_data()],
http_server:stop(ServerPid),
Config.
t_check_hooked(_) -> t_check_hooked(_) ->
{ok, Rules} = application:get_env(emqx_web_hook, rules), {ok, Rules} = application:get_env(emqx_web_hook, rules),
lists:foreach(fun({HookName, _Action}) -> lists:foreach(fun({HookName, _Action}) ->
@ -127,90 +137,69 @@ t_change_config(_) ->
application:set_env(emqx_web_hook, rules, Rules), application:set_env(emqx_web_hook, rules, Rules),
emqx_web_hook:load(). emqx_web_hook:load().
t_valid() ->
application:set_env(emqx_web_hook, headers, [{"k1","K1"}, {"k2", "K2"}]),
{ok, C} = emqtt:start_link([ {clientid, <<"simpleClient">>}
, {proto_ver, v5}
, {keepalive, 60}
]),
{ok, _} = emqtt:connect(C),
emqtt:subscribe(C, <<"TopicA">>, qos2),
emqtt:publish(C, <<"TopicA">>, <<"Payload...">>, qos2),
emqtt:unsubscribe(C, <<"TopicA">>),
emqtt:disconnect(C),
{Params, Headers} = get_http_message(),
[validate_hook_resp(A) || A <- Params],
?assertEqual(<<"K1">>, maps:get(<<"k1">>, Headers)),
?assertEqual(<<"K2">>, maps:get(<<"k2">>, Headers)).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Utils %% Utils
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
get_http_message() -> validate_hook_headers(Headers) ->
receive ?assertEqual(<<"K1">>, maps:get(<<"k1">>, Headers)),
{Params, Headers} -> ?assertEqual(<<"K2">>, maps:get(<<"k2">>, Headers)).
L = [B || {B, _} <- Params],
{lists:reverse([emqx_json:decode(E, [return_maps]) || E <- L]), Headers}
after 500 ->
{null, null}
end.
validate_hook_resp(Body = ?ACTION(<<"client_connect">>)) -> validate_hook_resp(Body = ?ACTION(<<"client_connect">>)) ->
?assertEqual(5, maps:get(<<"proto_ver">>, Body)), ?assertEqual(5, maps:get(<<"proto_ver">>, Body)),
?assertEqual(60, maps:get(<<"keepalive">>, Body)), ?assertEqual(60, maps:get(<<"keepalive">>, Body)),
?assertEqual(<<"127.0.0.1">>, maps:get(<<"ipaddress">>, Body)), ?assertEqual(<<"127.0.0.1">>, maps:get(<<"ipaddress">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)), ?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body); assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"client_connack">>)) -> validate_hook_resp(Body = ?ACTION(<<"client_connack">>)) ->
?assertEqual(5, maps:get(<<"proto_ver">>, Body)), ?assertEqual(5, maps:get(<<"proto_ver">>, Body)),
?assertEqual(60, maps:get(<<"keepalive">>, Body)), ?assertEqual(60, maps:get(<<"keepalive">>, Body)),
?assertEqual(<<"success">>, maps:get(<<"conn_ack">>, Body)), ?assertEqual(<<"success">>, maps:get(<<"conn_ack">>, Body)),
?assertEqual(<<"127.0.0.1">>, maps:get(<<"ipaddress">>, Body)), ?assertEqual(<<"127.0.0.1">>, maps:get(<<"ipaddress">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)), ?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body); assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"client_connected">>)) -> validate_hook_resp(Body = ?ACTION(<<"client_connected">>)) ->
_ = maps:get(<<"connected_at">>, Body), _ = maps:get(<<"connected_at">>, Body),
?assertEqual(5, maps:get(<<"proto_ver">>, Body)), ?assertEqual(5, maps:get(<<"proto_ver">>, Body)),
?assertEqual(60, maps:get(<<"keepalive">>, Body)), ?assertEqual(60, maps:get(<<"keepalive">>, Body)),
?assertEqual(<<"127.0.0.1">>, maps:get(<<"ipaddress">>, Body)), ?assertEqual(<<"127.0.0.1">>, maps:get(<<"ipaddress">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)), ?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body); assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"client_disconnected">>)) -> validate_hook_resp(Body = ?ACTION(<<"client_disconnected">>)) ->
?assertEqual(<<"normal">>, maps:get(<<"reason">>, Body)), ?assertEqual(<<"normal">>, maps:get(<<"reason">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)), ?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body); assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"client_subscribe">>)) -> validate_hook_resp(Body = ?ACTION(<<"client_subscribe">>)) ->
_ = maps:get(<<"opts">>, Body), _ = maps:get(<<"opts">>, Body),
?assertEqual(<<"TopicA">>, maps:get(<<"topic">>, Body)), ?assertEqual(<<"TopicA">>, maps:get(<<"topic">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)), ?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body); assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"client_unsubscribe">>)) -> validate_hook_resp(Body = ?ACTION(<<"client_unsubscribe">>)) ->
_ = maps:get(<<"opts">>, Body), _ = maps:get(<<"opts">>, Body),
?assertEqual(<<"TopicA">>, maps:get(<<"topic">>, Body)), ?assertEqual(<<"TopicA">>, maps:get(<<"topic">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)), ?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body); assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"session_subscribed">>)) -> validate_hook_resp(Body = ?ACTION(<<"session_subscribed">>)) ->
_ = maps:get(<<"opts">>, Body), _ = maps:get(<<"opts">>, Body),
?assertEqual(<<"TopicA">>, maps:get(<<"topic">>, Body)), ?assertEqual(<<"TopicA">>, maps:get(<<"topic">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)), ?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body); assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"session_unsubscribed">>)) -> validate_hook_resp(Body = ?ACTION(<<"session_unsubscribed">>)) ->
?assertEqual(<<"TopicA">>, maps:get(<<"topic">>, Body)), ?assertEqual(<<"TopicA">>, maps:get(<<"topic">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)), ?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body); assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"session_terminated">>)) -> validate_hook_resp(Body = ?ACTION(<<"session_terminated">>)) ->
?assertEqual(<<"normal">>, maps:get(<<"reason">>, Body)), ?assertEqual(<<"normal">>, maps:get(<<"reason">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)), ?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body); assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"message_publish">>)) -> validate_hook_resp(Body = ?ACTION(<<"message_publish">>)) ->
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)), ?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_messages_attrs(Body); assert_messages_attrs(Body);
validate_hook_resp(Body = ?ACTION(<<"message_delivered">>)) -> validate_hook_resp(Body = ?ACTION(<<"message_delivered">>)) ->
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)), ?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_messages_attrs(Body); assert_messages_attrs(Body);
validate_hook_resp(Body = ?ACTION(<<"message_acked">>)) -> validate_hook_resp(Body = ?ACTION(<<"message_acked">>)) ->
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)), ?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_messages_attrs(Body). assert_messages_attrs(Body).
assert_username_clientid(#{<<"clientid">> := ClientId, <<"username">> := Username}) -> assert_username_clientid(#{<<"clientid">> := ClientId, <<"username">> := Username}) ->

114
apps/emqx_web_hook/test/http_server.erl
View File

@ -3,73 +3,103 @@
%% %%
%% It will deliver the http-request params to initialer process %% It will deliver the http-request params to initialer process
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%%
%% Author:wwhai
%%
-module(http_server). -module(http_server).
-behaviour(gen_server).
-compile(export_all). -export([start_link/0]).
-compile(nowarn_export_all). -export([get_received_data/0]).
-export([stop/1]).
-export([code_change/3, handle_call/3, handle_cast/2, handle_info/2, init/1, terminate/2]).
-define(HTTP_PORT, 9999).
-define(HTTPS_PORT, 8888).
-record(state, {}).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% APIs %% APIs
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
start() ->
{ok, _} = application:ensure_all_started(cowboy),
cowboy_router:compile([
{'_', [
{"/", ?MODULE, self()}
]}
]).
start_link() ->
gen_server:start_link(?MODULE, [], []).
start_http() -> init([]) ->
{ok, _Pid1} = cowboy:start_clear(http, [{port, 8080}], #{ EtsOptions = [named_table, public, set, {write_concurrency, true},
env => #{dispatch => start()} {read_concurrency, true}],
}), emqx_web_hook_http_test = ets:new(emqx_web_hook_http_test, EtsOptions),
io:format("Start http server on 8080 successfully!~n"). ok = start_http(?HTTP_PORT),
ok = start_https(?HTTPS_PORT),
{ok, #state{}}.
start_https() -> handle_call(_Request, _From, State) ->
Path = emqx_ct_helpers:deps_path(emqx_web_hook, "test/emqx_web_hook_SUITE_data/"), {reply, ignored, State}.
SslOpts = [{keyfile, Path ++ "/server-key.pem"},
{cacertfile, Path ++ "/ca.pem"},
{certfile, Path ++ "/server-cert.pem"}],
{ok, _Pid2} = cowboy:start_tls(https, [{port, 8081}] ++ SslOpts, handle_cast(_Msg, State) ->
#{env => #{dispatch => start()}}), {noreply, State}.
io:format(standard_error, "Start https server on 8081 successfully!~n", []).
stop_http() -> handle_info(_Info, State) ->
ok = cowboy:stop_listener(http), {noreply, State}.
io:format("Stopped http server on 8080").
stop_https() -> terminate(_Reason, _State) ->
ok = cowboy:stop_listener(https), stop_http(),
io:format("Stopped https server on 8081"). stop_https().
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
get_received_data() ->
ets:tab2list(emqx_web_hook_http_test).
stop(Pid) ->
ok = gen_server:stop(Pid).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Callbacks %% Callbacks
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
init(Req, ReceiverPid) -> start_http(Port) ->
Req1 = handle_request(Req, ReceiverPid), {ok, _Pid1} = cowboy:start_clear(http, [{port, Port}], #{
{ok, Req1, ReceiverPid}. env => #{dispatch => compile_router()}
}),
io:format(standard_error, "[TEST LOG] Start http server on 9999 successfully!~n", []).
%% @private start_https(Port) ->
handle_request(Req, ReceiverPid) -> Path = emqx_ct_helpers:deps_path(emqx_web_hook, "test/emqx_web_hook_SUITE_data/"),
SslOpts = [{keyfile, Path ++ "/server-key.pem"},
{cacertfile, Path ++ "/ca.pem"},
{certfile, Path ++ "/server-cert.pem"}],
{ok, _Pid2} = cowboy:start_tls(https, [{port, Port}] ++ SslOpts,
#{env => #{dispatch => compile_router()}}),
io:format(standard_error, "[TEST LOG] Start https server on 8888 successfully!~n", []).
stop_http() ->
ok = cowboy:stop_listener(http),
io:format("[TEST LOG] Stopped http server on 9999").
stop_https() ->
ok = cowboy:stop_listener(https),
io:format("[TEST LOG] Stopped https server on 8888").
compile_router() ->
{ok, _} = application:ensure_all_started(cowboy),
cowboy_router:compile([
{'_', [{"/", ?MODULE, #{}}]}
]).
init(Req, State) ->
Method = cowboy_req:method(Req), Method = cowboy_req:method(Req),
Headers = cowboy_req:headers(Req), Headers = cowboy_req:headers(Req),
Params = [Params] = case Method of
case Method of
<<"GET">> -> cowboy_req:parse_qs(Req); <<"GET">> -> cowboy_req:parse_qs(Req);
<<"POST">> -> <<"POST">> ->
{ok, PostVals, _Req2} = cowboy_req:read_urlencoded_body(Req), {ok, PostVals, _} = cowboy_req:read_urlencoded_body(Req),
PostVals PostVals
end, end,
io:format("Request Data:~p~nHeaders :~p~n", [Params, Headers]), ets:insert(emqx_web_hook_http_test, {Params, Headers}),
erlang:send(ReceiverPid, {Params, Headers}), {ok, reply(Req, ok), State}.
reply(Req, ok).
%% @private
reply(Req, ok) -> reply(Req, ok) ->
cowboy_req:reply(200, #{<<"content-type">> => <<"text/plain">>}, <<"hello">>, Req); cowboy_req:reply(200, #{<<"content-type">> => <<"text/plain">>}, <<"ok">>, Req);
reply(Req, error) -> reply(Req, error) ->
cowboy_req:reply(404, #{<<"content-type">> => <<"text/plain">>}, <<"deny">>, Req). cowboy_req:reply(404, #{<<"content-type">> => <<"text/plain">>}, <<"deny">>, Req).