From a2bafd1a18ff333d54d1e5a1d2b61e08ec810d1f Mon Sep 17 00:00:00 2001
From: zhanghongtong <rory-z@outlook.com>
Date: Tue, 24 Aug 2021 15:08:21 +0800
Subject: [PATCH] feat(authz): support authorization config file part 2

---
 .github/workflows/build_packages.yaml         |   4 +-
 apps/emqx_authz/etc/emqx_authz.conf           |   7 +-
 apps/emqx_authz/include/emqx_authz.hrl        |   9 +-
 apps/emqx_authz/src/emqx_authz.erl            | 226 ++++--------------
 apps/emqx_authz/src/emqx_authz_api.erl        |  49 +++-
 apps/emqx_authz/src/emqx_authz_mongo.erl      |  27 +--
 apps/emqx_authz/src/emqx_authz_mysql.erl      |  42 ++--
 apps/emqx_authz/src/emqx_authz_pgsql.erl      |  40 +---
 apps/emqx_authz/src/emqx_authz_redis.erl      |  28 +--
 apps/emqx_authz/src/emqx_authz_rule.erl       |  28 ++-
 apps/emqx_authz/src/emqx_authz_schema.erl     |  34 +--
 apps/emqx_authz/test/emqx_authz_SUITE.erl     | 194 ++++++---------
 apps/emqx_authz/test/emqx_authz_api_SUITE.erl | 164 ++++++++-----
 .../emqx_authz/test/emqx_authz_http_SUITE.erl |  15 +-
 .../test/emqx_authz_mongo_SUITE.erl           |  21 +-
 .../test/emqx_authz_mysql_SUITE.erl           |  36 ++-
 .../test/emqx_authz_pgsql_SUITE.erl           |  35 ++-
 .../test/emqx_authz_redis_SUITE.erl           |  20 +-
 .../src/emqx_connector_mongo.erl              |  12 +-
 .../src/emqx_connector_redis.erl              |  11 +-
 20 files changed, 396 insertions(+), 606 deletions(-)

diff --git a/.github/workflows/build_packages.yaml b/.github/workflows/build_packages.yaml
index a96ef705e..f42835a18 100644
--- a/.github/workflows/build_packages.yaml
+++ b/.github/workflows/build_packages.yaml
@@ -20,8 +20,8 @@ jobs:
     container: ${{ matrix.container }}
 
     outputs:
-      profiles: ${{ steps.set_profile.outputs.profiles}}
-      old_vsns: ${{ steps.set_profile.outputs.old_vsns}}
+      profiles: ${{ steps.set_profile.outputs.profiles }}
+      old_vsns: ${{ steps.set_profile.outputs.old_vsns }}
 
     steps:
       - uses: actions/checkout@v2
diff --git a/apps/emqx_authz/etc/emqx_authz.conf b/apps/emqx_authz/etc/emqx_authz.conf
index baabd8a37..5928b1f97 100644
--- a/apps/emqx_authz/etc/emqx_authz.conf
+++ b/apps/emqx_authz/etc/emqx_authz.conf
@@ -68,11 +68,6 @@ authorization_rules {
        #     }
        #     collection: mqtt_authz
        #     find: { "$or": [ { "username": "%u" }, { "clientid": "%c" } ] }
-       # },
-       {
-           permission = allow
-           action = all
-           topics = ["#"]
-       }
+       # }
     ]
 }
diff --git a/apps/emqx_authz/include/emqx_authz.hrl b/apps/emqx_authz/include/emqx_authz.hrl
index 30297ac66..a4f10c5f9 100644
--- a/apps/emqx_authz/include/emqx_authz.hrl
+++ b/apps/emqx_authz/include/emqx_authz.hrl
@@ -19,8 +19,13 @@
 
 -define(APP, emqx_authz).
 
--define(ALLOW_DENY(A), ((A =:= allow) orelse (A =:= deny))).
--define(PUBSUB(A), ((A =:= subscribe) orelse (A =:= publish) orelse (A =:= all))).
+-define(ALLOW_DENY(A), ((A =:= allow) orelse (A =:= <<"allow">>) orelse
+                        (A =:= deny)  orelse (A =:= <<"deny">>)
+                       )).
+-define(PUBSUB(A), ((A =:= subscribe) orelse (A =:= <<"subscribe">>) orelse
+                    (A =:= publish)   orelse (A =:= <<"publish">>) orelse
+                    (A =:= all)       orelse (A =:= <<"all">>)
+                   )).
 
 -record(authz_metrics, {
         allow = 'client.authorize.allow',
diff --git a/apps/emqx_authz/src/emqx_authz.erl b/apps/emqx_authz/src/emqx_authz.erl
index 2c6395199..3c8a56629 100644
--- a/apps/emqx_authz/src/emqx_authz.erl
+++ b/apps/emqx_authz/src/emqx_authz.erl
@@ -27,13 +27,11 @@
 
 -export([ register_metrics/0
         , init/0
-        , init_rule/1
         , lookup/0
         , lookup/1
         , move/2
         , update/2
         , authorize/5
-        , match/4
         ]).
 
 -export([post_config_update/3, pre_config_update/2]).
@@ -47,7 +45,7 @@ register_metrics() ->
 init() ->
     ok = register_metrics(),
     emqx_config_handler:add_handler(?CONF_KEY_PATH, ?MODULE),
-    NRules = [init_rule(Rule) || Rule <- emqx:get_config(?CONF_KEY_PATH, [])],
+    NRules = [init_provider(Rule) || Rule <- emqx:get_config(?CONF_KEY_PATH, [])],
     ok = emqx_hooks:add('client.authorize', {?MODULE, authorize, [NRules]}, -1).
 
 lookup() ->
@@ -148,12 +146,12 @@ post_config_update({move, Id, #{<<"after">> := AfterId}}, _NewRules, _OldRules)
     ok = emqx_authz_cache:drain_cache();
 
 post_config_update({head, Rules}, _NewRules, _OldConf) ->
-    InitedRules = [init_rule(R) || R <- check_rules(Rules)],
+    InitedRules = [init_provider(R) || R <- check_rules(Rules)],
     ok = emqx_hooks:put('client.authorize', {?MODULE, authorize, [InitedRules ++ lookup()]}, -1),
     ok = emqx_authz_cache:drain_cache();
 
 post_config_update({tail, Rules}, _NewRules, _OldConf) ->
-    InitedRules = [init_rule(R) || R <- check_rules(Rules)],
+    InitedRules = [init_provider(R) || R <- check_rules(Rules)],
     emqx_hooks:put('client.authorize', {?MODULE, authorize, [lookup() ++ InitedRules]}, -1),
     ok = emqx_authz_cache:drain_cache();
 
@@ -167,14 +165,14 @@ post_config_update({{replace_once, Id}, Rule}, _NewRules, _OldConf) when is_map(
             ok = emqx_resource:remove(Id)
     end,
     {OldRules1, OldRules2 } = lists:split(Index, OldInitedRules),
-    InitedRules = [init_rule(R#{annotations => #{id => Id}}) || R <- check_rules([Rule])],
+    InitedRules = [init_provider(R#{annotations => #{id => Id}}) || R <- check_rules([Rule])],
     ok = emqx_hooks:put('client.authorize', {?MODULE, authorize, [lists:droplast(OldRules1) ++ InitedRules ++ OldRules2]}, -1),
     ok = emqx_authz_cache:drain_cache();
 
 post_config_update(_, NewRules, _OldConf) ->
     %% overwrite the entire config!
     OldInitedRules = lookup(),
-    InitedRules = [init_rule(Rule) || Rule <- NewRules],
+    InitedRules = [init_provider(Rule) || Rule <- NewRules],
     ok = emqx_hooks:put('client.authorize', {?MODULE, authorize, [InitedRules]}, -1),
     lists:foreach(fun (#{type := _Type, enable := true, annotations := #{id := Id}}) ->
                          ok = emqx_resource:remove(Id);
@@ -235,29 +233,11 @@ create_resource(#{type := DB,
         {error, Reason} -> {error, Reason}
     end.
 
--spec(init_rule(rule()) -> rule()).
-init_rule(#{topics := Topics,
-            action := Action,
-            permission := Permission,
-            principal := Principal,
-            annotations := #{id := Id}
-           } = Rule) when ?ALLOW_DENY(Permission), ?PUBSUB(Action), is_list(Topics) ->
-    Rule#{annotations =>
-            #{id => Id,
-              principal => compile_principal(Principal),
-              topics => [compile_topic(Topic) || Topic <- Topics]}
-         };
-init_rule(#{topics := Topics,
-            action := Action,
-            permission := Permission
-           } = Rule) when ?ALLOW_DENY(Permission), ?PUBSUB(Action), is_list(Topics) ->
-    init_rule(Rule#{annotations =>#{id => gen_id(simple)}});
-
-init_rule(#{principal := Principal,
-            enable := true,
-            type := file,
-            path := Path
-           } = Rule) ->
+-spec(init_provider(rule()) -> rule()).
+init_provider(#{enable := true,
+                type := file,
+                path := Path
+               } = Rule) ->
     Rules = case file:consult(Path) of
                 {ok, Terms} ->
                     [emqx_authz_rule:compile(Term) || Term <- Terms];
@@ -275,92 +255,42 @@ init_rule(#{principal := Principal,
             #{id => gen_id(file),
               rules => Rules
          }};
-init_rule(#{principal := Principal,
-            enable := true,
-            type := http,
-            config := #{url := Url} = Config
-           } = Rule) ->
+init_provider(#{enable := true,
+                type := http,
+                config := #{url := Url} = Config
+               } = Rule) ->
     NConfig = maps:merge(Config, #{base_url => maps:remove(query, Url)}),
     case create_resource(Rule#{config := NConfig}) of
         {error, Reason} -> error({load_config_error, Reason});
         Id -> Rule#{annotations =>
-                      #{id => Id,
-                        principal => compile_principal(Principal)
-                       }
+                      #{id => Id}
                    }
     end;
-
-init_rule(#{principal := Principal,
-            enable := true,
-            type := DB
-         } = Rule) when DB =:= redis;
-                        DB =:= mongo ->
+init_provider(#{enable := true,
+                type := DB
+               } = Rule) when DB =:= redis;
+                              DB =:= mongo ->
     case create_resource(Rule) of
         {error, Reason} -> error({load_config_error, Reason});
         Id -> Rule#{annotations =>
-                      #{id => Id,
-                        principal => compile_principal(Principal)
-                       }
+                      #{id => Id}
                    }
     end;
