diff --git a/.ci/acl_migration_test/build.sh b/.ci/acl_migration_test/build.sh new file mode 100755 index 000000000..b7c779f15 --- /dev/null +++ b/.ci/acl_migration_test/build.sh @@ -0,0 +1,14 @@ +#!/bin/bash + +set -xe + +cd "$EMQX_PATH" + +rm -rf _build _upgrade_base + +mkdir _upgrade_base +pushd _upgrade_base + wget "https://s3-us-west-2.amazonaws.com/packages.emqx/emqx-ce/v${EMQX_BASE}/emqx-ubuntu20.04-${EMQX_BASE}-amd64.zip" +popd + +make emqx-zip diff --git a/.ci/acl_migration_test/prepare.sh b/.ci/acl_migration_test/prepare.sh new file mode 100755 index 000000000..07706867a --- /dev/null +++ b/.ci/acl_migration_test/prepare.sh @@ -0,0 +1,15 @@ +#!/bin/bash + +set -xe + +mkdir -p "$TEST_PATH" +cd "$TEST_PATH" + +cp ../"$EMQX_PATH"/_upgrade_base/*.zip ./ +unzip ./*.zip + +cp ../"$EMQX_PATH"/_packages/emqx/*.zip ./emqx/releases/ + +git clone --depth 1 https://github.com/terry-xiaoyu/one_more_emqx.git + +./one_more_emqx/one_more_emqx.sh emqx2 diff --git a/.ci/acl_migration_test/suite.sh b/.ci/acl_migration_test/suite.sh new file mode 100755 index 000000000..69c024c8d --- /dev/null +++ b/.ci/acl_migration_test/suite.sh @@ -0,0 +1,17 @@ +#!/bin/bash + +set -xe + +export EMQX_PATH="$1" +export EMQX_BASE="$2" + +export TEST_PATH="emqx_test" + +./build.sh + +VERSION=$("$EMQX_PATH"/pkg-vsn.sh) +export VERSION + +./prepare.sh + +./test.sh diff --git a/.ci/acl_migration_test/test.sh b/.ci/acl_migration_test/test.sh new file mode 100755 index 000000000..b214a0a52 --- /dev/null +++ b/.ci/acl_migration_test/test.sh @@ -0,0 +1,121 @@ +#!/bin/bash + +set -e + +EMQX_ENDPOINT="http://localhost:8081/api/v4/acl" +EMQX2_ENDPOINT="http://localhost:8917/api/v4/acl" + +function run() { + emqx="$1" + shift + + echo "[$emqx]" "$@" + + pushd "$TEST_PATH/$emqx" + "$@" + popd +} + +function post_rule() { + endpoint="$1" + rule="$2" + echo -n "->($endpoint) " + curl -s -u admin:public -X POST "$endpoint" -d "$rule" + echo +} + +function verify_clientid_rule() { + endpoint="$1" + id="$2" + echo -n "<-($endpoint) " + curl -s -u admin:public "$endpoint/clientid/$id" | grep "$id" || (echo "verify rule for client $id failed" && return 1) +} + +# Run nodes + +run emqx ./bin/emqx start +run emqx2 ./bin/emqx start + +run emqx ./bin/emqx_ctl plugins load emqx_auth_mnesia +run emqx2 ./bin/emqx_ctl plugins load emqx_auth_mnesia + +run emqx2 ./bin/emqx_ctl cluster join 'emqx@127.0.0.1' + +# Add ACL rule to unupgraded EMQX nodes + +post_rule "$EMQX_ENDPOINT" '{"clientid": "CLIENT1_A","topic": "t", "action": "pub", "access": "allow"}' +post_rule "$EMQX2_ENDPOINT" '{"clientid": "CLIENT1_B","topic": "t", "action": "pub", "access": "allow"}' + +# Upgrade emqx2 node + +run emqx2 ./bin/emqx install "$VERSION" +sleep 60 + +# Verify upgrade blocked + +run emqx2 ./bin/emqx eval 'emqx_acl_mnesia_migrator:is_old_table_migrated().' | grep false || (echo "emqx2 shouldn't have migrated" && exit 1) + +# Verify old rules on both nodes + +verify_clientid_rule "$EMQX_ENDPOINT" 'CLIENT1_A' +verify_clientid_rule "$EMQX2_ENDPOINT" 'CLIENT1_A' + +verify_clientid_rule "$EMQX_ENDPOINT" 'CLIENT1_B' +verify_clientid_rule "$EMQX2_ENDPOINT" 'CLIENT1_B' + +# Add ACL on OLD and NEW node, verify on all nodes + +post_rule "$EMQX_ENDPOINT" '{"clientid": "CLIENT2_A","topic": "t", "action": "pub", "access": "allow"}' +post_rule "$EMQX2_ENDPOINT" '{"clientid": "CLIENT2_B","topic": "t", "action": "pub", "access": "allow"}' + +verify_clientid_rule "$EMQX_ENDPOINT" 'CLIENT2_A' +verify_clientid_rule "$EMQX2_ENDPOINT" 'CLIENT2_A' + +verify_clientid_rule "$EMQX_ENDPOINT" 'CLIENT2_B' +verify_clientid_rule "$EMQX2_ENDPOINT" 'CLIENT2_B' + +# Upgrade emqx node + +run emqx ./bin/emqx install "$VERSION" + +# Wait for upgrade + +sleep 60 + +# Verify if upgrade occured + +run emqx ./bin/emqx eval 'emqx_acl_mnesia_migrator:is_old_table_migrated().' | grep true || (echo "emqx should have migrated" && exit 1) +run emqx2 ./bin/emqx eval 'emqx_acl_mnesia_migrator:is_old_table_migrated().' | grep true || (echo "emqx2 should have migrated" && exit 1) + +# Verify rules are kept + +verify_clientid_rule "$EMQX_ENDPOINT" 'CLIENT1_A' +verify_clientid_rule "$EMQX2_ENDPOINT" 'CLIENT1_A' + +verify_clientid_rule "$EMQX_ENDPOINT" 'CLIENT1_B' +verify_clientid_rule "$EMQX2_ENDPOINT" 'CLIENT1_B' + +verify_clientid_rule "$EMQX_ENDPOINT" 'CLIENT2_A' +verify_clientid_rule "$EMQX2_ENDPOINT" 'CLIENT2_A' + +verify_clientid_rule "$EMQX_ENDPOINT" 'CLIENT2_B' +verify_clientid_rule "$EMQX2_ENDPOINT" 'CLIENT2_B' + +# Add ACL on OLD and NEW node, verify on all nodes + +post_rule "$EMQX_ENDPOINT" '{"clientid": "CLIENT3_A","topic": "t", "action": "pub", "access": "allow"}' +post_rule "$EMQX2_ENDPOINT" '{"clientid": "CLIENT3_B","topic": "t", "action": "pub", "access": "allow"}' + +verify_clientid_rule "$EMQX_ENDPOINT" 'CLIENT3_A' +verify_clientid_rule "$EMQX2_ENDPOINT" 'CLIENT3_A' + +verify_clientid_rule "$EMQX_ENDPOINT" 'CLIENT3_B' +verify_clientid_rule "$EMQX2_ENDPOINT" 'CLIENT3_B' + +# Stop nodes + +run emqx ./bin/emqx stop +run emqx2 ./bin/emqx stop + +echo "Success!" + diff --git a/.ci/fvt_tests/relup.lux b/.ci/fvt_tests/relup.lux index 93889dad5..2940f5ce0 100644 --- a/.ci/fvt_tests/relup.lux +++ b/.ci/fvt_tests/relup.lux @@ -23,10 +23,7 @@ ?SH-PROMPT !cd emqx - !export EMQX_LOG__CONSOLE_HANDLER__ENABLE=true - !export EMQX_LOG__CONSOLE_HANDLER__LEVEL=debug - !export EMQX_LOG__PRIMARY_LEVEL=debug - !export EMQX_ZONES__DEFAULT__LISTENERS__MQTT_WSS__BIND="0.0.0.0:8085" + !export EMQX_LOG__LEVEL=debug !./bin/emqx start ?EMQ X .* is started successfully! @@ -39,9 +36,7 @@ ?SH-PROMPT !cd emqx2 - !export EMQX_LOG__CONSOLE_HANDLER__ENABLE=true - !export EMQX_LOG__CONSOLE_HANDLER__LEVEL=debug - !export EMQX_LOG__PRIMARY_LEVEL=debug + !export EMQX_LOG__LEVEL=debug !./bin/emqx start ?EMQ X .* is started successfully! diff --git a/.github/workflows/run_acl_migration_tests.yaml b/.github/workflows/run_acl_migration_tests.yaml new file mode 100644 index 000000000..855d9463c --- /dev/null +++ b/.github/workflows/run_acl_migration_tests.yaml @@ -0,0 +1,22 @@ +name: ACL fix & migration integration tests + +on: workflow_dispatch + +jobs: + test: + runs-on: ubuntu-20.04 + container: emqx/build-env:erl23.2.7.2-emqx-2-ubuntu20.04 + strategy: + fail-fast: true + env: + BASE_VERSION: "4.3.0" + steps: + - uses: actions/checkout@v2 + with: + path: emqx + - name: Prepare scripts + run: | + cp ./emqx/.ci/acl_migration_test/*.sh ./ + - name: Run tests + run: | + ./suite.sh emqx "$BASE_VERSION" diff --git a/.tool-versions b/.tool-versions index b87853803..200caa93b 100644 --- a/.tool-versions +++ b/.tool-versions @@ -1 +1 @@ -erlang 24.0.1-emqx-1 +erlang 23.2.7.2-emqx-2 diff --git a/apps/emqx_auth_mnesia/include/emqx_auth_mnesia.hrl b/apps/emqx_auth_mnesia/include/emqx_auth_mnesia.hrl index 034bd4f30..143f6b61e 100644 --- a/apps/emqx_auth_mnesia/include/emqx_auth_mnesia.hrl +++ b/apps/emqx_auth_mnesia/include/emqx_auth_mnesia.hrl @@ -1,21 +1,47 @@ -define(APP, emqx_auth_mnesia). --type(login():: {clientid, binary()} +-type(login() :: {clientid, binary()} | {username, binary()}). +-type(acl_target() :: login() | all). + +-type(acl_target_type() :: clientid | username | all). + +-type(access():: allow | deny). +-type(action():: pub | sub). +-type(legacy_action():: action() | pubsub). +-type(created_at():: integer()). + -record(emqx_user, { login :: login(), password :: binary(), - created_at :: integer() + created_at :: created_at() }). --record(emqx_acl, { - filter:: {login() | all, emqx_topic:topic()}, - action :: pub | sub | pubsub, - access :: allow | deny, - created_at :: integer() +-define(ACL_TABLE, emqx_acl). + +-define(MIGRATION_MARK_KEY, emqx_acl2_migration_started). + +-record(?ACL_TABLE, { + filter :: {acl_target(), emqx_topic:topic()} | ?MIGRATION_MARK_KEY, + action :: legacy_action(), + access :: access(), + created_at :: created_at() }). +-define(MIGRATION_MARK_RECORD, #?ACL_TABLE{filter = ?MIGRATION_MARK_KEY, action = pub, access = deny, created_at = 0}). + +-type(rule() :: {access(), action(), emqx_topic:topic(), created_at()}). + +-define(ACL_TABLE2, emqx_acl2). + +-record(?ACL_TABLE2, { + who :: acl_target(), + rules :: [ rule() ] + }). + +-type(acl_record() :: {acl_target(), emqx_topic:topic(), action(), access(), created_at()}). + -record(auth_metrics, { success = 'client.auth.success', failure = 'client.auth.failure', diff --git a/apps/emqx_auth_mnesia/src/emqx_acl_mnesia.erl b/apps/emqx_auth_mnesia/src/emqx_acl_mnesia.erl index c21955182..1e29d9121 100644 --- a/apps/emqx_auth_mnesia/src/emqx_acl_mnesia.erl +++ b/apps/emqx_auth_mnesia/src/emqx_acl_mnesia.erl @@ -18,24 +18,16 @@ -include("emqx_auth_mnesia.hrl"). --include_lib("stdlib/include/ms_transform.hrl"). - --define(TABLE, emqx_acl). - %% ACL Callbacks -export([ init/0 , register_metrics/0 , check_acl/5 , description/0 - ]). + ]). init() -> - ok = ekka_mnesia:create_table(emqx_acl, [ - {type, bag}, - {disc_copies, [node()]}, - {attributes, record_info(fields, emqx_acl)}, - {storage_properties, [{ets, [{read_concurrency, true}]}]}]), - ok = ekka_mnesia:copy_table(emqx_acl, disc_copies). + ok = emqx_acl_mnesia_db:create_table(), + ok = emqx_acl_mnesia_db:create_table2(). -spec(register_metrics() -> ok). register_metrics() -> @@ -46,12 +38,12 @@ check_acl(ClientInfo = #{ clientid := Clientid }, PubSub, Topic, _NoMatchAction, Acls = case Username of undefined -> - emqx_acl_mnesia_cli:lookup_acl({clientid, Clientid}) ++ - emqx_acl_mnesia_cli:lookup_acl(all); + emqx_acl_mnesia_db:lookup_acl({clientid, Clientid}) ++ + emqx_acl_mnesia_db:lookup_acl(all); _ -> - emqx_acl_mnesia_cli:lookup_acl({clientid, Clientid}) ++ - emqx_acl_mnesia_cli:lookup_acl({username, Username}) ++ - emqx_acl_mnesia_cli:lookup_acl(all) + emqx_acl_mnesia_db:lookup_acl({clientid, Clientid}) ++ + emqx_acl_mnesia_db:lookup_acl({username, Username}) ++ + emqx_acl_mnesia_db:lookup_acl(all) end, case match(ClientInfo, PubSub, Topic, Acls) of @@ -83,7 +75,6 @@ match(ClientInfo, PubSub, Topic, [ {_, ACLTopic, Action, Access, _} | Acls]) -> match_topic(ClientInfo, Topic, ACLTopic) when is_binary(Topic) -> emqx_topic:match(Topic, feed_var(ClientInfo, ACLTopic)). -match_actions(_, pubsub) -> true; match_actions(subscribe, sub) -> true; match_actions(publish, pub) -> true; match_actions(_, _) -> false. diff --git a/apps/emqx_auth_mnesia/src/emqx_acl_mnesia_api.erl b/apps/emqx_auth_mnesia/src/emqx_acl_mnesia_api.erl index fbd044d3f..10615b3e0 100644 --- a/apps/emqx_auth_mnesia/src/emqx_acl_mnesia_api.erl +++ b/apps/emqx_auth_mnesia/src/emqx_acl_mnesia_api.erl @@ -16,8 +16,6 @@ -module(emqx_acl_mnesia_api). --include("emqx_auth_mnesia.hrl"). - -include_lib("stdlib/include/ms_transform.hrl"). -import(proplists, [ get_value/2 @@ -99,26 +97,22 @@ ]). list_clientid(_Bindings, Params) -> - MatchSpec = ets:fun2ms( - fun({emqx_acl, {{clientid, Clientid}, Topic}, Action, Access, CreatedAt}) -> {{clientid,Clientid}, Topic, Action,Access, CreatedAt} end), - return({ok, emqx_auth_mnesia_api:paginate(emqx_acl, MatchSpec, Params, fun emqx_acl_mnesia_cli:comparing/2, fun format/1)}). + Table = emqx_acl_mnesia_db:login_acl_table(clientid), + return({ok, emqx_auth_mnesia_api:paginate_qh(Table, count(Table), Params, fun emqx_acl_mnesia_db:comparing/2, fun format/1)}). list_username(_Bindings, Params) -> - MatchSpec = ets:fun2ms( - fun({emqx_acl, {{username, Username}, Topic}, Action, Access, CreatedAt}) -> {{username, Username}, Topic, Action,Access, CreatedAt} end), - return({ok, emqx_auth_mnesia_api:paginate(emqx_acl, MatchSpec, Params, fun emqx_acl_mnesia_cli:comparing/2, fun format/1)}). + Table = emqx_acl_mnesia_db:login_acl_table(username), + return({ok, emqx_auth_mnesia_api:paginate_qh(Table, count(Table), Params, fun emqx_acl_mnesia_db:comparing/2, fun format/1)}). list_all(_Bindings, Params) -> - MatchSpec = ets:fun2ms( - fun({emqx_acl, {all, Topic}, Action, Access, CreatedAt}) -> {all, Topic, Action,Access, CreatedAt}end - ), - return({ok, emqx_auth_mnesia_api:paginate(emqx_acl, MatchSpec, Params, fun emqx_acl_mnesia_cli:comparing/2, fun format/1)}). + Table = emqx_acl_mnesia_db:login_acl_table(all), + return({ok, emqx_auth_mnesia_api:paginate_qh(Table, count(Table), Params, fun emqx_acl_mnesia_db:comparing/2, fun format/1)}). lookup(#{clientid := Clientid}, _Params) -> - return({ok, format(emqx_acl_mnesia_cli:lookup_acl({clientid, urldecode(Clientid)}))}); + return({ok, format(emqx_acl_mnesia_db:lookup_acl({clientid, urldecode(Clientid)}))}); lookup(#{username := Username}, _Params) -> - return({ok, format(emqx_acl_mnesia_cli:lookup_acl({username, urldecode(Username)}))}). + return({ok, format(emqx_acl_mnesia_db:lookup_acl({username, urldecode(Username)}))}). add(_Bindings, Params) -> [ P | _] = Params, @@ -152,7 +146,7 @@ do_add(Params) -> Access = get_value(<<"access">>, Params), Re = case validate([login, topic, action, access], [Login, Topic, Action, Access]) of ok -> - emqx_acl_mnesia_cli:add_acl(Login, Topic, erlang:binary_to_atom(Action, utf8), erlang:binary_to_atom(Access, utf8)); + emqx_acl_mnesia_db:add_acl(Login, Topic, erlang:binary_to_atom(Action, utf8), erlang:binary_to_atom(Access, utf8)); Err -> Err end, maps:merge(#{topic => Topic, @@ -165,15 +159,19 @@ do_add(Params) -> end). delete(#{clientid := Clientid, topic := Topic}, _) -> - return(emqx_acl_mnesia_cli:remove_acl({clientid, urldecode(Clientid)}, urldecode(Topic))); + return(emqx_acl_mnesia_db:remove_acl({clientid, urldecode(Clientid)}, urldecode(Topic))); delete(#{username := Username, topic := Topic}, _) -> - return(emqx_acl_mnesia_cli:remove_acl({username, urldecode(Username)}, urldecode(Topic))); + return(emqx_acl_mnesia_db:remove_acl({username, urldecode(Username)}, urldecode(Topic))); delete(#{topic := Topic}, _) -> - return(emqx_acl_mnesia_cli:remove_acl(all, urldecode(Topic))). + return(emqx_acl_mnesia_db:remove_acl(all, urldecode(Topic))). %%------------------------------------------------------------------------------ %% Interval Funcs %%------------------------------------------------------------------------------ + +count(QH) -> + qlc:fold(fun(_, Count) -> Count + 1 end, 0, QH). + format({{clientid, Clientid}, Topic, Action, Access, _CreatedAt}) -> #{clientid => Clientid, topic => Topic, action => Action, access => Access}; format({{username, Username}, Topic, Action, Access, _CreatedAt}) -> diff --git a/apps/emqx_auth_mnesia/src/emqx_acl_mnesia_cli.erl b/apps/emqx_auth_mnesia/src/emqx_acl_mnesia_cli.erl index 302a81637..145f0ede8 100644 --- a/apps/emqx_auth_mnesia/src/emqx_acl_mnesia_cli.erl +++ b/apps/emqx_auth_mnesia/src/emqx_acl_mnesia_cli.erl @@ -16,110 +16,28 @@ -module(emqx_acl_mnesia_cli). --include("emqx_auth_mnesia.hrl"). --include_lib("emqx/include/logger.hrl"). --include_lib("stdlib/include/ms_transform.hrl"). --define(TABLE, emqx_acl). - -%% Acl APIs --export([ add_acl/4 - , lookup_acl/1 - , all_acls/0 - , all_acls/1 - , remove_acl/2 - ]). - -export([cli/1]). --export([comparing/2]). -%%-------------------------------------------------------------------- -%% Acl API -%%-------------------------------------------------------------------- - -%% @doc Add Acls --spec(add_acl(login() | all, emqx_topic:topic(), pub | sub | pubsub, allow | deny) -> - ok | {error, any()}). -add_acl(Login, Topic, Action, Access) -> - Filter = {Login, Topic}, - Acl = #?TABLE{ - filter = Filter, - action = Action, - access = Access, - created_at = erlang:system_time(millisecond) - }, - ret(mnesia:transaction( - fun() -> - OldRecords = mnesia:wread({?TABLE, Filter}), - case Action of - pubsub -> - update_permission(pub, Acl, OldRecords), - update_permission(sub, Acl, OldRecords); - _ -> - update_permission(Action, Acl, OldRecords) - end - end)). - -%% @doc Lookup acl by login --spec(lookup_acl(login() | all) -> list()). -lookup_acl(undefined) -> []; -lookup_acl(Login) -> - MatchSpec = ets:fun2ms(fun({?TABLE, {Filter, ACLTopic}, Action, Access, CreatedAt}) - when Filter =:= Login -> - {Filter, ACLTopic, Action, Access, CreatedAt} - end), - lists:sort(fun comparing/2, ets:select(?TABLE, MatchSpec)). - -%% @doc Remove acl --spec(remove_acl(login() | all, emqx_topic:topic()) -> ok | {error, any()}). -remove_acl(Login, Topic) -> - ret(mnesia:transaction(fun mnesia:delete/1, [{?TABLE, {Login, Topic}}])). - -%% @doc All logins --spec(all_acls() -> list()). -all_acls() -> - all_acls(clientid) ++ - all_acls(username) ++ - all_acls(all). - -all_acls(clientid) -> - MatchSpec = ets:fun2ms( - fun({?TABLE, {{clientid, Clientid}, Topic}, Action, Access, CreatedAt}) -> - {{clientid, Clientid}, Topic, Action, Access, CreatedAt} - end), - lists:sort(fun comparing/2, ets:select(?TABLE, MatchSpec)); -all_acls(username) -> - MatchSpec = ets:fun2ms( - fun({?TABLE, {{username, Username}, Topic}, Action, Access, CreatedAt}) -> - {{username, Username}, Topic, Action, Access, CreatedAt} - end), - lists:sort(fun comparing/2, ets:select(?TABLE, MatchSpec)); -all_acls(all) -> - MatchSpec = ets:fun2ms( - fun({?TABLE, {all, Topic}, Action, Access, CreatedAt}) -> - {all, Topic, Action, Access, CreatedAt} - end - ), - lists:sort(fun comparing/2, ets:select(?TABLE, MatchSpec)). %%-------------------------------------------------------------------- %% ACL Cli %%-------------------------------------------------------------------- cli(["list"]) -> - [print_acl(Acl) || Acl <- all_acls()]; + [print_acl(Acl) || Acl <- emqx_acl_mnesia_db:all_acls()]; cli(["list", "clientid"]) -> - [print_acl(Acl) || Acl <- all_acls(clientid)]; + [print_acl(Acl) || Acl <- emqx_acl_mnesia_db:all_acls(clientid)]; cli(["list", "username"]) -> - [print_acl(Acl) || Acl <- all_acls(username)]; + [print_acl(Acl) || Acl <- emqx_acl_mnesia_db:all_acls(username)]; cli(["list", "_all"]) -> - [print_acl(Acl) || Acl <- all_acls(all)]; + [print_acl(Acl) || Acl <- emqx_acl_mnesia_db:all_acls(all)]; cli(["add", "clientid", Clientid, Topic, Action, Access]) -> case validate(action, Action) andalso validate(access, Access) of true -> - case add_acl( + case emqx_acl_mnesia_db:add_acl( {clientid, iolist_to_binary(Clientid)}, iolist_to_binary(Topic), list_to_existing_atom(Action), @@ -135,7 +53,7 @@ cli(["add", "clientid", Clientid, Topic, Action, Access]) -> cli(["add", "username", Username, Topic, Action, Access]) -> case validate(action, Action) andalso validate(access, Access) of true -> - case add_acl( + case emqx_acl_mnesia_db:add_acl( {username, iolist_to_binary(Username)}, iolist_to_binary(Topic), list_to_existing_atom(Action), @@ -151,7 +69,7 @@ cli(["add", "username", Username, Topic, Action, Access]) -> cli(["add", "_all", Topic, Action, Access]) -> case validate(action, Action) andalso validate(access, Access) of true -> - case add_acl( + case emqx_acl_mnesia_db:add_acl( all, iolist_to_binary(Topic), list_to_existing_atom(Action), @@ -165,16 +83,16 @@ cli(["add", "_all", Topic, Action, Access]) -> end; cli(["show", "clientid", Clientid]) -> - [print_acl(Acl) || Acl <- lookup_acl({clientid, iolist_to_binary(Clientid)})]; + [print_acl(Acl) || Acl <- emqx_acl_mnesia_db:lookup_acl({clientid, iolist_to_binary(Clientid)})]; cli(["show", "username", Username]) -> - [print_acl(Acl) || Acl <- lookup_acl({username, iolist_to_binary(Username)})]; + [print_acl(Acl) || Acl <- emqx_acl_mnesia_db:lookup_acl({username, iolist_to_binary(Username)})]; cli(["del", "clientid", Clientid, Topic])-> cli(["delete", "clientid", Clientid, Topic]); cli(["delete", "clientid", Clientid, Topic])-> - case remove_acl({clientid, iolist_to_binary(Clientid)}, iolist_to_binary(Topic)) of + case emqx_acl_mnesia_db:remove_acl({clientid, iolist_to_binary(Clientid)}, iolist_to_binary(Topic)) of ok -> emqx_ctl:print("ok~n"); {error, Reason} -> emqx_ctl:print("Error: ~p~n", [Reason]) end; @@ -183,7 +101,7 @@ cli(["del", "username", Username, Topic])-> cli(["delete", "username", Username, Topic]); cli(["delete", "username", Username, Topic])-> - case remove_acl({username, iolist_to_binary(Username)}, iolist_to_binary(Topic)) of + case emqx_acl_mnesia_db:remove_acl({username, iolist_to_binary(Username)}, iolist_to_binary(Topic)) of ok -> emqx_ctl:print("ok~n"); {error, Reason} -> emqx_ctl:print("Error: ~p~n", [Reason]) end; @@ -192,7 +110,7 @@ cli(["del", "_all", Topic])-> cli(["delete", "_all", Topic]); cli(["delete", "_all", Topic])-> - case remove_acl(all, iolist_to_binary(Topic)) of + case emqx_acl_mnesia_db:remove_acl(all, iolist_to_binary(Topic)) of ok -> emqx_ctl:print("ok~n"); {error, Reason} -> emqx_ctl:print("Error: ~p~n", [Reason]) end; @@ -215,13 +133,6 @@ cli(_) -> %% Internal functions %%-------------------------------------------------------------------- -comparing({_, _, _, _, CreatedAt1}, - {_, _, _, _, CreatedAt2}) -> - CreatedAt1 >= CreatedAt2. - -ret({atomic, ok}) -> ok; -ret({aborted, Error}) -> {error, Error}. - validate(action, "pub") -> true; validate(action, "sub") -> true; validate(action, "pubsub") -> true; @@ -244,27 +155,3 @@ print_acl({all, Topic, Action, Access, _}) -> "Acl($all topic = ~p action = ~p access = ~p)~n", [Topic, Action, Access] ). - -update_permission(Action, Acl0, OldRecords) -> - Acl = Acl0 #?TABLE{action = Action}, - maybe_delete_shadowed_records(Action, OldRecords), - mnesia:write(Acl). - -maybe_delete_shadowed_records(_, []) -> - ok; -maybe_delete_shadowed_records(Action1, [Rec = #emqx_acl{action = Action2} | Rest]) -> - if Action1 =:= Action2 -> - ok = mnesia:delete_object(Rec); - Action2 =:= pubsub -> - %% Perform migration from the old data format on the - %% fly. This is needed only for the enterprise version, - %% delete this branch on 5.0 - mnesia:delete_object(Rec), - mnesia:write(Rec#?TABLE{action = other_action(Action1)}); - true -> - ok - end, - maybe_delete_shadowed_records(Action1, Rest). - -other_action(pub) -> sub; -other_action(sub) -> pub. diff --git a/apps/emqx_auth_mnesia/src/emqx_acl_mnesia_db.erl b/apps/emqx_auth_mnesia/src/emqx_acl_mnesia_db.erl new file mode 100644 index 000000000..b483e59df --- /dev/null +++ b/apps/emqx_auth_mnesia/src/emqx_acl_mnesia_db.erl @@ -0,0 +1,339 @@ +%%-------------------------------------------------------------------- +%% Copyright (c) 2020-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_acl_mnesia_db). + +-include("emqx_auth_mnesia.hrl"). +-include_lib("stdlib/include/ms_transform.hrl"). +-include_lib("stdlib/include/qlc.hrl"). + +%% ACL APIs +-export([ create_table/0 + , create_table2/0 + ]). + +-export([ add_acl/4 + , lookup_acl/1 + , all_acls_export/0 + , all_acls/0 + , all_acls/1 + , remove_acl/2 + , merge_acl_records/3 + , login_acl_table/1 + , is_migration_started/0 + ]). + +-export([comparing/2]). + +%%-------------------------------------------------------------------- +%% ACL API +%%-------------------------------------------------------------------- + +%% @doc Create table `emqx_acl` of old format rules +-spec(create_table() -> ok). +create_table() -> + ok = ekka_mnesia:create_table(?ACL_TABLE, [ + {type, bag}, + {disc_copies, [node()]}, + {attributes, record_info(fields, ?ACL_TABLE)}, + {storage_properties, [{ets, [{read_concurrency, true}]}]}]), + ok = ekka_mnesia:copy_table(?ACL_TABLE, disc_copies). + +%% @doc Create table `emqx_acl2` of new format rules +-spec(create_table2() -> ok). +create_table2() -> + ok = ekka_mnesia:create_table(?ACL_TABLE2, [ + {type, ordered_set}, + {disc_copies, [node()]}, + {attributes, record_info(fields, ?ACL_TABLE2)}, + {storage_properties, [{ets, [{read_concurrency, true}]}]}]), + ok = ekka_mnesia:copy_table(?ACL_TABLE2, disc_copies). + +%% @doc Add Acls +-spec(add_acl(acl_target(), emqx_topic:topic(), legacy_action(), access()) -> + ok | {error, any()}). +add_acl(Login, Topic, Action, Access) -> + ret(mnesia:transaction(fun() -> + case is_migration_started() of + true -> add_acl_new(Login, Topic, Action, Access); + false -> add_acl_old(Login, Topic, Action, Access) + end + end)). + +%% @doc Lookup acl by login +-spec(lookup_acl(acl_target()) -> list(acl_record())). +lookup_acl(undefined) -> []; +lookup_acl(Login) -> + % After migration to ?ACL_TABLE2, ?ACL_TABLE never has any rules. This lookup should be removed later. + MatchSpec = ets:fun2ms(fun(#?ACL_TABLE{filter = {Filter, _}} = Rec) + when Filter =:= Login -> Rec + end), + OldRecs = ets:select(?ACL_TABLE, MatchSpec), + + NewAcls = ets:lookup(?ACL_TABLE2, Login), + MergedAcl = merge_acl_records(Login, OldRecs, NewAcls), + lists:sort(fun comparing/2, acl_to_list(MergedAcl)). + +%% @doc Remove ACL +-spec remove_acl(acl_target(), emqx_topic:topic()) -> ok | {error, any()}. +remove_acl(Login, Topic) -> + ret(mnesia:transaction(fun() -> + mnesia:delete({?ACL_TABLE, {Login, Topic}}), + case mnesia:wread({?ACL_TABLE2, Login}) of + [] -> ok; + [#?ACL_TABLE2{rules = Rules} = Acl] -> + case delete_topic_rules(Topic, Rules) of + [] -> mnesia:delete({?ACL_TABLE2, Login}); + [_ | _] = RemainingRules -> + mnesia:write(Acl#?ACL_TABLE2{rules = RemainingRules}) + end + end + end)). + +%% @doc All ACL rules +-spec(all_acls() -> list(acl_record())). +all_acls() -> + all_acls(username) ++ + all_acls(clientid) ++ + all_acls(all). + +%% @doc All ACL rules of specified type +-spec(all_acls(acl_target_type()) -> list(acl_record())). +all_acls(AclTargetType) -> + lists:sort(fun comparing/2, qlc:eval(login_acl_table(AclTargetType))). + +%% @doc All ACL rules fetched transactionally +-spec(all_acls_export() -> list(acl_record())). +all_acls_export() -> + AclTargetTypes = [username, clientid, all], + MatchSpecNew = lists:flatmap(fun login_match_spec_new/1, AclTargetTypes), + MatchSpecOld = lists:flatmap(fun login_match_spec_old/1, AclTargetTypes), + + {atomic, Records} = mnesia:transaction( + fun() -> + QH = acl_table(MatchSpecNew, MatchSpecOld, fun mnesia:table/2, fun lookup_mnesia/2), + qlc:eval(QH) + end), + Records. + +%% @doc QLC table of logins matching spec +-spec(login_acl_table(acl_target_type()) -> qlc:query_handle()). +login_acl_table(AclTargetType) -> + MatchSpecNew = login_match_spec_new(AclTargetType), + MatchSpecOld = login_match_spec_old(AclTargetType), + acl_table(MatchSpecNew, MatchSpecOld, fun ets:table/2, fun lookup_ets/2). + +%% @doc Combine old `emqx_acl` ACL records with a new `emqx_acl2` ACL record for a given login +-spec(merge_acl_records(acl_target(), [#?ACL_TABLE{}], [#?ACL_TABLE2{}]) -> #?ACL_TABLE2{}). +merge_acl_records(Login, OldRecs, Acls) -> + OldRules = old_recs_to_rules(OldRecs), + NewRules = case Acls of + [] -> []; + [#?ACL_TABLE2{rules = Rules}] -> Rules + end, + #?ACL_TABLE2{who = Login, rules = merge_rules(NewRules, OldRules)}. + +%% @doc Checks if background migration of ACL rules from `emqx_acl` to `emqx_acl2` format started. +%% Should be run in transaction +-spec(is_migration_started() -> boolean()). +is_migration_started() -> + case mnesia:read({?ACL_TABLE, ?MIGRATION_MARK_KEY}) of + [?MIGRATION_MARK_RECORD | _] -> true; + [] -> false + end. + +%%-------------------------------------------------------------------- +%% Internal functions +%%-------------------------------------------------------------------- + +add_acl_new(Login, Topic, Action, Access) -> + Rule = {Access, Action, Topic, erlang:system_time(millisecond)}, + Rules = normalize_rule(Rule), + OldAcl = mnesia:wread({?ACL_TABLE2, Login}), + NewAcl = case OldAcl of + [#?ACL_TABLE2{rules = OldRules} = Acl] -> + Acl#?ACL_TABLE2{rules = merge_rules(Rules, OldRules)}; + [] -> + #?ACL_TABLE2{who = Login, rules = Rules} + end, + mnesia:write(NewAcl). + +add_acl_old(Login, Topic, Action, Access) -> + Filter = {Login, Topic}, + Acl = #?ACL_TABLE{ + filter = Filter, + action = Action, + access = Access, + created_at = erlang:system_time(millisecond) + }, + OldRecords = mnesia:wread({?ACL_TABLE, Filter}), + case Action of + pubsub -> + update_permission(pub, Acl, OldRecords), + update_permission(sub, Acl, OldRecords); + _ -> + update_permission(Action, Acl, OldRecords) + end. + +old_recs_to_rules(OldRecs) -> + lists:flatmap(fun old_rec_to_rules/1, OldRecs). + +old_rec_to_rules(#?ACL_TABLE{filter = {_, Topic}, action = Action, access = Access, created_at = CreatedAt}) -> + normalize_rule({Access, Action, Topic, CreatedAt}). + +normalize_rule({Access, pubsub, Topic, CreatedAt}) -> + [{Access, pub, Topic, CreatedAt}, {Access, sub, Topic, CreatedAt}]; +normalize_rule({Access, Action, Topic, CreatedAt}) -> + [{Access, Action, Topic, CreatedAt}]. + +merge_rules([], OldRules) -> OldRules; +merge_rules([NewRule | RestNewRules], OldRules) -> + merge_rules(RestNewRules, merge_rule(NewRule, OldRules)). + +merge_rule({_, Action, Topic, _ } = NewRule, OldRules) -> + [NewRule | lists:filter( + fun({_, OldAction, OldTopic, _}) -> + {Action, Topic} =/= {OldAction, OldTopic} + end, OldRules)]. + +acl_to_list(#?ACL_TABLE2{who = Login, rules = Rules}) -> + [{Login, Topic, Action, Access, CreatedAt} || {Access, Action, Topic, CreatedAt} <- Rules]. + +delete_topic_rules(Topic, Rules) -> + [Rule || {_, _, T, _} = Rule <- Rules, T =/= Topic]. + +comparing({_, _, _, _, CreatedAt} = Rec1, + {_, _, _, _, CreatedAt} = Rec2) -> + Rec1 >= Rec2; + +comparing({_, _, _, _, CreatedAt1}, + {_, _, _, _, CreatedAt2}) -> + CreatedAt1 >= CreatedAt2. + +login_match_spec_old(all) -> + ets:fun2ms(fun(#?ACL_TABLE{filter = {all, _}} = Record) -> + Record + end); + +login_match_spec_old(Type) when (Type =:= username) or (Type =:= clientid) -> + ets:fun2ms(fun(#?ACL_TABLE{filter = {{RecordType, _}, _}} = Record) + when RecordType =:= Type -> Record + end). + +login_match_spec_new(all) -> + ets:fun2ms(fun(#?ACL_TABLE2{who = all} = Record) -> + Record + end); + +login_match_spec_new(Type) when (Type =:= username) or (Type =:= clientid) -> + ets:fun2ms(fun(#?ACL_TABLE2{who = {RecordType, _}} = Record) + when RecordType =:= Type -> Record + end). + +acl_table(MatchSpecNew, MatchSpecOld, TableFun, LookupFun) -> + TraverseFun = + fun() -> + CursorNew = + qlc:cursor( + TableFun(?ACL_TABLE2, [{traverse, {select, MatchSpecNew}}])), + CursorOld = + qlc:cursor( + TableFun(?ACL_TABLE, [{traverse, {select, MatchSpecOld}}])), + traverse_new(CursorNew, CursorOld, #{}, LookupFun) + end, + + qlc:table(TraverseFun, []). + + +% These are traverse funs for qlc table created by `acl_table/4`. +% Traversing consumes memory: it collects logins present in `?ACL_TABLE` and +% at the same time having rules in `?ACL_TABLE2`. +% Such records appear if ACLs are inserted before migration started. +% After migration, number of such logins is zero, so traversing starts working in +% constant memory. + +traverse_new(CursorNew, CursorOld, FoundKeys, LookupFun) -> + Acls = qlc:next_answers(CursorNew, 1), + case Acls of + [] -> + qlc:delete_cursor(CursorNew), + traverse_old(CursorOld, FoundKeys); + [#?ACL_TABLE2{who = Login, rules = Rules} = Acl] -> + Keys = lists:usort([{Login, Topic} || {_, _, Topic, _} <- Rules]), + OldRecs = lists:flatmap(fun(Key) -> LookupFun(?ACL_TABLE, Key) end, Keys), + MergedAcl = merge_acl_records(Login, OldRecs, [Acl]), + NewFoundKeys = + lists:foldl(fun(#?ACL_TABLE{filter = Key}, Found) -> maps:put(Key, true, Found) end, + FoundKeys, + OldRecs), + case acl_to_list(MergedAcl) of + [] -> + traverse_new(CursorNew, CursorOld, NewFoundKeys, LookupFun); + List -> + List ++ fun() -> traverse_new(CursorNew, CursorOld, NewFoundKeys, LookupFun) end + end + end. + +traverse_old(CursorOld, FoundKeys) -> + OldAcls = qlc:next_answers(CursorOld), + case OldAcls of + [] -> + qlc:delete_cursor(CursorOld), + []; + _ -> + Records = [ {Login, Topic, Action, Access, CreatedAt} + || #?ACL_TABLE{filter = {Login, Topic}, action = LegacyAction, access = Access, created_at = CreatedAt} <- OldAcls, + {_, Action, _, _} <- normalize_rule({Access, LegacyAction, Topic, CreatedAt}), + not maps:is_key({Login, Topic}, FoundKeys) + ], + case Records of + [] -> traverse_old(CursorOld, FoundKeys); + List -> List ++ fun() -> traverse_old(CursorOld, FoundKeys) end + end + end. + +lookup_mnesia(Tab, Key) -> + mnesia:read({Tab, Key}). + +lookup_ets(Tab, Key) -> + ets:lookup(Tab, Key). + +update_permission(Action, Acl0, OldRecords) -> + Acl = Acl0 #?ACL_TABLE{action = Action}, + maybe_delete_shadowed_records(Action, OldRecords), + mnesia:write(Acl). + +maybe_delete_shadowed_records(_, []) -> + ok; +maybe_delete_shadowed_records(Action1, [Rec = #emqx_acl{action = Action2} | Rest]) -> + if Action1 =:= Action2 -> + ok = mnesia:delete_object(Rec); + Action2 =:= pubsub -> + %% Perform migration from the old data format on the + %% fly. This is needed only for the enterprise version, + %% delete this branch on 5.0 + mnesia:delete_object(Rec), + mnesia:write(Rec#?ACL_TABLE{action = other_action(Action1)}); + true -> + ok + end, + maybe_delete_shadowed_records(Action1, Rest). + +other_action(pub) -> sub; +other_action(sub) -> pub. + +ret({atomic, ok}) -> ok; +ret({aborted, Error}) -> {error, Error}. diff --git a/apps/emqx_auth_mnesia/src/emqx_acl_mnesia_migrator.erl b/apps/emqx_auth_mnesia/src/emqx_acl_mnesia_migrator.erl new file mode 100644 index 000000000..864f00884 --- /dev/null +++ b/apps/emqx_auth_mnesia/src/emqx_acl_mnesia_migrator.erl @@ -0,0 +1,215 @@ +%%-------------------------------------------------------------------- +%% Copyright (c) 2020-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_acl_mnesia_migrator). + +-include("emqx_auth_mnesia.hrl"). +-include_lib("emqx/include/logger.hrl"). +-include_lib("snabbkaffe/include/snabbkaffe.hrl"). + +-behaviour(gen_statem). + +-define(CHECK_ALL_NODES_INTERVAL, 60000). + +-type(migration_delay_reason() :: old_nodes | bad_nodes). + +-export([ + callback_mode/0, + init/1 +]). + +-export([ + waiting_all_nodes/3, + checking_old_table/3, + migrating/3 +]). + +-export([ + start_link/0, + start_link/1, + start_supervised/0, + stop_supervised/0, + migrate_records/0, + is_migrating_on_node/1, + is_old_table_migrated/0 +]). + +%%-------------------------------------------------------------------- +%% External interface +%%-------------------------------------------------------------------- + +start_link() -> + start_link(?MODULE). + +start_link(Name) when is_atom(Name) -> + start_link(#{ + name => Name + }); + +start_link(#{name := Name} = Opts) -> + gen_statem:start_link({local, Name}, ?MODULE, Opts, []). + +start_supervised() -> + try + {ok, _} = supervisor:restart_child(emqx_auth_mnesia_sup, ?MODULE), + ok + catch + exit:{noproc, _} -> ok + end. + +stop_supervised() -> + try + ok = supervisor:terminate_child(emqx_auth_mnesia_sup, ?MODULE), + ok = supervisor:delete_child(emqx_auth_mnesia_sup, ?MODULE) + catch + exit:{noproc, _} -> ok + end. + +%%-------------------------------------------------------------------- +%% gen_statem callbacks +%%-------------------------------------------------------------------- + +callback_mode() -> state_functions. + +init(Opts) -> + ok = emqx_acl_mnesia_db:create_table(), + ok = emqx_acl_mnesia_db:create_table2(), + Name = maps:get(name, Opts, ?MODULE), + CheckNodesInterval = maps:get(check_nodes_interval, Opts, ?CHECK_ALL_NODES_INTERVAL), + GetNodes = maps:get(get_nodes, Opts, fun all_nodes/0), + Data = + #{name => Name, + check_nodes_interval => CheckNodesInterval, + get_nodes => GetNodes}, + {ok, waiting_all_nodes, Data, [{state_timeout, 0, check_nodes}]}. + +%%-------------------------------------------------------------------- +%% state callbacks +%%-------------------------------------------------------------------- + +waiting_all_nodes(state_timeout, check_nodes, Data) -> + #{name := Name, check_nodes_interval := CheckNodesInterval, get_nodes := GetNodes} = Data, + case is_all_nodes_migrating(Name, GetNodes()) of + true -> + ?tp(info, emqx_acl_mnesia_migrator_check_old_table, #{}), + {next_state, checking_old_table, Data, [{next_event, internal, check_old_table}]}; + {false, Reason, Nodes} -> + ?tp(info, + emqx_acl_mnesia_migrator_bad_nodes_delay, + #{delay => CheckNodesInterval, + reason => Reason, + name => Name, + nodes => Nodes}), + {keep_state_and_data, [{state_timeout, CheckNodesInterval, check_nodes}]} + end. + +checking_old_table(internal, check_old_table, Data) -> + case is_old_table_migrated() of + true -> + ?tp(info, emqx_acl_mnesia_migrator_finish, #{}), + {next_state, finished, Data, [{hibernate, true}]}; + false -> + ?tp(info, emqx_acl_mnesia_migrator_start_migration, #{}), + {next_state, migrating, Data, [{next_event, internal, start_migration}]} + end. + +migrating(internal, start_migration, Data) -> + ok = migrate_records(), + {next_state, checking_old_table, Data, [{next_event, internal, check_old_table}]}. + +%% @doc Returns `true` if migration is started in the local node, otherwise crash. +-spec(is_migrating_on_node(atom()) -> true). +is_migrating_on_node(Name) -> + true = is_pid(erlang:whereis(Name)). + +%% @doc Run migration of records +-spec(migrate_records() -> ok). +migrate_records() -> + ok = add_migration_mark(), + Key = peek_record(), + do_migrate_records(Key). + +%% @doc Run migration of records +-spec(is_all_nodes_migrating(atom(), list(node())) -> true | {false, migration_delay_reason(), list(node())}). +is_all_nodes_migrating(Name, Nodes) -> + case rpc:multicall(Nodes, ?MODULE, is_migrating_on_node, [Name]) of + {Results, []} -> + OldNodes = [ Node || {Node, Result} <- lists:zip(Nodes, Results), Result =/= true ], + case OldNodes of + [] -> true; + _ -> {false, old_nodes, OldNodes} + end; + {_, [_BadNode | _] = BadNodes} -> + {false, bad_nodes, BadNodes} + end. + +%%-------------------------------------------------------------------- +%% Internal functions +%%-------------------------------------------------------------------- + +all_nodes() -> + ekka_mnesia:cluster_nodes(all). + +is_old_table_migrated() -> + Result = + mnesia:transaction(fun() -> + case mnesia:first(?ACL_TABLE) of + ?MIGRATION_MARK_KEY -> + case mnesia:next(?ACL_TABLE, ?MIGRATION_MARK_KEY) of + '$end_of_table' -> true; + _OtherKey -> false + end; + '$end_of_table' -> false; + _OtherKey -> false + end + end), + case Result of + {atomic, true} -> + true; + _ -> + false + end. + +add_migration_mark() -> + {atomic, ok} = mnesia:transaction(fun() -> mnesia:write(?MIGRATION_MARK_RECORD) end), + ok. + +peek_record() -> + Key = mnesia:dirty_first(?ACL_TABLE), + case Key of + ?MIGRATION_MARK_KEY -> + mnesia:dirty_next(?ACL_TABLE, Key); + _ -> Key + end. + +do_migrate_records('$end_of_table') -> ok; +do_migrate_records({_Login, _Topic} = Key) -> + ?tp(emqx_acl_mnesia_migrator_record_selected, #{key => Key}), + _ = mnesia:transaction(fun migrate_one_record/1, [Key]), + do_migrate_records(peek_record()). + +migrate_one_record({Login, _Topic} = Key) -> + case mnesia:wread({?ACL_TABLE, Key}) of + [] -> + ?tp(emqx_acl_mnesia_migrator_record_missed, #{key => Key}), + record_missing; + OldRecs -> + Acls = mnesia:wread({?ACL_TABLE2, Login}), + UpdatedAcl = emqx_acl_mnesia_db:merge_acl_records(Login, OldRecs, Acls), + ok = mnesia:write(UpdatedAcl), + ok = mnesia:delete({?ACL_TABLE, Key}), + ?tp(emqx_acl_mnesia_migrator_record_migrated, #{key => Key}) + end. diff --git a/apps/emqx_auth_mnesia/src/emqx_auth_mnesia.app.src b/apps/emqx_auth_mnesia/src/emqx_auth_mnesia.app.src index e61a22f0a..b15c7fdd3 100644 --- a/apps/emqx_auth_mnesia/src/emqx_auth_mnesia.app.src +++ b/apps/emqx_auth_mnesia/src/emqx_auth_mnesia.app.src @@ -1,6 +1,6 @@ {application, emqx_auth_mnesia, [{description, "EMQ X Authentication with Mnesia"}, - {vsn, "4.3.3"}, % strict semver, bump manually + {vsn, "4.3.4"}, % strict semver, bump manually {modules, []}, {registered, []}, {applications, [kernel,stdlib,mnesia]}, diff --git a/apps/emqx_auth_mnesia/src/emqx_auth_mnesia.appup.src b/apps/emqx_auth_mnesia/src/emqx_auth_mnesia.appup.src index d5cb8b49f..82df99b3a 100644 --- a/apps/emqx_auth_mnesia/src/emqx_auth_mnesia.appup.src +++ b/apps/emqx_auth_mnesia/src/emqx_auth_mnesia.appup.src @@ -1,30 +1,31 @@ -%% -*-: erlang -*- - +%% -*- mode: erlang -*- {VSN, [ - {"4.3.2", [ + {<<"4.3.[0-3]">>, [ + {add_module,emqx_acl_mnesia_db}, + {add_module,emqx_acl_mnesia_migrator, [emqx_acl_mnesia_db]}, + {update, emqx_auth_mnesia_sup, supervisor}, + {apply, {emqx_acl_mnesia_migrator, start_supervised, []}}, + {load_module,emqx_auth_mnesia_api, brutal_purge,soft_purge,[]}, + {load_module,emqx_acl_mnesia, brutal_purge,soft_purge,[]}, {load_module,emqx_acl_mnesia_api, brutal_purge,soft_purge,[]}, - {load_module,emqx_auth_mnesia_api, brutal_purge,soft_purge,[]} + {load_module,emqx_acl_mnesia_cli, brutal_purge,soft_purge,[]} ]}, - {"4.3.1", [ - {load_module,emqx_auth_mnesia_api, brutal_purge,soft_purge,[]} - ]}, - {"4.3.0", [ - {load_module,emqx_auth_mnesia_api, brutal_purge,soft_purge,[]} - ]}, - {<<".*">>, []} + {<<".*">>, [ + ]} ], [ - {"4.3.2", [ + {<<"4.3.[0-3]">>, [ + {apply, {emqx_acl_mnesia_migrator, stop_supervised, []}}, + {update, emqx_auth_mnesia_sup, supervisor}, + {load_module,emqx_acl_mnesia_cli, brutal_purge,soft_purge,[]}, {load_module,emqx_acl_mnesia_api, brutal_purge,soft_purge,[]}, - {load_module,emqx_auth_mnesia_api, brutal_purge,soft_purge,[]} + {load_module,emqx_auth_mnesia_api, brutal_purge,soft_purge,[]}, + {load_module,emqx_acl_mnesia, brutal_purge,soft_purge,[]}, + {delete_module,emqx_acl_mnesia_migrator}, + {delete_module,emqx_acl_mnesia_db} ]}, - {"4.3.1", [ - {load_module,emqx_auth_mnesia_api, brutal_purge,soft_purge,[]} - ]}, - {"4.3.0", [ - {load_module,emqx_auth_mnesia_api, brutal_purge,soft_purge,[]} - ]}, - {<<".*">>, []} + {<<".*">>, [ + ]} ] }. diff --git a/apps/emqx_auth_mnesia/src/emqx_auth_mnesia_api.erl b/apps/emqx_auth_mnesia/src/emqx_auth_mnesia_api.erl index 9d9fff6f6..da24ddd53 100644 --- a/apps/emqx_auth_mnesia/src/emqx_auth_mnesia_api.erl +++ b/apps/emqx_auth_mnesia/src/emqx_auth_mnesia_api.erl @@ -23,7 +23,7 @@ -import(proplists, [get_value/2]). -import(minirest, [return/1]). --export([paginate/5]). +-export([paginate_qh/5]). -export([ list_clientid/2 , lookup_clientid/2 @@ -212,9 +212,12 @@ delete_username(#{username := Username}, _) -> %% Paging Query %%------------------------------------------------------------------------------ -paginate(Tables, MatchSpec, Params, ComparingFun, RowFun) -> - Qh = query_handle(Tables, MatchSpec), - Count = count(Tables, MatchSpec), +paginate(Table, MatchSpec, Params, ComparingFun, RowFun) -> + Qh = query_handle(Table, MatchSpec), + Count = count(Table, MatchSpec), + paginate_qh(Qh, Count, Params, ComparingFun, RowFun). + +paginate_qh(Qh, Count, Params, ComparingFun, RowFun) -> Page = page(Params), Limit = limit(Params), Cursor = qlc:cursor(Qh), @@ -231,24 +234,12 @@ paginate(Tables, MatchSpec, Params, ComparingFun, RowFun) -> query_handle(Table, MatchSpec) when is_atom(Table) -> Options = {traverse, {select, MatchSpec}}, - qlc:q([R|| R <- ets:table(Table, Options)]); -query_handle([Table], MatchSpec) when is_atom(Table) -> - Options = {traverse, {select, MatchSpec}}, - qlc:q([R|| R <- ets:table(Table, Options)]); -query_handle(Tables, MatchSpec) -> - Options = {traverse, {select, MatchSpec}}, - qlc:append([qlc:q([E || E <- ets:table(T, Options)]) || T <- Tables]). + qlc:q([R || R <- ets:table(Table, Options)]). count(Table, MatchSpec) when is_atom(Table) -> [{MatchPattern, Where, _Re}] = MatchSpec, NMatchSpec = [{MatchPattern, Where, [true]}], - ets:select_count(Table, NMatchSpec); -count([Table], MatchSpec) when is_atom(Table) -> - [{MatchPattern, Where, _Re}] = MatchSpec, - NMatchSpec = [{MatchPattern, Where, [true]}], - ets:select_count(Table, NMatchSpec); -count(Tables, MatchSpec) -> - lists:sum([count(T, MatchSpec) || T <- Tables]). + ets:select_count(Table, NMatchSpec). page(Params) -> binary_to_integer(proplists:get_value(<<"_page">>, Params, <<"1">>)). diff --git a/apps/emqx_auth_mnesia/src/emqx_auth_mnesia_sup.erl b/apps/emqx_auth_mnesia/src/emqx_auth_mnesia_sup.erl index 3784eaaf6..2099eba8c 100644 --- a/apps/emqx_auth_mnesia/src/emqx_auth_mnesia_sup.erl +++ b/apps/emqx_auth_mnesia/src/emqx_auth_mnesia_sup.erl @@ -33,4 +33,16 @@ start_link() -> %%-------------------------------------------------------------------- init([]) -> - {ok, {{one_for_one, 10, 100}, []}}. \ No newline at end of file + {ok, {{one_for_one, 10, 100}, [ + child_spec(emqx_acl_mnesia_migrator, worker, []) + ]}}. + +child_spec(M, worker, Args) -> + #{id => M, + start => {M, start_link, Args}, + restart => permanent, + shutdown => 5000, + type => worker, + modules => [M] + }. + diff --git a/apps/emqx_auth_mnesia/test/emqx_acl_mnesia_SUITE.erl b/apps/emqx_auth_mnesia/test/emqx_acl_mnesia_SUITE.erl index 8ace680da..eb1ea74f3 100644 --- a/apps/emqx_auth_mnesia/test/emqx_acl_mnesia_SUITE.erl +++ b/apps/emqx_auth_mnesia/test/emqx_acl_mnesia_SUITE.erl @@ -22,6 +22,7 @@ -include("emqx_auth_mnesia.hrl"). -include_lib("eunit/include/eunit.hrl"). -include_lib("common_test/include/ct.hrl"). +-include_lib("snabbkaffe/include/snabbkaffe.hrl"). -import(emqx_ct_http, [ request_api/3 , request_api/5 @@ -39,10 +40,15 @@ all() -> emqx_ct:all(?MODULE). groups() -> - []. + [{async_migration_tests, [sequence], [ + t_old_and_new_acl_migration_by_migrator, + t_old_and_new_acl_migration_repeated_by_migrator, + t_migration_concurrency + ]}]. init_per_suite(Config) -> emqx_ct_helpers:start_apps([emqx_modules, emqx_management, emqx_auth_mnesia], fun set_special_configs/1), + supervisor:terminate_child(emqx_auth_mnesia_sup, emqx_acl_mnesia_migrator), create_default_app(), Config. @@ -50,14 +56,32 @@ end_per_suite(_Config) -> delete_default_app(), emqx_ct_helpers:stop_apps([emqx_modules, emqx_management, emqx_auth_mnesia]). -init_per_testcase(t_check_acl_as_clientid, Config) -> +init_per_testcase_clean(_, Config) -> + mnesia:clear_table(?ACL_TABLE), + mnesia:clear_table(?ACL_TABLE2), + Config. + +init_per_testcase_emqx_hook(t_check_acl_as_clientid, Config) -> emqx:hook('client.check_acl', fun emqx_acl_mnesia:check_acl/5, [#{key_as => clientid}]), Config; - -init_per_testcase(_, Config) -> +init_per_testcase_emqx_hook(_, Config) -> emqx:hook('client.check_acl', fun emqx_acl_mnesia:check_acl/5, [#{key_as => username}]), Config. +init_per_testcase_migration(t_management_before_migration, Config) -> + Config; +init_per_testcase_migration(_, Config) -> + emqx_acl_mnesia_migrator:migrate_records(), + Config. + +init_per_testcase(Case, Config) -> + PerTestInitializers = [ + fun init_per_testcase_clean/2, + fun init_per_testcase_migration/2, + fun init_per_testcase_emqx_hook/2 + ], + lists:foldl(fun(Init, Conf) -> Init(Case, Conf) end, Config, PerTestInitializers). + end_per_testcase(_, Config) -> emqx:unhook('client.check_acl', fun emqx_acl_mnesia:check_acl/5), Config. @@ -76,25 +100,34 @@ set_special_configs(_App) -> %% Testcases %%------------------------------------------------------------------------------ -t_management(_Config) -> - clean_all_acls(), - ?assertEqual("Acl with Mnesia", emqx_acl_mnesia:description()), - ?assertEqual([], emqx_acl_mnesia_cli:all_acls()), +t_management_before_migration(_Config) -> + {atomic, IsStarted} = mnesia:transaction(fun emqx_acl_mnesia_db:is_migration_started/0), + ?assertNot(IsStarted), + run_acl_tests(). - ok = emqx_acl_mnesia_cli:add_acl({clientid, <<"test_clientid">>}, <<"topic/%c">>, sub, allow), - ok = emqx_acl_mnesia_cli:add_acl({clientid, <<"test_clientid">>}, <<"topic/+">>, pub, deny), - ok = emqx_acl_mnesia_cli:add_acl({username, <<"test_username">>}, <<"topic/%u">>, sub, deny), - ok = emqx_acl_mnesia_cli:add_acl({username, <<"test_username">>}, <<"topic/+">>, pub, allow), - ok = emqx_acl_mnesia_cli:add_acl(all, <<"#">>, pubsub, deny), +t_management_after_migration(_Config) -> + {atomic, IsStarted} = mnesia:transaction(fun emqx_acl_mnesia_db:is_migration_started/0), + ?assert(IsStarted), + run_acl_tests(). + +run_acl_tests() -> + ?assertEqual("Acl with Mnesia", emqx_acl_mnesia:description()), + ?assertEqual([], emqx_acl_mnesia_db:all_acls()), + + ok = emqx_acl_mnesia_db:add_acl({clientid, <<"test_clientid">>}, <<"topic/%c">>, sub, allow), + ok = emqx_acl_mnesia_db:add_acl({clientid, <<"test_clientid">>}, <<"topic/+">>, pub, deny), + ok = emqx_acl_mnesia_db:add_acl({username, <<"test_username">>}, <<"topic/%u">>, sub, deny), + ok = emqx_acl_mnesia_db:add_acl({username, <<"test_username">>}, <<"topic/+">>, pub, allow), + ok = emqx_acl_mnesia_db:add_acl(all, <<"#">>, pubsub, deny), %% Sleeps below are needed to hide the race condition between %% mnesia and ets dirty select in check_acl, that make this test %% flaky timer:sleep(100), - ?assertEqual(2, length(emqx_acl_mnesia_cli:lookup_acl({clientid, <<"test_clientid">>}))), - ?assertEqual(2, length(emqx_acl_mnesia_cli:lookup_acl({username, <<"test_username">>}))), - ?assertEqual(2, length(emqx_acl_mnesia_cli:lookup_acl(all))), - ?assertEqual(6, length(emqx_acl_mnesia_cli:all_acls())), + ?assertEqual(2, length(emqx_acl_mnesia_db:lookup_acl({clientid, <<"test_clientid">>}))), + ?assertEqual(2, length(emqx_acl_mnesia_db:lookup_acl({username, <<"test_username">>}))), + ?assertEqual(2, length(emqx_acl_mnesia_db:lookup_acl(all))), + ?assertEqual(6, length(emqx_acl_mnesia_db:all_acls())), User1 = #{zone => external, clientid => <<"test_clientid">>}, User2 = #{zone => external, clientid => <<"no_exist">>, username => <<"test_username">>}, @@ -110,30 +143,30 @@ t_management(_Config) -> deny = emqx_access_control:check_acl(User3, publish, <<"topic/A/B">>), %% Test merging of pubsub capability: - ok = emqx_acl_mnesia_cli:add_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>, pubsub, deny), + ok = emqx_acl_mnesia_db:add_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>, pubsub, deny), timer:sleep(100), deny = emqx_access_control:check_acl(User1, subscribe, <<"topic/mix">>), deny = emqx_access_control:check_acl(User1, publish, <<"topic/mix">>), - ok = emqx_acl_mnesia_cli:add_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>, pub, allow), + ok = emqx_acl_mnesia_db:add_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>, pub, allow), timer:sleep(100), deny = emqx_access_control:check_acl(User1, subscribe, <<"topic/mix">>), allow = emqx_access_control:check_acl(User1, publish, <<"topic/mix">>), - ok = emqx_acl_mnesia_cli:add_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>, pubsub, allow), + ok = emqx_acl_mnesia_db:add_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>, pubsub, allow), timer:sleep(100), allow = emqx_access_control:check_acl(User1, subscribe, <<"topic/mix">>), allow = emqx_access_control:check_acl(User1, publish, <<"topic/mix">>), - ok = emqx_acl_mnesia_cli:add_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>, sub, deny), + ok = emqx_acl_mnesia_db:add_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>, sub, deny), timer:sleep(100), deny = emqx_access_control:check_acl(User1, subscribe, <<"topic/mix">>), allow = emqx_access_control:check_acl(User1, publish, <<"topic/mix">>), - ok = emqx_acl_mnesia_cli:add_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>, pub, deny), + ok = emqx_acl_mnesia_db:add_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>, pub, deny), timer:sleep(100), deny = emqx_access_control:check_acl(User1, subscribe, <<"topic/mix">>), deny = emqx_access_control:check_acl(User1, publish, <<"topic/mix">>), %% Test implicit migration of pubsub to pub and sub: - ok = emqx_acl_mnesia_cli:remove_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>), - ok = mnesia:dirty_write(#emqx_acl{ + ok = emqx_acl_mnesia_db:remove_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>), + ok = mnesia:dirty_write(#?ACL_TABLE{ filter = {{clientid, <<"test_clientid">>}, <<"topic/mix">>}, action = pubsub, access = allow, @@ -142,24 +175,130 @@ t_management(_Config) -> timer:sleep(100), allow = emqx_access_control:check_acl(User1, subscribe, <<"topic/mix">>), allow = emqx_access_control:check_acl(User1, publish, <<"topic/mix">>), - ok = emqx_acl_mnesia_cli:add_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>, pub, deny), + ok = emqx_acl_mnesia_db:add_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>, pub, deny), timer:sleep(100), allow = emqx_access_control:check_acl(User1, subscribe, <<"topic/mix">>), deny = emqx_access_control:check_acl(User1, publish, <<"topic/mix">>), - ok = emqx_acl_mnesia_cli:add_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>, sub, deny), + ok = emqx_acl_mnesia_db:add_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>, sub, deny), timer:sleep(100), deny = emqx_access_control:check_acl(User1, subscribe, <<"topic/mix">>), deny = emqx_access_control:check_acl(User1, publish, <<"topic/mix">>), - ok = emqx_acl_mnesia_cli:remove_acl({clientid, <<"test_clientid">>}, <<"topic/%c">>), - ok = emqx_acl_mnesia_cli:remove_acl({clientid, <<"test_clientid">>}, <<"topic/+">>), - ok = emqx_acl_mnesia_cli:remove_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>), - ok = emqx_acl_mnesia_cli:remove_acl({username, <<"test_username">>}, <<"topic/%u">>), - ok = emqx_acl_mnesia_cli:remove_acl({username, <<"test_username">>}, <<"topic/+">>), - ok = emqx_acl_mnesia_cli:remove_acl(all, <<"#">>), + ok = emqx_acl_mnesia_db:remove_acl({clientid, <<"test_clientid">>}, <<"topic/%c">>), + ok = emqx_acl_mnesia_db:remove_acl({clientid, <<"test_clientid">>}, <<"topic/+">>), + ok = emqx_acl_mnesia_db:remove_acl({clientid, <<"test_clientid">>}, <<"topic/mix">>), + ok = emqx_acl_mnesia_db:remove_acl({username, <<"test_username">>}, <<"topic/%u">>), + ok = emqx_acl_mnesia_db:remove_acl({username, <<"test_username">>}, <<"topic/+">>), + ok = emqx_acl_mnesia_db:remove_acl(all, <<"#">>), timer:sleep(100), - ?assertEqual([], emqx_acl_mnesia_cli:all_acls()). + ?assertEqual([], emqx_acl_mnesia_db:all_acls()). + +t_old_and_new_acl_combination(_Config) -> + create_conflicting_records(), + + ?assertEqual(combined_conflicting_records(), emqx_acl_mnesia_db:all_acls()), + ?assertEqual( + lists:usort(combined_conflicting_records()), + lists:usort(emqx_acl_mnesia_db:all_acls_export())). + +t_old_and_new_acl_migration(_Config) -> + create_conflicting_records(), + emqx_acl_mnesia_migrator:migrate_records(), + + ?assertEqual(combined_conflicting_records(), emqx_acl_mnesia_db:all_acls()), + ?assertEqual( + lists:usort(combined_conflicting_records()), + lists:usort(emqx_acl_mnesia_db:all_acls_export())), + + % check that old table is not popoulated anymore + ok = emqx_acl_mnesia_db:add_acl({clientid, <<"test_clientid">>}, <<"topic/%c">>, sub, allow), + ?assert(emqx_acl_mnesia_migrator:is_old_table_migrated()). + + +t_migration_concurrency(_Config) -> + Key = {{clientid,<<"client6">>}, <<"t">>}, + Record = #?ACL_TABLE{filter = Key, action = pubsub, access = deny, created_at = 0}, + {atomic, ok} = mnesia:transaction(fun mnesia:write/1, [Record]), + + LockWaitAndDelete = + fun() -> + [_Rec] = mnesia:wread({?ACL_TABLE, Key}), + {{Pid, Ref}, _} = + ?wait_async_action(spawn_monitor(fun emqx_acl_mnesia_migrator:migrate_records/0), + #{?snk_kind := emqx_acl_mnesia_migrator_record_selected}, + 1000), + mnesia:delete({?ACL_TABLE, Key}), + {Pid, Ref} + end, + + ?check_trace( + begin + {atomic, {Pid, Ref}} = mnesia:transaction(LockWaitAndDelete), + receive {'DOWN', Ref, process, Pid, _} -> ok end + end, + fun(_, Trace) -> + ?assertMatch([_], ?of_kind(emqx_acl_mnesia_migrator_record_missed, Trace)) + end), + + ?assert(emqx_acl_mnesia_migrator:is_old_table_migrated()), + ?assertEqual([], emqx_acl_mnesia_db:all_acls()). + + +t_old_and_new_acl_migration_by_migrator(_Config) -> + create_conflicting_records(), + + meck:new(fake_nodes, [non_strict]), + meck:expect(fake_nodes, all, fun() -> [node(), 'somebadnode@127.0.0.1'] end), + + ?check_trace( + begin + % check all nodes every 30 ms + {ok, _} = emqx_acl_mnesia_migrator:start_link(#{ + name => ct_migrator, + check_nodes_interval => 30, + get_nodes => fun fake_nodes:all/0 + }), + timer:sleep(100) + end, + fun(_, Trace) -> + ?assertEqual([], ?of_kind(emqx_acl_mnesia_migrator_start_migration, Trace)) + end), + + ?check_trace( + begin + meck:expect(fake_nodes, all, fun() -> [node()] end), + timer:sleep(100) + end, + fun(_, Trace) -> + ?assertMatch([_], ?of_kind(emqx_acl_mnesia_migrator_finish, Trace)) + end), + + meck:unload(fake_nodes), + + ?assertEqual(combined_conflicting_records(), emqx_acl_mnesia_db:all_acls()), + ?assert(emqx_acl_mnesia_migrator:is_old_table_migrated()). + +t_old_and_new_acl_migration_repeated_by_migrator(_Config) -> + create_conflicting_records(), + emqx_acl_mnesia_migrator:migrate_records(), + + ?check_trace( + begin + {ok, _} = emqx_acl_mnesia_migrator:start_link(ct_migrator), + timer:sleep(100) + end, + fun(_, Trace) -> + ?assertEqual([], ?of_kind(emqx_acl_mnesia_migrator_start_migration, Trace)), + ?assertMatch([_], ?of_kind(emqx_acl_mnesia_migrator_finish, Trace)) + end). + +t_start_stop_supervised(_Config) -> + ?assertEqual(undefined, whereis(emqx_acl_mnesia_migrator)), + ok = emqx_acl_mnesia_migrator:start_supervised(), + ?assert(is_pid(whereis(emqx_acl_mnesia_migrator))), + ok = emqx_acl_mnesia_migrator:stop_supervised(), + ?assertEqual(undefined, whereis(emqx_acl_mnesia_migrator)). t_acl_cli(_Config) -> meck:new(emqx_ctl, [non_strict, passthrough]), @@ -168,8 +307,6 @@ t_acl_cli(_Config) -> meck:expect(emqx_ctl, usage, fun(Usages) -> emqx_ctl:format_usage(Usages) end), meck:expect(emqx_ctl, usage, fun(Cmd, Descr) -> emqx_ctl:format_usage(Cmd, Descr) end), - clean_all_acls(), - ?assertEqual(0, length(emqx_acl_mnesia_cli:cli(["list"]))), emqx_acl_mnesia_cli:cli(["add", "clientid", "test_clientid", "topic/A", "pub", "deny"]), @@ -202,8 +339,6 @@ t_acl_cli(_Config) -> meck:unload(emqx_ctl). t_rest_api(_Config) -> - clean_all_acls(), - Params1 = [#{<<"clientid">> => <<"test_clientid">>, <<"topic">> => <<"topic/A">>, <<"action">> => <<"pub">>, @@ -273,13 +408,24 @@ t_rest_api(_Config) -> {ok, Res3} = request_http_rest_list(["$all"]), ?assertMatch([], get_http_data(Res3)). -%%------------------------------------------------------------------------------ -%% Helpers -%%------------------------------------------------------------------------------ -clean_all_acls() -> - [ mnesia:dirty_delete({emqx_acl, Login}) - || Login <- mnesia:dirty_all_keys(emqx_acl)]. +create_conflicting_records() -> + Records = [ + #?ACL_TABLE{filter = {{clientid,<<"client6">>}, <<"t">>}, action = pubsub, access = deny, created_at = 0}, + #?ACL_TABLE{filter = {{clientid,<<"client5">>}, <<"t">>}, action = pubsub, access = deny, created_at = 1}, + #?ACL_TABLE2{who = {clientid,<<"client5">>}, rules = [{allow, sub, <<"t">>, 2}]} + ], + mnesia:transaction(fun() -> lists:foreach(fun mnesia:write/1, Records) end). + + +combined_conflicting_records() -> + % pubsub's are split, ACL_TABLE2 rules shadow ACL_TABLE rules + [ + {{clientid,<<"client5">>},<<"t">>,sub,allow,2}, + {{clientid,<<"client5">>},<<"t">>,pub,deny,1}, + {{clientid,<<"client6">>},<<"t">>,sub,deny,0}, + {{clientid,<<"client6">>},<<"t">>,pub,deny,0} + ]. %%-------------------------------------------------------------------- %% HTTP Request diff --git a/apps/emqx_lwm2m/etc/emqx_lwm2m.conf b/apps/emqx_lwm2m/etc/emqx_lwm2m.conf index 968b8fd19..0aa061b1c 100644 --- a/apps/emqx_lwm2m/etc/emqx_lwm2m.conf +++ b/apps/emqx_lwm2m/etc/emqx_lwm2m.conf @@ -146,4 +146,4 @@ lwm2m.dtls.ciphers = ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384,E ## Note that 'lwm2m.dtls.ciphers' and 'lwm2m.dtls.psk_ciphers' cannot ## be configured at the same time. ## See 'https://tools.ietf.org/html/rfc4279#section-2'. -#lwm2m.dtls.psk_ciphers = PSK-AES128-CBC-SHA,PSK-AES256-CBC-SHA,PSK-3DES-EDE-CBC-SHA,PSK-RC4-SHA +#lwm2m.dtls.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 diff --git a/apps/emqx_lwm2m/priv/emqx_lwm2m.schema b/apps/emqx_lwm2m/priv/emqx_lwm2m.schema index bf5f144e0..ded81df05 100644 --- a/apps/emqx_lwm2m/priv/emqx_lwm2m.schema +++ b/apps/emqx_lwm2m/priv/emqx_lwm2m.schema @@ -185,7 +185,7 @@ end}. OldCert = cuttlefish:conf_get("lwm2m.certfile", Conf, undefined), %% Ciphers - SplitFun = fun(undefined) -> undefined; (S) -> string:tokens(S, ",") end, + SplitFun = fun(undefined) -> []; (S) -> string:tokens(S, ",") end, Ciphers = case cuttlefish:conf_get("lwm2m.dtls.ciphers", Conf, undefined) of undefined -> @@ -198,16 +198,17 @@ end}. undefined -> []; C2 -> - Psk = 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, SplitFun(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, <<>>}}] end, Ciphers /= [] - andalso PskCiphers /= [] - andalso cuttlefish:invalid("The 'lwm2m.dtls.ciphers' and 'lwm2m.dtls.psk_ciphers' cannot exist simultaneously."), + andalso PskCiphers /= [] + andalso cuttlefish:invalid("The 'lwm2m.dtls.ciphers' and 'lwm2m.dtls.psk_ciphers' cannot coexist"), NCiphers = Ciphers ++ PskCiphers, diff --git a/apps/emqx_lwm2m/src/emqx_lwm2m.app.src b/apps/emqx_lwm2m/src/emqx_lwm2m.app.src index f4afe8fbc..551cf8d07 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.3"}, % strict semver, bump manually! + {vsn, "4.3.4"}, % 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 07af339fd..600cf236b 100644 --- a/apps/emqx_lwm2m/src/emqx_lwm2m.appup.src +++ b/apps/emqx_lwm2m/src/emqx_lwm2m.appup.src @@ -1,19 +1,21 @@ %% -*-: erlang -*- -{"4.3.3", +{"4.3.4", [ - {<<"4.3.[0-1]">>, [ + {<<"4\\.3\\.[0-1]">>, [ {restart_application, emqx_lwm2m} ]}, {"4.3.2", [ {load_module, emqx_lwm2m_message, brutal_purge, soft_purge, []} - ]} + ]}, + {"4.3.3", []} %% only config change ], [ - {<<"4.3.[0-1]">>, [ + {<<"4\\.3\\.[0-1]">>, [ {restart_application, emqx_lwm2m} ]}, {"4.3.2", [ {load_module, emqx_lwm2m_message, brutal_purge, soft_purge, []} - ]} + ]}, + {"4.3.3", []} %% only config change ] }. diff --git a/apps/emqx_management/src/emqx_management.app.src b/apps/emqx_management/src/emqx_management.app.src index 94c22f693..405b4c244 100644 --- a/apps/emqx_management/src/emqx_management.app.src +++ b/apps/emqx_management/src/emqx_management.app.src @@ -1,6 +1,6 @@ {application, emqx_management, [{description, "EMQ X Management API and CLI"}, - {vsn, "4.3.7"}, % strict semver, bump manually! + {vsn, "4.3.8"}, % strict semver, bump manually! {modules, []}, {registered, [emqx_management_sup]}, {applications, [kernel,stdlib,minirest]}, diff --git a/apps/emqx_management/src/emqx_management.appup.src b/apps/emqx_management/src/emqx_management.appup.src index 1cc55817b..1463334b4 100644 --- a/apps/emqx_management/src/emqx_management.appup.src +++ b/apps/emqx_management/src/emqx_management.appup.src @@ -1,13 +1,13 @@ %% -*- mode: erlang -*- {VSN, - [ {<<"4.3.[0-6]">>, + [ {<<"4\\.3\\.[0-9]+">>, [ {apply,{minirest,stop_http,['http:management']}}, {apply,{minirest,stop_http,['https:management']}}, {restart_application, emqx_management} ]}, {<<".*">>, []} ], - [ {<<"4.3.[0-6]">>, + [ {<<"4\\.3\\.[0-9]+">>, [ {apply,{minirest,stop_http,['http:management']}}, {apply,{minirest,stop_http,['https:management']}}, {restart_application, emqx_management} diff --git a/apps/emqx_management/src/emqx_mgmt_api_clients.erl b/apps/emqx_management/src/emqx_mgmt_api_clients.erl index 2fe6a5ccb..1ddd87a3d 100644 --- a/apps/emqx_management/src/emqx_mgmt_api_clients.erl +++ b/apps/emqx_management/src/emqx_mgmt_api_clients.erl @@ -334,7 +334,7 @@ query({Qs, Fuzzy}, Start, Limit) -> match_fun(Ms, Fuzzy) -> MsC = ets:match_spec_compile(Ms), REFuzzy = lists:map(fun({K, like, S}) -> - {ok, RE} = re:compile(S), + {ok, RE} = re:compile(escape(S)), {K, like, RE} end, Fuzzy), fun(Rows) -> @@ -347,6 +347,9 @@ match_fun(Ms, Fuzzy) -> end end. +escape(B) when is_binary(B) -> + re:replace(B, <<"\\\\">>, <<"\\\\\\\\">>, [{return, binary}, global]). + run_fuzzy_match(_, []) -> true; run_fuzzy_match(E = {_, #{clientinfo := ClientInfo}, _}, [{Key, _, RE}|Fuzzy]) -> @@ -450,4 +453,9 @@ params2qs_test() -> [{{'$1', #{}, '_'}, [], ['$_']}] = qs2ms([]). +escape_test() -> + Str = <<"\\n">>, + {ok, Re} = re:compile(escape(Str)), + {match, _} = re:run(<<"\\name">>, Re). + -endif. diff --git a/apps/emqx_management/src/emqx_mgmt_api_pubsub.erl b/apps/emqx_management/src/emqx_mgmt_api_pubsub.erl index 84763f403..53ca022bb 100644 --- a/apps/emqx_management/src/emqx_mgmt_api_pubsub.erl +++ b/apps/emqx_management/src/emqx_mgmt_api_pubsub.erl @@ -158,7 +158,7 @@ do_subscribe(ClientId, Topics, QoS) -> _ -> ok end. -do_publish(ClientId, _Topics, _Qos, _Retain, _Payload) when not is_binary(ClientId) -> +do_publish(ClientId, _Topics, _Qos, _Retain, _Payload) when not (is_binary(ClientId) or (ClientId =:= undefined)) -> {ok, ?ERROR8, <<"bad clientid: must be string">>}; do_publish(_ClientId, [], _Qos, _Retain, _Payload) -> {ok, ?ERROR15, bad_topic}; diff --git a/apps/emqx_management/src/emqx_mgmt_data_backup.erl b/apps/emqx_management/src/emqx_mgmt_data_backup.erl index 3e2fe784b..6e467a8ba 100644 --- a/apps/emqx_management/src/emqx_mgmt_data_backup.erl +++ b/apps/emqx_management/src/emqx_mgmt_data_backup.erl @@ -118,18 +118,18 @@ export_auth_mnesia() -> end. export_acl_mnesia() -> - case ets:info(emqx_acl) of + case ets:info(emqx_acl2) of undefined -> []; _ -> - lists:map(fun({_, Filter, Action, Access, CreatedAt}) -> - Filter1 = case Filter of - {{Type, TypeValue}, Topic} -> + lists:map(fun({Login, Topic, Action, Access, CreatedAt}) -> + Filter1 = case Login of + {Type, TypeValue} -> [{type, Type}, {type_value, TypeValue}, {topic, Topic}]; - {Type, Topic} -> + Type -> [{type, Type}, {topic, Topic}] end, Filter1 ++ [{action, Action}, {access, Access}, {created_at, CreatedAt}] - end, ets:tab2list(emqx_acl)) + end, emqx_acl_mnesia_db:all_acls_export()) end. -ifdef(EMQX_ENTERPRISE). @@ -473,10 +473,9 @@ do_import_auth_mnesia(Auths) -> end. do_import_acl_mnesia_by_old_data(Acls) -> - case ets:info(emqx_acl) of + case ets:info(emqx_acl2) of undefined -> ok; _ -> - CreatedAt = erlang:system_time(millisecond), lists:foreach(fun(#{<<"login">> := Login, <<"topic">> := Topic, <<"allow">> := Allow, @@ -485,11 +484,11 @@ do_import_acl_mnesia_by_old_data(Acls) -> true -> allow; false -> deny end, - mnesia:dirty_write({emqx_acl, {{get_old_type(), Login}, Topic}, any_to_atom(Action), Allow1, CreatedAt}) + emqx_acl_mnesia_db:add_acl({get_old_type(), Login}, Topic, any_to_atom(Action), Allow1) end, Acls) end. do_import_acl_mnesia(Acls) -> - case ets:info(emqx_acl) of + case ets:info(emqx_acl2) of undefined -> ok; _ -> lists:foreach(fun(Map = #{<<"action">> := Action, @@ -501,7 +500,7 @@ do_import_acl_mnesia(Acls) -> Value -> {any_to_atom(maps:get(<<"type">>, Map)), Value} end, - emqx_acl_mnesia_cli:add_acl(Login, Topic, any_to_atom(Action), any_to_atom(Access)) + emqx_acl_mnesia_db:add_acl(Login, Topic, any_to_atom(Action), any_to_atom(Access)) end, Acls) end. diff --git a/apps/emqx_management/test/emqx_auth_mnesia_migration_SUITE.erl b/apps/emqx_management/test/emqx_auth_mnesia_migration_SUITE.erl index 838529f03..7ccba161b 100644 --- a/apps/emqx_management/test/emqx_auth_mnesia_migration_SUITE.erl +++ b/apps/emqx_management/test/emqx_auth_mnesia_migration_SUITE.erl @@ -30,7 +30,7 @@ matrix() -> , Version <- ["v4.2.10", "v4.1.5"]]. all() -> - [t_import_4_0, t_import_4_1, t_import_4_2]. + [t_import_4_0, t_import_4_1, t_import_4_2, t_export_import]. groups() -> [{username, [], cases()}, {clientid, [], cases()}]. @@ -52,7 +52,8 @@ init_per_testcase(_, Config) -> Config. end_per_testcase(_, _Config) -> - {atomic,ok} = mnesia:clear_table(emqx_acl), + {atomic,ok} = mnesia:clear_table(?ACL_TABLE), + {atomic,ok} = mnesia:clear_table(?ACL_TABLE2), {atomic,ok} = mnesia:clear_table(emqx_user), ok. -ifdef(EMQX_ENTERPRISE). @@ -138,25 +139,50 @@ t_import_4_2(Config) -> test_import(clientid, {<<"client_for_test">>, <<"public">>}), test_import(username, {<<"user_for_test">>, <<"public">>}), - ?assertMatch([#emqx_acl{ - filter = {{Type,<<"emqx_c">>}, <<"Topic/A">>}, - action = pub, - access = allow - }, - #emqx_acl{ - filter = {{Type,<<"emqx_c">>}, <<"Topic/A">>}, - action = sub, - access = allow - }], - lists:sort(ets:tab2list(emqx_acl))). + ?assertMatch([ + {{username, <<"emqx_c">>}, <<"Topic/A">>, pub, allow, _}, + {{username, <<"emqx_c">>}, <<"Topic/A">>, sub, allow, _} + ], + lists:sort(emqx_acl_mnesia_db:all_acls())). -endif. +t_export_import(_Config) -> + emqx_acl_mnesia_migrator:migrate_records(), + + Records = [ + #?ACL_TABLE2{who = {clientid,<<"client1">>}, rules = [{allow, sub, <<"t1">>, 1}]}, + #?ACL_TABLE2{who = {clientid,<<"client2">>}, rules = [{allow, pub, <<"t2">>, 2}]} + ], + mnesia:transaction(fun() -> lists:foreach(fun mnesia:write/1, Records) end), + timer:sleep(100), + + AclData = emqx_json:encode(emqx_mgmt_data_backup:export_acl_mnesia()), + + mnesia:transaction(fun() -> + lists:foreach(fun(#?ACL_TABLE2{who = Who}) -> + mnesia:delete({?ACL_TABLE2, Who}) + end, + Records) + end), + + ?assertEqual([], emqx_acl_mnesia_db:all_acls()), + + emqx_mgmt_data_backup:import_acl_mnesia(emqx_json:decode(AclData, [return_maps]), "4.3"), + timer:sleep(100), + + ?assertMatch([ + {{clientid, <<"client1">>}, <<"t1">>, sub, allow, _}, + {{clientid, <<"client2">>}, <<"t2">>, pub, allow, _} + ], lists:sort(emqx_acl_mnesia_db:all_acls())). + do_import(File, Config) -> do_import(File, Config, "{}"). do_import(File, Config, Overrides) -> - mnesia:clear_table(emqx_acl), + mnesia:clear_table(?ACL_TABLE), + mnesia:clear_table(?ACL_TABLE2), mnesia:clear_table(emqx_user), + emqx_acl_mnesia_migrator:migrate_records(), Filename = filename:join(proplists:get_value(data_dir, Config), File), emqx_mgmt_data_backup:import(Filename, Overrides). @@ -172,4 +198,4 @@ test_import(clientid, {ClientID, Password}) -> Req = #{clientid => ClientID, password => Password}, ?assertMatch({stop, #{auth_result := success}}, - emqx_auth_mnesia:check(Req, #{}, #{hash_type => sha256})). \ No newline at end of file + emqx_auth_mnesia:check(Req, #{}, #{hash_type => sha256})). diff --git a/apps/emqx_management/test/emqx_mgmt_api_SUITE.erl b/apps/emqx_management/test/emqx_mgmt_api_SUITE.erl index e0754a522..a739fa3c7 100644 --- a/apps/emqx_management/test/emqx_mgmt_api_SUITE.erl +++ b/apps/emqx_management/test/emqx_mgmt_api_SUITE.erl @@ -447,6 +447,19 @@ t_pubsub(_) -> after 100 -> false end), + + % no clientid + {ok, Code} = request_api(post, api_path(["mqtt/publish"]), [], auth_header_(), + #{<<"topic">> => <<"mytopic">>, + <<"qos">> => 1, + <<"payload">> => <<"hello">>}), + ?assert(receive + {publish, #{payload := <<"hello">>}} -> + true + after 100 -> + false + end), + %% json payload {ok, Code} = request_api(post, api_path(["mqtt/publish"]), [], auth_header_(), #{<<"clientid">> => ClientId, @@ -491,9 +504,9 @@ t_pubsub(_) -> ok = emqtt:disconnect(C1), - ?assertEqual(2, emqx_metrics:val('messages.qos1.received') - Qos1Received), + ?assertEqual(3, emqx_metrics:val('messages.qos1.received') - Qos1Received), ?assertEqual(2, emqx_metrics:val('messages.qos2.received') - Qos2Received), - ?assertEqual(4, emqx_metrics:val('messages.received') - Received). + ?assertEqual(5, emqx_metrics:val('messages.received') - Received). loop([]) -> []; diff --git a/apps/emqx_web_hook/src/emqx_web_hook.appup.src b/apps/emqx_web_hook/src/emqx_web_hook.appup.src index db245b2ee..ec716f45c 100644 --- a/apps/emqx_web_hook/src/emqx_web_hook.appup.src +++ b/apps/emqx_web_hook/src/emqx_web_hook.appup.src @@ -1,28 +1,20 @@ -%% -*-: erlang -*- - +%% -*- mode: erlang -*- {VSN, - [ - {<<"4.3.[0-2]">>, [ - {apply, {application, stop,[emqx_web_hook]}}, - {load_module, emqx_web_hook_app, brutal_purge, soft_purge, []}, - {load_module, emqx_web_hook, brutal_purge, soft_purge, []}, - {load_module, emqx_web_hook_actions, brutal_purge, soft_purge, []} - ]}, - {<<"4.3.[3-5]">>, [ - {load_module, emqx_web_hook_actions, brutal_purge, soft_purge, []} - ]}, - {<<".*">>, []} - ], - [ - {<<"4.3.[0-2]">>, [ - {apply, {application, stop, [emqx_web_hook]}}, - {load_module, emqx_web_hook_app, brutal_purge, soft_purge, []}, - {load_module, emqx_web_hook, brutal_purge, soft_purge, []}, - {load_module, emqx_web_hook_actions, brutal_purge, soft_purge, []} - ]}, - {<<"4.3.[3-5]">>, [ - {load_module, emqx_web_hook_actions, brutal_purge, soft_purge, []} - ]}, - {<<".*">>, []} - ] -}. + [{"4.3.5",[{load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]}, + {<<"4.3.[0-2]">>, + [{apply,{application,stop,[emqx_web_hook]}}, + {load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]}, + {load_module,emqx_web_hook,brutal_purge,soft_purge,[]}, + {load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]}, + {<<"4.3.[3-4]">>, + [{load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]}, + {<<".*">>,[]}], + [{"4.3.5",[{load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]}, + {<<"4.3.[0-2]">>, + [{apply,{application,stop,[emqx_web_hook]}}, + {load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]}, + {load_module,emqx_web_hook,brutal_purge,soft_purge,[]}, + {load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]}, + {<<"4.3.[3-4]">>, + [{load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]}, + {<<".*">>,[]}]}. diff --git a/bin/emqx b/bin/emqx index e8fda9c2b..ff12afcac 100755 --- a/bin/emqx +++ b/bin/emqx @@ -238,12 +238,22 @@ generate_config() { sed '/^#/d' "$CUTTLE_GEN_ARG_FILE" | sed '/^$/d' | while IFS='' read -r ARG_LINE || [ -n "$ARG_LINE" ]; do ARG_KEY=$(echo "$ARG_LINE" | awk '{$NF="";print}') ARG_VALUE=$(echo "$ARG_LINE" | awk '{print $NF}') - TMP_ARG_VALUE=$(grep "^$ARG_KEY" "$TMP_ARG_FILE" | awk '{print $NF}') - if [ "$ARG_VALUE" != "$TMP_ARG_VALUE" ] ; then - if [ -n "$TMP_ARG_VALUE" ]; then - sh -c "$SED_REPLACE 's/^$ARG_KEY.*$/$ARG_LINE/' $TMP_ARG_FILE" - else - echo "$ARG_LINE" >> "$TMP_ARG_FILE" + if [ "$ARG_KEY" = '' ]; then + ## for the flags, e.g. -heart -emu_args etc + ARG_KEY=$(echo "$ARG_LINE" | awk '{print $1}') + ARG_VALUE='' + TMP_ARG_KEY=$(grep "^$ARG_KEY" "$TMP_ARG_FILE" | awk '{print $1}') + if [ "$TMP_ARG_KEY" = '' ]; then + echo "$ARG_KEY" >> "$TMP_ARG_FILE" + fi + else + TMP_ARG_VALUE=$(grep "^$ARG_KEY" "$TMP_ARG_FILE" | awk '{print $NF}') + if [ "$ARG_VALUE" != "$TMP_ARG_VALUE" ] ; then + if [ -n "$TMP_ARG_VALUE" ]; then + sh -c "$SED_REPLACE 's/^$ARG_KEY.*$/$ARG_LINE/' $TMP_ARG_FILE" + else + echo "$ARG_LINE" >> "$TMP_ARG_FILE" + fi fi fi done diff --git a/etc/emqx.conf b/etc/emqx.conf index 6043dc361..152f28215 100644 --- a/etc/emqx.conf +++ b/etc/emqx.conf @@ -199,6 +199,16 @@ node.data_dir = {{ platform_data_dir }} ## Heartbeat monitoring of an Erlang runtime system. Comment the line to disable ## heartbeat, or set the value as 'on' ## +## Turning this on may cause the node to restart if it becomes unresponsive to +## the heartbeat pings. +## +## NOTE: When managed by systemd (or other supervision tools like systemd), +## heart will probably only cause EMQ X to stop, but restart or not will +## depend on systemd's restart strategy. +## NOTE: When running in docker, the container will die as soon as the the +## heart process kills EMQ X, but restart or not will depend on container +## supervision strategy, such as k8s restartPolicy. +## ## Value: on ## ## vm.args: -heart diff --git a/lib-ce/emqx_dashboard/src/emqx_dashboard.appup.src b/lib-ce/emqx_dashboard/src/emqx_dashboard.appup.src index 7d0ffd5ea..4dc02511c 100644 --- a/lib-ce/emqx_dashboard/src/emqx_dashboard.appup.src +++ b/lib-ce/emqx_dashboard/src/emqx_dashboard.appup.src @@ -1,6 +1,6 @@ %% -*- mode: erlang -*- {VSN, - [ {<<"4.3.[0-4]">>, + [ {<<"4.3.[0-9]">>, %% load all plugins %% NOTE: this depends on the fact that emqx_dashboard is always %% the last application gets upgraded @@ -10,7 +10,7 @@ ]}, {<<".*">>, []} ], - [ {<<"4.3.[0-4]">>, + [ {<<"4.3.[0-9]">>, [ {apply, {emqx_rule_engine, load_providers, []}} , {restart_application, emqx_dashboard} , {apply, {emqx_plugins, load, []}} diff --git a/rebar.config b/rebar.config index f56e8e2c2..c49733535 100644 --- a/rebar.config +++ b/rebar.config @@ -55,7 +55,7 @@ , {recon, {git, "https://github.com/ferd/recon", {tag, "2.5.1"}}} , {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.13.0"}}} + , {snabbkaffe, {git, "https://github.com/kafka4beam/snabbkaffe.git", {tag, "0.15.0"}}} ]}. {xref_ignores, diff --git a/scripts/apps-version-check.sh b/scripts/apps-version-check.sh index b070e3c45..596f7404a 100755 --- a/scripts/apps-version-check.sh +++ b/scripts/apps-version-check.sh @@ -18,7 +18,7 @@ while read -r app; do changed="$(git diff --name-only "$latest_release"...HEAD \ -- "$app_path/src" \ -- "$app_path/priv" \ - -- "$app_path/c_src" | wc -l)" + -- "$app_path/c_src" | { grep -v -E 'appup\.src' || true; } | wc -l)" if [ "$changed" -gt 0 ]; then echo "$src_file needs a vsn bump" bad_app_count=$(( bad_app_count + 1)) diff --git a/scripts/one-more-emqx-ee.sh b/scripts/one-more-emqx-ee.sh new file mode 100644 index 000000000..f94681056 --- /dev/null +++ b/scripts/one-more-emqx-ee.sh @@ -0,0 +1,106 @@ +#!/bin/bash +# shellcheck disable=2090 +############### +## args and env validation +############### + +if ! [ -d "emqx" ]; then + echo "[error] this script must be run at the same dir as the emqx" + exit 1 +fi + +if [ $# -eq 0 ] + then + echo "[error] a new emqx name should be provided!" + echo "Usage: ./one_more_emqx " + echo " e.g. ./one_more_emqx emqx2" + exit 1 +fi + +NEW_EMQX=$1 +if [ -d "$NEW_EMQX" ]; then + echo "[error] a dir named ${NEW_EMQX} already exists!" + exit 2 +fi +echo creating "$NEW_EMQX" ... + +SED_REPLACE="sed -i " +# shellcheck disable=2089 +case $(sed --help 2>&1) in + *GNU*) SED_REPLACE="sed -i ";; + *) SED_REPLACE="sed -i ''";; +esac + +PORT_INC_=$(cksum <<< "$NEW_EMQX" | cut -f 1 -d ' ') +PORT_INC=$((PORT_INC_ % 1000)) +echo using increment factor: $PORT_INC + +############### +## helpers +############### +process_emqx_conf() { + echo "processing config file: $1" + $SED_REPLACE '/^#/d' "$1" + $SED_REPLACE '/^$/d' "$1" + + for entry_ in "${entries_to_be_inc[@]}" + do + echo inc port for "$entry_" + ip_port_=$(grep -E "$entry_"'[ \t]*=' "$1" 2> /dev/null | tail -1 | cut -d = -f 2- | sed -e 's/^[[:space:]]*//' -e 's/[[:space:]]*$//') + echo -- from: "$ip_port_" + ip_=$(echo "$ip_port_" | cut -sd : -f 1) + port_=$(echo "$ip_port_" | cut -sd : -f 2) + if [ -z "$ip_" ] + then + new_ip_port=$(( ip_port_ + PORT_INC )) + else + new_ip_port="${ip_}:$(( port_ + PORT_INC ))" + fi + echo -- to: "$new_ip_port" + $SED_REPLACE 's|'"$entry_"'[ \t]*=.*|'"$entry_"' = '"$new_ip_port"'|g' "$1" + done +} + +############### +## main +############### + +cp -r emqx "$NEW_EMQX" + +## change the rpc ports +$SED_REPLACE 's|tcp_server_port[ \t]*=.*|tcp_server_port = 5369|g' emqx/etc/rpc.conf +$SED_REPLACE 's|tcp_client_port[ \t]*=.*|tcp_client_port = 5370|g' emqx/etc/rpc.conf +$SED_REPLACE 's|tcp_client_port[ \t]*=.*|tcp_client_port = 5369|g' "$NEW_EMQX/etc/rpc.conf" +$SED_REPLACE 's|tcp_server_port[ \t]*=.*|tcp_server_port = 5370|g' "$NEW_EMQX/etc/rpc.conf" +$SED_REPLACE 's|.*node\.name.*|node.name='"$NEW_EMQX"'@127.0.0.1|g' "$NEW_EMQX/etc/emqx.conf" + +conf_ext="*.conf" + +find "$NEW_EMQX" -name "${conf_ext}" | while read -r conf; do + if [ "${conf##*/}" = 'emqx.conf' ] + then + declare -a entries_to_be_inc=("node.dist_listen_min" + "node.dist_listen_max") + process_emqx_conf "$conf" "${entries_to_be_inc[@]}" + elif [ "${conf##*/}" = 'listeners.conf' ] + then + declare -a entries_to_be_inc=("listener.tcp.external" + "listener.tcp.internal" + "listener.ssl.external" + "listener.ws.external" + "listener.wss.external") + process_emqx_conf "$conf" "${entries_to_be_inc[@]}" + elif [ "${conf##*/}" = 'emqx_management.conf' ] + then + declare -a entries_to_be_inc=("management.listener.http" + "management.listener.https") + process_emqx_conf "$conf" "${entries_to_be_inc[@]}" + elif [ "${conf##*/}" = 'emqx_dashboard.conf' ] + then + declare -a entries_to_be_inc=("dashboard.listener.http" + "dashboard.listener.https") + process_emqx_conf "$conf" "${entries_to_be_inc[@]}" + else + echo "." + fi +done diff --git a/scripts/one-more-emqx.sh b/scripts/one-more-emqx.sh new file mode 100644 index 000000000..d905f64c4 --- /dev/null +++ b/scripts/one-more-emqx.sh @@ -0,0 +1,102 @@ +#!/bin/bash +# shellcheck disable=2090 +############### +## args and env validation +############### + +if ! [ -d "emqx" ]; then + echo "[error] this script must be run at the same dir as the emqx" + exit 1 +fi + +if [ $# -eq 0 ] + then + echo "[error] a new emqx name should be provided!" + echo "Usage: ./one_more_emqx " + echo " e.g. ./one_more_emqx emqx2" + exit 1 +fi + +NEW_EMQX=$1 +if [ -d "$NEW_EMQX" ]; then + echo "[error] a dir named ${NEW_EMQX} already exists!" + exit 2 +fi +echo creating "$NEW_EMQX" ... + +SED_REPLACE="sed -i " +# shellcheck disable=2089 +case $(sed --help 2>&1) in + *GNU*) SED_REPLACE="sed -i ";; + *) SED_REPLACE="sed -i ''";; +esac + +PORT_INC_=$(cksum <<< "$NEW_EMQX" | cut -f 1 -d ' ') +PORT_INC=$((PORT_INC_ % 1000)) +echo using increment factor: "$PORT_INC" + +############### +## helpers +############### +process_emqx_conf() { + echo "processing config file: $1" + $SED_REPLACE '/^#/d' "$1" + $SED_REPLACE '/^$/d' "$1" + $SED_REPLACE 's|.*node\.name.*|node.name='"$NEW_EMQX"'@127.0.0.1|g' "$1" + + for entry_ in "${entries_to_be_inc[@]}" + do + echo inc port for "$entry_" + ip_port_=$(grep -E "$entry_"'[ \t]*=' "$1" 2> /dev/null | tail -1 | cut -d = -f 2- | sed -e 's/^[[:space:]]*//' -e 's/[[:space:]]*$//') + echo -- from: "$ip_port_" + ip_=$(echo "$ip_port_" | cut -sd : -f 1) + port_=$(echo "$ip_port_" | cut -sd : -f 2) + if [ -z "$ip_" ] + then + new_ip_port=$(( ip_port_ + PORT_INC )) + else + new_ip_port="${ip_}:$(( port_ + PORT_INC ))" + fi + echo -- to: "$new_ip_port" + $SED_REPLACE 's|'"$entry_"'[ \t]*=.*|'"$entry_"' = '"$new_ip_port"'|g' "$1" + done +} + +############### +## main +############### + +cp -r emqx "$NEW_EMQX" + +## change the rpc ports +$SED_REPLACE 's|tcp_server_port[ \t]*=.*|tcp_server_port = 5369|g' emqx/etc/emqx.conf +$SED_REPLACE 's|tcp_client_port[ \t]*=.*|tcp_client_port = 5370|g' emqx/etc/emqx.conf +$SED_REPLACE 's|tcp_client_port[ \t]*=.*|tcp_client_port = 5369|g' "$NEW_EMQX/etc/emqx.conf" +$SED_REPLACE 's|tcp_server_port[ \t]*=.*|tcp_server_port = 5370|g' "$NEW_EMQX/etc/emqx.conf" + +conf_ext="*.conf" +find "$NEW_EMQX" -name "${conf_ext}" | while read -r conf; do + if [ "${conf##*/}" = 'emqx.conf' ] + then + declare -a entries_to_be_inc=("node.dist_listen_min" + "dist_listen_max" + "listener.tcp.external" + "listener.tcp.internal" + "listener.ssl.external" + "listener.ws.external" + "listener.wss.external") + process_emqx_conf "$conf" "${entries_to_be_inc[@]}" + elif [ "${conf##*/}" = 'emqx_management.conf' ] + then + declare -a entries_to_be_inc=("management.listener.http" + "management.listener.https") + process_emqx_conf "$conf" "${entries_to_be_inc[@]}" + elif [ "${conf##*/}" = 'emqx_dashboard.conf' ] + then + declare -a entries_to_be_inc=("dashboard.listener.http" + "dashboard.listener.https") + process_emqx_conf "$conf" "${entries_to_be_inc[@]}" + else + echo "." + fi +done diff --git a/scripts/update_appup.escript b/scripts/update_appup.escript index 61e6ae717..c8061f9de 100755 --- a/scripts/update_appup.escript +++ b/scripts/update_appup.escript @@ -25,14 +25,16 @@ Usage: Options: - --check Don't update the appfile, just check that they are complete - --prev-tag Specify the previous release tag. Otherwise the previous patch version is used - --repo Upsteam git repo URL - --remote Get upstream repo URL from the specified git remote - --skip-build Don't rebuild the releases. May produce wrong results - --make-command A command used to assemble the release - --release-dir Release directory - --src-dirs Directories where source code is found. Defaults to '{src,apps,lib-*}/**/' + --check Don't update the appfile, just check that they are complete + --prev-tag Specify the previous release tag. Otherwise the previous patch version is used + --repo Upsteam git repo URL + --remote Get upstream repo URL from the specified git remote + --skip-build Don't rebuild the releases. May produce wrong results + --make-command A command used to assemble the release + --release-dir Release directory + --src-dirs Directories where source code is found. Defaults to '{src,apps,lib-*}/**/' + --binary-rel-url Binary release URL pattern. %TAG% variable is substituted with the release tag. + E.g. \"https://github.com/emqx/emqx/releases/download/v4.3.8/emqx-centos7-%TAG%-amd64.zip\" ". -record(app, @@ -41,18 +43,26 @@ Options: }). default_options() -> - #{ clone_url => find_upstream_repo("origin") - , make_command => "make emqx-rel" - , beams_dir => "_build/emqx/rel/emqx/lib/" - , check => false - , prev_tag => undefined - , src_dirs => "{src,apps,lib-*}/**/" + #{ clone_url => find_upstream_repo("origin") + , make_command => "make emqx-rel" + , beams_dir => "_build/emqx/rel/emqx/lib/" + , check => false + , prev_tag => undefined + , src_dirs => "{src,apps,lib-*}/**/" + , binary_rel_url => undefined }. +%% App-specific actions that should be added unconditionally to any update/downgrade: +app_specific_actions(_) -> + []. + +ignored_apps() -> + [emqx_dashboard, emqx_management] ++ otp_standard_apps(). + main(Args) -> #{current_release := CurrentRelease} = Options = parse_args(Args, default_options()), init_globals(Options), - case find_pred_tag(CurrentRelease) of + case find_prev_tag(CurrentRelease) of {ok, Baseline} -> main(Options, Baseline); undefined -> @@ -78,6 +88,8 @@ parse_args(["--src-dirs", Pattern|Rest], State) -> parse_args(Rest, State#{src_dirs => Pattern}); parse_args(["--prev-tag", Tag|Rest], State) -> parse_args(Rest, State#{prev_tag => Tag}); +parse_args(["--binary-rel-url", URL|Rest], State) -> + parse_args(Rest, State#{binary_rel_url => {ok, URL}}); parse_args(_, _) -> fail(usage()). @@ -88,7 +100,7 @@ main(Options, Baseline) -> "~n===================================~n"), CurrAppsIdx = index_apps(CurrRelDir), PrevAppsIdx = index_apps(PrevRelDir), - %% log("Curr: ~p~nPrev: ~p~n", [CurrApps, PrevApps]), + %% log("Curr: ~p~nPrev: ~p~n", [CurrAppsIdx, PrevAppsIdx]), AppupChanges = find_appup_actions(CurrAppsIdx, PrevAppsIdx), case getopt(check) of true -> @@ -115,17 +127,25 @@ warn_and_exit(false) -> log("~nERROR: Incomplete appups found. Please inspect the output for more details.~n"), halt(1). -prepare(Baseline, Options = #{make_command := MakeCommand, beams_dir := BeamDir}) -> +prepare(Baseline, Options = #{make_command := MakeCommand, beams_dir := BeamDir, binary_rel_url := BinRel}) -> log("~n===================================~n" "Baseline: ~s" "~n===================================~n", [Baseline]), log("Building the current version...~n"), bash(MakeCommand), log("Downloading and building the previous release...~n"), - {ok, PrevRootDir} = build_pred_release(Baseline, Options), - {BeamDir, filename:join(PrevRootDir, BeamDir)}. + PrevRelDir = + case BinRel of + undefined -> + {ok, PrevRootDir} = build_prev_release(Baseline, Options), + filename:join(PrevRootDir, BeamDir); + {ok, _URL} -> + {ok, PrevRootDir} = download_prev_release(Baseline, Options), + PrevRootDir + end, + {BeamDir, PrevRelDir}. -build_pred_release(Baseline, #{clone_url := Repo, make_command := MakeCommand}) -> +build_prev_release(Baseline, #{clone_url := Repo, make_command := MakeCommand}) -> BaseDir = "/tmp/emqx-baseline/", Dir = filename:basename(Repo, ".git") ++ [$-|Baseline], %% TODO: shallow clone @@ -137,10 +157,22 @@ build_pred_release(Baseline, #{clone_url := Repo, make_command := MakeCommand}) bash(Script, Env), {ok, filename:join(BaseDir, Dir)}. +download_prev_release(Tag, #{binary_rel_url := {ok, URL0}, clone_url := Repo}) -> + URL = string:replace(URL0, "%TAG%", Tag, all), + BaseDir = "/tmp/emqx-baseline-bin/", + Dir = filename:basename(Repo, ".git") ++ [$-|Tag], + Filename = filename:join(BaseDir, Dir), + Script = "mkdir -p ${OUTFILE} && + { [ -f ${OUTFILE}.zip ] || wget -O ${OUTFILE}.zip ${URL}; } && + unzip -n -d ${OUTFILE} ${OUTFILE}.zip", + Env = [{"TAG", Tag}, {"OUTFILE", Filename}, {"URL", URL}], + bash(Script, Env), + {ok, Filename}. + find_upstream_repo(Remote) -> string:trim(os:cmd("git remote get-url " ++ Remote)). -find_pred_tag(CurrentRelease) -> +find_prev_tag(CurrentRelease) -> case getopt(prev_tag) of undefined -> {Maj, Min, Patch} = parse_semver(CurrentRelease), @@ -172,8 +204,8 @@ find_appup_actions(_App, AppIdx, AppIdx) -> []; find_appup_actions(App, CurrAppIdx, PrevAppIdx = #app{version = PrevVersion}) -> {OldUpgrade, OldDowngrade} = find_old_appup_actions(App, PrevVersion), - Upgrade = merge_update_actions(diff_app(App, CurrAppIdx, PrevAppIdx), OldUpgrade), - Downgrade = merge_update_actions(diff_app(App, PrevAppIdx, CurrAppIdx), OldDowngrade), + Upgrade = merge_update_actions(App, diff_app(App, CurrAppIdx, PrevAppIdx), OldUpgrade), + Downgrade = merge_update_actions(App, diff_app(App, PrevAppIdx, CurrAppIdx), OldDowngrade), if OldUpgrade =:= Upgrade andalso OldDowngrade =:= Downgrade -> %% The appup file has been already updated: []; @@ -183,7 +215,7 @@ find_appup_actions(App, CurrAppIdx, PrevAppIdx = #app{version = PrevVersion}) -> find_old_appup_actions(App, PrevVersion) -> {Upgrade0, Downgrade0} = - case locate(App, ".appup.src") of + case locate(ebin_current, App, ".appup") of {ok, AppupFile} -> {_, U, D} = read_appup(AppupFile), {U, D}; @@ -192,22 +224,24 @@ find_old_appup_actions(App, PrevVersion) -> end, {ensure_version(PrevVersion, Upgrade0), ensure_version(PrevVersion, Downgrade0)}. -merge_update_actions(Changes, Vsns) -> +merge_update_actions(App, Changes, Vsns) -> lists:map(fun(Ret = {<<".*">>, _}) -> Ret; ({Vsn, Actions}) -> - {Vsn, do_merge_update_actions(Changes, Actions)} + {Vsn, do_merge_update_actions(App, Changes, Actions)} end, Vsns). -do_merge_update_actions({New0, Changed0, Deleted0}, OldActions) -> +do_merge_update_actions(App, {New0, Changed0, Deleted0}, OldActions) -> + AppSpecific = app_specific_actions(App) -- OldActions, AlreadyHandled = lists:flatten(lists:map(fun process_old_action/1, OldActions)), New = New0 -- AlreadyHandled, Changed = Changed0 -- AlreadyHandled, Deleted = Deleted0 -- AlreadyHandled, [{load_module, M, brutal_purge, soft_purge, []} || M <- Changed ++ New] ++ OldActions ++ - [{delete_module, M} || M <- Deleted]. + [{delete_module, M} || M <- Deleted] ++ + AppSpecific. %% @doc Process the existing actions to exclude modules that are @@ -222,12 +256,13 @@ process_old_action(LoadModule) when is_tuple(LoadModule) andalso process_old_action(_) -> []. -ensure_version(Version, Versions) -> - case lists:keyfind(Version, 1, Versions) of +ensure_version(Version, OldInstructions) -> + OldVersions = [ensure_string(element(1, I)) || I <- OldInstructions], + case lists:member(Version, OldVersions) of false -> - [{Version, []}|Versions]; + [{Version, []}|OldInstructions]; _ -> - Versions + OldInstructions end. read_appup(File) -> @@ -251,7 +286,7 @@ update_appups(Changes) -> Changes). do_update_appup(App, Upgrade, Downgrade) -> - case locate(App, ".appup.src") of + case locate(src, App, ".appup.src") of {ok, AppupFile} -> render_appfile(AppupFile, Upgrade, Downgrade); undefined -> @@ -260,7 +295,7 @@ do_update_appup(App, Upgrade, Downgrade) -> render_appfile(AppupFile, Upgrade, Downgrade); false -> set_invalid(), - log("ERROR: Appup file for the external dependency '~p' is not complete.~n Missing changes: ~p", [App, Upgrade]) + log("ERROR: Appup file for the external dependency '~p' is not complete.~n Missing changes: ~p~n", [App, Upgrade]) end end. @@ -273,7 +308,7 @@ render_appfile(File, Upgrade, Downgrade) -> ok = file:write_file(File, IOList). create_stub(App) -> - case locate(App, ".app.src") of + case locate(src, App, ".app.src") of {ok, AppSrc} -> AppupFile = filename:basename(AppSrc) ++ ".appup.src", Default = {<<".*">>, []}, @@ -288,8 +323,9 @@ create_stub(App) -> %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% index_apps(ReleaseDir) -> - maps:from_list([index_app(filename:join(ReleaseDir, AppFile)) || - AppFile <- filelib:wildcard("**/ebin/*.app", ReleaseDir)]). + Apps0 = maps:from_list([index_app(filename:join(ReleaseDir, AppFile)) || + AppFile <- filelib:wildcard("**/ebin/*.app", ReleaseDir)]), + maps:without(ignored_apps(), Apps0). index_app(AppFile) -> {ok, [{application, App, Properties}]} = file:consult(AppFile), @@ -320,7 +356,10 @@ diff_app(App, #app{version = NewVersion, modules = NewModules}, #app{version = O NChanges = length(New) + length(Changed) + length(Deleted), if NewVersion =:= OldVersion andalso NChanges > 0 -> set_invalid(), - log("ERROR: Application '~p' contains changes, but its version is not updated", [App]); + log("ERROR: Application '~p' contains changes, but its version is not updated~n", [App]); + NewVersion > OldVersion -> + log("INFO: Application '~p' has been updated: ~p -> ~p~n", [App, OldVersion, NewVersion]), + ok; true -> ok end, @@ -372,7 +411,16 @@ semver(Maj, Min, Patch) -> lists:flatten(io_lib:format("~p.~p.~p", [Maj, Min, Patch])). %% Locate a file in a specified application -locate(App, Suffix) -> +locate(ebin_current, App, Suffix) -> + ReleaseDir = getopt(beams_dir), + AppStr = atom_to_list(App), + case filelib:wildcard(ReleaseDir ++ "/**/ebin/" ++ AppStr ++ Suffix) of + [File] -> + {ok, File}; + [] -> + undefined + end; +locate(src, App, Suffix) -> AppStr = atom_to_list(App), SrcDirs = getopt(src_dirs), case filelib:wildcard(SrcDirs ++ AppStr ++ Suffix) of @@ -425,3 +473,11 @@ log(Msg) -> log(Msg, Args) -> io:format(standard_error, Msg, Args). + +ensure_string(Str) when is_binary(Str) -> + binary_to_list(Str); +ensure_string(Str) when is_list(Str) -> + Str. + +otp_standard_apps() -> + [ssl, mnesia, kernel, asn1, stdlib]. diff --git a/src/emqx.appup.src b/src/emqx.appup.src index e4eb459a9..d4f9d43ce 100644 --- a/src/emqx.appup.src +++ b/src/emqx.appup.src @@ -1,36 +1,40 @@ %% -*- mode: erlang -*- -Instructions = -{"4.3.10", - [ - %% app 4.3.9 was released in e4.3.4(enterprise) but not v4.3.9(opensource) - {"4.3.9", [ - {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.8", [ +{VSN, + [{"4.3.9", + [{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]}, {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.7", [ + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.8", + [{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]}, + {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, + {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, + {load_module,emqx_frame,brutal_purge,soft_purge,[]}, + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.7", + [{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]}, {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.6", [ + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.6", + [{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]}, {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_ctl,brutal_purge,soft_purge,[]}, {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.5", [ + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.5", + [{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]}, {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_cm,brutal_purge,soft_purge,[]}, @@ -39,9 +43,10 @@ Instructions = {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.4", [ + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.4", + [{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]}, {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_cm,brutal_purge,soft_purge,[]}, @@ -51,10 +56,10 @@ Instructions = {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.3", [ - {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.3", + [{load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_packet,brutal_purge,soft_purge,[]}, {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]}, @@ -65,10 +70,10 @@ Instructions = {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.2", [ - {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.2", + [{load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_packet,brutal_purge,soft_purge,[]}, {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]}, @@ -82,10 +87,10 @@ Instructions = {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.1", [ - {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.1", + [{load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_packet,brutal_purge,soft_purge,[]}, {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]}, @@ -103,9 +108,11 @@ Instructions = {load_module,emqx_ctl,brutal_purge,soft_purge,[]}, {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.0", [ + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.0", + [{load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, + {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_packet,brutal_purge,soft_purge,[]}, @@ -126,43 +133,58 @@ Instructions = {load_module,emqx_http_lib,brutal_purge,soft_purge,[]}, {load_module,emqx_access_rule,brutal_purge,soft_purge,[]}, {load_module,emqx_ctl,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, {<<".*">>,[]}], - [ - {"4.3.9", [ - {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.8", [ + [{"4.3.9", + [{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]}, {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.7", [ + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.8", + [{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]}, + {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, + {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, + {load_module,emqx_frame,brutal_purge,soft_purge,[]}, + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.7", + [{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]}, + {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, + {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.6", [ + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.6", + [{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]}, + {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, + {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_ctl,brutal_purge,soft_purge,[]}, {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.5", [ + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.5", + [{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]}, + {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, + {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_cm,brutal_purge,soft_purge,[]}, {load_module,emqx_access_rule,brutal_purge,soft_purge,[]}, {load_module,emqx_ctl,brutal_purge,soft_purge,[]}, {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.4", [ + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.4", + [{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]}, + {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, + {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_cm,brutal_purge,soft_purge,[]}, @@ -170,9 +192,11 @@ Instructions = {load_module,emqx_access_rule,brutal_purge,soft_purge,[]}, {load_module,emqx_ctl,brutal_purge,soft_purge,[]}, {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.3", [ + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.3", + [{load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, + {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_packet,brutal_purge,soft_purge,[]}, @@ -182,9 +206,11 @@ Instructions = {load_module,emqx_access_rule,brutal_purge,soft_purge,[]}, {load_module,emqx_ctl,brutal_purge,soft_purge,[]}, {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.2", [ + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.2", + [{load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, + {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_packet,brutal_purge,soft_purge,[]}, @@ -197,9 +223,11 @@ Instructions = {load_module,emqx_access_rule,brutal_purge,soft_purge,[]}, {load_module,emqx_ctl,brutal_purge,soft_purge,[]}, {load_module,emqx_frame,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.1", [ + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.1", + [{load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, + {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_packet,brutal_purge,soft_purge,[]}, @@ -216,10 +244,10 @@ Instructions = {load_module,emqx_http_lib,brutal_purge,soft_purge,[]}, {load_module,emqx_access_rule,brutal_purge,soft_purge,[]}, {load_module,emqx_ctl,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {"4.3.0", [ - {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {"4.3.0", + [{load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]}, {load_module,emqx_misc,brutal_purge,soft_purge,[]}, {load_module,emqx_packet,brutal_purge,soft_purge,[]}, {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]}, @@ -240,23 +268,6 @@ Instructions = {load_module,emqx_ctl,brutal_purge,soft_purge,[]}, {load_module,emqx_pqueue,brutal_purge,soft_purge,[]}, {load_module,emqx_mqueue,brutal_purge,soft_purge,[]}, - {load_module,emqx_rpc,brutal_purge,soft_purge,[]} - ]}, - {<<".*">>,[]}]}, - -%% Always reload emqx_app for emqx_app:get_release/0 to return the correct version -Mandatory = [{load_module,emqx_app,brutal_purge,soft_purge,[]}], - -Append = fun - ({<<".*">>, Instrs}) -> - {<<".*">>, Instrs}; - ({Vsn, Instrs}) -> - {Vsn, Instrs ++ Mandatory} -end, - -PostProcess = fun({Vsn, UpList, DownList}) -> - {Vsn, [Append(Up) || Up <- UpList], - [Append(Dn) || Dn <- DownList]} -end, - -PostProcess(Instructions). + {load_module,emqx_rpc,brutal_purge,soft_purge,[]}, + {load_module,emqx_app,brutal_purge,soft_purge,[]}]}, + {<<".*">>,[]}]}. diff --git a/src/emqx_ws_connection.erl b/src/emqx_ws_connection.erl index d686e1611..07d437b4b 100644 --- a/src/emqx_ws_connection.erl +++ b/src/emqx_ws_connection.erl @@ -242,7 +242,7 @@ parse_header_fun_origin(Req, Opts) -> Origins = proplists:get_value(check_origins, Opts, []), case lists:member(Value, Origins) of true -> ok; - false -> {origin_not_allowed, Value} + false -> {error, {origin_not_allowed, Value}} end end. diff --git a/test/emqx_ws_connection_SUITE.erl b/test/emqx_ws_connection_SUITE.erl index 6db831972..56d038c23 100644 --- a/test/emqx_ws_connection_SUITE.erl +++ b/test/emqx_ws_connection_SUITE.erl @@ -247,7 +247,7 @@ t_ws_check_origin(_) -> ?assertMatch({gun_upgrade, _}, start_ws_client(#{protocols => [<<"mqtt">>], headers => [{<<"origin">>, <<"http://localhost:18083">>}]})), - ?assertMatch({gun_response, {_, 500, _}}, + ?assertMatch({gun_response, {_, 403, _}}, start_ws_client(#{protocols => [<<"mqtt">>], headers => [{<<"origin">>, <<"http://localhost:18080">>}]})), emqx_ct_helpers:stop_apps([]).