feat: add jwt worker for rule actions

Will be used in GCP PubSub resource for EE.

apps/emqx_rule_engine/include/rule_actions.hrl

@@ -28,3 +28,5 @@
 -define(bound_v(Key, ENVS0),
     maps:get(Key,
         maps:get(?BINDING_KEYS, ENVS0, #{}))).
+
+-define(JWT_TABLE, emqx_rule_engine_jwt_table).

apps/emqx_rule_engine/src/emqx_rule_engine.app.src

@@ -2,7 +2,7 @@
  [{description, "EMQ X Rule Engine"},
   {vsn, "4.4.11"}, % strict semver, bump manually!
   {modules, []},
-  {registered, [emqx_rule_engine_sup, emqx_rule_registry]},
+  {registered, [emqx_rule_engine_sup, emqx_rule_registry, emqx_rule_engine_jwt_sup]},
   {applications, [kernel,stdlib,rulesql,getopt]},
   {mod, {emqx_rule_engine_app, []}},
   {env, []},

apps/emqx_rule_engine/src/emqx_rule_engine.appup.src

@@ -2,7 +2,11 @@
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
   [{"4.4.10",
-    [{load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_rule_engine_jwt_worker},
+     {add_module,emqx_rule_engine_jwt_sup},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
+     {apply,{emqx_rule_engine_sup,start_jwt_sup,[]}},
+     {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
@@ -12,7 +16,11 @@
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
    {"4.4.9",
-    [{load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_rule_engine_jwt_worker},
+     {add_module,emqx_rule_engine_jwt_sup},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
+     {apply,{emqx_rule_engine_sup,start_jwt_sup,[]}},
+     {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
@@ -23,7 +31,11 @@
      {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]}]},
    {"4.4.8",
-    [{load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_rule_engine_jwt_worker},
+     {add_module,emqx_rule_engine_jwt_sup},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
+     {apply,{emqx_rule_engine_sup,start_jwt_sup,[]}},
+     {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
@@ -35,7 +47,11 @@
      {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]}]},
    {<<"4\\.4\\.[6-7]">>,
-    [{load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_rule_engine_jwt_worker},
+     {add_module,emqx_rule_engine_jwt_sup},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
+     {apply,{emqx_rule_engine_sup,start_jwt_sup,[]}},
+     {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
@@ -48,7 +64,11 @@
      {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
    {"4.4.5",
-    [{load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_rule_engine_jwt_worker},
+     {add_module,emqx_rule_engine_jwt_sup},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
+     {apply,{emqx_rule_engine_sup,start_jwt_sup,[]}},
+     {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
@@ -62,7 +82,11 @@
      {load_module,emqx_rule_validator,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]}]},
    {"4.4.4",
-    [{load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_rule_engine_jwt_worker},
+     {add_module,emqx_rule_engine_jwt_sup},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
+     {apply,{emqx_rule_engine_sup,start_jwt_sup,[]}},
+     {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
@@ -76,7 +100,11 @@
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]}]},
    {"4.4.3",
-    [{load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_rule_engine_jwt_worker},
+     {add_module,emqx_rule_engine_jwt_sup},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
+     {apply,{emqx_rule_engine_sup,start_jwt_sup,[]}},
+     {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
@@ -92,7 +120,11 @@
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]}]},
    {"4.4.2",
-    [{load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_rule_engine_jwt_worker},
+     {add_module,emqx_rule_engine_jwt_sup},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
+     {apply,{emqx_rule_engine_sup,start_jwt_sup,[]}},
+     {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_validator,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
@@ -109,7 +141,11 @@
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]}]},
    {"4.4.1",
-    [{load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_rule_engine_jwt_worker},
+     {add_module,emqx_rule_engine_jwt_sup},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
+     {apply,{emqx_rule_engine_sup,start_jwt_sup,[]}},
+     {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_validator,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
@@ -126,7 +162,11 @@
      {add_module,emqx_rule_date},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
    {"4.4.0",
-    [{load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_rule_engine_jwt_worker},
+     {add_module,emqx_rule_engine_jwt_sup},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
+     {apply,{emqx_rule_engine_sup,start_jwt_sup,[]}},
+     {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_validator,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
@@ -149,37 +189,47 @@
      {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
-     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {delete_module,emqx_rule_engine_jwt_sup},
+     {delete_module,emqx_rule_engine_jwt_worker}]},
    {"4.4.9",
     [{load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
-     {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
+     {delete_module,emqx_rule_engine_jwt_sup},
+     {delete_module,emqx_rule_engine_jwt_worker}]},
    {"4.4.8",
     [{load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
-     {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
+     {delete_module,emqx_rule_engine_jwt_sup},
+     {delete_module,emqx_rule_engine_jwt_worker}]},
    {<<"4\\.4\\.[6-7]">>,
     [{load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
@@ -188,11 +238,14 @@
      {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
-     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {delete_module,emqx_rule_engine_jwt_sup},
+     {delete_module,emqx_rule_engine_jwt_worker}]},
    {"4.4.5",
     [{load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
@@ -202,11 +255,14 @@
      {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_validator,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
-     {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
+     {delete_module,emqx_rule_engine_jwt_sup},
+     {delete_module,emqx_rule_engine_jwt_worker}]},
    {"4.4.4",
     [{load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
@@ -216,10 +272,13 @@
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
-     {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
+     {delete_module,emqx_rule_engine_jwt_sup},
+     {delete_module,emqx_rule_engine_jwt_worker}]},
    {"4.4.3",
     [{load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_validator,brutal_purge,soft_purge,[]},
@@ -232,10 +291,13 @@
      {load_module,emqx_rule_maps,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
-     {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
+     {delete_module,emqx_rule_engine_jwt_sup},
+     {delete_module,emqx_rule_engine_jwt_worker}]},
    {"4.4.2",
     [{load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_validator,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
@@ -249,10 +311,13 @@
      {delete_module,emqx_rule_date},
      {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
-     {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
+     {delete_module,emqx_rule_engine_jwt_sup},
+     {delete_module,emqx_rule_engine_jwt_worker}]},
    {"4.4.1",
     [{load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_validator,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_maps,brutal_purge,soft_purge,[]},
@@ -266,10 +331,13 @@
      {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
      {delete_module,emqx_rule_date},
-     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {delete_module,emqx_rule_engine_jwt_sup},
+     {delete_module,emqx_rule_engine_jwt_worker}]},
    {"4.4.0",
     [{load_module,emqx_rule_monitor,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine_sup,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_validator,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_maps,brutal_purge,soft_purge,[]},
@@ -283,5 +351,7 @@
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
-     {delete_module,emqx_rule_date}]},
+     {delete_module,emqx_rule_date},
+     {delete_module,emqx_rule_engine_jwt_sup},
+     {delete_module,emqx_rule_engine_jwt_worker}]},
    {<<".*">>,[]}]}.

apps/emqx_rule_engine/src/emqx_rule_engine_jwt_sup.erl

@@ -0,0 +1,57 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 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_rule_engine_jwt_sup).
+
+-behaviour(supervisor).
+
+-export([ start_link/0
+        , start_worker/2
+        , stop_worker/1
+        ]).
+
+-export([init/1]).
+
+start_link() ->
+    supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+init([]) ->
+    SupFlags = #{ strategy => one_for_one
+                , intensity => 10
+                , period => 5
+                , auto_shutdown => never
+                },
+    ChildSpecs = [],
+    {ok, {SupFlags, ChildSpecs}}.
+
+start_worker(Id, Config) ->
+    Ref = erlang:alias([reply]),
+    ChildSpec = jwt_worker_child_spec(Id, Config, Ref),
+    {ok, Pid} = supervisor:start_child(?MODULE, ChildSpec),
+    {Ref, Pid}.
+
+stop_worker(Id) ->
+    supervisor:terminate_child(?MODULE, Id).
+
+jwt_worker_child_spec(Id, Config, Ref) ->
+    #{ id => Id
+     , start => {emqx_rule_engine_jwt_worker, start_link, [Config, Ref]}
+     , restart => permanent
+     , type => worker
+     , significant => false
+     , shutdown => brutal_kill
+     , modules => [emqx_rule_engine_jwt_worker]
+     }.

apps/emqx_rule_engine/src/emqx_rule_engine_jwt_worker.erl

@@ -0,0 +1,220 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 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_rule_engine_jwt_worker).
+
+-behaviour(gen_server).
+
+%% API
+-export([ start_link/2
+        , lookup_jwt/1
+        , lookup_jwt/2
+        ]).
+
+%% gen_server API
+-export([ init/1
+        , handle_continue/2
+        , handle_call/3
+        , handle_cast/2
+        , handle_info/2
+        , format_status/1
+        , format_status/2
+        ]).
+
+-include_lib("jose/include/jose_jwk.hrl").
+-include_lib("emqx_rule_engine/include/rule_engine.hrl").
+-include_lib("emqx_rule_engine/include/rule_actions.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+
+-type config() :: #{ private_key := binary()
+                   , resource_id := resource_id()
+                   , expiration := timer:time()
+                   , table := ets:table()
+                   , iss := binary()
+                   , sub := binary()
+                   , aud := binary()
+                   , kid := binary()
+                   , alg := binary()
+                   }.
+-type jwt() :: binary().
+-type state() :: #{ refresh_timer := undefined | timer:tref()
+                  , resource_id := resource_id()
+                  , expiration := timer:time()
+                  , table := ets:table()
+                  , jwt := undefined | jwt()
+                    %% only undefined during startup
+                  , jwk := undefined | jose_jwk:key()
+                  , iss := binary()
+                  , sub := binary()
+                  , aud := binary()
+                  , kid := binary()
+                  , alg := binary()
+                  }.
+
+-define(refresh_jwt, refresh_jwt).
+
+%%-----------------------------------------------------------------------------------------
+%% API
+%%-----------------------------------------------------------------------------------------
+
+-spec start_link(config(), reference()) -> gen_server:start_ret().
+start_link(#{ private_key := _
+            , expiration := _
+            , resource_id := _
+            , table := _
+            , iss := _
+            , sub := _
+            , aud := _
+            , kid := _
+            , alg := _
+            } = Config,
+           Ref) ->
+    gen_server:start_link(?MODULE, {Config, Ref}, []).
+
+-spec lookup_jwt(resource_id()) -> {ok, jwt()} | {error, not_found}.
+lookup_jwt(ResourceId) ->
+    ?MODULE:lookup_jwt(?JWT_TABLE, ResourceId).
+
+-spec lookup_jwt(ets:table(), resource_id()) -> {ok, jwt()} | {error, not_found}.
+lookup_jwt(TId, ResourceId) ->
+    try
+        case ets:lookup(TId, {ResourceId, jwt}) of
+            [{{ResourceId, jwt}, JWT}] ->
+                {ok, JWT};
+            [] ->
+                {error, not_found}
+        end
+    catch
+        error:badarg ->
+            {error, not_found}
+    end.
+
+%%-----------------------------------------------------------------------------------------
+%% gen_server API
+%%-----------------------------------------------------------------------------------------
+
+-spec init({config(), Ref}) -> {ok, state(), {continue, {make_key, binary(), Ref}}}
+                               | {stop, {error, term()}}
+              when Ref :: reference().
+init({#{private_key := PrivateKeyPEM} = Config, Ref}) ->
+    {ok, _} = application:ensure_all_started(jose),
+    State0 = maps:without([private_key], Config),
+    State = State0#{ jwk => undefined
+                   , jwt => undefined
+                   , refresh_timer => undefined
+                   },
+    {ok, State, {continue, {make_key, PrivateKeyPEM, Ref}}}.
+
+handle_continue({make_key, PrivateKeyPEM, Ref}, State0) ->
+    case jose_jwk:from_pem(PrivateKeyPEM) of
+        JWK = #jose_jwk{} ->
+            State = State0#{jwk := JWK},
+            {noreply, State, {continue, {create_token, Ref}}};
+        [] ->
+            Ref ! {Ref, {error, {invalid_private_key, empty_key}}},
+            {stop, {error, empty_key}, State0};
+        {error, Reason} ->
+            Ref ! {Ref, {error, {invalid_private_key, Reason}}},
+            {stop, {error, Reason}, State0};
+        Error ->
+            Ref ! {Ref, {error, {invalid_private_key, Error}}},
+            {stop, {error, Error}, State0}
+    end;
+handle_continue({create_token, Ref}, State0) ->
+    JWT = do_generate_jwt(State0),
+    store_jwt(State0, JWT),
+    State1 = State0#{jwt := JWT},
+    State = ensure_timer(State1),
+    Ref ! {Ref, token_created},
+    {noreply, State}.
+
+handle_call(_Req, _From, State) ->
+    {reply, {error, bad_call}, State}.
+
+handle_cast(_Req, State) ->
+    {noreply, State}.
+
+handle_info({timeout, TRef, ?refresh_jwt}, State0 = #{refresh_timer := TRef}) ->
+    JWT = do_generate_jwt(State0),
+    store_jwt(State0, JWT),
+    ?tp(rule_engine_jwt_worker_refresh, #{}),
+    State1 = State0#{jwt := JWT},
+    State = ensure_timer(State1#{refresh_timer := undefined}),
+    {noreply, State};
+handle_info(_Msg, State) ->
+    {noreply, State}.
+
+format_status(State) ->
+    censor_secrets(State).
+
+format_status(_Opt, [_PDict, State0]) ->
+    State = censor_secrets(State0),
+    [{data, [{"State", State}]}].
+
+%%-----------------------------------------------------------------------------------------
+%% Helper fns
+%%-----------------------------------------------------------------------------------------
+
+-spec do_generate_jwt(state()) -> jwt().
+do_generate_jwt(#{ expiration := ExpirationMS
+                 , iss := Iss
+                 , sub := Sub
+                 , aud := Aud
+                 , kid := KId
+                 , alg := Alg
+                 , jwk := JWK
+                 } = _State) ->
+    Headers = #{ <<"alg">> => Alg
+               , <<"kid">> => KId
+               },
+    Now = erlang:system_time(seconds),
+    ExpirationS = erlang:convert_time_unit(ExpirationMS, millisecond, second),
+    Claims = #{ <<"iss">> => Iss
+              , <<"sub">> => Sub
+              , <<"aud">> => Aud
+              , <<"iat">> => Now
+              , <<"exp">> => Now + ExpirationS
+              },
+    JWT0 = jose_jwt:sign(JWK, Headers, Claims),
+    {_, JWT} = jose_jws:compact(JWT0),
+    JWT.
+
+-spec store_jwt(state(), jwt()) -> ok.
+store_jwt(#{resource_id := ResourceId, table := TId}, JWT) ->
+    true = ets:insert(TId, {{ResourceId, jwt}, JWT}),
+    ?tp(jwt_worker_token_stored, #{resource_id => ResourceId}),
+    ok.
+
+-spec ensure_timer(state()) -> state().
+ensure_timer(State = #{ refresh_timer := undefined
+                      , expiration := ExpirationMS0
+                      }) ->
+    ExpirationMS = max(5_000, ExpirationMS0 - 5_000),
+    TRef = erlang:start_timer(ExpirationMS, self(), ?refresh_jwt),
+    State#{refresh_timer => TRef};
+ensure_timer(State) ->
+    State.
+
+-spec censor_secrets(state()) -> map().
+censor_secrets(State) ->
+    maps:map(
+     fun(Key, _Value) when Key =:= jwt;
+                           Key =:= jwk ->
+             "******";
+        (_Key, Value) ->
+             Value
+     end,
+     State).

apps/emqx_rule_engine/src/emqx_rule_engine_sup.erl

@@ -22,7 +22,9 @@
 
 -export([start_link/0]).
 
--export([start_locker/0]).
+-export([ start_locker/0
+        , start_jwt_sup/0
+        ]).
 
 -export([init/1]).
 
@@ -31,8 +33,12 @@ start_link() ->
 
 init([]) ->
     Opts = [public, named_table, set, {read_concurrency, true}],
-    _ = ets:new(?ACTION_INST_PARAMS_TAB, [{keypos, #action_instance_params.id}|Opts]),
+    ensure_table(?ACTION_INST_PARAMS_TAB, [{keypos, #action_instance_params.id}|Opts]),
-    _ = ets:new(?RES_PARAMS_TAB, [{keypos, #resource_params.id}|Opts]),
+    ensure_table(?RES_PARAMS_TAB, [{keypos, #resource_params.id}|Opts]),
+    SupFlags = #{ strategy => one_for_one
+                , intensity => 10
+                , period => 10
+                },
     Registry = #{id => emqx_rule_registry,
                  start => {emqx_rule_registry, start_link, []},
                  restart => permanent,
@@ -51,7 +57,8 @@ init([]) ->
                 shutdown => 5000,
                 type => worker,
                 modules => [emqx_rule_monitor]},
-    {ok, {{one_for_one, 10, 10}, [Registry, Metrics, Monitor]}}.
+    JWTSup = jwt_sup_child_spec(),
+    {ok, {SupFlags, [Registry, Metrics, Monitor, JWTSup]}}.
 
 start_locker() ->
     Locker = #{id => emqx_rule_locker,
@@ -61,3 +68,32 @@ start_locker() ->
                type => worker,
                modules => [emqx_rule_locker]},
     supervisor:start_child(?MODULE, Locker).
+
+start_jwt_sup() ->
+    JWTSup = jwt_sup_child_spec(),
+    supervisor:start_child(?MODULE, JWTSup).
+
+jwt_sup_child_spec() ->
+    #{ id => emqx_rule_engine_jwt_sup
+     , start => {emqx_rule_engine_jwt_sup, start_link, []}
+     , type => supervisor
+     , restart => permanent
+     , shutdown => 5_000
+     , modules => [emqx_rule_engine_jwt_sup]
+     }.
+
+ensure_table(Name, Opts) ->
+    try
+        case ets:whereis(name) of
+            undefined ->
+                _ = ets:new(Name, Opts),
+                ok;
+            _ ->
+                ok
+        end
+    catch
+        %% stil the table exists (somehow can happen in hot-upgrade,
+        %% it seems).
+        error:badarg ->
+            ok
+    end.

apps/emqx_rule_engine/src/emqx_rule_metrics.erl

@@ -19,6 +19,7 @@
 -behaviour(gen_server).
 
 -include("rule_engine.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
 
 %% API functions
 -export([ start_link/0
@@ -222,7 +223,9 @@ inc(Id, Metric, Val) ->
             counters:add(couters_ref(Id), metrics_idx(Metric), Val);
         Ref ->
             counters:add(Ref, metrics_idx(Metric), Val)
-    end.
+    end,
+    ?tp(rule_metrics_inc, #{id => Id, metric => Metric, value => Val}),
+    ok.
 
 inc_actions_taken(Id) ->
     inc_actions_taken(Id, 1).

apps/emqx_rule_engine/test/emqx_rule_engine_jwt_worker_SUITE.erl

@@ -0,0 +1,246 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 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_rule_engine_jwt_worker_SUITE).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+-include_lib("emqx_rule_engine/include/rule_engine.hrl").
+-include_lib("jose/include/jose_jwt.hrl").
+-include_lib("jose/include/jose_jws.hrl").
+
+-compile([export_all, nowarn_export_all]).
+
+%%-----------------------------------------------------------------------------
+%% CT boilerplate
+%%-----------------------------------------------------------------------------
+
+all() ->
+    emqx_ct:all(?MODULE).
+
+init_per_suite(Config) ->
+    Config.
+
+end_per_suite(_Config) ->
+    ok.
+
+%%-----------------------------------------------------------------------------
+%% Helper fns
+%%-----------------------------------------------------------------------------
+
+generate_private_key_pem() ->
+    PublicExponent = 65537,
+    Size = 2048,
+    Key = public_key:generate_key({rsa, Size, PublicExponent}),
+    DERKey = public_key:der_encode('PrivateKeyInfo', Key),
+    public_key:pem_encode([{'PrivateKeyInfo', DERKey, not_encrypted}]).
+
+generate_config() ->
+    PrivateKeyPEM = generate_private_key_pem(),
+    ResourceID = emqx_guid:gen(),
+    #{ private_key => PrivateKeyPEM
+     , expiration => timer:hours(1)
+     , resource_id => ResourceID
+     , table => ets:new(test_jwt_table, [ordered_set, public])
+     , iss => <<"issuer">>
+     , sub => <<"subject">>
+     , aud => <<"audience">>
+     , kid => <<"key id">>
+     , alg => <<"RS256">>
+     }.
+
+is_expired(JWT) ->
+    #jose_jwt{fields = #{<<"exp">> := Exp}} = jose_jwt:peek(JWT),
+    Now = erlang:system_time(seconds),
+    Now >= Exp.
+
+%%-----------------------------------------------------------------------------
+%% Test cases
+%%-----------------------------------------------------------------------------
+
+t_create_success(_Config) ->
+    Ref = alias([reply]),
+    Config = generate_config(),
+    ?assertMatch({ok, _}, emqx_rule_engine_jwt_worker:start_link(Config, Ref)),
+    receive
+        {Ref, token_created} ->
+            ok
+    after
+        1_000 ->
+            ct:fail("should have confirmed token creation; msgs: ~0p",
+                    [process_info(self(), messages)])
+    end,
+    ok.
+
+t_empty_key(_Config) ->
+    Ref = alias([reply]),
+    Config0 = generate_config(),
+    Config = Config0#{private_key := <<>>},
+    process_flag(trap_exit, true),
+    ?assertMatch({ok, _}, emqx_rule_engine_jwt_worker:start_link(Config, Ref)),
+    receive
+        {Ref, {error, {invalid_private_key, empty_key}}} ->
+            ok
+    after
+        1_000 ->
+            ct:fail("should have errored; msgs: ~0p",
+                    [process_info(self(), messages)])
+    end,
+    ok.
+
+t_invalid_pem(_Config) ->
+    Ref = alias([reply]),
+    Config0 = generate_config(),
+    InvalidPEM = public_key:pem_encode([{'PrivateKeyInfo', <<"xxxxxx">>, not_encrypted},
+                                        {'PrivateKeyInfo', <<"xxxxxx">>, not_encrypted}]),
+    Config = Config0#{private_key := InvalidPEM},
+    process_flag(trap_exit, true),
+    ?assertMatch({ok, _}, emqx_rule_engine_jwt_worker:start_link(Config, Ref)),
+    receive
+        {Ref, {error, {invalid_private_key, _}}} ->
+            ok
+    after
+        1_000 ->
+            ct:fail("should have errored; msgs: ~0p",
+                    [process_info(self(), messages)])
+    end,
+    ok.
+
+t_refresh(_Config) ->
+    Ref = alias([reply]),
+    Config0 = #{ table := Table
+               , resource_id := ResourceId
+               } = generate_config(),
+    Config = Config0#{expiration => 5_000},
+    ?check_trace(
+       begin
+         {{ok, _Pid}, {ok, _Event}} =
+           ?wait_async_action(
+              emqx_rule_engine_jwt_worker:start_link(Config, Ref),
+              #{?snk_kind := jwt_worker_token_stored},
+              5_000),
+         {ok, FirstJWT} = emqx_rule_engine_jwt_worker:lookup_jwt(Table, ResourceId),
+         ?block_until(#{?snk_kind := rule_engine_jwt_worker_refresh}, 15_000),
+         {ok, SecondJWT} = emqx_rule_engine_jwt_worker:lookup_jwt(Table, ResourceId),
+         ?assertNot(is_expired(SecondJWT)),
+         ?assert(is_expired(FirstJWT)),
+         {FirstJWT, SecondJWT}
+       end,
+       fun({FirstJWT, SecondJWT}, Trace) ->
+         ?assertMatch([_, _ | _],
+                      ?of_kind(jwt_worker_token_stored, Trace)),
+         ?assertNotEqual(FirstJWT, SecondJWT),
+         ok
+       end),
+    ok.
+
+t_format_status(_Config) ->
+    Ref = alias([reply]),
+    Config = generate_config(),
+    {ok, Pid} = emqx_rule_engine_jwt_worker:start_link(Config, Ref),
+    {status, _, _, Props} = sys:get_status(Pid),
+    [State] = [State
+               || Info = [_ | _] <- Props,
+                  {data, Data = [_ | _]} <- Info,
+                  {"State", State} <- Data],
+    ?assertMatch(
+      #{ jwt := "******"
+       , jwk := "******"
+       },
+       State),
+    ok.
+
+t_lookup_ok(_Config) ->
+    Ref = alias([reply]),
+    Config = #{ table := Table
+              , resource_id := ResourceId
+              , private_key := PrivateKeyPEM
+              , aud := Aud
+              , iss := Iss
+              , sub := Sub
+              , kid := KId
+              } = generate_config(),
+    {ok, _} = emqx_rule_engine_jwt_worker:start_link(Config, Ref),
+    receive
+        {Ref, token_created} ->
+            ok
+    after
+        500 ->
+            error(timeout)
+    end,
+    Res = emqx_rule_engine_jwt_worker:lookup_jwt(Table, ResourceId),
+    ?assertMatch({ok, _}, Res),
+    {ok, JWT} = Res,
+    ?assert(is_binary(JWT)),
+    JWK = jose_jwk:from_pem(PrivateKeyPEM),
+    {IsValid, ParsedJWT, JWS} = jose_jwt:verify_strict(JWK, [<<"RS256">>], JWT),
+    ?assertMatch(
+      #jose_jwt{
+         fields = #{ <<"aud">> := Aud
+                   , <<"iss">> := Iss
+                   , <<"sub">> := Sub
+                   , <<"exp">> := _
+                   , <<"iat">> := _
+                   }},
+      ParsedJWT),
+    ?assertNot(is_expired(JWT)),
+    ?assertMatch(
+      #jose_jws{
+         alg = {_, 'RS256'},
+         fields = #{ <<"kid">> := KId
+                   , <<"typ">> := <<"JWT">>
+                   }},
+      JWS),
+    ?assert(IsValid),
+    ok.
+
+t_lookup_not_found(_Config) ->
+    Table = ets:new(test_jwt_table, [ordered_set, public]),
+    InexistentResource = <<"xxx">>,
+    ?assertEqual({error, not_found},
+                 emqx_rule_engine_jwt_worker:lookup_jwt(Table, InexistentResource)),
+    ok.
+
+t_lookup_badarg(_Config) ->
+    InexistentTable = i_dont_exist,
+    InexistentResource = <<"xxx">>,
+    ?assertEqual({error, not_found},
+                 emqx_rule_engine_jwt_worker:lookup_jwt(InexistentTable, InexistentResource)),
+    ok.
+
+t_start_supervised_worker(_Config) ->
+    {ok, _} = emqx_rule_engine_jwt_sup:start_link(),
+    Config = #{resource_id := ResourceId} = generate_config(),
+    {Ref, Pid} = emqx_rule_engine_jwt_sup:start_worker(ResourceId, Config),
+    receive
+        {Ref, token_created} ->
+            ok
+    after
+        1_000 ->
+            ct:fail("timeout")
+    end,
+    MRef = monitor(process, Pid),
+    ?assert(is_process_alive(Pid)),
+    ok = emqx_rule_engine_jwt_sup:stop_worker(ResourceId),
+    receive
+        {'DOWN', MRef, process, Pid, _} ->
+            ok
+    after
+        1_000 ->
+            ct:fail("timeout")
+    end,
+    ok.

changes/v4.4.11-en.md

@@ -6,6 +6,8 @@
     - Added support for OCSP (Online Certificate Status Protocol) Stapling
     - Added CRL (Certificate Revocation List) cache auto refresh
 
+- Added a JWT worker for creating and refreshing JWT tokens in rule engine actions. [#9241](https://github.com/emqx/emqx/pull/9241)
+
 ### Bug fixes
 
 - Fix get trace list crash when trace not initialize. [#9156](https://github.com/emqx/emqx/pull/9156)

changes/v4.4.11-zh.md

@@ -6,6 +6,8 @@
     - 增加了对 OCSP (Online Certificate Status Protocol) Stapling 的支持
     - 增加了 CRL（证书吊销列表）缓存的自动刷新功能
 
+- 增加了一个JWT工作者，用于在规则引擎动作中创建和刷新JWT令牌。[#9241](https://github.com/emqx/emqx/pull/9241)
+
 ### 修复
 
 - 修复日志追踪模块没开启时，GET Trace 列表接口报错的问题。[#9156](https://github.com/emqx/emqx/pull/9156)

test/emqx_node_helpers.erl

@@ -31,12 +31,23 @@ start_slave(Name) ->
     start_slave(Name, #{}).
 
 start_slave(Name, Opts) ->
+    SlaveMod = maps:get(slave_mod, Opts, ct_slave),
     Node = make_node_name(Name),
-    case ct_slave:start(Node, [{kill_if_fail, true},
+    DoStart =
+        fun() ->
+          case SlaveMod of
+              ct_slave ->
+                  ct_slave:start(Node,
+                                 [{kill_if_fail, true},
                                   {monitor_master, true},
                                   {init_timeout, 10000},
                                   {startup_timeout, 10000},
-                               {erl_flags, ebin_path()}]) of
+                                  {erl_flags, ebin_path()}]);
+              slave ->
+                  slave:start_link(host(), Name, ebin_path())
+          end
+        end,
+    case DoStart() of
         {ok, _} ->
             ok;
         {error, started_not_connected, _} ->
@@ -115,6 +126,9 @@ setup_node(Node, #{} = Opts) ->
     ?assertEqual( node()
                 , gen_rpc:call(Node, gen_rpc, call, [node(), erlang, node, []])
                 ),
+
+    ok = snabbkaffe:forward_trace(Node),
+
     ok.
 
 %% Routes are replicated async.