-
-init_rule(#{principal := Principal,
-            enable := true,
-            type := DB,
-            sql := SQL
-         } = Rule) when DB =:= mysql;
-                        DB =:= pgsql ->
+init_provider(#{enable := true,
+                type := DB,
+                sql := SQL
+               } = Rule) when DB =:= mysql;
+                              DB =:= pgsql ->
     Mod = list_to_existing_atom(io_lib:format("~s_~s",[?APP, DB])),
     case create_resource(Rule) of
         {error, Reason} -> error({load_config_error, Reason});
         Id -> Rule#{annotations =>
                       #{id => Id,
-                        principal => compile_principal(Principal),
                         sql => Mod:parse_query(SQL)
                        }
                    }
     end;
-
-init_rule(#{enable := false,
-            type := _DB
-         } = Rule) ->
-    Rule.
-
-compile_principal(all) -> all;
-compile_principal(#{username := Username}) ->
-    {ok, MP} = re:compile(bin(Username)),
-    #{username => MP};
-compile_principal(#{clientid := Clientid}) ->
-    {ok, MP} = re:compile(bin(Clientid)),
-    #{clientid => MP};
-compile_principal(#{ipaddress := IpAddress}) ->
-    #{ipaddress => esockd_cidr:parse(b2l(IpAddress), true)};
-compile_principal(#{'and' := Principals}) when is_list(Principals) ->
-    #{'and' => [compile_principal(Principal) || Principal <- Principals]};
-compile_principal(#{'or' := Principals}) when is_list(Principals) ->
-    #{'or' => [compile_principal(Principal) || Principal <- Principals]}.
-
-compile_topic(<<"eq ", Topic/binary>>) ->
-    compile_topic(#{'eq' => Topic});
-compile_topic(#{'eq' := Topic}) ->
-    #{'eq' => emqx_topic:words(bin(Topic))};
-compile_topic(Topic) when is_binary(Topic)->
-    Words = emqx_topic:words(bin(Topic)),
-    case pattern(Words) of
-        true  -> #{pattern => Words};
-        false -> Words
-    end.
-
-pattern(Words) ->
-    lists:member(<<"%u">>, Words) orelse lists:member(<<"%c">>, Words).
-
-bin(A) when is_atom(A) -> atom_to_binary(A, utf8);
-bin(B) when is_binary(B) -> B;
-bin(L) when is_list(L) -> list_to_binary(L);
-bin(X) -> X.
-
-b2l(B) when is_list(B) -> B;
-b2l(B) when is_binary(B) -> binary_to_list(B).
+init_provider(#{enable := false} = Rule) ->Rule.
 
 %%--------------------------------------------------------------------
 %% AuthZ callbacks
@@ -387,97 +317,21 @@ authorize(#{username := Username,
     end.
 
 do_authorize(Client, PubSub, Topic,
-               [Connector = #{type := DB,
-                              enable := true,
-                              annotations := #{principal := Principal}
-                             } | Tail] ) ->
-    case match_principal(Client, Principal) of
-        true ->
-            Mod = list_to_existing_atom(io_lib:format("~s_~s",[emqx_authz, DB])),
-            case Mod:authorize(Client, PubSub, Topic, Connector) of
-                nomatch -> do_authorize(Client, PubSub, Topic, Tail);
-                Matched -> Matched
-            end;
-        false -> do_authorize(Client, PubSub, Topic, Tail)
+               [#{type := file,
+                  enable := true,
+                  annotations := #{rule := Rules}
+                 } | Tail] ) ->
+    case emqx_authz_rule:match(Client, PubSub, Topic, Rules) of
+        nomatch -> do_authorize(Client, PubSub, Topic, Tail);
+        Matched -> Matched
     end;
 do_authorize(Client, PubSub, Topic,
-               [#{permission := Permission} = Rule | Tail]) ->
-    case match(Client, PubSub, Topic, Rule) of
-        true -> {matched, Permission};
-        false -> do_authorize(Client, PubSub, Topic, Tail)
+               [Connector = #{type := Type,
+                              enable := true
+                             } | Tail] ) ->
+    Mod = list_to_existing_atom(io_lib:format("~s_~s",[emqx_authz, Type])),
+    case Mod:authorize(Client, PubSub, Topic, Connector) of
+        nomatch -> do_authorize(Client, PubSub, Topic, Tail);
+        Matched -> Matched
     end;
 do_authorize(_Client, _PubSub, _Topic, []) -> nomatch.
-
-match(Client, PubSub, Topic,
-      #{action := Action,
-        annotations := #{
-            principal := Principal,
-            topics := TopicFilters
-        }
-       }) ->
-    match_action(PubSub, Action) andalso
-    match_principal(Client, Principal) andalso
-    match_topics(Client, Topic, TopicFilters).
-
-match_action(publish, publish) -> true;
-match_action(subscribe, subscribe) -> true;
-match_action(_, all) -> true;
-match_action(_, _) -> false.
-
-match_principal(_, all) -> true;
-match_principal(#{username := undefined}, #{username := _MP}) ->
-    false;
-match_principal(#{username := Username}, #{username := MP}) ->
-    case re:run(Username, MP) of
-        {match, _} -> true;
-        _ -> false
-    end;
-match_principal(#{clientid := Clientid}, #{clientid := MP}) ->
-    case re:run(Clientid, MP) of
-        {match, _} -> true;
-        _ -> false
-    end;
-match_principal(#{peerhost := undefined}, #{ipaddress := _CIDR}) ->
-    false;
-match_principal(#{peerhost := IpAddress}, #{ipaddress := CIDR}) ->
-    esockd_cidr:match(IpAddress, CIDR);
-match_principal(ClientInfo, #{'and' := Principals}) when is_list(Principals) ->
-    lists:foldl(fun(Principal, Permission) ->
-                  match_principal(ClientInfo, Principal) andalso Permission
-                end, true, Principals);
-match_principal(ClientInfo, #{'or' := Principals}) when is_list(Principals) ->
-    lists:foldl(fun(Principal, Permission) ->
-                  match_principal(ClientInfo, Principal) orelse Permission
-                end, false, Principals);
-match_principal(_, _) -> false.
-
-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 match_topics(ClientInfo, Topic, Filters);
-match_topics(ClientInfo, Topic, [TopicFilter|Filters]) ->
-   match_topic(emqx_topic:words(Topic), TopicFilter)
-       orelse match_topics(ClientInfo, Topic, Filters).
-
-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}, [<<"%c">>|Words], Acc) ->
-    feed_var(ClientInfo, Words, [<<"%c">>|Acc]);
-feed_var(ClientInfo = #{clientid := ClientId}, [<<"%c">>|Words], Acc) ->
-    feed_var(ClientInfo, Words, [ClientId |Acc]);
-feed_var(ClientInfo = #{username := undefined}, [<<"%u">>|Words], Acc) ->
-    feed_var(ClientInfo, Words, [<<"%u">>|Acc]);
-feed_var(ClientInfo = #{username := Username}, [<<"%u">>|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/src/emqx_authz_api.erl b/apps/emqx_authz/src/emqx_authz_api.erl
index 15aaed65a..1646a9af2 100644
--- a/apps/emqx_authz/src/emqx_authz_api.erl
+++ b/apps/emqx_authz/src/emqx_authz_api.erl
@@ -419,12 +419,26 @@ move_rule_api() ->
     {"/authorization/:id/move", Metadata, move_rule}.
 
 rules(get, #{query_string := Query}) ->
-    Rules = lists:foldl(fun (#{type := _Type, enable := true, annotations := #{id := Id} = Annotations} = Rule, AccIn) ->
+    Rules = lists:foldl(fun (#{type := _Type, enable := true, config := #{server := Server} = Config, annotations := #{id := Id}} = Rule, AccIn) ->
                                 NRule = case emqx_resource:health_check(Id) of
                                     ok ->
-                                        Rule#{annotations => Annotations#{status => healthy}};
+                                        Rule#{config => Config#{server => emqx_connector_schema_lib:ip_port_to_string(Server)},
+                                              annotations => #{id => Id,
+                                                               status => healthy}};
                                     _ ->
-                                        Rule#{annotations => Annotations#{status => unhealthy}}
+                                        Rule#{config => Config#{server => emqx_connector_schema_lib:ip_port_to_string(Server)},
+                                              annotations => #{id => Id,
+                                                               status => unhealthy}}
+                                end,
+                                lists:append(AccIn, [NRule]);
+                            (#{type := _Type, enable := true, annotations := #{id := Id}} = Rule, AccIn) ->
+                                NRule = case emqx_resource:health_check(Id) of
+                                    ok ->
+                                        Rule#{annotations => #{id => Id,
+                                                               status => healthy}};
+                                    _ ->
+                                        Rule#{annotations => #{id => Id,
+                                                               status => unhealthy}}
                                 end,
                                 lists:append(AccIn, [NRule]);
                             (Rule, AccIn) ->
@@ -462,17 +476,26 @@ rules(put, #{body := RawConfig}) ->
 rule(get, #{bindings := #{id := Id}}) ->
     case emqx_authz:lookup(Id) of
         {error, Reason} -> {404, #{messgae => atom_to_binary(Reason)}};
-        Rule ->
-            case maps:get(type, Rule, undefined) of
-                undefined -> {200, Rule};
+        #{type := file} = Rule -> {200, Rule};
+        #{config := #{server := Server} = Config} = Rule ->
+            case emqx_resource:health_check(Id) of
+                ok ->
+                    {200, Rule#{config => Config#{server => emqx_connector_schema_lib:ip_port_to_string(Server)},
+                                annotations => #{id => Id,
+                                                 status => healthy}}};
                 _ ->
-                    case emqx_resource:health_check(Id) of
-                        ok ->
-                            {200, Rule#{annotations => #{status => healthy}}};
-                        _ ->
-                            {200, Rule#{annotations => #{status => unhealthy}}}
-                    end
-
+                    {200, Rule#{config => Config#{server => emqx_connector_schema_lib:ip_port_to_string(Server)},
+                                annotations => #{id => Id,
+                                                 status => unhealthy}}}
+            end;
+        Rule ->
+            case emqx_resource:health_check(Id) of
+                ok ->
+                    {200, Rule#{annotations => #{id => Id,
+                                                 status => healthy}}};
+                _ ->
+                    {200, Rule#{annotations => #{id => Id,
+                                                 status => unhealthy}}}
             end
     end;
 rule(put, #{bindings := #{id := RuleId}, body := RawConfig}) ->
diff --git a/apps/emqx_authz/src/emqx_authz_mongo.erl b/apps/emqx_authz/src/emqx_authz_mongo.erl
index c015f8208..25a787b8f 100644
--- a/apps/emqx_authz/src/emqx_authz_mongo.erl
+++ b/apps/emqx_authz/src/emqx_authz_mongo.erl
@@ -44,38 +44,19 @@ authorize(Client, PubSub, Topic,
             nomatch;
         [] -> nomatch;
         Rows ->
-            do_authorize(Client, PubSub, Topic, Rows)
+            Rules = [ emqx_authz_rule:compile({Permission, all, Action, Topics})
+                     || #{<<"topics">> := Topics, <<"permission">> := Permission, <<"action">> := Action} <- Rows],
+            do_authorize(Client, PubSub, Topic, Rules)
     end.
 
 do_authorize(_Client, _PubSub, _Topic, []) ->
     nomatch;
 do_authorize(Client, PubSub, Topic, [Rule | Tail]) ->
-    case match(Client, PubSub, Topic, Rule) of
+    case emqx_authz_rule:match(Client, PubSub, Topic, Rule) of
         {matched, Permission} -> {matched, Permission};
         nomatch -> do_authorize(Client, PubSub, Topic, Tail)
     end.
 
-match(Client, PubSub, Topic,
-      #{<<"topics">> := Topics,
-        <<"permission">> := Permission,
-        <<"action">> := Action
-       }) ->
-    Rule = #{<<"permission">> => Permission,
-             <<"topics">> => Topics,
-             <<"action">> => Action
-            },
-    #{simple_rule :=
-      #{permission := NPermission} = NRule
-     } = hocon_schema:check_plain(
-            emqx_authz_schema,
-            #{<<"simple_rule">> => Rule},
-            #{atom_key => true},
-            [simple_rule]),
-    case emqx_authz:match(Client, PubSub, Topic, emqx_authz:init_rule(NRule)) of
-        true -> {matched, NPermission};
-        false -> nomatch
-    end.
-
 replvar(Find, #{clientid := Clientid,
                 username := Username,
                 peerhost := IpAddress
diff --git a/apps/emqx_authz/src/emqx_authz_mysql.erl b/apps/emqx_authz/src/emqx_authz_mysql.erl
index 2ce991eba..d5550b2fb 100644
--- a/apps/emqx_authz/src/emqx_authz_mysql.erl
+++ b/apps/emqx_authz/src/emqx_authz_mysql.erl
@@ -62,39 +62,25 @@ authorize(Client, PubSub, Topic,
 do_authorize(_Client, _PubSub, _Topic, _Columns, []) ->
     nomatch;
 do_authorize(Client, PubSub, Topic, Columns, [Row | Tail]) ->
-    case match(Client, PubSub, Topic, format_result(Columns, Row)) of
+    case emqx_authz_rule:match(Client, PubSub, Topic,
+                               emqx_authz_rule:compile(format_result(Columns, Row))
+                              ) of
         {matched, Permission} -> {matched, Permission};
         nomatch -> do_authorize(Client, PubSub, Topic, Columns, Tail)
     end.
 
-format_result(Columns, Row) ->
-    L = [ begin
-              K = lists:nth(I, Columns),
-              V = lists:nth(I, Row),
-              {K, V}
-          end || I <- lists:seq(1, length(Columns)) ],
-    maps:from_list(L).
 
-match(Client, PubSub, Topic,
-      #{<<"permission">> := Permission,
-        <<"action">> := Action,
-        <<"topic">> := TopicFilter
-       }) ->
-    Rule = #{<<"topics">> => [TopicFilter],
-             <<"action">> => Action,
-             <<"permission">> =>  Permission
-            },
-    #{simple_rule :=
-      #{permission := NPermission} = NRule
-     } = hocon_schema:check_plain(
-            emqx_authz_schema,
-            #{<<"simple_rule">> => Rule},
-            #{atom_key => true},
-            [simple_rule]),
-    case emqx_authz:match(Client, PubSub, Topic, emqx_authz:init_rule(NRule)) of
-        true -> {matched, NPermission};
-        false -> nomatch
-    end.
+format_result(Columns, Row) ->
+    Permission = lists:nth(index(<<"permission">>, Columns), Row),
+    Action = lists:nth(index(<<"action">>, Columns), Row),
+    Topic = lists:nth(index(<<"topic">>, Columns), Row),
+    {Permission, all, Action, [Topic]}.
+
+index(Elem, List) ->
+    index(Elem, List, 1).
+index(_Elem, [], _Index) -> {error, not_found};
+index(Elem, [ Elem | _List], Index) -> Index;
+index(Elem, [ _ | List], Index) -> index(Elem, List, Index + 1).
 
 replvar(Params, ClientInfo) ->
     replvar(Params, ClientInfo, []).
diff --git a/apps/emqx_authz/src/emqx_authz_pgsql.erl b/apps/emqx_authz/src/emqx_authz_pgsql.erl
index f3e793763..d9555b85d 100644
--- a/apps/emqx_authz/src/emqx_authz_pgsql.erl
+++ b/apps/emqx_authz/src/emqx_authz_pgsql.erl
@@ -66,39 +66,25 @@ authorize(Client, PubSub, Topic,
 do_authorize(_Client, _PubSub, _Topic, _Columns, []) ->
     nomatch;
 do_authorize(Client, PubSub, Topic, Columns, [Row | Tail]) ->
-    case match(Client, PubSub, Topic, format_result(Columns, Row)) of
+    case emqx_authz_rule:match(Client, PubSub, Topic,
+                               emqx_authz_rule:compile(format_result(Columns, Row))
+                              ) of
         {matched, Permission} -> {matched, Permission};
         nomatch -> do_authorize(Client, PubSub, Topic, Columns, Tail)
     end.
 
 format_result(Columns, Row) ->
-    L = [ begin
-              {column, K, _, _, _, _, _, _, _} = lists:nth(I, Columns),
-              V = lists:nth(I, tuple_to_list(Row)),
-              {K, V}
-          end || I <- lists:seq(1, length(Columns)) ],
-    maps:from_list(L).
+    Permission = lists:nth(index(<<"permission">>, 2, Columns), erlang:tuple_to_list(Row)),
+    Action = lists:nth(index(<<"action">>, 2, Columns), erlang:tuple_to_list(Row)),
+    Topic = lists:nth(index(<<"topic">>, 2, Columns), erlang:tuple_to_list(Row)),
+    {Permission, all, Action, [Topic]}.
 
-match(Client, PubSub, Topic,
-      #{<<"permission">> := Permission,
-        <<"action">> := Action,
-        <<"topic">> := TopicFilter
-       }) ->
-    Rule = #{<<"topics">> => [TopicFilter],
-             <<"action">> => Action,
-             <<"permission">> =>  Permission
-            },
-    #{simple_rule :=
-      #{permission := NPermission} = NRule
-     } = hocon_schema:check_plain(
-            emqx_authz_schema,
-            #{<<"simple_rule">> => Rule},
-            #{atom_key => true},
-            [simple_rule]),
-    case emqx_authz:match(Client, PubSub, Topic, emqx_authz:init_rule(NRule)) of
-        true -> {matched, NPermission};
-        false -> nomatch
-    end.
+index(Key, N, TupleList) when is_integer(N) ->
+    Tuple = lists:keyfind(Key, N, TupleList),
+    index(Tuple, TupleList, 1);
+index(_Tuple, [], _Index) -> {error, not_found};
+index(Tuple, [Tuple | _TupleList], Index) -> Index;
+index(Tuple, [_ | TupleList], Index) -> index(Tuple, TupleList, Index + 1).
 
 replvar(Params, ClientInfo) ->
     replvar(Params, ClientInfo, []).
diff --git a/apps/emqx_authz/src/emqx_authz_redis.erl b/apps/emqx_authz/src/emqx_authz_redis.erl
index 8f6731fd8..3ac7d7e3f 100644
--- a/apps/emqx_authz/src/emqx_authz_redis.erl
+++ b/apps/emqx_authz/src/emqx_authz_redis.erl
@@ -50,35 +50,13 @@ authorize(Client, PubSub, Topic,
 do_authorize(_Client, _PubSub, _Topic, []) ->
     nomatch;
 do_authorize(Client, PubSub, Topic, [TopicFilter, Action | Tail]) ->
-    case match(Client, PubSub, Topic,
-               #{topics => TopicFilter,
-                 action => Action
-                })
-    of
+    case emqx_authz_rule:match(Client, PubSub, Topic,
+                               emqx_authz_rule:compile({allow, all, Action, [TopicFilter]})
+                              )of
         {matched, Permission} -> {matched, Permission};
         nomatch -> do_authorize(Client, PubSub, Topic, Tail)
     end.
 
-match(Client, PubSub, Topic,
-      #{topics := TopicFilter,
-        action := Action
-       }) ->
-    Rule = #{<<"principal">> => all,
-             <<"topics">> => [TopicFilter],
-             <<"action">> => Action,
-             <<"permission">> => allow
-            },
-    #{simple_rule := NRule
-     } = hocon_schema:check_plain(
-            emqx_authz_schema,
-            #{<<"simple_rule">> => Rule},
-            #{atom_key => true},
-            [simple_rule]),
-    case emqx_authz:match(Client, PubSub, Topic, emqx_authz:init_rule(NRule)) of
-        true -> {matched, allow};
-        false -> nomatch
-    end.
-
 replvar(Cmd, Client = #{cn := CN}) ->
     replvar(repl(Cmd, "%C", CN), maps:remove(cn, Client));
 replvar(Cmd, Client = #{dn := DN}) ->
diff --git a/apps/emqx_authz/src/emqx_authz_rule.erl b/apps/emqx_authz/src/emqx_authz_rule.erl
index 8fd7b9721..4786bf39d 100644
--- a/apps/emqx_authz/src/emqx_authz_rule.erl
+++ b/apps/emqx_authz/src/emqx_authz_rule.erl
@@ -26,15 +26,14 @@
 
 %% APIs
 -export([ match/4
+        , matches/4
         , compile/1
         ]).
 
 -export_type([rule/0]).
 
 compile({Permission, Who, Action, TopicFilters}) when ?ALLOW_DENY(Permission), ?PUBSUB(Action), is_list(TopicFilters) ->
-    {Permission, compile_who(Who), Action, [compile_topic(Topic) || Topic <- TopicFilters]};
-compile({Permission, Who, Action, Topic}) when ?ALLOW_DENY(Permission), ?PUBSUB(Action) ->
-    {Permission, compile_who(Who), Action, [compile_topic(Topic)]}.
+    {atom(Permission), compile_who(Who), atom(Action), [compile_topic(Topic) || Topic <- TopicFilters]}.
 
 compile_who(all) -> all;
 compile_who({username, Username}) ->
@@ -52,6 +51,8 @@ compile_who({'and', L}) when is_list(L) ->
 compile_who({'or', L}) when is_list(L) ->
     {'or', [compile_who(Who) || Who <- L]}.
 
+compile_topic(<<"eq ", Topic/binary>>) ->
+    {eq, emqx_topic:words(Topic)};
 compile_topic({eq, Topic}) ->
     {eq, emqx_topic:words(bin(Topic))};
 compile_topic(Topic) ->
@@ -64,11 +65,32 @@ compile_topic(Topic) ->
 pattern(Words) ->
     lists:member(<<"%u">>, Words) orelse lists:member(<<"%c">>, Words).
 
+atom(B) when is_binary(B) ->
+    try binary_to_existing_atom(B, utf8)
+    catch
+        _ -> binary_to_atom(B)
+    end;
+atom(L) when is_list(L) ->
+    try list_to_existing_atom(L)
+    catch
+        _ -> list_to_atom(L)
+    end;
+atom(A) when is_atom(A) -> A.
+
 bin(L) when is_list(L) ->
     list_to_binary(L);
 bin(B) when is_binary(B) ->
     B.
 
+-spec(matches(emqx_types:clientinfo(), emqx_types:pubsub(), emqx_types:topic(), [rule()])
+      -> {matched, allow} | {matched, deny} | nomatch).
+matches(Client, PubSub, Topic, []) -> nomatch;
+matches(Client, PubSub, Topic, [{Permission, Who, Action, TopicFilters} | Tail]) ->
+    case match(Client, PubSub, Topic, {Permission, Who, Action, TopicFilters}) of
+        nomatch -> matches(Client, PubSub, Topic, Tail);
+        Matched -> Matched
+    end.
+
 -spec(match(emqx_types:clientinfo(), emqx_types:pubsub(), emqx_types:topic(), rule())
       -> {matched, allow} | {matched, deny} | nomatch).
 match(Client, PubSub, Topic, {Permission, Who, Action, TopicFilters}) ->
diff --git a/apps/emqx_authz/src/emqx_authz_schema.erl b/apps/emqx_authz/src/emqx_authz_schema.erl
index 958ad9dec..a09046572 100644
--- a/apps/emqx_authz/src/emqx_authz_schema.erl
+++ b/apps/emqx_authz/src/emqx_authz_schema.erl
@@ -23,8 +23,7 @@ fields("authorization_rules") ->
     [ {rules, rules()}
     ];
 fields(file) ->
-    [ {principal, principal()}
-    , {type, #{type => http}}
+    [ {type, #{type => http}}
     , {enable, #{type => boolean(),
                  default => true}}
     , {path, #{type => string(),
@@ -36,8 +35,7 @@ fields(file) ->
               }}
     ];
 fields(http) ->
-    [ {principal, principal()}
-    , {type, #{type => http}}
+    [ {type, #{type => http}}
     , {enable, #{type => boolean(),
                  default => true}}
     , {config, #{type => hoconsc:union([ hoconsc:ref(?MODULE, http_get)
@@ -113,16 +111,6 @@ fields(mysql) ->
 fields(pgsql) ->
     connector_fields(pgsql) ++
     [ {sql, query()} ];
-fields(simple_rule) ->
-    [ {permission,   #{type => permission()}}
-    , {action,   #{type => action()}}
-    , {topics,   #{type => union_array(
-                             [ binary()
-                             , hoconsc:ref(?MODULE, eq_topic)
-                             ]
-                            )}}
-    , {principal, principal()}
-    ];
 fields(username) ->
     [{username, #{type => binary()}}];
 fields(clientid) ->
@@ -160,8 +148,7 @@ union_array(Item) when is_list(Item) ->
 
 rules() ->
     #{type => union_array(
-                [ hoconsc:ref(?MODULE, simple_rule)
-                , hoconsc:ref(?MODULE, file)
+                [ hoconsc:ref(?MODULE, file)
                 , hoconsc:ref(?MODULE, http)
                 , hoconsc:ref(?MODULE, mysql)
                 , hoconsc:ref(?MODULE, pgsql)
@@ -170,18 +157,6 @@ rules() ->
                 ])
     }.
 
-principal() ->
-    #{default => all,
-      type => hoconsc:union(
-                [ all
-                , hoconsc:ref(?MODULE, username)
-                , hoconsc:ref(?MODULE, clientid)
-                , hoconsc:ref(?MODULE, ipaddress)
-                , hoconsc:ref(?MODULE, andlist)
-                , hoconsc:ref(?MODULE, orlist)
-                ])
-     }.
-
 query() ->
     #{type => binary(),
       validator => fun(S) ->
@@ -202,8 +177,7 @@ connector_fields(DB) ->
               Error ->
                   erlang:error(Error)
           end,
-    [ {principal, principal()}
-    , {type, #{type => DB}}
+    [ {type, #{type => DB}}
     , {enable, #{type => boolean(),
                  default => true}}
     ] ++ Mod:fields("").
diff --git a/apps/emqx_authz/test/emqx_authz_SUITE.erl b/apps/emqx_authz/test/emqx_authz_SUITE.erl
index bcc855a59..30848f3d1 100644
--- a/apps/emqx_authz/test/emqx_authz_SUITE.erl
+++ b/apps/emqx_authz/test/emqx_authz_SUITE.erl
@@ -31,6 +31,10 @@ groups() ->
     [].
 
 init_per_suite(Config) ->
+    meck:new(emqx_resource, [non_strict, passthrough, no_history, no_link]),
+    meck:expect(emqx_resource, create, fun(_, _, _) -> {ok, meck_data} end),
+    meck:expect(emqx_resource, remove, fun(_) -> ok end ),
+
     ok = emqx_config:init_load(emqx_authz_schema, ?CONF_DEFAULT),
     ok = emqx_ct_helpers:start_apps([emqx_authz]),
     {ok, _} = emqx:update_config([authorization, cache, enable], false),
@@ -39,43 +43,63 @@ init_per_suite(Config) ->
 
 end_per_suite(_Config) ->
     {ok, _} = emqx_authz:update(replace, []),
-    emqx_ct_helpers:stop_apps([emqx_authz]),
+    emqx_ct_helpers:stop_apps([emqx_authz, emqx_resource]),
+    meck:unload(emqx_resource),
     ok.
 
 init_per_testcase(_, Config) ->
     {ok, _} = emqx_authz:update(replace, []),
     Config.
 
--define(RULE1, #{<<"principal">> => <<"all">>,
-                 <<"topics">> => [<<"#">>],
-                 <<"action">> => <<"all">>,
-                 <<"permission">> => <<"deny">>}
-       ).
--define(RULE2, #{<<"principal">> =>
-                    #{<<"ipaddress">> => <<"127.0.0.1">>},
-                 <<"topics">> =>
-                        [#{<<"eq">> => <<"#">>},
-                         #{<<"eq">> => <<"+">>}
-                        ] ,
-                 <<"action">> => <<"all">>,
-                 <<"permission">> => <<"allow">>}
-       ).
--define(RULE3,#{<<"principal">> =>
-                    #{<<"and">> => [#{<<"username">> => <<"^test?">>},
-                                    #{<<"clientid">> => <<"^test?">>}
-                                   ]},
-                <<"topics">> => [<<"test">>],
-                <<"action">> => <<"publish">>,
-                <<"permission">> => <<"allow">>}
-       ).
--define(RULE4,#{<<"principal">> =>
-                    #{<<"or">> => [#{<<"username">> => <<"^test">>},
-                                   #{<<"clientid">> => <<"test?">>}
-                                  ]},
-                <<"topics">> => [<<"%u">>,<<"%c">>],
-                <<"action">> => <<"publish">>,
-                <<"permission">> => <<"deny">>}
-       ).
+-define(RULE1, #{<<"type">> => <<"http">>,
+                 <<"config">> => #{
+                    <<"url">> => <<"https://fake.com:443/">>,
+                    <<"headers">> => #{},
+                    <<"method">> => <<"get">>,
+                    <<"request_timeout">> => 5000}
+                }).
+-define(RULE2, #{<<"type">> => <<"mongo">>,
+                 <<"config">> => #{
+                        <<"mongo_type">> => <<"single">>,
+                        <<"server">> => <<"127.0.0.1:27017">>,
+                        <<"pool_size">> => 1,
+                        <<"database">> => <<"mqtt">>,
+                        <<"ssl">> => #{<<"enable">> => false}},
+                 <<"collection">> => <<"fake">>,
+                 <<"find">> => #{<<"a">> => <<"b">>}
+                }).
+-define(RULE3, #{<<"type">> => <<"mysql">>,
+                 <<"config">> => #{
+                     <<"server">> => <<"127.0.0.1:27017">>,
+                     <<"pool_size">> => 1,
+                     <<"database">> => <<"mqtt">>,
+                     <<"username">> => <<"xx">>,
+                     <<"password">> => <<"ee">>,
+                     <<"auto_reconnect">> => true,
+                     <<"ssl">> => #{<<"enable">> => false}},
+                 <<"sql">> => <<"abcb">>
+                }).
+-define(RULE4, #{<<"type">> => <<"pgsql">>,
+                 <<"config">> => #{
+                     <<"server">> => <<"127.0.0.1:27017">>,
+                     <<"pool_size">> => 1,
+                     <<"database">> => <<"mqtt">>,
+                     <<"username">> => <<"xx">>,
+                     <<"password">> => <<"ee">>,
+                     <<"auto_reconnect">> => true,
+                     <<"ssl">> => #{<<"enable">> => false}},
+                 <<"sql">> => <<"abcb">>
+                }).
+-define(RULE5, #{<<"type">> => <<"redis">>,
+                 <<"config">> => #{
+                     <<"server">> => <<"127.0.0.1:27017">>,
+                     <<"pool_size">> => 1,
+                     <<"database">> => 0,
+                     <<"password">> => <<"ee">>,
+                     <<"auto_reconnect">> => true,
+                     <<"ssl">> => #{<<"enable">> => false}},
+                 <<"cmd">> => <<"HGETALL mqtt_authz:%u">>
+                }).
 
 %%------------------------------------------------------------------------------
 %% Testcases
