chore(authz): test Mysql backend with real Mysql

apps/emqx_authz/include/emqx_authz.hrl

@@ -21,7 +21,7 @@
 
 -define(CONF_KEY_PATH, [authorization, sources]).
 
--define(RE_PLACEHOLDER, "\\$\\{[a-z0-9\\-]+\\}").
+-define(RE_PLACEHOLDER, "\\$\\{[a-z0-9_]+\\}").
 
 -define(USERNAME_RULES_EXAMPLE, #{username => user1,
                                   rules => [ #{topic => <<"test/toopic/1">>,

apps/emqx_authz/src/emqx_authz_mysql.erl

@@ -53,17 +53,6 @@ dry_run(Source) ->
 destroy(#{annotations := #{id := Id}}) ->
     ok = emqx_resource:remove(Id).
 
-parse_query(undefined) ->
-    undefined;
-parse_query(Sql) ->
-    case re:run(Sql, ?RE_PLACEHOLDER, [global, {capture, all, list}]) of
-        {match, Variables} ->
-            Params = [Var || [Var] <- Variables],
-            {re:replace(Sql, ?RE_PLACEHOLDER, "?", [global, {return, list}]), Params};
-        nomatch ->
-            {Sql, []}
-    end.
-
 authorize(Client, PubSub, Topic,
             #{annotations := #{id := ResourceID,
                                query := {Query, Params}
@@ -80,6 +69,15 @@ authorize(Client, PubSub, Topic,
             nomatch
     end.
 
+parse_query(Sql) ->
+    case re:run(Sql, ?RE_PLACEHOLDER, [global, {capture, all, list}]) of
+        {match, Variables} ->
+            Params = [Var || [Var] <- Variables],
+            {re:replace(Sql, ?RE_PLACEHOLDER, "?", [global, {return, list}]), Params};
+        nomatch ->
+            {Sql, []}
+    end.
+
 do_authorize(_Client, _PubSub, _Topic, _Columns, []) ->
     nomatch;
 do_authorize(Client, PubSub, Topic, Columns, [Row | Tail]) ->

apps/emqx_authz/test/emqx_authz_mongodb_SUITE.erl

@@ -33,14 +33,6 @@ all() ->
 groups() ->
     [].
 
-init_per_testcase(_TestCase, Config) ->
-    {ok, _} = mc_worker_api:connect(mongo_config()),
-    Config.
-
-end_per_testcase(_TestCase, _Config) ->
-    ok = reset_samples(),
-    ok = mc_worker_api:disconnect(?MONGO_CLIENT).
-
 init_per_suite(Config) ->
     case emqx_authz_test_lib:is_tcp_server_available(?MONGO_HOST, ?MONGO_PORT) of
         true ->
@@ -55,7 +47,7 @@ init_per_suite(Config) ->
     end.
 
 end_per_suite(_Config) ->
-    ok = emqx_authz_test_lib:reset_authorizers(),
+    ok = emqx_authz_test_lib:restore_authorizers(),
     ok = stop_apps([emqx_resource, emqx_connector]),
     ok = emqx_common_test_helpers:stop_apps([emqx_authz]).
 
@@ -65,6 +57,15 @@ set_special_configs(emqx_authz) ->
 set_special_configs(_) ->
     ok.
 
+init_per_testcase(_TestCase, Config) ->
+    {ok, _} = mc_worker_api:connect(mongo_config()),
+    ok = emqx_authz_test_lib:reset_authorizers(),
+    Config.
+
+end_per_testcase(_TestCase, _Config) ->
+    ok = reset_samples(),
+    ok = mc_worker_api:disconnect(?MONGO_CLIENT).
+
 %%------------------------------------------------------------------------------
 %% Testcases
 %%------------------------------------------------------------------------------

apps/emqx_authz/test/emqx_authz_mysql_SUITE.erl

@@ -23,7 +23,10 @@
 -include_lib("common_test/include/ct.hrl").
 -include_lib("emqx/include/emqx_placeholder.hrl").
 
--define(CONF_DEFAULT, <<"authorization: {sources: []}">>).
+
+-define(MYSQL_HOST, "mysql").
+-define(MYSQL_PORT, 3306).
+-define(MYSQL_RESOURCE, <<"emqx_authz_mysql_SUITE">>).
 
 all() ->
     emqx_common_test_helpers:all(?MODULE).
@@ -32,101 +35,240 @@ groups() ->
     [].
 
 init_per_suite(Config) ->
-    meck:new(emqx_resource, [non_strict, passthrough, no_history, no_link]),
+    case emqx_authn_test_lib:is_tcp_server_available(?MYSQL_HOST, ?MYSQL_PORT) of
-    meck:expect(emqx_resource, create, fun(_, _, _) -> {ok, meck_data} end ),
+        true ->
-    meck:expect(emqx_resource, remove, fun(_) -> ok end ),
+            ok = emqx_common_test_helpers:start_apps(
-
+                   [emqx_conf, emqx_authz],
-    ok = emqx_common_test_helpers:start_apps(
+                   fun set_special_configs/1
-           [emqx_conf, emqx_authz],
+                  ),
-           fun set_special_configs/1),
+            ok = start_apps([emqx_resource, emqx_connector]),
-
+            {ok, _} = emqx_resource:create_local(
-    Rules = [#{<<"type">> => <<"mysql">>,
+              ?MYSQL_RESOURCE,
-               <<"server">> => <<"127.0.0.1:27017">>,
+              emqx_connector_mysql,
-               <<"pool_size">> => 1,
+              mysql_config()),
-               <<"database">> => <<"mqtt">>,
+            Config;
-               <<"username">> => <<"xx">>,
+        false ->
-               <<"password">> => <<"ee">>,
+            {skip, no_mysql}
-               <<"auto_reconnect">> => true,
+    end.
-               <<"ssl">> => #{<<"enable">> => false},
-               <<"query">> => <<"abcb">>
-              }],
-    {ok, _} = emqx_authz:update(replace, Rules),
-    Config.
 
 end_per_suite(_Config) ->
-    {ok, _} = emqx:update_config(
+    ok = emqx_authz_test_lib:restore_authorizers(),
-                [authorization],
+    ok = emqx_resource:remove_local(?MYSQL_RESOURCE),
-                #{<<"no_match">> => <<"allow">>,
+    ok = stop_apps([emqx_resource, emqx_connector]),
-                  <<"cache">> => #{<<"enable">> => <<"true">>},
+    ok = emqx_common_test_helpers:stop_apps([emqx_authz]).
-                  <<"sources">> => []}),
+
-    emqx_common_test_helpers:stop_apps([emqx_authz, emqx_conf]),
+init_per_testcase(Config) ->
-    meck:unload(emqx_resource),
+    ok = emqx_authz_test_lib:reset_authorizers(),
-    ok.
+    Config.
 
 set_special_configs(emqx_authz) ->
-    {ok, _} = emqx:update_config([authorization, cache, enable], false),
+    ok = emqx_authz_test_lib:reset_authorizers();
-    {ok, _} = emqx:update_config([authorization, no_match], deny),
-    {ok, _} = emqx:update_config([authorization, sources], []),
-    ok;
-set_special_configs(_App) ->
-    ok.
 
--define(COLUMNS, [ <<"action">>
+set_special_configs(_) ->
-                 , <<"permission">>
+    ok.
-                 , <<"topic">>
-                 ]).
--define(SOURCE1, [[<<"all">>, <<"deny">>, <<"#">>]]).
--define(SOURCE2, [[<<"all">>, <<"allow">>, <<"eq #">>]]).
--define(SOURCE3, [[<<"subscribe">>, <<"allow">>, <<"test/", ?PH_CLIENTID/binary>>]]).
--define(SOURCE4, [[<<"publish">>, <<"allow">>, <<"test/", ?PH_USERNAME/binary>>]]).
 
 %%------------------------------------------------------------------------------
 %% Testcases
 %%------------------------------------------------------------------------------
 
-t_authz(_) ->
+t_topic_rules(_Config) ->
-    ClientInfo1 = #{clientid => <<"test">>,
+    ClientInfo = #{clientid => <<"clientid">>,
-                    username => <<"test">>,
+                   username => <<"username">>,
-                    peerhost => {127,0,0,1},
+                   peerhost => {127,0,0,1},
-                    zone => default,
+                   zone => default,
-                    listener => {tcp, default}
+                   listener => {tcp, default}
-                   },
+                  },
-    ClientInfo2 = #{clientid => <<"test_clientid">>,
-                    username => <<"test_username">>,
-                    peerhost => {192,168,0,10},
-                    zone => default,
-                    listener => {tcp, default}
-                   },
-    ClientInfo3 = #{clientid => <<"test_clientid">>,
-                    username => <<"fake_username">>,
-                    peerhost => {127,0,0,1},
-                    zone => default,
-                    listener => {tcp, default}
-                   },
 
