emqx/apps/emqx_exhook/test/props/prop_exhook_hooks.erl

681 lines
22 KiB
Erlang

%%--------------------------------------------------------------------
%% Copyright (c) 2020-2024 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(prop_exhook_hooks).
-include_lib("proper/include/proper.hrl").
-include_lib("eunit/include/eunit.hrl").
-include_lib("emqx/include/emqx_access_control.hrl").
-import(
emqx_proper_types,
[
conninfo/0,
clientinfo/0,
sessioninfo/0,
message/0,
connack_return_code/0,
topictab/0,
topic/0,
subopts/0,
pubsub/0
]
).
-define(CONF_DEFAULT, <<
"\n"
"exhook {\n"
" servers =\n"
" [ { name = default,\n"
" url = \"http://127.0.0.1:9000\"\n"
" }\n"
" ]\n"
"}\n"
>>).
-define(ALL(Vars, Types, Exprs),
?SETUP(
fun() ->
State = do_setup(),
fun() -> do_teardown(State) end
end,
?FORALL(Vars, Types, Exprs)
)
).
-define(DEFAULT_CLUSTER_NAME_ATOM, emqxcl).
-define(DEFAULT_CLUSTER_NAME_BIN, <<"emqxcl">>).
%%--------------------------------------------------------------------
%% Properties
%%--------------------------------------------------------------------
prop_client_connect() ->
?ALL(
{ConnInfo, ConnProps, Meta},
{conninfo(), conn_properties(), request_meta()},
begin
ok = emqx_hooks:run('client.connect', [ConnInfo, ConnProps]),
{'on_client_connect', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
props => properties(ConnProps),
conninfo => from_conninfo(ConnInfo),
meta => Meta
},
?assertEqual(Expected, Resp),
true
end
).
prop_client_connack() ->
?ALL(
{ConnInfo, Rc, AckProps, Meta},
{conninfo(), connack_return_code(), ack_properties(), request_meta()},
begin
ok = emqx_hooks:run('client.connack', [ConnInfo, Rc, AckProps]),
{'on_client_connack', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
props => properties(AckProps),
result_code => atom_to_binary(Rc, utf8),
conninfo => from_conninfo(ConnInfo),
meta => Meta
},
?assertEqual(Expected, Resp),
true
end
).
prop_client_authenticate() ->
?ALL(
{ClientInfo0, AuthResult, Meta},
{clientinfo(), authresult(), request_meta()},
begin
ClientInfo = inject_magic_into(username, ClientInfo0),
OutAuthResult = emqx_hooks:run_fold('client.authenticate', [ClientInfo], AuthResult),
ExpectedAuthResult =
case maps:get(username, ClientInfo) of
<<"baduser">> ->
{error, not_authorized};
<<"gooduser">> ->
ok;
<<"normaluser">> ->
ok;
_ ->
case AuthResult of
ok -> ok;
_ -> {error, not_authorized}
end
end,
?assertEqual(ExpectedAuthResult, OutAuthResult),
{'on_client_authenticate', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
result => authresult_to_bool(AuthResult),
clientinfo => from_clientinfo(ClientInfo),
meta => Meta
},
?assertEqual(Expected, Resp),
true
end
).
prop_client_authorize() ->
MkResult = fun(Result) -> #{result => Result, from => exhook} end,
?ALL(
{ClientInfo0, PubSub, Topic, Result, Meta},
{
clientinfo(),
pubsub(),
topic(),
oneof([MkResult(allow), MkResult(deny)]),
request_meta()
},
begin
ClientInfo = inject_magic_into(username, ClientInfo0),
OutResult = emqx_hooks:run_fold(
'client.authorize',
[ClientInfo, PubSub, Topic],
Result
),
ExpectedOutResult =
case maps:get(username, ClientInfo) of
<<"baduser">> -> MkResult(deny);
<<"gooduser">> -> MkResult(allow);
<<"normaluser">> -> MkResult(allow);
_ -> Result
end,
?assertEqual(ExpectedOutResult, OutResult),
{'on_client_authorize', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
result => aclresult_to_bool(Result),
type => pubsub_to_enum(PubSub),
topic => Topic,
clientinfo => from_clientinfo(ClientInfo),
meta => Meta
},
?assertEqual(Expected, Resp),
true
end
).
prop_client_connected() ->
?ALL(
{ClientInfo, ConnInfo, Meta},
{clientinfo(), conninfo(), request_meta()},
begin
ok = emqx_hooks:run('client.connected', [ClientInfo, ConnInfo]),
{'on_client_connected', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
clientinfo => from_clientinfo(ClientInfo),
meta => Meta
},
?assertEqual(Expected, Resp),
true
end
).
prop_client_disconnected() ->
?ALL(
{ClientInfo, Reason, ConnInfo, Meta},
{clientinfo(), shutdown_reason(), conninfo(), request_meta()},
begin
ok = emqx_hooks:run('client.disconnected', [ClientInfo, Reason, ConnInfo]),
{'on_client_disconnected', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
reason => stringfy(Reason),
clientinfo => from_clientinfo(ClientInfo),
meta => Meta
},
?assertEqual(Expected, Resp),
true
end
).
prop_client_subscribe() ->
?ALL(
{ClientInfo, SubProps, TopicTab, Meta},
{clientinfo(), sub_properties(), topictab(), request_meta()},
begin
ok = emqx_hooks:run('client.subscribe', [ClientInfo, SubProps, TopicTab]),
{'on_client_subscribe', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
props => properties(SubProps),
topic_filters => topicfilters(TopicTab),
clientinfo => from_clientinfo(ClientInfo),
meta => Meta
},
?assertEqual(Expected, Resp),
true
end
).
prop_client_unsubscribe() ->
?ALL(
{ClientInfo, UnSubProps, TopicTab, Meta},
{clientinfo(), unsub_properties(), topictab(), request_meta()},
begin
ok = emqx_hooks:run('client.unsubscribe', [ClientInfo, UnSubProps, TopicTab]),
{'on_client_unsubscribe', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
props => properties(UnSubProps),
topic_filters => topicfilters(TopicTab),
clientinfo => from_clientinfo(ClientInfo),
meta => Meta
},
?assertEqual(Expected, Resp),
true
end
).
prop_session_created() ->
?ALL(
{ClientInfo, SessInfo, Meta},
{clientinfo(), sessioninfo(), request_meta()},
begin
ok = emqx_hooks:run('session.created', [ClientInfo, SessInfo]),
{'on_session_created', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
clientinfo => from_clientinfo(ClientInfo),
meta => Meta
},
?assertEqual(Expected, Resp),
true
end
).
prop_session_subscribed() ->
?ALL(
{ClientInfo, Topic, SubOpts, Meta},
{clientinfo(), topic(), subopts(), request_meta()},
begin
ok = emqx_hooks:run('session.subscribed', [ClientInfo, Topic, SubOpts]),
{'on_session_subscribed', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
topic => Topic,
subopts => subopts(SubOpts),
clientinfo => from_clientinfo(ClientInfo),
meta => Meta
},
?assertEqual(Expected, Resp),
true
end
).
prop_session_unsubscribed() ->
?ALL(
{ClientInfo, Topic, SubOpts, Meta},
{clientinfo(), topic(), subopts(), request_meta()},
begin
ok = emqx_hooks:run('session.unsubscribed', [ClientInfo, Topic, SubOpts]),
{'on_session_unsubscribed', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
topic => Topic,
clientinfo => from_clientinfo(ClientInfo),
meta => Meta
},
?assertEqual(Expected, Resp),
true
end
).
prop_session_resumed() ->
?ALL(
{ClientInfo, SessInfo, Meta},
{clientinfo(), sessioninfo(), request_meta()},
begin
ok = emqx_hooks:run('session.resumed', [ClientInfo, SessInfo]),
{'on_session_resumed', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
clientinfo => from_clientinfo(ClientInfo),
meta => Meta
},
?assertEqual(Expected, Resp),
true
end
).
prop_session_discared() ->
?ALL(
{ClientInfo, SessInfo, Meta},
{clientinfo(), sessioninfo(), request_meta()},
begin
ok = emqx_hooks:run('session.discarded', [ClientInfo, SessInfo]),
{'on_session_discarded', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{clientinfo => from_clientinfo(ClientInfo), meta => Meta},
?assertEqual(Expected, Resp),
true
end
).
prop_session_takenover() ->
?ALL(
{ClientInfo, SessInfo, Meta},
{clientinfo(), sessioninfo(), request_meta()},
begin
ok = emqx_hooks:run('session.takenover', [ClientInfo, SessInfo]),
{'on_session_takenover', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{clientinfo => from_clientinfo(ClientInfo), meta => Meta},
?assertEqual(Expected, Resp),
true
end
).
prop_session_terminated() ->
?ALL(
{ClientInfo, Reason, SessInfo, Meta},
{clientinfo(), shutdown_reason(), sessioninfo(), request_meta()},
begin
ok = emqx_hooks:run('session.terminated', [ClientInfo, Reason, SessInfo]),
{'on_session_terminated', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
reason => stringfy(Reason),
clientinfo => from_clientinfo(ClientInfo),
meta => Meta
},
?assertEqual(Expected, Resp),
true
end
).
prop_message_publish() ->
?ALL(
{Msg0, Meta},
{message(), request_meta()},
begin
Msg = emqx_message:from_map(
inject_magic_into(from, emqx_message:to_map(Msg0))
),
OutMsg = emqx_hooks:run_fold('message.publish', [], Msg),
case emqx_topic:match(emqx_message:topic(Msg), <<"$SYS/#">>) of
true ->
?assertEqual(Msg, OutMsg),
skip;
_ ->
ExpectedOutMsg =
case emqx_message:from(Msg) of
<<"baduser">> ->
MsgMap =
#{headers := Headers} =
emqx_message:to_map(Msg),
emqx_message:from_map(
MsgMap#{
qos => 0,
topic => <<"">>,
payload => <<"">>,
headers => maps:put(allow_publish, false, Headers)
}
);
<<"gooduser">> = From ->
MsgMap =
#{headers := Headers} =
emqx_message:to_map(Msg),
emqx_message:from_map(
MsgMap#{
topic => From,
payload => From,
headers => maps:put(allow_publish, true, Headers)
}
);
_ ->
Msg
end,
?assertEqual(ExpectedOutMsg, OutMsg),
{'on_message_publish', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
message => from_message(Msg),
meta => Meta
},
?assertEqual(Expected, Resp)
end,
true
end
).
prop_message_dropped() ->
?ALL(
{Msg, By, Reason, Meta},
{message(), hardcoded, shutdown_reason(), request_meta()},
begin
ok = emqx_hooks:run('message.dropped', [Msg, By, Reason]),
case emqx_topic:match(emqx_message:topic(Msg), <<"$SYS/#">>) of
true ->
skip;
_ ->
{'on_message_dropped', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
reason => stringfy(Reason),
message => from_message(Msg),
meta => Meta
},
?assertEqual(Expected, Resp)
end,
true
end
).
prop_message_delivered() ->
?ALL(
{ClientInfo, Msg, Meta},
{clientinfo(), message(), request_meta()},
begin
ok = emqx_hooks:run('message.delivered', [ClientInfo, Msg]),
case emqx_topic:match(emqx_message:topic(Msg), <<"$SYS/#">>) of
true ->
skip;
_ ->
{'on_message_delivered', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
clientinfo => from_clientinfo(ClientInfo),
message => from_message(Msg),
meta => Meta
},
?assertEqual(Expected, Resp)
end,
true
end
).
prop_message_acked() ->
?ALL(
{ClientInfo, Msg, Meta},
{clientinfo(), message(), request_meta()},
begin
ok = emqx_hooks:run('message.acked', [ClientInfo, Msg]),
case emqx_topic:match(emqx_message:topic(Msg), <<"$SYS/#">>) of
true ->
skip;
_ ->
{'on_message_acked', Resp} = emqx_exhook_demo_svr:take(),
Expected =
#{
clientinfo => from_clientinfo(ClientInfo),
message => from_message(Msg),
meta => Meta
},
?assertEqual(Expected, Resp)
end,
true
end
).
nodestr() ->
stringfy(node()).
peerhost(#{peername := {Host, _}}) ->
ntoa(Host).
peerport(#{peername := {_, Port}}) ->
Port.
sockport(#{sockname := {_, Port}}) ->
Port.
%% copied from emqx_exhook
ntoa({0, 0, 0, 0, 0, 16#ffff, AB, CD}) ->
list_to_binary(inet_parse:ntoa({AB bsr 8, AB rem 256, CD bsr 8, CD rem 256}));
ntoa(IP) ->
list_to_binary(inet_parse:ntoa(IP)).
option(undefined) -> <<>>;
option(B) -> B.
properties(undefined) ->
[];
properties(M) when is_map(M) ->
maps:fold(
fun(K, V, Acc) ->
[
#{
name => stringfy(K),
value => stringfy(V)
}
| Acc
]
end,
[],
M
).
topicfilters(Tfs) when is_list(Tfs) ->
[
#{name => emqx_topic:maybe_format_share(Topic), subopts => subopts(SubOpts)}
|| {Topic, SubOpts} <- Tfs
].
%% @private
stringfy(Term) when is_binary(Term) ->
Term;
stringfy(Term) when is_integer(Term) ->
integer_to_binary(Term);
stringfy(Term) when is_atom(Term) ->
atom_to_binary(Term, utf8);
stringfy(Term) when is_list(Term) ->
list_to_binary(Term);
stringfy(Term) ->
unicode:characters_to_binary((io_lib:format("~0p", [Term]))).
subopts(SubOpts) ->
#{
qos => maps:get(qos, SubOpts, 0),
rh => maps:get(rh, SubOpts, 0),
rap => maps:get(rap, SubOpts, 0),
nl => maps:get(nl, SubOpts, 0)
}.
authresult_to_bool(AuthResult) ->
AuthResult == ok.
aclresult_to_bool(#{result := Result}) ->
Result == allow.
pubsub_to_enum(?authz_action(publish)) -> 'PUBLISH';
pubsub_to_enum(?authz_action(subscribe)) -> 'SUBSCRIBE'.
from_conninfo(ConnInfo) ->
#{
node => nodestr(),
clientid => maps:get(clientid, ConnInfo),
username => option(maps:get(username, ConnInfo, <<>>)),
peerhost => peerhost(ConnInfo),
peerport => peerport(ConnInfo),
sockport => sockport(ConnInfo),
proto_name => maps:get(proto_name, ConnInfo),
proto_ver => stringfy(maps:get(proto_ver, ConnInfo)),
keepalive => maps:get(keepalive, ConnInfo)
}.
from_clientinfo(ClientInfo) ->
#{
node => nodestr(),
clientid => maps:get(clientid, ClientInfo),
username => option(maps:get(username, ClientInfo, <<>>)),
password => option(maps:get(password, ClientInfo, <<>>)),
peerhost => ntoa(maps:get(peerhost, ClientInfo)),
peerport => maps:get(peerport, ClientInfo),
sockport => maps:get(sockport, ClientInfo),
protocol => stringfy(maps:get(protocol, ClientInfo)),
mountpoint => option(maps:get(mountpoint, ClientInfo, <<>>)),
is_superuser => maps:get(is_superuser, ClientInfo, false),
anonymous => maps:get(anonymous, ClientInfo, true),
cn => option(maps:get(cn, ClientInfo, <<>>)),
dn => option(maps:get(dn, ClientInfo, <<>>))
}.
from_message(Msg) ->
#{
node => nodestr(),
id => emqx_guid:to_hexstr(emqx_message:id(Msg)),
qos => emqx_message:qos(Msg),
from => stringfy(emqx_message:from(Msg)),
topic => emqx_message:topic(Msg),
payload => emqx_message:payload(Msg),
timestamp => emqx_message:timestamp(Msg),
headers => emqx_exhook_handler:headers(
emqx_message:get_headers(Msg)
)
}.
%%--------------------------------------------------------------------
%% Helper
%%--------------------------------------------------------------------
do_setup() ->
logger:set_primary_config(#{level => warning}),
ok = ekka:start(),
application:set_env(ekka, cluster_name, ?DEFAULT_CLUSTER_NAME_ATOM),
_ = emqx_exhook_demo_svr:start(),
ok = emqx_config:init_load(emqx_exhook_schema, ?CONF_DEFAULT),
emqx_common_test_helpers:start_apps([emqx_exhook]),
%% waiting first loaded event
{'on_provider_loaded', _} = emqx_exhook_demo_svr:take(),
ok.
do_teardown(_) ->
emqx_common_test_helpers:stop_apps([emqx_exhook]),
%% waiting last unloaded event
{'on_provider_unloaded', _} = emqx_exhook_demo_svr:take(),
_ = emqx_exhook_demo_svr:stop(),
logger:set_primary_config(#{level => notice}),
timer:sleep(2000),
ok.
%%--------------------------------------------------------------------
%% Generators
%%--------------------------------------------------------------------
conn_properties() ->
#{}.
ack_properties() ->
#{}.
sub_properties() ->
#{}.
unsub_properties() ->
#{}.
shutdown_reason() ->
oneof([utf8(), {shutdown, emqx_proper_types:limited_latin_atom()}]).
authresult() ->
?LET(
RC,
connack_return_code(),
case RC of
success -> ok;
_ -> {error, RC}
end
).
inject_magic_into(Key, Object) ->
case castspell() of
muggles -> Object;
Spell -> Object#{Key => Spell}
end.
castspell() ->
L = [<<"baduser">>, <<"gooduser">>, <<"normaluser">>, muggles],
lists:nth(rand:uniform(length(L)), L).
request_meta() ->
#{
node => nodestr(),
version => stringfy(emqx_sys:version()),
sysdescr => stringfy(emqx_sys:sysdescr()),
cluster_name => ?DEFAULT_CLUSTER_NAME_BIN
}.