@@ -86,73 +110,50 @@ t_update_rule(_) ->
     {ok, _} = emqx_authz:update(head, [?RULE1]),
     {ok, _} = emqx_authz:update(tail, [?RULE3]),
 
+    dbg:tracer(),dbg:p(all,c),
+    dbg:tpl(hocon_schema, check, cx),
     Lists1 = emqx_authz:check_rules([?RULE1, ?RULE2, ?RULE3]),
     ?assertMatch(Lists1, emqx:get_config([authorization_rules, rules], [])),
 
-    [#{annotations := #{id := Id1,
-                        principal := all,
-                        topics := [['#']]}
-      },
-     #{annotations := #{id := Id2,
-                        principal := #{ipaddress := {{127,0,0,1},{127,0,0,1},32}},
-                        topics := [#{eq := ['#']}, #{eq := ['+']}]}
-      },
-     #{annotations := #{id := Id3,
-                        principal :=
-                             #{'and' := [#{username := {re_pattern, _, _, _, _}},
-                                         #{clientid := {re_pattern, _, _, _, _}}
-                                        ]
-                              },
-                        topics := [[<<"test">>]]}
-      }
+    [#{annotations := #{id := Id1}, type := http},
+     #{annotations := #{id := Id2}, type := mongo},
+     #{annotations := #{id := Id3}, type := mysql}
     ] = emqx_authz:lookup(),
 
