Merge pull request #9290 from lafirest/feat/dealyed_as_capability

fix(acl): Check the real topic in delayed messages

apps/emqx/src/emqx_access_control.erl

@@ -24,6 +24,11 @@
     authorize/3
 ]).
 
+-ifdef(TEST).
+-compile(export_all).
+-compile(nowarn_export_all).
+-endif.
+
 %%--------------------------------------------------------------------
 %% APIs
 %%--------------------------------------------------------------------
@@ -45,6 +50,19 @@ authenticate(Credential) ->
 %% @doc Check Authorization
 -spec authorize(emqx_types:clientinfo(), emqx_types:pubsub(), emqx_types:topic()) ->
     allow | deny.
+authorize(ClientInfo, PubSub, <<"$delayed/", Data/binary>> = RawTopic) ->
+    case binary:split(Data, <<"/">>) of
+        [_, Topic] ->
+            authorize(ClientInfo, PubSub, Topic);
+        _ ->
+            ?SLOG(warning, #{
+                msg => "invalid_dealyed_topic_format",
+                expected_example => "$delayed/1/t/foo",
+                got => RawTopic
+            }),
+            inc_authz_metrics(deny),
+            deny
+    end;
 authorize(ClientInfo, PubSub, Topic) ->
     Result =
         case emqx_authz_cache:is_enabled() of

apps/emqx/test/emqx_access_control_SUITE.erl

@@ -32,6 +32,12 @@ init_per_suite(Config) ->
 end_per_suite(_Config) ->
     emqx_common_test_helpers:stop_apps([]).
 
+end_per_testcase(t_delayed_authorize, Config) ->
+    meck:unload(emqx_access_control),
+    Config;
+end_per_testcase(_, Config) ->
+    Config.
+
 t_authenticate(_) ->
     ?assertMatch({ok, _}, emqx_access_control:authenticate(clientinfo())).
 
@@ -39,6 +45,28 @@ t_authorize(_) ->
     Publish = ?PUBLISH_PACKET(?QOS_0, <<"t">>, 1, <<"payload">>),
     ?assertEqual(allow, emqx_access_control:authorize(clientinfo(), Publish, <<"t">>)).
 
+t_delayed_authorize(_) ->
+    RawTopic = "$dealyed/1/foo/2",
+    InvalidTopic = "$dealyed/1/foo/3",
+    Topic = "foo/2",
+
+    ok = meck:new(emqx_access_control, [passthrough, no_history, no_link]),
+    ok = meck:expect(
+        emqx_access_control,
+        do_authorize,
+        fun
+            (_, _, Topic) -> allow;
+            (_, _, _) -> deny
+        end
+    ),
+
+    Publish1 = ?PUBLISH_PACKET(?QOS_0, RawTopic, 1, <<"payload">>),
+    ?assertEqual(allow, emqx_access_control:authorize(clientinfo(), Publish1, RawTopic)),
+
+    Publish2 = ?PUBLISH_PACKET(?QOS_0, InvalidTopic, 1, <<"payload">>),
+    ?assertEqual(allow, emqx_access_control:authorize(clientinfo(), Publish2, InvalidTopic)),
+    ok.
+
 %%--------------------------------------------------------------------
 %% Helper functions
 %%--------------------------------------------------------------------

changes/v5.0.10-en.md

@@ -32,3 +32,6 @@
   the encoding (to JSON) of the event will fail.
 
 - Fix bad HTTP response status code for `/gateways` API, when Gateway name is unknown, it should return `404` instead of `400` [#9268](https://github.com/emqx/emqx/pull/9268).
+
+- Fix incorrect topic authorize checking of delayed messages [#9290](https://github.com/emqx/emqx/pull/9290).
+  Now will determine the actual topic of the delayed messages, e.g. `$delayed/1/t/foo` will be treated as `t/foo` in authorize checks.

changes/v5.0.10-zh.md

@@ -30,3 +30,6 @@
   `$events/message_dropped`, 如果消息事件是共享订阅产生的，在编码（到 JSON 格式）过程中会失败。
 
 - 修复 HTTP API `/gateways` 的返回状态码，未知 Gateway 名字应返回 `404` 而不是 `400` [#9268](https://github.com/emqx/emqx/pull/9268)。
+
+- 修复延迟消息的主题授权判断不正确的问题 [#9290](https://github.com/emqx/emqx/pull/9290)。
+  现在将会对延迟消息中的真实主题进行授权判断，比如，`$delayed/1/t/foo` 会被当作 `t/foo` 进行判断。