-    meck:expect(emqx_resource, query, fun(_, _) -> {ok, ?COLUMNS, []} end),
+    ok = emqx_authz_test_lib:test_no_topic_rules(ClientInfo, fun setup_client_samples/2),
-    ?assertEqual(deny, emqx_access_control:authorize(ClientInfo1, subscribe, <<"#">>)), % nomatch
-    ?assertEqual(deny, emqx_access_control:authorize(ClientInfo1, publish, <<"#">>)), % nomatch
 
-    meck:expect(emqx_resource, query, fun(_, _) -> {ok, ?COLUMNS, ?SOURCE1 ++ ?SOURCE2} end),
+    ok = emqx_authz_test_lib:test_allow_topic_rules(ClientInfo, fun setup_client_samples/2),
-    ?assertEqual(deny, emqx_access_control:authorize(ClientInfo1, subscribe, <<"+">>)),
-    ?assertEqual(deny, emqx_access_control:authorize(ClientInfo1, publish, <<"+">>)),
 
-    meck:expect(emqx_resource, query, fun(_, _) -> {ok, ?COLUMNS, ?SOURCE2 ++ ?SOURCE1} end),
+    ok = emqx_authz_test_lib:test_deny_topic_rules(ClientInfo, fun setup_client_samples/2).
-    ?assertEqual(allow, emqx_access_control:authorize(ClientInfo1, subscribe, <<"#">>)),
-    ?assertEqual(deny, emqx_access_control:authorize(ClientInfo1, subscribe, <<"+">>)),
 