+    {ok, _} = emqx_authz:update({replace_once, Id1}, ?RULE5),
     {ok, _} = emqx_authz:update({replace_once, Id3}, ?RULE4),
     Lists2 = emqx_authz:check_rules([?RULE1, ?RULE2, ?RULE4]),
     ?assertMatch(Lists2, emqx:get_config([authorization_rules, rules], [])),
 
-    [#{annotations := #{id := Id1,
-                        principal := all,
-                        topics := [['#']]}
-      },
-     #{annotations := #{id := Id2,
-                        principal := #{ipaddress := {{127,0,0,1},{127,0,0,1},32}},
-                        topics := [#{eq := ['#']},
-                                   #{eq := ['+']}]}
-      },
-     #{annotations := #{id := Id3,
-                        principal :=
-                              #{'or' := [#{username := {re_pattern, _, _, _, _}},
-                                         #{clientid := {re_pattern, _, _, _, _}}
-                                        ]
-                               },
-                        topics := [#{pattern := [<<"%u">>]},
-                                   #{pattern := [<<"%c">>]}
-                                  ]}
-      }
+    [#{annotations := #{id := Id1}, type := redis},
+     #{annotations := #{id := Id2}, type := mongo},
+     #{annotations := #{id := Id3}, type := pgsql}
     ] = emqx_authz:lookup(),
 
     {ok, _} = emqx_authz:update(replace, []).
 
 t_move_rule(_) ->
