diff --git a/.github/workflows/build_and_push_docker_images.yaml b/.github/workflows/build_and_push_docker_images.yaml index 71515f699..213ab27fd 100644 --- a/.github/workflows/build_and_push_docker_images.yaml +++ b/.github/workflows/build_and_push_docker_images.yaml @@ -25,7 +25,7 @@ jobs: prepare: runs-on: ubuntu-20.04 # prepare source with any OTP version, no need for a matrix - container: "ghcr.io/emqx/emqx-builder/5.0-26:1.13.4-24.3.4.2-1-ubuntu20.04" + container: "ghcr.io/emqx/emqx-builder/5.0-27:1.13.4-24.3.4.2-1-ubuntu20.04" outputs: PROFILE: ${{ steps.get_profile.outputs.PROFILE }} @@ -129,7 +129,7 @@ jobs: # NOTE: 'otp' and 'elixir' are to configure emqx-builder image # only support latest otp and elixir, not a matrix builder: - - 5.0-26 # update to latest + - 5.0-27 # update to latest otp: - 24.3.4.2-1 # switch to 25 once ready to release 5.1 elixir: diff --git a/.github/workflows/build_packages.yaml b/.github/workflows/build_packages.yaml index 8764c7097..2ec7dac3e 100644 --- a/.github/workflows/build_packages.yaml +++ b/.github/workflows/build_packages.yaml @@ -23,7 +23,7 @@ on: jobs: prepare: runs-on: ubuntu-20.04 - container: ghcr.io/emqx/emqx-builder/5.0-26:1.13.4-24.3.4.2-1-ubuntu20.04 + container: ghcr.io/emqx/emqx-builder/5.0-27:1.13.4-24.3.4.2-1-ubuntu20.04 outputs: BUILD_PROFILE: ${{ steps.get_profile.outputs.BUILD_PROFILE }} IS_EXACT_TAG: ${{ steps.get_profile.outputs.IS_EXACT_TAG }} @@ -204,6 +204,7 @@ jobs: - amd64 - arm64 os: + - ubuntu22.04 - ubuntu20.04 - ubuntu18.04 - debian11 @@ -215,7 +216,7 @@ jobs: - aws-arm64 - ubuntu-20.04 builder: - - 5.0-26 + - 5.0-27 elixir: - 1.13.4 exclude: @@ -227,17 +228,17 @@ jobs: - profile: emqx otp: 25.1.2-2 arch: amd64 - os: ubuntu20.04 - build_machine: ubuntu-20.04 - builder: 5.0-26 + os: ubuntu22.04 + build_machine: ubuntu-22.04 + builder: 5.0-27 elixir: 1.13.4 release_with: elixir - profile: emqx otp: 25.1.2-2 arch: amd64 os: amzn2 - build_machine: ubuntu-20.04 - builder: 5.0-26 + build_machine: ubuntu-22.04 + builder: 5.0-27 elixir: 1.13.4 release_with: elixir diff --git a/.github/workflows/build_slim_packages.yaml b/.github/workflows/build_slim_packages.yaml index 88e454860..c7b5b0c83 100644 --- a/.github/workflows/build_slim_packages.yaml +++ b/.github/workflows/build_slim_packages.yaml @@ -30,10 +30,12 @@ jobs: matrix: profile: - ["emqx", "24.3.4.2-1", "el7"] - - ["emqx", "25.1.2-2", "ubuntu20.04"] + - ["emqx", "24.3.4.2-1", "ubuntu20.04"] + - ["emqx", "25.1.2-2", "ubuntu22.04"] - ["emqx-enterprise", "24.3.4.2-1", "ubuntu20.04"] + - ["emqx-enterprise", "25.1.2-2", "ubuntu22.04"] builder: - - 5.0-26 + - 5.0-27 elixir: - 1.13.4 diff --git a/.github/workflows/check_deps_integrity.yaml b/.github/workflows/check_deps_integrity.yaml index 59fc01d74..6d4f53dbc 100644 --- a/.github/workflows/check_deps_integrity.yaml +++ b/.github/workflows/check_deps_integrity.yaml @@ -5,7 +5,7 @@ on: [pull_request, push] jobs: check_deps_integrity: runs-on: ubuntu-20.04 - container: ghcr.io/emqx/emqx-builder/5.0-26:1.13.4-25.1.2-2-ubuntu20.04 + container: ghcr.io/emqx/emqx-builder/5.0-27:1.13.4-25.1.2-2-ubuntu20.04 steps: - uses: actions/checkout@v3 diff --git a/.github/workflows/code_style_check.yaml b/.github/workflows/code_style_check.yaml index 6ab8ea8b6..38e3a5b8b 100644 --- a/.github/workflows/code_style_check.yaml +++ b/.github/workflows/code_style_check.yaml @@ -5,7 +5,7 @@ on: [pull_request] jobs: code_style_check: runs-on: ubuntu-20.04 - container: "ghcr.io/emqx/emqx-builder/5.0-26:1.13.4-25.1.2-2-ubuntu20.04" + container: "ghcr.io/emqx/emqx-builder/5.0-27:1.13.4-25.1.2-2-ubuntu20.04" steps: - uses: actions/checkout@v3 with: diff --git a/.github/workflows/elixir_apps_check.yaml b/.github/workflows/elixir_apps_check.yaml index e912db10f..892cd3a05 100644 --- a/.github/workflows/elixir_apps_check.yaml +++ b/.github/workflows/elixir_apps_check.yaml @@ -8,7 +8,7 @@ jobs: elixir_apps_check: runs-on: ubuntu-latest # just use the latest builder - container: "ghcr.io/emqx/emqx-builder/5.0-26:1.13.4-25.1.2-2-ubuntu20.04" + container: "ghcr.io/emqx/emqx-builder/5.0-27:1.13.4-25.1.2-2-ubuntu20.04" strategy: fail-fast: false diff --git a/.github/workflows/elixir_deps_check.yaml b/.github/workflows/elixir_deps_check.yaml index 5e64d69c4..40d70a902 100644 --- a/.github/workflows/elixir_deps_check.yaml +++ b/.github/workflows/elixir_deps_check.yaml @@ -7,7 +7,7 @@ on: [pull_request, push] jobs: elixir_deps_check: runs-on: ubuntu-20.04 - container: ghcr.io/emqx/emqx-builder/5.0-26:1.13.4-25.1.2-2-ubuntu20.04 + container: ghcr.io/emqx/emqx-builder/5.0-27:1.13.4-25.1.2-2-ubuntu20.04 steps: - name: Checkout diff --git a/.github/workflows/elixir_release.yml b/.github/workflows/elixir_release.yml index f53051d1d..b93b3d63c 100644 --- a/.github/workflows/elixir_release.yml +++ b/.github/workflows/elixir_release.yml @@ -17,7 +17,7 @@ jobs: profile: - emqx - emqx-enterprise - container: ghcr.io/emqx/emqx-builder/5.0-26:1.13.4-25.1.2-2-ubuntu20.04 + container: ghcr.io/emqx/emqx-builder/5.0-27:1.13.4-25.1.2-2-ubuntu20.04 steps: - name: Checkout uses: actions/checkout@v3 diff --git a/.github/workflows/run_emqx_app_tests.yaml b/.github/workflows/run_emqx_app_tests.yaml index ddbb14609..0b3f21b4a 100644 --- a/.github/workflows/run_emqx_app_tests.yaml +++ b/.github/workflows/run_emqx_app_tests.yaml @@ -12,7 +12,7 @@ jobs: strategy: matrix: builder: - - 5.0-26 + - 5.0-27 otp: - 24.3.4.2-1 - 25.1.2-2 diff --git a/.github/workflows/run_fvt_tests.yaml b/.github/workflows/run_fvt_tests.yaml index 1a4568725..f233562d1 100644 --- a/.github/workflows/run_fvt_tests.yaml +++ b/.github/workflows/run_fvt_tests.yaml @@ -16,7 +16,7 @@ jobs: prepare: runs-on: ubuntu-20.04 # prepare source with any OTP version, no need for a matrix - container: ghcr.io/emqx/emqx-builder/5.0-26:1.13.4-24.3.4.2-1-alpine3.15.1 + container: ghcr.io/emqx/emqx-builder/5.0-27:1.13.4-24.3.4.2-1-alpine3.15.1 steps: - uses: actions/checkout@v3 @@ -49,7 +49,7 @@ jobs: os: - ["alpine3.15.1", "alpine:3.15.1"] builder: - - 5.0-26 + - 5.0-27 otp: - 24.3.4.2-1 elixir: @@ -122,7 +122,7 @@ jobs: os: - ["debian11", "debian:11-slim"] builder: - - 5.0-26 + - 5.0-27 otp: - 24.3.4.2-1 elixir: diff --git a/.github/workflows/run_relup_tests.yaml b/.github/workflows/run_relup_tests.yaml index a6854aa40..9ae82a706 100644 --- a/.github/workflows/run_relup_tests.yaml +++ b/.github/workflows/run_relup_tests.yaml @@ -15,7 +15,7 @@ on: jobs: relup_test_plan: runs-on: ubuntu-20.04 - container: "ghcr.io/emqx/emqx-builder/5.0-26:1.13.4-24.3.4.2-1-ubuntu20.04" + container: "ghcr.io/emqx/emqx-builder/5.0-27:1.13.4-24.3.4.2-1-ubuntu20.04" outputs: CUR_EE_VSN: ${{ steps.find-versions.outputs.CUR_EE_VSN }} OLD_VERSIONS: ${{ steps.find-versions.outputs.OLD_VERSIONS }} diff --git a/.github/workflows/run_test_cases.yaml b/.github/workflows/run_test_cases.yaml index ac0edef13..6b4357abc 100644 --- a/.github/workflows/run_test_cases.yaml +++ b/.github/workflows/run_test_cases.yaml @@ -30,12 +30,12 @@ jobs: MATRIX="$(echo "${APPS}" | jq -c ' [ (.[] | select(.profile == "emqx") | . + { - builder: "5.0-26", + builder: "5.0-27", otp: "25.1.2-2", elixir: "1.13.4" }), (.[] | select(.profile == "emqx-enterprise") | . + { - builder: "5.0-26", + builder: "5.0-27", otp: ["24.3.4.2-1", "25.1.2-2"][], elixir: "1.13.4" }) @@ -223,7 +223,7 @@ jobs: - ct - ct_docker runs-on: ubuntu-20.04 - container: "ghcr.io/emqx/emqx-builder/5.0-26:1.13.4-24.3.4.2-1-ubuntu20.04" + container: "ghcr.io/emqx/emqx-builder/5.0-27:1.13.4-24.3.4.2-1-ubuntu20.04" steps: - uses: AutoModality/action-clean@v1 - uses: actions/download-artifact@v3 diff --git a/apps/emqx/src/emqx_broker.erl b/apps/emqx/src/emqx_broker.erl index 1c31d86c2..d56620123 100644 --- a/apps/emqx/src/emqx_broker.erl +++ b/apps/emqx/src/emqx_broker.erl @@ -196,13 +196,13 @@ do_unsubscribe(Topic, SubPid, SubOpts) -> true = ets:delete(?SUBOPTION, {Topic, SubPid}), true = ets:delete_object(?SUBSCRIPTION, {SubPid, Topic}), Group = maps:get(share, SubOpts, undefined), - do_unsubscribe(Group, Topic, SubPid, SubOpts), - emqx_exclusive_subscription:unsubscribe(Topic, SubOpts). + do_unsubscribe(Group, Topic, SubPid, SubOpts). do_unsubscribe(undefined, Topic, SubPid, SubOpts) -> case maps:get(shard, SubOpts, 0) of 0 -> true = ets:delete_object(?SUBSCRIBER, {Topic, SubPid}), + emqx_exclusive_subscription:unsubscribe(Topic, SubOpts), cast(pick(Topic), {unsubscribed, Topic}); I -> true = ets:delete_object(?SUBSCRIBER, {{shard, Topic, I}, SubPid}), @@ -366,14 +366,7 @@ subscriber_down(SubPid) -> SubOpts when is_map(SubOpts) -> _ = emqx_broker_helper:reclaim_seq(Topic), true = ets:delete(?SUBOPTION, {Topic, SubPid}), - case maps:get(shard, SubOpts, 0) of - 0 -> - true = ets:delete_object(?SUBSCRIBER, {Topic, SubPid}), - ok = cast(pick(Topic), {unsubscribed, Topic}); - I -> - true = ets:delete_object(?SUBSCRIBER, {{shard, Topic, I}, SubPid}), - ok = cast(pick({Topic, I}), {unsubscribed, Topic, I}) - end; + do_unsubscribe(undefined, Topic, SubPid, SubOpts); undefined -> ok end diff --git a/apps/emqx/src/emqx_exclusive_subscription.erl b/apps/emqx/src/emqx_exclusive_subscription.erl index afb6317b7..a1f7f76ae 100644 --- a/apps/emqx/src/emqx_exclusive_subscription.erl +++ b/apps/emqx/src/emqx_exclusive_subscription.erl @@ -32,7 +32,8 @@ -export([ check_subscribe/2, - unsubscribe/2 + unsubscribe/2, + clear/0 ]). %% Internal exports (RPC) @@ -77,7 +78,7 @@ on_add_module() -> mnesia(boot). on_delete_module() -> - mria:clear_table(?EXCLUSIVE_SHARD). + clear(). %%-------------------------------------------------------------------- %% APIs @@ -101,6 +102,9 @@ unsubscribe(Topic, #{is_exclusive := true}) -> unsubscribe(_Topic, _SubOpts) -> ok. +clear() -> + mria:clear_table(?TAB). + %%-------------------------------------------------------------------- %% Internal functions %%-------------------------------------------------------------------- diff --git a/apps/emqx/test/emqx_exclusive_sub_SUITE.erl b/apps/emqx/test/emqx_exclusive_sub_SUITE.erl new file mode 100644 index 000000000..79dfc9de6 --- /dev/null +++ b/apps/emqx/test/emqx_exclusive_sub_SUITE.erl @@ -0,0 +1,159 @@ +%%-------------------------------------------------------------------- +%% Copyright (c) 2018-2023 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_exclusive_sub_SUITE). + +-compile(export_all). +-compile(nowarn_export_all). + +-include_lib("emqx/include/emqx.hrl"). +-include_lib("emqx/include/emqx_mqtt.hrl"). +-include_lib("eunit/include/eunit.hrl"). + +-define(EXCLUSIVE_TOPIC, <<"$exclusive/t/1">>). +-define(NORMAL_TOPIC, <<"t/1">>). + +-define(CHECK_SUB(Client, Code), ?CHECK_SUB(Client, ?EXCLUSIVE_TOPIC, Code)). +-define(CHECK_SUB(Client, Topic, Code), + {ok, _, [Code]} = emqtt:subscribe(Client, Topic, []) +). + +all() -> emqx_common_test_helpers:all(?MODULE). + +init_per_suite(Config) -> + emqx_common_test_helpers:start_apps([]), + ok = ekka:start(), + OldConf = emqx:get_config([zones], #{}), + emqx_config:put_zone_conf(default, [mqtt, exclusive_subscription], true), + timer:sleep(50), + [{old_conf, OldConf} | Config]. + +end_per_suite(Config) -> + emqx_config:put([zones], proplists:get_value(old_conf, Config)), + ekka:stop(), + mria:stop(), + mria_mnesia:delete_schema(), + emqx_common_test_helpers:stop_apps([]). + +end_per_testcase(_TestCase, _Config) -> + emqx_exclusive_subscription:clear(). + +t_exclusive_sub(_) -> + {ok, C1} = emqtt:start_link([ + {clientid, <<"client1">>}, + {clean_start, false}, + {proto_ver, v5}, + {properties, #{'Session-Expiry-Interval' => 100}} + ]), + {ok, _} = emqtt:connect(C1), + ?CHECK_SUB(C1, 0), + + {ok, C2} = emqtt:start_link([ + {clientid, <<"client2">>}, + {clean_start, false}, + {proto_ver, v5} + ]), + {ok, _} = emqtt:connect(C2), + ?CHECK_SUB(C2, ?RC_QUOTA_EXCEEDED), + + %% keep exclusive even disconnected + ok = emqtt:disconnect(C1), + timer:sleep(1000), + + ?CHECK_SUB(C2, ?RC_QUOTA_EXCEEDED), + + ok = emqtt:disconnect(C2). + +t_allow_normal_sub(_) -> + {ok, C1} = emqtt:start_link([ + {clientid, <<"client1">>}, + {proto_ver, v5} + ]), + {ok, _} = emqtt:connect(C1), + ?CHECK_SUB(C1, 0), + + {ok, C2} = emqtt:start_link([ + {clientid, <<"client2">>}, + {proto_ver, v5} + ]), + {ok, _} = emqtt:connect(C2), + ?CHECK_SUB(C2, ?NORMAL_TOPIC, 0), + + ok = emqtt:disconnect(C1), + ok = emqtt:disconnect(C2). + +t_unsub(_) -> + {ok, C1} = emqtt:start_link([ + {clientid, <<"client1">>}, + {proto_ver, v5} + ]), + {ok, _} = emqtt:connect(C1), + ?CHECK_SUB(C1, 0), + + {ok, C2} = emqtt:start_link([ + {clientid, <<"client2">>}, + {proto_ver, v5} + ]), + {ok, _} = emqtt:connect(C2), + ?CHECK_SUB(C2, ?RC_QUOTA_EXCEEDED), + + {ok, #{}, [0]} = emqtt:unsubscribe(C1, ?EXCLUSIVE_TOPIC), + + ?CHECK_SUB(C2, 0), + + ok = emqtt:disconnect(C1), + ok = emqtt:disconnect(C2). + +t_clean_session(_) -> + erlang:process_flag(trap_exit, true), + {ok, C1} = emqtt:start_link([ + {clientid, <<"client1">>}, + {clean_start, true}, + {proto_ver, v5}, + {properties, #{'Session-Expiry-Interval' => 0}} + ]), + {ok, _} = emqtt:connect(C1), + ?CHECK_SUB(C1, 0), + + {ok, C2} = emqtt:start_link([ + {clientid, <<"client2">>}, + {proto_ver, v5} + ]), + {ok, _} = emqtt:connect(C2), + ?CHECK_SUB(C2, ?RC_QUOTA_EXCEEDED), + + %% auto clean when session was cleand + ok = emqtt:disconnect(C1), + + timer:sleep(1000), + + ?CHECK_SUB(C2, 0), + + ok = emqtt:disconnect(C2). + +t_feat_disabled(_) -> + OldConf = emqx:get_config([zones], #{}), + emqx_config:put_zone_conf(default, [mqtt, exclusive_subscription], false), + + {ok, C1} = emqtt:start_link([ + {clientid, <<"client1">>}, + {proto_ver, v5} + ]), + {ok, _} = emqtt:connect(C1), + ?CHECK_SUB(C1, ?RC_TOPIC_FILTER_INVALID), + ok = emqtt:disconnect(C1), + + emqx_config:put([zones], OldConf). diff --git a/apps/emqx_authz/src/emqx_authz.app.src b/apps/emqx_authz/src/emqx_authz.app.src index 6a4b721e9..3fea50147 100644 --- a/apps/emqx_authz/src/emqx_authz.app.src +++ b/apps/emqx_authz/src/emqx_authz.app.src @@ -1,7 +1,7 @@ %% -*- mode: erlang -*- {application, emqx_authz, [ {description, "An OTP application"}, - {vsn, "0.1.12"}, + {vsn, "0.1.13"}, {registered, []}, {mod, {emqx_authz_app, []}}, {applications, [ diff --git a/apps/emqx_authz/src/emqx_authz_rule.erl b/apps/emqx_authz/src/emqx_authz_rule.erl index 4aa0983e6..306ca9433 100644 --- a/apps/emqx_authz/src/emqx_authz_rule.erl +++ b/apps/emqx_authz/src/emqx_authz_rule.erl @@ -100,15 +100,17 @@ compile_topic(<<"eq ", Topic/binary>>) -> compile_topic({eq, Topic}) -> {eq, emqx_topic:words(bin(Topic))}; compile_topic(Topic) -> - Words = emqx_topic:words(bin(Topic)), - case pattern(Words) of - true -> {pattern, Words}; - false -> Words + TopicBin = bin(Topic), + case + emqx_placeholder:preproc_tmpl( + TopicBin, + #{placeholders => [?PH_USERNAME, ?PH_CLIENTID]} + ) + of + [{str, _}] -> emqx_topic:words(TopicBin); + Tokens -> {pattern, Tokens} end. -pattern(Words) -> - lists:member(?PH_USERNAME, Words) orelse lists:member(?PH_CLIENTID, Words). - atom(B) when is_binary(B) -> try binary_to_existing_atom(B, utf8) @@ -202,8 +204,8 @@ match_who(_, _) -> match_topics(_ClientInfo, _Topic, []) -> false; match_topics(ClientInfo, Topic, [{pattern, PatternFilter} | Filters]) -> - TopicFilter = feed_var(ClientInfo, PatternFilter), - match_topic(emqx_topic:words(Topic), TopicFilter) orelse + TopicFilter = emqx_placeholder:proc_tmpl(PatternFilter, ClientInfo), + match_topic(emqx_topic:words(Topic), emqx_topic:words(TopicFilter)) orelse match_topics(ClientInfo, Topic, Filters); match_topics(ClientInfo, Topic, [TopicFilter | Filters]) -> match_topic(emqx_topic:words(Topic), TopicFilter) orelse @@ -213,18 +215,3 @@ match_topic(Topic, {'eq', TopicFilter}) -> Topic =:= TopicFilter; match_topic(Topic, TopicFilter) -> emqx_topic:match(Topic, TopicFilter). - -feed_var(ClientInfo, Pattern) -> - feed_var(ClientInfo, Pattern, []). -feed_var(_ClientInfo, [], Acc) -> - lists:reverse(Acc); -feed_var(ClientInfo = #{clientid := undefined}, [?PH_CLIENTID | Words], Acc) -> - feed_var(ClientInfo, Words, [?PH_CLIENTID | Acc]); -feed_var(ClientInfo = #{clientid := ClientId}, [?PH_CLIENTID | Words], Acc) -> - feed_var(ClientInfo, Words, [ClientId | Acc]); -feed_var(ClientInfo = #{username := undefined}, [?PH_USERNAME | Words], Acc) -> - feed_var(ClientInfo, Words, [?PH_USERNAME | Acc]); -feed_var(ClientInfo = #{username := Username}, [?PH_USERNAME | Words], Acc) -> - feed_var(ClientInfo, Words, [Username | Acc]); -feed_var(ClientInfo, [W | Words], Acc) -> - feed_var(ClientInfo, Words, [W | Acc]). diff --git a/apps/emqx_authz/test/emqx_authz_rule_SUITE.erl b/apps/emqx_authz/test/emqx_authz_rule_SUITE.erl index 77f8617ee..76e5677ce 100644 --- a/apps/emqx_authz/test/emqx_authz_rule_SUITE.erl +++ b/apps/emqx_authz/test/emqx_authz_rule_SUITE.erl @@ -35,6 +35,7 @@ ]}, publish, [?PH_S_USERNAME, ?PH_S_CLIENTID]} ). +-define(SOURCE6, {allow, {username, "test"}, publish, ["t/foo${username}boo"]}). all() -> emqx_common_test_helpers:all(?MODULE). @@ -80,7 +81,7 @@ t_compile(_) -> {{127, 0, 0, 1}, {127, 0, 0, 1}, 32}, {{192, 168, 1, 0}, {192, 168, 1, 255}, 24} ]}, - subscribe, [{pattern, [?PH_CLIENTID]}]}, + subscribe, [{pattern, [{var, {var, <<"clientid">>}}]}]}, emqx_authz_rule:compile(?SOURCE3) ), @@ -97,9 +98,18 @@ t_compile(_) -> {username, {re_pattern, _, _, _, _}}, {clientid, {re_pattern, _, _, _, _}} ]}, - publish, [{pattern, [?PH_USERNAME]}, {pattern, [?PH_CLIENTID]}]}, + publish, [ + {pattern, [{var, {var, <<"username">>}}]}, {pattern, [{var, {var, <<"clientid">>}}]} + ]}, emqx_authz_rule:compile(?SOURCE5) ), + + ?assertEqual( + {allow, {username, {eq, <<"test">>}}, publish, [ + {pattern, [{str, <<"t/foo">>}, {var, {var, <<"username">>}}, {str, <<"boo">>}]} + ]}, + emqx_authz_rule:compile(?SOURCE6) + ), ok. t_match(_) -> @@ -307,4 +317,24 @@ t_match(_) -> emqx_authz_rule:compile(?SOURCE5) ) ), + + ?assertEqual( + nomatch, + emqx_authz_rule:match( + ClientInfo1, + publish, + <<"t/foo${username}boo">>, + emqx_authz_rule:compile(?SOURCE6) + ) + ), + + ?assertEqual( + {matched, allow}, + emqx_authz_rule:match( + ClientInfo4, + publish, + <<"t/footestboo">>, + emqx_authz_rule:compile(?SOURCE6) + ) + ), ok. diff --git a/apps/emqx_dashboard/i18n/emqx_dashboard_i18n.conf b/apps/emqx_dashboard/i18n/emqx_dashboard_i18n.conf index 9f6e2e6b0..0b2a3026f 100644 --- a/apps/emqx_dashboard/i18n/emqx_dashboard_i18n.conf +++ b/apps/emqx_dashboard/i18n/emqx_dashboard_i18n.conf @@ -96,6 +96,16 @@ The configuration is only valid when the inet6 is true.""" zh: "IPv6 only" } } + proxy_header { + desc { + en: "Enable support for `HAProxy` header. Be aware once enabled regular HTTP requests can't be handled anymore." + zh: "开启对 `HAProxy` 的支持,注意:一旦开启了这个功能,就无法再处理普通的 HTTP 请求了。" + } + label: { + en: "Enable support for HAProxy header" + zh: "开启对 `HAProxy` 的支持" + } + } desc_dashboard { desc { en: "Configuration for EMQX dashboard." diff --git a/apps/emqx_dashboard/src/emqx_dashboard.app.src b/apps/emqx_dashboard/src/emqx_dashboard.app.src index 57d63247b..b6c95ca97 100644 --- a/apps/emqx_dashboard/src/emqx_dashboard.app.src +++ b/apps/emqx_dashboard/src/emqx_dashboard.app.src @@ -2,7 +2,7 @@ {application, emqx_dashboard, [ {description, "EMQX Web Dashboard"}, % strict semver, bump manually! - {vsn, "5.0.12"}, + {vsn, "5.0.13"}, {modules, []}, {registered, [emqx_dashboard_sup]}, {applications, [kernel, stdlib, mnesia, minirest, emqx]}, diff --git a/apps/emqx_dashboard/src/emqx_dashboard.erl b/apps/emqx_dashboard/src/emqx_dashboard.erl index 36c7660cc..2afc8b362 100644 --- a/apps/emqx_dashboard/src/emqx_dashboard.erl +++ b/apps/emqx_dashboard/src/emqx_dashboard.erl @@ -92,8 +92,8 @@ start_listeners(Listeners) -> }, Res = lists:foldl( - fun({Name, Protocol, Bind, RanchOptions}, Acc) -> - Minirest = BaseMinirest#{protocol => Protocol}, + fun({Name, Protocol, Bind, RanchOptions, ProtoOpts}, Acc) -> + Minirest = BaseMinirest#{protocol => Protocol, protocol_options => ProtoOpts}, case minirest:start(Name, RanchOptions, Minirest) of {ok, _} -> ?ULOG("Listener ~ts on ~ts started.~n", [ @@ -125,7 +125,7 @@ stop_listeners(Listeners) -> ?SLOG(warning, #{msg => "stop_listener_failed", name => Name, port => Port}) end end - || {Name, _, Port, _} <- listeners(Listeners) + || {Name, _, Port, _, _} <- listeners(Listeners) ], ok. @@ -164,7 +164,13 @@ listeners(Listeners) -> maps:get(enable, Conf) andalso begin {Conf1, Bind} = ip_port(Conf), - {true, {listener_name(Protocol), Protocol, Bind, ranch_opts(Conf1)}} + {true, { + listener_name(Protocol), + Protocol, + Bind, + ranch_opts(Conf1), + proto_opts(Conf1) + }} end end, maps:to_list(Listeners) @@ -197,7 +203,7 @@ ranch_opts(Options) -> SocketOpts = maps:fold( fun filter_false/3, [], - maps:without([enable, inet6, ipv6_v6only | Keys], Options) + maps:without([enable, inet6, ipv6_v6only, proxy_header | Keys], Options) ), InetOpts = case Options of @@ -210,6 +216,9 @@ ranch_opts(Options) -> end, RanchOpts#{socket_opts => InetOpts ++ SocketOpts}. +proto_opts(Options) -> + maps:with([proxy_header], Options). + filter_false(_K, false, S) -> S; filter_false(K, V, S) -> [{K, V} | S]. diff --git a/apps/emqx_dashboard/src/emqx_dashboard_schema.erl b/apps/emqx_dashboard/src/emqx_dashboard_schema.erl index 5a2eb590b..ceb2415f8 100644 --- a/apps/emqx_dashboard/src/emqx_dashboard_schema.erl +++ b/apps/emqx_dashboard/src/emqx_dashboard_schema.erl @@ -160,6 +160,14 @@ common_listener_fields() -> default => false, desc => ?DESC(ipv6_v6only) } + )}, + {"proxy_header", + ?HOCON( + boolean(), + #{ + desc => ?DESC(proxy_header), + default => false + } )} ]. diff --git a/apps/emqx_dashboard/test/emqx_dashboard_api_test_helpers.erl b/apps/emqx_dashboard/test/emqx_dashboard_api_test_helpers.erl index 87a3654ac..8df130897 100644 --- a/apps/emqx_dashboard/test/emqx_dashboard_api_test_helpers.erl +++ b/apps/emqx_dashboard/test/emqx_dashboard_api_test_helpers.erl @@ -19,6 +19,7 @@ -export([ set_default_config/0, set_default_config/1, + set_default_config/2, request/2, request/3, request/4, @@ -36,6 +37,9 @@ set_default_config() -> set_default_config(<<"admin">>). set_default_config(DefaultUsername) -> + set_default_config(DefaultUsername, false). + +set_default_config(DefaultUsername, HAProxyEnabled) -> Config = #{ listeners => #{ http => #{ @@ -46,7 +50,8 @@ set_default_config(DefaultUsername) -> max_connections => 512, num_acceptors => 4, send_timeout => 5000, - backlog => 512 + backlog => 512, + proxy_header => HAProxyEnabled } }, default_username => DefaultUsername, diff --git a/apps/emqx_dashboard/test/emqx_dashboard_haproxy_SUITE.erl b/apps/emqx_dashboard/test/emqx_dashboard_haproxy_SUITE.erl new file mode 100644 index 000000000..a05de339b --- /dev/null +++ b/apps/emqx_dashboard/test/emqx_dashboard_haproxy_SUITE.erl @@ -0,0 +1,100 @@ +%%-------------------------------------------------------------------- +%% Copyright (c) 2020-2023 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_dashboard_haproxy_SUITE). + +-compile(nowarn_export_all). +-compile(export_all). + +-include_lib("eunit/include/eunit.hrl"). +-include("emqx_dashboard.hrl"). + +all() -> + emqx_common_test_helpers:all(?MODULE). + +init_per_suite(Config) -> + emqx_common_test_helpers:start_apps( + [emqx_management, emqx_dashboard], + fun set_special_configs/1 + ), + Config. + +set_special_configs(emqx_dashboard) -> + emqx_dashboard_api_test_helpers:set_default_config(<<"admin">>, true), + ok; +set_special_configs(_) -> + ok. + +end_per_suite(Config) -> + application:unload(emqx_management), + mnesia:clear_table(?ADMIN), + emqx_common_test_helpers:stop_apps([emqx_dashboard, emqx_management]), + mria:stop(), + Config. + +t_status(_Config) -> + ProxyInfo = #{ + version => 1, + command => proxy, + transport_family => ipv4, + transport_protocol => stream, + src_address => {127, 0, 0, 1}, + src_port => 444, + dest_address => {192, 168, 0, 1}, + dest_port => 443 + }, + {ok, Socket} = gen_tcp:connect( + "localhost", + 18083, + [binary, {active, false}, {packet, raw}] + ), + ok = gen_tcp:send(Socket, ranch_proxy_header:header(ProxyInfo)), + {ok, Token} = emqx_dashboard_admin:sign_token(<<"admin">>, <<"public">>), + ok = gen_tcp:send( + Socket, + "GET /status HTTP/1.1\r\n" + "Host: localhost\r\n" + "Authorization: Bearer " ++ binary_to_list(Token) ++ + "\r\n" + "\r\n" + ), + {_, 200, _, Rest0} = cow_http:parse_status_line(raw_recv_head(Socket)), + {Headers, Body0} = cow_http:parse_headers(Rest0), + {_, LenBin} = lists:keyfind(<<"content-length">>, 1, Headers), + Len = binary_to_integer(LenBin), + Body = + if + byte_size(Body0) =:= Len -> + Body0; + true -> + {ok, Body1} = gen_tcp:recv(Socket, Len - byte_size(Body0), 5000), + <> + end, + ?assertMatch({match, _}, re:run(Body, "Node .+ is started\nemqx is running")), + ok. + +raw_recv_head(Socket) -> + {ok, Data} = gen_tcp:recv(Socket, 0, 10000), + raw_recv_head(Socket, Data). + +raw_recv_head(Socket, Buffer) -> + case binary:match(Buffer, <<"\r\n\r\n">>) of + nomatch -> + {ok, Data} = gen_tcp:recv(Socket, 0, 10000), + raw_recv_head(Socket, <>); + {_, _} -> + Buffer + end. diff --git a/apps/emqx_modules/test/emqx_delayed_SUITE.erl b/apps/emqx_modules/test/emqx_delayed_SUITE.erl index ab35e37dc..ffea436bb 100644 --- a/apps/emqx_modules/test/emqx_delayed_SUITE.erl +++ b/apps/emqx_modules/test/emqx_delayed_SUITE.erl @@ -229,6 +229,14 @@ t_banned_delayed(_) -> }), snabbkaffe:start_trace(), + {ok, SubRef} = + snabbkaffe:subscribe( + ?match_event(#{?snk_kind := ignore_delayed_message_publish}), + _NEvents = 2, + _Timeout = 10000, + 0 + ), + lists:foreach( fun(ClientId) -> Msg = emqx_message:make(ClientId, <<"$delayed/1/bc">>, <<"payload">>), @@ -237,8 +245,7 @@ t_banned_delayed(_) -> [ClientId1, ClientId1, ClientId1, ClientId2, ClientId2] ), - timer:sleep(2000), - Trace = snabbkaffe:collect_trace(), + {ok, Trace} = snabbkaffe:receive_events(SubRef), snabbkaffe:stop(), emqx_banned:delete(Who), mnesia:clear_table(emqx_delayed), diff --git a/apps/emqx_retainer/test/emqx_retainer_SUITE.erl b/apps/emqx_retainer/test/emqx_retainer_SUITE.erl index 845f07802..09d1f77da 100644 --- a/apps/emqx_retainer/test/emqx_retainer_SUITE.erl +++ b/apps/emqx_retainer/test/emqx_retainer_SUITE.erl @@ -687,11 +687,19 @@ t_deliver_when_banned(_) -> }), timer:sleep(100), - snabbkaffe:start_trace(), - {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained/+">>, [{qos, 0}, {rh, 0}]), - timer:sleep(500), - Trace = snabbkaffe:collect_trace(), + snabbkaffe:start_trace(), + {ok, SubRef} = + snabbkaffe:subscribe( + ?match_event(#{?snk_kind := ignore_retained_message_deliver}), + _NEvents = 3, + _Timeout = 10000, + 0 + ), + + {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained/+">>, [{qos, 0}, {rh, 0}]), + + {ok, Trace} = snabbkaffe:receive_events(SubRef), ?assertEqual(3, length(?of_kind(ignore_retained_message_deliver, Trace))), snabbkaffe:stop(), emqx_banned:delete(Who), diff --git a/changes/v5.0.16/feat-9802.en.md b/changes/v5.0.16/feat-9802.en.md new file mode 100644 index 000000000..ac314879a --- /dev/null +++ b/changes/v5.0.16/feat-9802.en.md @@ -0,0 +1 @@ +Support HAProxy protocol for dashboard API. diff --git a/changes/v5.0.16/feat-9802.zh.md b/changes/v5.0.16/feat-9802.zh.md new file mode 100644 index 000000000..9afcd8f17 --- /dev/null +++ b/changes/v5.0.16/feat-9802.zh.md @@ -0,0 +1 @@ +现在 dashboard 增加了对 `HAProxy` 协议的支持。 diff --git a/changes/v5.0.16/feat-9871.en.md b/changes/v5.0.16/feat-9871.en.md new file mode 100644 index 000000000..b907aa3f1 --- /dev/null +++ b/changes/v5.0.16/feat-9871.en.md @@ -0,0 +1,3 @@ +Allow the placeholder to be anywhere in the topic for `authz` rules. +e.g: +`{allow, {username, "who"}, publish, ["t/foo${username}boo/${clientid}xxx"]}.` diff --git a/changes/v5.0.16/feat-9871.zh.md b/changes/v5.0.16/feat-9871.zh.md new file mode 100644 index 000000000..ecd526a93 --- /dev/null +++ b/changes/v5.0.16/feat-9871.zh.md @@ -0,0 +1,3 @@ +允许占位符出现在 `authz` 规则中的主题里的任意位置。 +例如: +`{allow, {username, "who"}, publish, ["t/foo${username}boo/${clientid}xxx"]}.` diff --git a/changes/v5.0.16/fix-9864.en.md b/changes/v5.0.16/fix-9864.en.md new file mode 100644 index 000000000..ad5cd53c9 --- /dev/null +++ b/changes/v5.0.16/fix-9864.en.md @@ -0,0 +1 @@ +Fix the exclusive topics aren't removed when the session has already been cleaned. diff --git a/changes/v5.0.16/fix-9864.zh.md b/changes/v5.0.16/fix-9864.zh.md new file mode 100644 index 000000000..b79d81988 --- /dev/null +++ b/changes/v5.0.16/fix-9864.zh.md @@ -0,0 +1 @@ +修复会话清除后相关的排他订阅主题没有被清理的问题。 diff --git a/mix.exs b/mix.exs index 6fbddb28b..18e11fdd7 100644 --- a/mix.exs +++ b/mix.exs @@ -56,7 +56,7 @@ defmodule EMQXUmbrella.MixProject do {:ekka, github: "emqx/ekka", tag: "0.13.9", override: true}, {:gen_rpc, github: "emqx/gen_rpc", tag: "2.8.1", override: true}, {:grpc, github: "emqx/grpc-erl", tag: "0.6.7", override: true}, - {:minirest, github: "emqx/minirest", tag: "1.3.7", override: true}, + {:minirest, github: "emqx/minirest", tag: "1.3.8", override: true}, {:ecpool, github: "emqx/ecpool", tag: "0.5.3", override: true}, {:replayq, github: "emqx/replayq", tag: "0.3.7", override: true}, {:pbkdf2, github: "emqx/erlang-pbkdf2", tag: "2.0.4", override: true}, diff --git a/rebar.config b/rebar.config index e61aac544..1a9d651dc 100644 --- a/rebar.config +++ b/rebar.config @@ -58,7 +58,7 @@ , {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.13.9"}}} , {gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "2.8.1"}}} , {grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.7"}}} - , {minirest, {git, "https://github.com/emqx/minirest", {tag, "1.3.7"}}} + , {minirest, {git, "https://github.com/emqx/minirest", {tag, "1.3.8"}}} , {ecpool, {git, "https://github.com/emqx/ecpool", {tag, "0.5.3"}}} , {replayq, {git, "https://github.com/emqx/replayq.git", {tag, "0.3.7"}}} , {pbkdf2, {git, "https://github.com/emqx/erlang-pbkdf2.git", {tag, "2.0.4"}}}