-    meck:expect(emqx_resource, query, fun(_, _) -> {ok, ?COLUMNS, ?SOURCE3 ++ ?SOURCE4} end),
+
-    ?assertEqual(allow, emqx_access_control:authorize(
+t_lookups(_Config) ->
-                          ClientInfo2, subscribe, <<"test/test_clientid">>)),
+    ClientInfo = #{clientid => <<"clientid">>,
-    ?assertEqual(deny,  emqx_access_control:authorize(
+                   cn => <<"cn">>,
-                          ClientInfo2, publish,   <<"test/test_clientid">>)),
+                   dn => <<"dn">>,
-    ?assertEqual(deny,  emqx_access_control:authorize(
+                   username => <<"username">>,
-                          ClientInfo2, subscribe, <<"test/test_username">>)),
+                   peerhost => {127,0,0,1},
-    ?assertEqual(allow, emqx_access_control:authorize(
+                   zone => default,
-                          ClientInfo2, publish,   <<"test/test_username">>)),
+                   listener => {tcp, default}
-    ?assertEqual(deny,  emqx_access_control:authorize(
+                  },
-                          ClientInfo3, subscribe, <<"test">>)), % nomatch
+
-    ?assertEqual(deny,  emqx_access_control:authorize(
+    %% by clientid
-                          ClientInfo3, publish,   <<"test">>)), % nomatch
+
-    ok.
+    ok = init_table(),
+    ok = q(<<"INSERT INTO acl(clientid, topic, permission, action)"
+             "VALUES(?, ?, ?, ?)">>,
+           [<<"clientid">>, <<"a">>, <<"allow">>, <<"subscribe">>]),
+
+    ok = setup_config(
+      #{<<"query">> => <<"SELECT permission, action, topic "
+                         "FROM acl WHERE clientid = ${clientid}">>}),
+
+    ok = emqx_authz_test_lib:test_samples(
+           ClientInfo,
+           [{allow, subscribe, <<"a">>},
+            {deny, subscribe, <<"b">>}]),
+
+    %% by peerhost
+
+    ok = init_table(),
+    ok = q(<<"INSERT INTO acl(peerhost, topic, permission, action)"
+             "VALUES(?, ?, ?, ?)">>,
+           [<<"127.0.0.1">>, <<"a">>, <<"allow">>, <<"subscribe">>]),
+
+    ok = setup_config(
+      #{<<"query">> => <<"SELECT permission, action, topic "
+                         "FROM acl WHERE peerhost = ${peerhost}">>}),
+
+    ok = emqx_authz_test_lib:test_samples(
+           ClientInfo,
+           [{allow, subscribe, <<"a">>},
+            {deny, subscribe, <<"b">>}]),
+
+    %% by cn
+
+    ok = init_table(),
+    ok = q(<<"INSERT INTO acl(cn, topic, permission, action)"
+             "VALUES(?, ?, ?, ?)">>,
+           [<<"cn">>, <<"a">>, <<"allow">>, <<"subscribe">>]),
+
+    ok = setup_config(
+      #{<<"query">> => <<"SELECT permission, action, topic "
+                         "FROM acl WHERE cn = ${cert_common_name}">>}),
+
+    ok = emqx_authz_test_lib:test_samples(
+           ClientInfo,
+           [{allow, subscribe, <<"a">>},
+            {deny, subscribe, <<"b">>}]),
+
+    %% by dn
+
+    ok = init_table(),
+    ok = q(<<"INSERT INTO acl(dn, topic, permission, action)"
+             "VALUES(?, ?, ?, ?)">>,
+           [<<"dn">>, <<"a">>, <<"allow">>, <<"subscribe">>]),
+
+    ok = setup_config(
+      #{<<"query">> => <<"SELECT permission, action, topic "
+                         "FROM acl WHERE dn = ${cert_subject}">>}),
+
+    ok = emqx_authz_test_lib:test_samples(
+           ClientInfo,
+           [{allow, subscribe, <<"a">>},
+            {deny, subscribe, <<"b">>}]).
+
+t_mysql_error(_Config) ->
+    ClientInfo = #{clientid => <<"clientid">>,
+                   username => <<"username">>,
+                   peerhost => {127,0,0,1},
+                   zone => default,
+                   listener => {tcp, default}
+                  },
+
+    ok = setup_config(
+      #{<<"query">> => <<"SOME INVALID STATEMENT">>}),
+
+    ok = emqx_authz_test_lib:test_samples(
+           ClientInfo,
+           [{deny, subscribe, <<"a">>}]).
+
+
+t_create_invalid(_Config) ->
+    BadConfig = maps:merge(
+                  raw_mysql_authz_config(),
+                  #{<<"server">> => <<"255.255.255.255:33333">>}),
+    {error, _} = emqx_authz:update(?CMD_REPLACE, [BadConfig]),
+
+    [] = emqx_authz:lookup().
+
+%%------------------------------------------------------------------------------
+%% Helpers
+%%------------------------------------------------------------------------------
+
+raw_mysql_authz_config() ->
+    #{
+        <<"enable">> => <<"true">>,
+
+        <<"type">> => <<"mysql">>,
+        <<"database">> => <<"mqtt">>,
+        <<"username">> => <<"root">>,
+        <<"password">> => <<"public">>,
+
+        <<"query">> => <<"SELECT permission, action, topic "
+                         "FROM acl WHERE username = ${username}">>,
+
+        <<"server">> => mysql_server()
+    }.
+
+q(Sql) ->
+    emqx_resource:query(
+      ?MYSQL_RESOURCE,
+      {sql, Sql}).
+
+q(Sql, Params) ->
+    emqx_resource:query(
+      ?MYSQL_RESOURCE,
+      {sql, Sql, Params}).
+
+init_table() ->
+    ok = drop_table(),
+    ok = q("CREATE TABLE acl(
+                       username VARCHAR(255),
+                       clientid VARCHAR(255),
+                       peerhost VARCHAR(255),
+                       cn VARCHAR(255),
+                       dn VARCHAR(255),
+                       topic VARCHAR(255),
+                       permission VARCHAR(255),
+                       action VARCHAR(255))").
+
+drop_table() ->
+    ok = q("DROP TABLE IF EXISTS acl").
+
+setup_client_samples(ClientInfo, Samples) ->
+    #{username := Username} = ClientInfo,
+    ok = init_table(),
+    ok = lists:foreach(
+           fun(#{topics := Topics, permission := Permission, action := Action}) ->
+                   lists:foreach(
+                     fun(Topic) ->
+                             q(<<"INSERT INTO acl(username, topic, permission, action)"
+                                 "VALUES(?, ?, ?, ?)">>,
+                               [Username, Topic, Permission, Action])
+                     end,
+                     Topics)
+           end,
+           Samples),
+    setup_config(
+      #{<<"query">> => <<"SELECT permission, action, topic "
+                         "FROM acl WHERE username = ${username}">>}).
+
+setup_config(SpecialParams) ->
+    emqx_authz_test_lib:setup_config(
+      raw_mysql_authz_config(),
+      SpecialParams).
+
+mysql_server() ->
+    iolist_to_binary(
+      io_lib:format(
+        "~s:~b",
+        [?MYSQL_HOST, ?MYSQL_PORT])).
+
+mysql_config() ->
+    #{auto_reconnect => true,
+      database => <<"mqtt">>,
+      username => <<"root">>,
+      password => <<"public">>,
+      pool_size => 8,
+      server => {?MYSQL_HOST, ?MYSQL_PORT},
+      ssl => #{enable => false}
+     }.
+
+start_apps(Apps) ->
+    lists:foreach(fun application:ensure_all_started/1, Apps).
+
+stop_apps(Apps) ->
+    lists:foreach(fun application:stop/1, Apps).