-    {ok, _} = emqx_authz:update(replace, [?RULE1, ?RULE2, ?RULE3, ?RULE4]),
+    {ok, _} = emqx_authz:update(replace, [?RULE1, ?RULE2, ?RULE3, ?RULE4, ?RULE5]),
     [#{annotations := #{id := Id1}},
      #{annotations := #{id := Id2}},
      #{annotations := #{id := Id3}},
-     #{annotations := #{id := Id4}}
+     #{annotations := #{id := Id4}},
+     #{annotations := #{id := Id5}}
     ] = emqx_authz:lookup(),
 
     {ok, _} = emqx_authz:move(Id4, <<"top">>),
     ?assertMatch([#{annotations := #{id := Id4}},
                   #{annotations := #{id := Id1}},
                   #{annotations := #{id := Id2}},
-                  #{annotations := #{id := Id3}}
+                  #{annotations := #{id := Id3}},
+                  #{annotations := #{id := Id5}}
                  ], emqx_authz:lookup()),
 
     {ok, _} = emqx_authz:move(Id1, <<"bottom">>),
     ?assertMatch([#{annotations := #{id := Id4}},
                   #{annotations := #{id := Id2}},
                   #{annotations := #{id := Id3}},
+                  #{annotations := #{id := Id5}},
                   #{annotations := #{id := Id1}}
                  ], emqx_authz:lookup()),
 
@@ -160,66 +161,15 @@ t_move_rule(_) ->
     ?assertMatch([#{annotations := #{id := Id3}},
                   #{annotations := #{id := Id4}},
                   #{annotations := #{id := Id2}},
+                  #{annotations := #{id := Id5}},
                   #{annotations := #{id := Id1}}
                  ], emqx_authz:lookup()),
 
     {ok, _} = emqx_authz:move(Id2, #{<<"after">> => Id1}),
     ?assertMatch([#{annotations := #{id := Id3}},
                   #{annotations := #{id := Id4}},
+                  #{annotations := #{id := Id5}},
                   #{annotations := #{id := Id1}},
                   #{annotations := #{id := Id2}}
                  ], emqx_authz:lookup()),
     ok.
-
-t_authz(_) ->
-    ClientInfo1 = #{clientid => <<"test">>,
-                    username => <<"test">>,
-                    peerhost => {127,0,0,1},
-                    zone => default,
-                    listener => mqtt_tcp
-                   },
-    ClientInfo2 = #{clientid => <<"test">>,
-                    username => <<"test">>,
-                    peerhost => {192,168,0,10},
-                    zone => default,
-                    listener => mqtt_tcp
-                   },
-    ClientInfo3 = #{clientid => <<"test">>,
-                    username => <<"fake">>,
-                    peerhost => {127,0,0,1},
-                    zone => default,
-                    listener => mqtt_tcp
-                   },
-    ClientInfo4 = #{clientid => <<"fake">>,
-                    username => <<"test">>,
-                    peerhost => {127,0,0,1},
-                    zone => default,
-                    listener => mqtt_tcp
-                   },
-
-    Rules1 = [emqx_authz:init_rule(Rule) || Rule <- emqx_authz:check_rules([?RULE1, ?RULE2])],
-    Rules2 = [emqx_authz:init_rule(Rule) || Rule <- emqx_authz:check_rules([?RULE2, ?RULE1])],
-    Rules3 = [emqx_authz:init_rule(Rule) || Rule <- emqx_authz:check_rules([?RULE3, ?RULE4])],
-    Rules4 = [emqx_authz:init_rule(Rule) || Rule <- emqx_authz:check_rules([?RULE4, ?RULE1])],
-
-    ?assertEqual({stop, deny},
-        emqx_authz:authorize(ClientInfo1, subscribe, <<"#">>, deny, [])),
-    ?assertEqual({stop, deny},
-        emqx_authz:authorize(ClientInfo1, subscribe, <<"+">>, deny, Rules1)),
-    ?assertEqual({stop, allow},
-        emqx_authz:authorize(ClientInfo1, subscribe, <<"+">>, deny, Rules2)),
-    ?assertEqual({stop, allow},
-        emqx_authz:authorize(ClientInfo1, publish, <<"test">>, deny, Rules3)),
-    ?assertEqual({stop, deny},
-        emqx_authz:authorize(ClientInfo1, publish, <<"test">>, deny, Rules4)),
-    ?assertEqual({stop, deny},
-        emqx_authz:authorize(ClientInfo2, subscribe, <<"#">>, deny, Rules2)),
-    ?assertEqual({stop, deny},
-        emqx_authz:authorize(ClientInfo3, publish, <<"test">>, deny, Rules3)),
-    ?assertEqual({stop, deny},
-        emqx_authz:authorize(ClientInfo3, publish, <<"fake">>, deny, Rules4)),
-    ?assertEqual({stop, deny},
-        emqx_authz:authorize(ClientInfo4, publish, <<"test">>, deny, Rules3)),
-    ?assertEqual({stop, deny},
-        emqx_authz:authorize(ClientInfo4, publish, <<"fake">>, deny, Rules4)),
-    ok.
diff --git a/apps/emqx_authz/test/emqx_authz_api_SUITE.erl b/apps/emqx_authz/test/emqx_authz_api_SUITE.erl
index 9b6153465..77d620ccb 100644
--- a/apps/emqx_authz/test/emqx_authz_api_SUITE.erl
+++ b/apps/emqx_authz/test/emqx_authz_api_SUITE.erl
@@ -37,46 +37,100 @@
 -define(API_VERSION, "v5").
 -define(BASE_PATH, "api").
 
--define(CONF_DEFAULT, <<"authorization: {rules: []}">>).
+% -define(RULE1, #{<<"principal">> => <<"all">>,
+%                  <<"topics">> => [<<"#">>],
+%                  <<"action">> => <<"all">>,
+%                  <<"permission">> => <<"deny">>}
+%        ).
+% -define(RULE2, #{<<"principal">> =>
+%                     #{<<"ipaddress">> => <<"127.0.0.1">>},
+%                  <<"topics">> =>
+%                         [#{<<"eq">> => <<"#">>},
+%                          #{<<"eq">> => <<"+">>}
+%                         ] ,
+%                  <<"action">> => <<"all">>,
+%                  <<"permission">> => <<"allow">>}
+%        ).
+% -define(RULE3,#{<<"principal">> =>
+%                     #{<<"and">> => [#{<<"username">> => <<"^test?">>},
+%                                     #{<<"clientid">> => <<"^test?">>}
+%                                    ]},
+%                 <<"topics">> => [<<"test">>],
+%                 <<"action">> => <<"publish">>,
+%                 <<"permission">> => <<"allow">>}
+%        ).
+% -define(RULE4,#{<<"principal">> =>
+%                     #{<<"or">> => [#{<<"username">> => <<"^test">>},
+%                                    #{<<"clientid">> => <<"test?">>}
+%                                   ]},
+%                 <<"topics">> => [<<"%u">>,<<"%c">>],
+%                 <<"action">> => <<"publish">>,
+%                 <<"permission">> => <<"deny">>}
+%        ).
 
--define(RULE1, #{<<"principal">> => <<"all">>,
-                 <<"topics">> => [<<"#">>],
-                 <<"action">> => <<"all">>,
-                 <<"permission">> => <<"deny">>}
-       ).
--define(RULE2, #{<<"principal">> =>
-                    #{<<"ipaddress">> => <<"127.0.0.1">>},
-                 <<"topics">> =>
-                        [#{<<"eq">> => <<"#">>},
-                         #{<<"eq">> => <<"+">>}
-                        ] ,
-                 <<"action">> => <<"all">>,
-                 <<"permission">> => <<"allow">>}
-       ).
--define(RULE3,#{<<"principal">> =>
-                    #{<<"and">> => [#{<<"username">> => <<"^test?">>},
-                                    #{<<"clientid">> => <<"^test?">>}
-                                   ]},
-                <<"topics">> => [<<"test">>],
-                <<"action">> => <<"publish">>,
-                <<"permission">> => <<"allow">>}
-       ).
--define(RULE4,#{<<"principal">> =>
-                    #{<<"or">> => [#{<<"username">> => <<"^test">>},
-                                   #{<<"clientid">> => <<"test?">>}
-                                  ]},
-                <<"topics">> => [<<"%u">>,<<"%c">>],
-                <<"action">> => <<"publish">>,
-                <<"permission">> => <<"deny">>}
-       ).
+-define(RULE1, #{<<"type">> => <<"http">>,
+                 <<"config">> => #{
+                    <<"url">> => <<"https://fake.com:443/">>,
+                    <<"headers">> => #{},
+                    <<"method">> => <<"get">>,
+                    <<"request_timeout">> => 5000}
+                }).
+-define(RULE2, #{<<"type">> => <<"mongo">>,
+                 <<"config">> => #{
+                        <<"mongo_type">> => <<"single">>,
+                        <<"server">> => <<"127.0.0.1:27017">>,
+                        <<"pool_size">> => 1,
+                        <<"database">> => <<"mqtt">>,
+                        <<"ssl">> => #{<<"enable">> => false}},
+                 <<"collection">> => <<"fake">>,
+                 <<"find">> => #{<<"a">> => <<"b">>}
+                }).
+-define(RULE3, #{<<"type">> => <<"mysql">>,
+                 <<"config">> => #{
+                     <<"server">> => <<"127.0.0.1:27017">>,
+                     <<"pool_size">> => 1,
+                     <<"database">> => <<"mqtt">>,
+                     <<"username">> => <<"xx">>,
+                     <<"password">> => <<"ee">>,
+                     <<"auto_reconnect">> => true,
+                     <<"ssl">> => #{<<"enable">> => false}},
+                 <<"sql">> => <<"abcb">>
+                }).
+-define(RULE4, #{<<"type">> => <<"pgsql">>,
+                 <<"config">> => #{
+                     <<"server">> => <<"127.0.0.1:27017">>,
+                     <<"pool_size">> => 1,
+                     <<"database">> => <<"mqtt">>,
+                     <<"username">> => <<"xx">>,
+                     <<"password">> => <<"ee">>,
+                     <<"auto_reconnect">> => true,
+                     <<"ssl">> => #{<<"enable">> => false}},
+                 <<"sql">> => <<"abcb">>
+                }).
+-define(RULE5, #{<<"type">> => <<"redis">>,
+                 <<"config">> => #{
+                     <<"server">> => <<"127.0.0.1:27017">>,
+                     <<"pool_size">> => 1,
+                     <<"database">> => 0,
+                     <<"password">> => <<"ee">>,
+                     <<"auto_reconnect">> => true,
+                     <<"ssl">> => #{<<"enable">> => false}},
+                 <<"cmd">> => <<"HGETALL mqtt_authz:%u">>
+                }).
 
 all() ->
-    emqx_ct:all(?MODULE).
+    % emqx_ct:all(?MODULE).
+    [].
 
 groups() ->
     [].
 
 init_per_suite(Config) ->
+    meck:new(emqx_resource, [non_strict, passthrough, no_history, no_link]),
+    meck:expect(emqx_resource, create, fun(_, _, _) -> {ok, meck_data} end),
+    meck:expect(emqx_resource, update, fun(_, _, _, _) -> {ok, meck_data} end),
+    meck:expect(emqx_resource, remove, fun(_) -> ok end ),
+
     ekka_mnesia:start(),
     emqx_mgmt_auth:mnesia(boot),
 
@@ -89,7 +143,8 @@ init_per_suite(Config) ->
 
 end_per_suite(_Config) ->
     {ok, _} = emqx_authz:update(replace, []),
-    emqx_ct_helpers:stop_apps([emqx_authz, emqx_management]),
+    emqx_ct_helpers:stop_apps([emqx_resource, emqx_authz, emqx_management]),
+    meck:unload(emqx_resource),
     ok.
 
 set_special_configs(emqx_management) ->
@@ -111,12 +166,7 @@ t_api(_) ->
     ?assertEqual([], get_rules(Result1)),
 
     lists:foreach(fun(_) ->
-                        {ok, 204, _} = request(post, uri(["authorization"]),
-                                         #{<<"action">> => <<"all">>,
-                                           <<"permission">> => <<"deny">>,
-                                           <<"principal">> => <<"all">>,
-                                           <<"topics">> => [<<"#">>]}
-                                        )
+                        {ok, 204, _} = request(post, uri(["authorization"]), ?RULE1)
                   end, lists:seq(1, 20)),
     {ok, 200, Result2} = request(get, uri(["authorization"]), []),
     ?assertEqual(20, length(get_rules(Result2))),
@@ -128,30 +178,23 @@ t_api(_) ->
                           ?assertEqual(10, length(get_rules(Result)))
                   end, lists:seq(1, 2)),
 
-    {ok, 204, _} = request(put, uri(["authorization"]),
-                           [ #{<<"action">> => <<"all">>, <<"permission">> => <<"allow">>, <<"principal">> => <<"all">>, <<"topics">> => [<<"#">>]}
-                           , #{<<"action">> => <<"all">>, <<"permission">> => <<"allow">>, <<"principal">> => <<"all">>, <<"topics">> => [<<"#">>]}
-                           , #{<<"action">> => <<"all">>, <<"permission">> => <<"allow">>, <<"principal">> => <<"all">>, <<"topics">> => [<<"#">>]}
-                           ]),
+    {ok, 204, _} = request(put, uri(["authorization"]), [?RULE1, ?RULE2, ?RULE3, ?RULE4]),
 
     {ok, 200, Result3} = request(get, uri(["authorization"]), []),
     Rules = get_rules(Result3),