apps/emqx_authz/test/emqx_authz_redis_SUITE.erl

@@ -52,11 +52,15 @@ init_per_suite(Config) ->
     end.
 
 end_per_suite(_Config) ->
-    ok = emqx_authz_test_lib:reset_authorizers(),
+    ok = emqx_authz_test_lib:restore_authorizers(),
     ok = emqx_resource:remove_local(?REDIS_RESOURCE),
     ok = stop_apps([emqx_resource, emqx_connector]),
     ok = emqx_common_test_helpers:stop_apps([emqx_authz]).
 
+init_per_testcase(Config) ->
+    ok = emqx_authz_test_lib:reset_authorizers(),
+    Config.
+
 set_special_configs(emqx_authz) ->
     ok = emqx_authz_test_lib:reset_authorizers();
 
@@ -199,7 +203,6 @@ setup_client_samples(ClientInfo, Samples) ->
     setup_config(#{}).
 
 setup_config(SpecialParams) ->
-    ok = emqx_authz_test_lib:reset_authorizers(deny, false),
     Config = maps:merge(raw_redis_authz_config(), SpecialParams),
     {ok, _} = emqx_authz:update(?CMD_REPLACE, [Config]),
     ok.

apps/emqx_authz/test/emqx_authz_test_lib.erl

@@ -25,6 +25,9 @@
 -define(DEFAULT_CHECK_AVAIL_TIMEOUT, 1000).
 
 reset_authorizers() ->
+    reset_authorizers(deny, false).
+
+restore_authorizers() ->
     reset_authorizers(allow, true).
 
 reset_authorizers(Nomatch, ChacheEnabled) ->
@@ -36,7 +39,6 @@ reset_authorizers(Nomatch, ChacheEnabled) ->
     ok.
 
 setup_config(BaseConfig, SpecialParams) ->
-    ok = reset_authorizers(deny, false),
     Config = maps:merge(BaseConfig, SpecialParams),
     {ok, _} = emqx_authz:update(?CMD_REPLACE, [Config]),
     ok.
@@ -101,6 +103,7 @@ test_allow_topic_rules(ClientInfo, SetupSamples) ->
                 }
               ],
 
+    ok = reset_authorizers(deny, false),
     ok = SetupSamples(ClientInfo, Samples),
 
     ok = test_samples(
@@ -161,11 +164,6 @@ test_allow_topic_rules(ClientInfo, SetupSamples) ->
 
 test_deny_topic_rules(ClientInfo, SetupSamples) ->
     Samples = [
-               #{
-                 topics => [<<"#">>],
-                 permission => <<"allow">>,
-                 action => <<"all">>
-                },
                #{
                  topics => [<<"eq testpub1/${username}">>,
                             <<"testpub2/${clientid}">>,
@@ -190,6 +188,7 @@ test_deny_topic_rules(ClientInfo, SetupSamples) ->
                 }
               ],
 
+    ok = reset_authorizers(allow, false),
     ok = SetupSamples(ClientInfo, Samples),
 
     ok = test_samples(