-    ?assertEqual(3, length(Rules)),
-
-    lists:foreach(fun(#{<<"permission">> := Allow}) ->
-                          ?assertEqual(<<"allow">>, Allow)
-                  end, Rules),
+    ?assertEqual(4, length(Rules)),
+    ?assertMatch([ #{<<"type">> := <<"http">>}
+                 , #{<<"type">> := <<"mongo">>}
+                 , #{<<"type">> := <<"mysql">>}
+                 , #{<<"type">> := <<"pgsql">>}
+                 ], Rules),
 
     #{<<"annotations">> := #{<<"id">> := Id}} = lists:nth(2, Rules),
 
-    {ok, 204, _} = request(put, uri(["authorization", binary_to_list(Id)]),
-                           #{<<"action">> => <<"all">>, <<"permission">> => <<"deny">>,
-                             <<"principal">> => <<"all">>, <<"topics">> => [<<"#">>]}),
+    {ok, 204, _} = request(put, uri(["authorization", binary_to_list(Id)]), ?RULE5),
 
     {ok, 200, Result4} = request(get, uri(["authorization", binary_to_list(Id)]), []),
-    ?assertMatch(#{<<"annotations">> := #{<<"id">> := Id},
-                   <<"permission">> := <<"deny">>
-                  }, jsx:decode(Result4)),
+    ?assertMatch(#{<<"type">> := <<"redis">>}, jsx:decode(Result4)),
 
     lists:foreach(fun(#{<<"annotations">> := #{<<"id">> := Id0}}) ->
                     {ok, 204, _} = request(delete, uri(["authorization", binary_to_list(Id0)]), [])
@@ -161,11 +204,12 @@ t_api(_) ->
     ok.
 
 t_move_rule(_) ->
-    {ok, _} = emqx_authz:update(replace, [?RULE1, ?RULE2, ?RULE3, ?RULE4]),
+    {ok, _} = emqx_authz:update(replace, [?RULE1, ?RULE2, ?RULE3, ?RULE4, ?RULE5]),
     [#{annotations := #{id := Id1}},
      #{annotations := #{id := Id2}},
      #{annotations := #{id := Id3}},
-     #{annotations := #{id := Id4}}
+     #{annotations := #{id := Id4}},
+     #{annotations := #{id := Id5}}
     ] = emqx_authz:lookup(),
 
     {ok, 204, _} = request(post, uri(["authorization", Id4, "move"]),
@@ -173,7 +217,8 @@ t_move_rule(_) ->
     ?assertMatch([#{annotations := #{id := Id4}},
                   #{annotations := #{id := Id1}},
                   #{annotations := #{id := Id2}},
-                  #{annotations := #{id := Id3}}
+                  #{annotations := #{id := Id3}},
+                  #{annotations := #{id := Id5}}
                  ], emqx_authz:lookup()),
 
     {ok, 204, _} = request(post, uri(["authorization", Id1, "move"]),
@@ -181,6 +226,7 @@ t_move_rule(_) ->
     ?assertMatch([#{annotations := #{id := Id4}},
                   #{annotations := #{id := Id2}},
                   #{annotations := #{id := Id3}},
+                  #{annotations := #{id := Id5}},
                   #{annotations := #{id := Id1}}
                  ], emqx_authz:lookup()),
 
@@ -189,6 +235,7 @@ t_move_rule(_) ->
     ?assertMatch([#{annotations := #{id := Id3}},
                   #{annotations := #{id := Id4}},
                   #{annotations := #{id := Id2}},
+                  #{annotations := #{id := Id5}},
                   #{annotations := #{id := Id1}}
                  ], emqx_authz:lookup()),
 
@@ -196,6 +243,7 @@ t_move_rule(_) ->
                            #{<<"position">> => #{<<"after">> => Id1}}),
     ?assertMatch([#{annotations := #{id := Id3}},
                   #{annotations := #{id := Id4}},
+                  #{annotations := #{id := Id5}},
                   #{annotations := #{id := Id1}},
                   #{annotations := #{id := Id2}}
                  ], emqx_authz:lookup()),
diff --git a/apps/emqx_authz/test/emqx_authz_http_SUITE.erl b/apps/emqx_authz/test/emqx_authz_http_SUITE.erl
index fb95c1b00..c7a3fc449 100644
--- a/apps/emqx_authz/test/emqx_authz_http_SUITE.erl
+++ b/apps/emqx_authz/test/emqx_authz_http_SUITE.erl
@@ -41,14 +41,13 @@ init_per_suite(Config) ->
 
     {ok, _} = emqx:update_config([authorization, cache, enable], false),
     {ok, _} = emqx:update_config([authorization, no_match], deny),
-    Rules = [#{ <<"config">> => #{
-                    <<"url">> => <<"https://fake.com:443/">>,
-                    <<"headers">> => #{},
-                    <<"method">> => <<"get">>,
-                    <<"request_timeout">> => 5000
-                },
-                <<"principal">> => <<"all">>,
-                <<"type">> => <<"http">>}
+    Rules = [#{<<"type">> => <<"http">>,
+               <<"config">> => #{
+                   <<"url">> => <<"https://fake.com:443/">>,
+                   <<"headers">> => #{},
+                   <<"method">> => <<"get">>,
+                   <<"request_timeout">> => 5000
+               }}
             ],
     {ok, _} = emqx_authz:update(replace, Rules),
     Config.
diff --git a/apps/emqx_authz/test/emqx_authz_mongo_SUITE.erl b/apps/emqx_authz/test/emqx_authz_mongo_SUITE.erl
index cffc0ad76..dac106b37 100644
--- a/apps/emqx_authz/test/emqx_authz_mongo_SUITE.erl
+++ b/apps/emqx_authz/test/emqx_authz_mongo_SUITE.erl
@@ -39,17 +39,16 @@ init_per_suite(Config) ->
     ok = emqx_ct_helpers:start_apps([emqx_authz]),
     {ok, _} = emqx:update_config([authorization, cache, enable], false),
     {ok, _} = emqx:update_config([authorization, no_match], deny),
-    Rules = [#{ <<"config">> => #{
-                        <<"mongo_type">> => <<"single">>,
-                        <<"server">> => <<"127.0.0.1:27017">>,
-                        <<"pool_size">> => 1,
-                        <<"database">> => <<"mqtt">>,
-                        <<"ssl">> => #{<<"enable">> => false}},
-                <<"principal">> => <<"all">>,
-                <<"collection">> => <<"fake">>,
-                <<"find">> => #{<<"a">> => <<"b">>},
-                <<"type">> => <<"mongo">>}
-            ],
+    Rules = [#{<<"type">> => <<"mongo">>,
+               <<"config">> => #{
+                      <<"mongo_type">> => <<"single">>,
+                      <<"server">> => <<"127.0.0.1:27017">>,
+                      <<"pool_size">> => 1,
+                      <<"database">> => <<"mqtt">>,
+                      <<"ssl">> => #{<<"enable">> => false}},
+               <<"collection">> => <<"fake">>,
+               <<"find">> => #{<<"a">> => <<"b">>}
+              }],
     {ok, _} = emqx_authz:update(replace, Rules),
     Config.
 
diff --git a/apps/emqx_authz/test/emqx_authz_mysql_SUITE.erl b/apps/emqx_authz/test/emqx_authz_mysql_SUITE.erl
index dcc0e47d7..0fba033a6 100644
--- a/apps/emqx_authz/test/emqx_authz_mysql_SUITE.erl
+++ b/apps/emqx_authz/test/emqx_authz_mysql_SUITE.erl
@@ -40,18 +40,17 @@ init_per_suite(Config) ->
 
     {ok, _} = emqx:update_config([authorization, cache, enable], false),
     {ok, _} = emqx:update_config([authorization, no_match], deny),
-    Rules = [#{ <<"config">> => #{
-                    <<"server">> => <<"127.0.0.1:27017">>,
-                    <<"pool_size">> => 1,
-                    <<"database">> => <<"mqtt">>,
-                    <<"username">> => <<"xx">>,
-                    <<"password">> => <<"ee">>,
-                    <<"auto_reconnect">> => true,
-                    <<"ssl">> => #{<<"enable">> => false}
-                },
-                <<"principal">> => <<"all">>,
-                <<"sql">> => <<"abcb">>,
-                <<"type">> => <<"mysql">> }],
+    Rules = [#{<<"type">> => <<"mysql">>,
+               <<"config">> => #{
+                   <<"server">> => <<"127.0.0.1:27017">>,
+                   <<"pool_size">> => 1,
+                   <<"database">> => <<"mqtt">>,
+                   <<"username">> => <<"xx">>,
+                   <<"password">> => <<"ee">>,
+                   <<"auto_reconnect">> => true,
+                   <<"ssl">> => #{<<"enable">> => false}},
+               <<"sql">> => <<"abcb">>
+              }],
     {ok, _} = emqx_authz:update(replace, Rules),
     Config.
 
@@ -60,17 +59,14 @@ end_per_suite(_Config) ->
     emqx_ct_helpers:stop_apps([emqx_authz, emqx_resource]),
     meck:unload(emqx_resource).
 
--define(COLUMNS, [ <<"ipaddress">>
-                 , <<"username">>
-                 , <<"clientid">>
-                 , <<"action">>
+-define(COLUMNS, [ <<"action">>
                  , <<"permission">>
                  , <<"topic">>
                  ]).
--define(RULE1, [[<<"127.0.0.1">>, <<>>, <<>>, <<"all">>, <<"deny">>, <<"#">>]]).
--define(RULE2, [[<<"127.0.0.1">>, <<>>, <<>>, <<"all">>, <<"allow">>, <<"eq #">>]]).
--define(RULE3, [[<<>>, <<"^test">>, <<"^test">> ,<<"subscribe">>, <<"allow">>, <<"test/%c">>]]).
--define(RULE4, [[<<>>, <<"^test">>, <<"^test">> ,<<"publish">>, <<"allow">>, <<"test/%u">>]]).
+-define(RULE1, [[<<"all">>, <<"deny">>, <<"#">>]]).
+-define(RULE2, [[<<"all">>, <<"allow">>, <<"eq #">>]]).
+-define(RULE3, [[<<"subscribe">>, <<"allow">>, <<"test/%c">>]]).
+-define(RULE4, [[<<"publish">>, <<"allow">>, <<"test/%u">>]]).
 
 %%------------------------------------------------------------------------------
 %% Testcases
diff --git a/apps/emqx_authz/test/emqx_authz_pgsql_SUITE.erl b/apps/emqx_authz/test/emqx_authz_pgsql_SUITE.erl
index b4383e21e..d21caa223 100644
--- a/apps/emqx_authz/test/emqx_authz_pgsql_SUITE.erl
+++ b/apps/emqx_authz/test/emqx_authz_pgsql_SUITE.erl
@@ -40,17 +40,17 @@ init_per_suite(Config) ->
 
     {ok, _} = emqx:update_config([authorization, cache, enable], false),
     {ok, _} = emqx:update_config([authorization, no_match], deny),
-    Rules = [#{ <<"config">> => #{
-                    <<"server">> => <<"127.0.0.1:27017">>,
-                    <<"pool_size">> => 1,
-                    <<"database">> => <<"mqtt">>,
-                    <<"username">> => <<"xx">>,
-                    <<"password">> => <<"ee">>,
-                    <<"auto_reconnect">> => true,
-                    <<"ssl">> => #{<<"enable">> => false}
-                },
-                <<"sql">> => <<"abcb">>,
-                <<"type">> => <<"pgsql">> }],
+    Rules = [#{<<"type">> => <<"pgsql">>,
+               <<"config">> => #{
+                   <<"server">> => <<"127.0.0.1:27017">>,
+                   <<"pool_size">> => 1,
+                   <<"database">> => <<"mqtt">>,
+                   <<"username">> => <<"xx">>,
+                   <<"password">> => <<"ee">>,
+                   <<"auto_reconnect">> => true,
+                   <<"ssl">> => #{<<"enable">> => false}},
+               <<"sql">> => <<"abcb">>
+              }],
     {ok, _} = emqx_authz:update(replace, Rules),
     Config.
 
@@ -59,17 +59,14 @@ end_per_suite(_Config) ->
     emqx_ct_helpers:stop_apps([emqx_authz, emqx_resource]),
     meck:unload(emqx_resource).
 
--define(COLUMNS, [ {column, <<"ipaddress">>, meck, meck, meck, meck, meck, meck, meck}
-                 , {column, <<"username">>, meck, meck, meck, meck, meck, meck, meck}
-                 , {column, <<"clientid">>, meck, meck, meck, meck, meck, meck, meck}
-                 , {column, <<"action">>, meck, meck, meck, meck, meck, meck, meck}
+-define(COLUMNS, [ {column, <<"action">>, meck, meck, meck, meck, meck, meck, meck}
                  , {column, <<"permission">>, meck, meck, meck, meck, meck, meck, meck}
                  , {column, <<"topic">>, meck, meck, meck, meck, meck, meck, meck}
                  ]).
--define(RULE1, [{<<"127.0.0.1">>, <<>>, <<>>, <<"all">>, <<"deny">>, <<"#">>}]).
--define(RULE2, [{<<"127.0.0.1">>, <<>>, <<>>, <<"all">>, <<"allow">>, <<"eq #">>}]).
--define(RULE3, [{<<>>, <<"^test">>, <<"^test">> ,<<"subscribe">>, <<"allow">>, <<"test/%c">>}]).
--define(RULE4, [{<<>>, <<"^test">>, <<"^test">> ,<<"publish">>, <<"allow">>, <<"test/%u">>}]).
+-define(RULE1, [{<<"all">>, <<"deny">>, <<"#">>}]).
+-define(RULE2, [{<<"all">>, <<"allow">>, <<"eq #">>}]).
+-define(RULE3, [{<<"subscribe">>, <<"allow">>, <<"test/%c">>}]).
+-define(RULE4, [{<<"publish">>, <<"allow">>, <<"test/%u">>}]).
 
 %%------------------------------------------------------------------------------
 %% Testcases
diff --git a/apps/emqx_authz/test/emqx_authz_redis_SUITE.erl b/apps/emqx_authz/test/emqx_authz_redis_SUITE.erl
index d3eebeb2e..073c339ae 100644
--- a/apps/emqx_authz/test/emqx_authz_redis_SUITE.erl
+++ b/apps/emqx_authz/test/emqx_authz_redis_SUITE.erl
@@ -41,16 +41,16 @@ init_per_suite(Config) ->
 
     {ok, _} = emqx:update_config([authorization, cache, enable], false),
     {ok, _} = emqx:update_config([authorization, no_match], deny),
-    Rules = [#{ <<"config">> => #{
-                    <<"server">> => <<"127.0.0.1:27017">>,
-                    <<"pool_size">> => 1,
-                    <<"database">> => 0,
-                    <<"password">> => <<"ee">>,
-                    <<"auto_reconnect">> => true,
-                    <<"ssl">> => #{<<"enable">> => false}
-                },
-                <<"cmd">> => <<"HGETALL mqtt_authz:%u">>,
-                <<"type">> => <<"redis">> }],
+    Rules = [#{<<"type">> => <<"redis">>,
+               <<"config">> => #{
+                   <<"server">> => <<"127.0.0.1:27017">>,
+                   <<"pool_size">> => 1,
+                   <<"database">> => 0,
+                   <<"password">> => <<"ee">>,
+                   <<"auto_reconnect">> => true,
+                   <<"ssl">> => #{<<"enable">> => false}},
+               <<"cmd">> => <<"HGETALL mqtt_authz:%u">>
+              }],
     {ok, _} = emqx_authz:update(replace, Rules),
     Config.
 
diff --git a/apps/emqx_connector/src/emqx_connector_mongo.erl b/apps/emqx_connector/src/emqx_connector_mongo.erl
index 36ea01db2..6f69fafc4 100644
--- a/apps/emqx_connector/src/emqx_connector_mongo.erl
+++ b/apps/emqx_connector/src/emqx_connector_mongo.erl
@@ -19,8 +19,9 @@
 -include_lib("typerefl/include/types.hrl").
 -include_lib("emqx_resource/include/emqx_resource_behaviour.hrl").
 
--type server() :: string().
+-type server() :: emqx_schema:ip_port().
 -reflect_type([server/0]).
+-typerefl_from_string({server/0, emqx_connector_schema_lib, to_ip_port}).
 
 %% callbacks of behaviour emqx_resource
 -export([ on_start/2
@@ -95,7 +96,7 @@ on_start(InstId, Config = #{server := Server,
                             mongo_type := single}) ->
     logger:info("starting mongodb connector: ~p, config: ~p", [InstId, Config]),
     Opts = [{type, single},
-            {hosts, [Server]}
+            {hosts, [emqx_connector_schema_lib:ip_port_to_string(Server)]}
             ],
     do_start(InstId, Opts, Config);
 
@@ -104,14 +105,17 @@ on_start(InstId, Config = #{servers := Servers,
                             replica_set_name := RsName}) ->
     logger:info("starting mongodb connector: ~p, config: ~p", [InstId, Config]),
     Opts = [{type,  {rs, RsName}},
-            {hosts, Servers}],
+            {hosts, [emqx_connector_schema_lib:ip_port_to_string(S)
+                     || S <- Servers]}
+           ],
     do_start(InstId, Opts, Config);
 
 on_start(InstId, Config = #{servers := Servers,
                             mongo_type := sharded}) ->
     logger:info("starting mongodb connector: ~p, config: ~p", [InstId, Config]),
     Opts = [{type, sharded},
-            {hosts, Servers}
+            {hosts, [emqx_connector_schema_lib:ip_port_to_string(S)
+                     || S <- Servers]}
             ],
     do_start(InstId, Opts, Config).
 
diff --git a/apps/emqx_connector/src/emqx_connector_redis.erl b/apps/emqx_connector/src/emqx_connector_redis.erl
index 1ea31ced8..60087188f 100644
--- a/apps/emqx_connector/src/emqx_connector_redis.erl
+++ b/apps/emqx_connector/src/emqx_connector_redis.erl
@@ -19,10 +19,9 @@
 -include_lib("typerefl/include/types.hrl").
 -include_lib("emqx_resource/include/emqx_resource_behaviour.hrl").
 
--type server() :: tuple().
+-type server() :: emqx_schema:ip_port().
 -reflect_type([server/0]).
--typerefl_from_string({server/0, ?MODULE, to_server}).
--export([to_server/1]).
+-typerefl_from_string({server/0, emqx_connector_schema_lib, to_ip_port}).
 
 -export([structs/0, fields/1]).
 
@@ -170,9 +169,3 @@ redis_fields() ->
                    default => 0}}
     , {auto_reconnect, fun emqx_connector_schema_lib:auto_reconnect/1}
     ].
-
-to_server(Server) ->
-    case string:tokens(Server, ":") of
-        [Host, Port] -> {ok, {Host, list_to_integer(Port)}};
-        _ -> {error, Server}
-    end.