chore(authn): add MySQL & PostgreSQL backend tests

.ci/docker-compose-file/docker-compose-mysql-tcp.yaml

@@ -5,6 +5,8 @@ services:
     container_name: mysql
     image: mysql:${MYSQL_TAG}
     restart: always
+    ports:
+      - "3306:3306"
     environment:
       MYSQL_ROOT_PASSWORD: public
       MYSQL_DATABASE: mqtt

.github/workflows/run_test_cases.yaml

@@ -67,10 +67,14 @@ jobs:
         - uses: actions/checkout@v2
         - name: docker compose up
           env:
+            MYSQL_TAG: 8
+            PGSQL_TAG: 13
             REDIS_TAG: 6
             GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           run: |
             docker-compose \
+                -f .ci/docker-compose-file/docker-compose-mysql-tcp.yaml \
+                -f .ci/docker-compose-file/docker-compose-pgsql-tcp.yaml \
                 -f .ci/docker-compose-file/docker-compose-redis-single-tcp.yaml \
                 -f .ci/docker-compose-file/docker-compose.yaml \
                 up -d --build

apps/emqx_authn/src/emqx_authn_utils.erl

@@ -89,6 +89,12 @@ is_superuser(#{<<"is_superuser">> := <<"">>}) ->
     #{is_superuser => false};
 is_superuser(#{<<"is_superuser">> := <<"0">>}) ->
     #{is_superuser => false};
+is_superuser(#{<<"is_superuser">> := 0}) ->
+    #{is_superuser => false};
+is_superuser(#{<<"is_superuser">> := null}) ->
+    #{is_superuser => false};
+is_superuser(#{<<"is_superuser">> := false}) ->
+    #{is_superuser => false};
 is_superuser(#{<<"is_superuser">> := _}) ->
     #{is_superuser => true};
 is_superuser(#{}) ->

apps/emqx_authn/test/emqx_authn_mysql_SUITE.erl

@@ -4,7 +4,8 @@
 %% 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
+%%
+%%     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,
@@ -23,70 +24,396 @@
 -include_lib("common_test/include/ct.hrl").
 -include_lib("emqx/include/emqx_placeholder.hrl").
 
+
+-define(MYSQL_HOST, "mysql").
+-define(MYSQL_PORT, 3306).
+-define(MYSQL_RESOURCE, <<"emqx_authn_mysql_SUITE">>).
+
+-define(PATH, [authentication]).
+
 all() ->
-    emqx_common_test_helpers:all(?MODULE).
+    [{group, require_seeds}, t_create, t_create_invalid].
 
 groups() ->
-    [].
+    [{require_seeds, [], [t_authenticate, t_update, t_destroy]}].
 
-init_per_suite(Config) ->
-    ok = emqx_common_test_helpers:start_apps([emqx_authn]),
+init_per_testcase(_, Config) ->
+    emqx_authentication:initialize_authentication(?GLOBAL, []),
+    emqx_authn_test_lib:delete_authenticators(
+      [authentication],
+      ?GLOBAL),
     Config.
 
-end_per_suite(_Config) ->
-    emqx_common_test_helpers:stop_apps([emqx_authn]),
-    ok.
+init_per_group(require_seeds, Config) ->
+    ok = init_seeds(),
+    Config.
 
-init_per_testcase(t_authn, Config) ->
-    meck:new(emqx_resource, [non_strict, passthrough, no_history, no_link]),
-    meck:expect(emqx_resource, create_local, fun(_, _, _) -> {ok, undefined} end),
-    Config;
-init_per_testcase(_, Config) ->
+end_per_group(require_seeds, Config) ->
+    ok = drop_seeds(),
     Config.
 
-end_per_testcase(t_authn, _Config) ->
-    meck:unload(emqx_resource),
-    ok;
-end_per_testcase(_, _Config) ->
-    ok.
+init_per_suite(Config) ->
+    case emqx_authn_test_lib:is_tcp_server_available(?MYSQL_HOST, ?MYSQL_PORT) of
+        true ->
+            ok = emqx_common_test_helpers:start_apps([emqx_authn]),
+            ok = start_apps([emqx_resource, emqx_connector]),
+            {ok, _} = emqx_resource:create_local(
+              ?MYSQL_RESOURCE,
+              emqx_connector_mysql,
+              mysql_config()),
+            Config;
+        false ->
+            {skip, no_mysql}
+    end.
+
+end_per_suite(_Config) ->
+    emqx_authn_test_lib:delete_authenticators(
+      [authentication],
+      ?GLOBAL),
+    ok = emqx_resource:remove_local(?MYSQL_RESOURCE),
+    ok = stop_apps([emqx_resource, emqx_connector]),
+    ok = emqx_common_test_helpers:stop_apps([emqx_authn]).
 
 %%------------------------------------------------------------------------------
-%% Testcases
+%% Tests
 %%------------------------------------------------------------------------------
 
-t_authn(_) ->
-    Password = <<"test">>,
-    Salt = <<"salt">>,
-    PasswordHash = emqx_authn_utils:hash(sha256, Password, Salt, prefix),
-
-    Config = #{<<"mechanism">> => <<"password-based">>,
-               <<"backend">> => <<"mysql">>,
-               <<"server">> => <<"127.0.0.1:3306">>,
-               <<"database">> => <<"mqtt">>,
-               <<"query">> =>
-                   <<"SELECT password_hash, salt FROM users where username = ",
-                        ?PH_USERNAME/binary, " LIMIT 1">>
-               },
-    {ok, _} = update_config([authentication], {create_authenticator, ?GLOBAL, Config}),
-
-    meck:expect(emqx_resource, query,
-        fun(_, {sql, _, [<<"good">>], _}) ->
-            {ok, [<<"password_hash">>, <<"salt">>], [[PasswordHash, Salt]]};
-           (_, {sql, _, _, _}) ->
-            {error, this_is_a_fictitious_reason}
-        end),
-
-    ClientInfo = #{zone => default,
-                   listener => 'tcp:default',
-                   protocol => mqtt,
-                   username => <<"good">>,
-                   password => Password},
-    ?assertEqual({ok, #{is_superuser => false}}, emqx_access_control:authenticate(ClientInfo)),
-
-    ClientInfo2 = ClientInfo#{username => <<"bad">>},
-    ?assertEqual({error, not_authorized}, emqx_access_control:authenticate(ClientInfo2)),
-    emqx_authn_test_lib:delete_config(<<"password-based:mysql">>),
-    ?AUTHN:delete_chain(?GLOBAL).
-
-update_config(Path, ConfigRequest) ->
-    emqx:update_config(Path, ConfigRequest, #{rawconf_with_defaults => true}).
+t_create(_Config) ->
+    AuthConfig = raw_mysql_auth_config(),
+
+    {ok, _} = emqx:update_config(
+                ?PATH,
+                {create_authenticator, ?GLOBAL, AuthConfig}),
+
+    {ok, [#{provider := emqx_authn_mysql}]} = emqx_authentication:list_authenticators(?GLOBAL).
+
+t_create_invalid(_Config) ->
+    AuthConfig = raw_mysql_auth_config(),
+
+    InvalidConfigs =
+        [
+         maps:without([server], AuthConfig),
+         AuthConfig#{server => <<"unknownhost:3333">>},
+         AuthConfig#{password => <<"wrongpass">>},
+         AuthConfig#{database => <<"wrongdatabase">>}
+        ],
+
+    lists:foreach(
+      fun(Config) ->
+              {error, _} = emqx:update_config(
+                             ?PATH,
+                             {create_authenticator, ?GLOBAL, Config}),
+
+              {ok, []} = emqx_authentication:list_authenticators(?GLOBAL)
+      end,
+      InvalidConfigs).
+
+t_authenticate(_Config) ->
+    ok = lists:foreach(
+           fun(Sample) ->
+                   ct:pal("test_user_auth sample: ~p", [Sample]),
+                   test_user_auth(Sample)
+           end,
+           user_seeds()).
+
+test_user_auth(#{credentials := Credentials0,
+                 config_params := SpecificConfgParams,
+                 result := Result}) ->
+    AuthConfig = maps:merge(raw_mysql_auth_config(), SpecificConfgParams),
+
+    {ok, _} = emqx:update_config(
+                ?PATH,
+                {create_authenticator, ?GLOBAL, AuthConfig}),
+
+    Credentials = Credentials0#{
+                    listener => 'tcp:default',
+                    protocol => mqtt
+                   },
+
+    ?assertEqual(Result, emqx_access_control:authenticate(Credentials)),
+
+    emqx_authn_test_lib:delete_authenticators(
+      [authentication],
+      ?GLOBAL).
+
+t_destroy(_Config) ->
+    AuthConfig = raw_mysql_auth_config(),
+
+    {ok, _} = emqx:update_config(
+                ?PATH,
+                {create_authenticator, ?GLOBAL, AuthConfig}),
+
+    {ok, [#{provider := emqx_authn_mysql, state := State}]}
+        = emqx_authentication:list_authenticators(?GLOBAL),
+
+    {ok, _} = emqx_authn_mysql:authenticate(
+                #{username => <<"plain">>,
+                  password => <<"plain">>
+                 },
+                State),
+
+    emqx_authn_test_lib:delete_authenticators(
+      [authentication],
+      ?GLOBAL),
+
+    % Authenticator should not be usable anymore
+    ?assertException(
+       error,
+       _,
+       emqx_authn_mysql:authenticate(
+         #{username => <<"plain">>,
+           password => <<"plain">>
+          },
+         State)).
+
+t_update(_Config) ->
+    CorrectConfig = raw_mysql_auth_config(),
+    IncorrectConfig =
+        CorrectConfig#{
+            query => <<"SELECT password_hash, salt, is_superuser_str as is_superuser
+                          FROM wrong_table where username = ${username} LIMIT 1">>},
+
+    {ok, _} = emqx:update_config(
+                ?PATH,
+                {create_authenticator, ?GLOBAL, IncorrectConfig}),
+
+    {error, not_authorized} = emqx_access_control:authenticate(
+                                #{username => <<"plain">>,
+                                  password => <<"plain">>,
+                                  listener => 'tcp:default',
+                                  protocol => mqtt
+                                 }),
+
+    % We update with config with correct query, provider should update and work properly
+    {ok, _} = emqx:update_config(
+                ?PATH,
+                {update_authenticator, ?GLOBAL, <<"password-based:mysql">>, CorrectConfig}),
+
+    {ok,_} = emqx_access_control:authenticate(
+               #{username => <<"plain">>,
+                 password => <<"plain">>,
+                 listener => 'tcp:default',
+                 protocol => mqtt
+                }).
+
+%%------------------------------------------------------------------------------
+%% Helpers
+%%------------------------------------------------------------------------------
+
+raw_mysql_auth_config() ->
+    #{
+        mechanism => <<"password-based">>,
+        password_hash_algorithm => <<"plain">>,
+        salt_position => <<"suffix">>,
+        enable => <<"true">>,
+
+        backend => <<"mysql">>,
+        database => <<"mqtt">>,
+        username => <<"root">>,
+        password => <<"public">>,
+
+        query => <<"SELECT password_hash, salt, is_superuser_str as is_superuser
+                      FROM users where username = ${username} LIMIT 1">>,
+        server => mysql_server()
+    }.
+
+user_seeds() ->
+    [#{data => #{
+                 username => "plain",
+                 password_hash => "plainsalt",
+                 salt => "salt",
+                 is_superuser_str => "1"
+                },
+       credentials => #{
+                        username => <<"plain">>,
+                        password => <<"plain">>},
+       config_params => #{},
+       result => {ok,#{is_superuser => true}}
+      },
+
+     #{data => #{
+                 username => "md5",
+                 password_hash => "9b4d0c43d206d48279e69b9ad7132e22",
+                 salt => "salt",
+                 is_superuser_str => "0"
+                },
+       credentials => #{
+                        username => <<"md5">>,
+                        password => <<"md5">>
+                       },
+       config_params => #{
+                          password_hash_algorithm => <<"md5">>,
+                          salt_position => <<"suffix">>
+                         },
+       result => {ok,#{is_superuser => false}}
+      },
+
+     #{data => #{
+         username => "sha256",
+         password_hash => "ac63a624e7074776d677dd61a003b8c803eb11db004d0ec6ae032a5d7c9c5caf",
+         salt => "salt",
+         is_superuser_int => 1
+        },
+       credentials => #{
+                        clientid => <<"sha256">>,
+                        password => <<"sha256">>
+                       },
+       config_params => #{
+              query => <<"SELECT password_hash, salt, is_superuser_int as is_superuser
+                            FROM users where username = ${clientid} LIMIT 1">>,
+              password_hash_algorithm => <<"sha256">>,
+              salt_position => <<"prefix">>
+             },
+       result => {ok,#{is_superuser => true}}
+      },
+
+     #{data => #{
+                 username => <<"bcrypt">>,
+                 password_hash => "$2b$12$wtY3h20mUjjmeaClpqZVveDWGlHzCGsvuThMlneGHA7wVeFYyns2u",
+                 salt => "$2b$12$wtY3h20mUjjmeaClpqZVve",
+                 is_superuser_int => 0
+                },
+       credentials => #{
+                        username => <<"bcrypt">>,
+                        password => <<"bcrypt">>
+                       },
+       config_params => #{
+              query => <<"SELECT password_hash, salt, is_superuser_int as is_superuser
+                            FROM users where username = ${username} LIMIT 1">>,
+              password_hash_algorithm => <<"bcrypt">>,
+              salt_position => <<"suffix">> % should be ignored
+             },
+       result => {ok,#{is_superuser => false}}
+      },
+
+     #{data => #{
+                 username => <<"bcrypt">>,
+                 password_hash => "$2b$12$wtY3h20mUjjmeaClpqZVveDWGlHzCGsvuThMlneGHA7wVeFYyns2u",
+                 salt => "$2b$12$wtY3h20mUjjmeaClpqZVve"
+                },
+       credentials => #{
+                        username => <<"bcrypt">>,
+                        password => <<"bcrypt">>
+                       },
+       config_params => #{
+              query => <<"SELECT password_hash, salt, is_superuser_int as is_superuser
+                            FROM users where username = ${username} LIMIT 1">>,
+              password_hash_algorithm => <<"bcrypt">>,
+              salt_position => <<"suffix">> % should be ignored
+             },
+       result => {ok,#{is_superuser => false}}
+      },
+
+     #{data => #{
+                 username => <<"bcrypt0">>,
+                 password_hash => "$2b$12$wtY3h20mUjjmeaClpqZVveDWGlHzCGsvuThMlneGHA7wVeFYyns2u",
+                 salt => "$2b$12$wtY3h20mUjjmeaClpqZVve",
+                 is_superuser_str => "0"
+                },
+       credentials => #{
+                        username => <<"bcrypt0">>,
+                        password => <<"bcrypt">>
+                       },
+       config_params => #{
+              % clientid variable & username credentials
+              query => <<"SELECT password_hash, salt, is_superuser_int as is_superuser
+                            FROM users where username = ${clientid} LIMIT 1">>,
+              password_hash_algorithm => <<"bcrypt">>,
+              salt_position => <<"suffix">>
+             },
+       result => {error,not_authorized}
+      },
+
+     #{data => #{
+                 username => <<"bcrypt1">>,
+                 password_hash => "$2b$12$wtY3h20mUjjmeaClpqZVveDWGlHzCGsvuThMlneGHA7wVeFYyns2u",
+                 salt => "$2b$12$wtY3h20mUjjmeaClpqZVve",
+                 is_superuser_str => "0"
+                },
+       credentials => #{
+                        username => <<"bcrypt1">>,
+                        password => <<"bcrypt">>
+                       },
+       config_params => #{
+              % Bad keys in query
+              query => <<"SELECT 1 AS unknown_field
+                            FROM users where username = ${username} LIMIT 1">>,
+              password_hash_algorithm => <<"bcrypt">>,
+              salt_position => <<"suffix">>
+             },
+       result => {error,not_authorized}
+      },
+
+     #{data => #{
+                 username => <<"bcrypt2">>,
+                 password_hash => "$2b$12$wtY3h20mUjjmeaClpqZVveDWGlHzCGsvuThMlneGHA7wVeFYyns2u",
+                 salt => "$2b$12$wtY3h20mUjjmeaClpqZVve",
+                 is_superuser => "0"
+                },
+       credentials => #{
+                        username => <<"bcrypt2">>,
+                        % Wrong password
+                        password => <<"wrongpass">>
+                       },
+       config_params => #{
+              password_hash_algorithm => <<"bcrypt">>,
+              salt_position => <<"suffix">>
+             },
+       result => {error,bad_username_or_password}
+      }
+    ].
+
+init_seeds() ->
+    ok = drop_seeds(),
+    ok = q("CREATE TABLE users(
+                       username VARCHAR(255),
+                       password_hash VARCHAR(255),
+                       salt VARCHAR(255),
+                       is_superuser_str VARCHAR(255),
+                       is_superuser_int TINYINT)"),
+
+    Fields = [username, password_hash, salt, is_superuser_str, is_superuser_int],
+    InsertQuery = "INSERT INTO users(username, password_hash, salt, "
+    " is_superuser_str, is_superuser_int) VALUES(?, ?, ?, ?, ?)",
+
+    lists:foreach(
+      fun(#{data := Values}) ->
+              Params = [maps:get(F, Values, null) || F <- Fields],
+              ok = q(InsertQuery, Params)
+      end,
+      user_seeds()).
+
+q(Sql) ->
+    emqx_resource:query(
+      ?MYSQL_RESOURCE,
+      {sql, Sql}).
+
+q(Sql, Params) ->
+    emqx_resource:query(
+      ?MYSQL_RESOURCE,
+      {sql, Sql, Params}).
+
+drop_seeds() ->
+    ok = q("DROP TABLE IF EXISTS users").
+
+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_authn/test/emqx_authn_pgsql_SUITE.erl

@@ -4,7 +4,8 @@
 %% 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
+%%
+%%     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,
@@ -24,37 +25,234 @@
 -include_lib("epgsql/include/epgsql.hrl").
 -include_lib("emqx/include/emqx_placeholder.hrl").
 
+-define(PGSQL_HOST, "pgsql").
+-define(PGSQL_PORT, 5432).
+-define(PGSQL_RESOURCE, <<"emqx_authn_pgsql_SUITE">>).
+
+-define(PATH, [authentication]).
+
 all() ->
-    emqx_common_test_helpers:all(?MODULE).
+    [{group, require_seeds}, t_create, t_create_invalid, t_parse_query].
 
 groups() ->
-    [].
+    [{require_seeds, [], [t_authenticate, t_update, t_destroy, t_is_superuser]}].
 
-init_per_suite(Config) ->
-    ok = emqx_common_test_helpers:start_apps([emqx_authn]),
+init_per_testcase(_, Config) ->
+    emqx_authentication:initialize_authentication(?GLOBAL, []),
+    emqx_authn_test_lib:delete_authenticators(
+      [authentication],
+      ?GLOBAL),
     Config.
 
-end_per_suite(_Config) ->
-    emqx_common_test_helpers:stop_apps([emqx_authn]),
-    ok.
+init_per_group(require_seeds, Config) ->
+    ok = init_seeds(),
+    Config.
 
-init_per_testcase(t_authn, Config) ->
-    meck:new(emqx_resource, [non_strict, passthrough, no_history, no_link]),
-    meck:expect(emqx_resource, create_local, fun(_, _, _) -> {ok, undefined} end),
-    Config;
-init_per_testcase(_, Config) ->
+end_per_group(require_seeds, Config) ->
+    ok = drop_seeds(),
     Config.
 
-end_per_testcase(t_authn, _Config) ->
-    meck:unload(emqx_resource),
-    ok;
-end_per_testcase(_, _Config) ->
-    ok.
+init_per_suite(Config) ->
+    case emqx_authn_test_lib:is_tcp_server_available(?PGSQL_HOST, ?PGSQL_PORT) of
+        true ->
+            ok = emqx_common_test_helpers:start_apps([emqx_authn]),
+            ok = start_apps([emqx_resource, emqx_connector]),
+            {ok, _} = emqx_resource:create_local(
+              ?PGSQL_RESOURCE,
+              emqx_connector_pgsql,
+              pgsql_config()),
+            Config;
+        false ->
+            {skip, no_pgsql}
+    end.
+
+end_per_suite(_Config) ->
+    emqx_authn_test_lib:delete_authenticators(
+      [authentication],
+      ?GLOBAL),
+    ok = emqx_resource:remove_local(?PGSQL_RESOURCE),
+    ok = stop_apps([emqx_resource, emqx_connector]),
+    ok = emqx_common_test_helpers:stop_apps([emqx_authn]).
 
 %%------------------------------------------------------------------------------
-%% Testcases
+%% Tests
 %%------------------------------------------------------------------------------
 
+t_create(_Config) ->
+    AuthConfig = raw_pgsql_auth_config(),
+
+    {ok, _} = emqx:update_config(
+                ?PATH,
+                {create_authenticator, ?GLOBAL, AuthConfig}),
+
+    {ok, [#{provider := emqx_authn_pgsql}]} = emqx_authentication:list_authenticators(?GLOBAL).
+
+t_create_invalid(_Config) ->
+    AuthConfig = raw_pgsql_auth_config(),
+
+    InvalidConfigs =
+        [
+         maps:without([server], AuthConfig),
+         AuthConfig#{server => <<"unknownhost:3333">>},
+         AuthConfig#{password => <<"wrongpass">>},
+         AuthConfig#{database => <<"wrongdatabase">>}
+        ],
+
+    lists:foreach(
+      fun(Config) ->
+              {error, _} = emqx:update_config(
+                             ?PATH,
+                             {create_authenticator, ?GLOBAL, Config}),
+
+              {ok, []} = emqx_authentication:list_authenticators(?GLOBAL)
+      end,
+      InvalidConfigs).
+
+t_authenticate(_Config) ->
+    ok = lists:foreach(
+           fun(Sample) ->
+                   ct:pal("test_user_auth sample: ~p", [Sample]),
+                   test_user_auth(Sample)
+           end,
+           user_seeds()).
+
+test_user_auth(#{credentials := Credentials0,
+                 config_params := SpecificConfgParams,
+                 result := Result}) ->
+    AuthConfig = maps:merge(raw_pgsql_auth_config(), SpecificConfgParams),
+
+    {ok, _} = emqx:update_config(
+                ?PATH,
+                {create_authenticator, ?GLOBAL, AuthConfig}),
+
+    Credentials = Credentials0#{
+                    listener => 'tcp:default',
+                    protocol => mqtt
+                   },
+
+    ?assertEqual(Result, emqx_access_control:authenticate(Credentials)),
+
+    emqx_authn_test_lib:delete_authenticators(
+      [authentication],
+      ?GLOBAL).
+
+t_destroy(_Config) ->
+    AuthConfig = raw_pgsql_auth_config(),
+
+    {ok, _} = emqx:update_config(
+                ?PATH,
+                {create_authenticator, ?GLOBAL, AuthConfig}),
+
+    {ok, [#{provider := emqx_authn_pgsql, state := State}]}
+        = emqx_authentication:list_authenticators(?GLOBAL),
+
+    {ok, _} = emqx_authn_pgsql:authenticate(
+                #{username => <<"plain">>,
+                  password => <<"plain">>
+                 },
+                State),
+
+    emqx_authn_test_lib:delete_authenticators(
+      [authentication],
+      ?GLOBAL),
+
+    % Authenticator should not be usable anymore
+    ?assertException(
+       error,
+       _,
+       emqx_authn_pgsql:authenticate(
+         #{username => <<"plain">>,
+           password => <<"plain">>
+          },
+         State)).
+
+t_update(_Config) ->
+    CorrectConfig = raw_pgsql_auth_config(),
+    IncorrectConfig =
+        CorrectConfig#{
+            query => <<"SELECT password_hash, salt, is_superuser_str as is_superuser
+                          FROM wrong_table where username = ${username} LIMIT 1">>},
+
+    {ok, _} = emqx:update_config(
+                ?PATH,
+                {create_authenticator, ?GLOBAL, IncorrectConfig}),
+
+    {error, not_authorized} = emqx_access_control:authenticate(
+                                #{username => <<"plain">>,
+                                  password => <<"plain">>,
+                                  listener => 'tcp:default',
+                                  protocol => mqtt
+                                 }),
+
+    % We update with config with correct query, provider should update and work properly
+    {ok, _} = emqx:update_config(
+                ?PATH,
+                {update_authenticator, ?GLOBAL, <<"password-based:postgresql">>, CorrectConfig}),
+
+    {ok,_} = emqx_access_control:authenticate(
+               #{username => <<"plain">>,
+                 password => <<"plain">>,
+                 listener => 'tcp:default',
+                 protocol => mqtt
+                }).
+
+t_is_superuser(_Config) ->
+    Config = raw_pgsql_auth_config(),
+    {ok, _} = emqx:update_config(
+                ?PATH,
+                {create_authenticator, ?GLOBAL, Config}),
+
+    Checks = [
+              {is_superuser_str, "0", false},
+              {is_superuser_str, "",  false},
+              {is_superuser_str, null, false},
+              {is_superuser_str, "1", true},
+              {is_superuser_str, "val", true},
+
+              {is_superuser_int, 0, false},
+              {is_superuser_int, null, false},
+              {is_superuser_int, 1, true},
+              {is_superuser_int, 123, true},
+
+              {is_superuser_bool, false, false},
+              {is_superuser_bool, null, false},
+              {is_superuser_bool, true, true}
+             ],
+
+    lists:foreach(fun test_is_superuser/1, Checks).
+
+test_is_superuser({Field, Value, ExpectedValue}) ->
+    {ok, _} = q("DELETE FROM users"),
+
+    UserData = #{
+                 username => "user",
+                 password_hash => "plainsalt",
+                 salt => "salt",
+                 Field => Value
+                },
+
+    ok = create_user(UserData),
+
+    Query = "SELECT password_hash, salt, " ++ atom_to_list(Field) ++ " as is_superuser "
+    "FROM users where username = ${username} LIMIT 1",
+
+    Config = maps:put(query, Query, raw_pgsql_auth_config()),
+    {ok, _} = emqx:update_config(
+                ?PATH,
+                {update_authenticator, ?GLOBAL, <<"password-based:postgresql">>, Config}),
+
+    Credentials = #{
+                    listener => 'tcp:default',
+                    protocol => mqtt,
+                    username => <<"user">>,
+                    password => <<"plain">>
+                   },
+
+    ?assertEqual(
+       {ok, #{is_superuser => ExpectedValue}},
+       emqx_access_control:authenticate(Credentials)).
+
+
 t_parse_query(_) ->
     Query1 = ?PH_USERNAME,
     ?assertEqual({<<"$1">>, [?PH_USERNAME]}, emqx_authn_pgsql:parse_query(Query1)),
@@ -66,42 +264,214 @@ t_parse_query(_) ->
     Query3 = <<"nomatch">>,
     ?assertEqual({<<"nomatch">>, []}, emqx_authn_pgsql:parse_query(Query3)).
 
-t_authn(_) ->
-    Password = <<"test">>,
-    Salt = <<"salt">>,
-    PasswordHash = emqx_authn_utils:hash(sha256, Password, Salt, prefix),
-
-    Config = #{<<"mechanism">> => <<"password-based">>,
-               <<"backend">> => <<"postgresql">>,
-               <<"server">> => <<"127.0.0.1:5432">>,
-               <<"database">> => <<"mqtt">>,
-               <<"query">> =>
-                   <<"SELECT password_hash, salt FROM users where username = ",
-                    ?PH_USERNAME/binary, " LIMIT 1">>
-               },
-    {ok, _} = update_config([authentication], {create_authenticator, ?GLOBAL, Config}),
-
-    meck:expect(emqx_resource, query,
-        fun(_, {sql, _, [<<"good">>]}) ->
-            {ok, [#column{name = <<"password_hash">>}, #column{name = <<"salt">>}],
-                    [{PasswordHash, Salt}]};
-           (_, {sql, _, _}) ->
-            {error, this_is_a_fictitious_reason}
-        end),
-
-    ClientInfo = #{zone => default,
-                   listener => 'tcp:default',
-                   protocol => mqtt,
-                   username => <<"good">>,
-			       password => Password},
-    ?assertEqual({ok, #{is_superuser => false}}, emqx_access_control:authenticate(ClientInfo)),
-
-    ClientInfo2 = ClientInfo#{username => <<"bad">>},
-    ?assertEqual({error, not_authorized}, emqx_access_control:authenticate(ClientInfo2)),
-
-    emqx_authn_test_lib:delete_config(<<"password-based:postgresql">>),
-    ?AUTHN:delete_chain(?GLOBAL).
-
-update_config(Path, ConfigRequest) ->
-    emqx:update_config(Path, ConfigRequest, #{rawconf_with_defaults => true}).
+%%------------------------------------------------------------------------------
+%% Helpers
+%%------------------------------------------------------------------------------
+
+raw_pgsql_auth_config() ->
+    #{
+        mechanism => <<"password-based">>,
+        password_hash_algorithm => <<"plain">>,
+        salt_position => <<"suffix">>,
+        enable => <<"true">>,
+
+        backend => <<"postgresql">>,
+        database => <<"mqtt">>,
+        username => <<"root">>,
+        password => <<"public">>,
+
+        query => <<"SELECT password_hash, salt, is_superuser_str as is_superuser
+                      FROM users where username = ${username} LIMIT 1">>,
+        server => pgsql_server()
+    }.
+
+user_seeds() ->
+    [#{data => #{
+                 username => "plain",
+                 password_hash => "plainsalt",
+                 salt => "salt",
+                 is_superuser_str => "1"
+                },
+       credentials => #{
+                        username => <<"plain">>,
+                        password => <<"plain">>},
+       config_params => #{},
+       result => {ok,#{is_superuser => true}}
+      },
+
+     #{data => #{
+                 username => "md5",
+                 password_hash => "9b4d0c43d206d48279e69b9ad7132e22",
+                 salt => "salt",
+                 is_superuser_str => "0"
+                },
+       credentials => #{
+                        username => <<"md5">>,
+                        password => <<"md5">>
+                       },
+       config_params => #{
+                          password_hash_algorithm => <<"md5">>,
+                          salt_position => <<"suffix">>
+                         },
+       result => {ok,#{is_superuser => false}}
+      },
+
+     #{data => #{
+         username => "sha256",
+         password_hash => "ac63a624e7074776d677dd61a003b8c803eb11db004d0ec6ae032a5d7c9c5caf",
+         salt => "salt",
+         is_superuser_int => 1
+        },
+       credentials => #{
+                        clientid => <<"sha256">>,
+                        password => <<"sha256">>
+                       },
+       config_params => #{
+              query => <<"SELECT password_hash, salt, is_superuser_int as is_superuser
+                            FROM users where username = ${clientid} LIMIT 1">>,
+              password_hash_algorithm => <<"sha256">>,
+              salt_position => <<"prefix">>
+             },
+       result => {ok,#{is_superuser => true}}
+      },
+
+     #{data => #{
+                 username => <<"bcrypt">>,
+                 password_hash => "$2b$12$wtY3h20mUjjmeaClpqZVveDWGlHzCGsvuThMlneGHA7wVeFYyns2u",
+                 salt => "$2b$12$wtY3h20mUjjmeaClpqZVve",
+                 is_superuser_int => 0
+                },
+       credentials => #{
+                        username => <<"bcrypt">>,
+                        password => <<"bcrypt">>
+                       },
+       config_params => #{
+              query => <<"SELECT password_hash, salt, is_superuser_int as is_superuser
+                            FROM users where username = ${username} LIMIT 1">>,
+              password_hash_algorithm => <<"bcrypt">>,
+              salt_position => <<"suffix">> % should be ignored
+             },
+       result => {ok,#{is_superuser => false}}
+      },
+
+     #{data => #{
+                 username => <<"bcrypt0">>,
+                 password_hash => "$2b$12$wtY3h20mUjjmeaClpqZVveDWGlHzCGsvuThMlneGHA7wVeFYyns2u",
+                 salt => "$2b$12$wtY3h20mUjjmeaClpqZVve",
+                 is_superuser_str => "0"
+                },
+       credentials => #{
+                        username => <<"bcrypt0">>,
+                        password => <<"bcrypt">>
+                       },
+       config_params => #{
+              % clientid variable & username credentials
+              query => <<"SELECT password_hash, salt, is_superuser_int as is_superuser
+                            FROM users where username = ${clientid} LIMIT 1">>,
+              password_hash_algorithm => <<"bcrypt">>,
+              salt_position => <<"suffix">>
+             },
+       result => {error,not_authorized}
+      },
+
+     #{data => #{
+                 username => <<"bcrypt1">>,
+                 password_hash => "$2b$12$wtY3h20mUjjmeaClpqZVveDWGlHzCGsvuThMlneGHA7wVeFYyns2u",
+                 salt => "$2b$12$wtY3h20mUjjmeaClpqZVve",
+                 is_superuser_str => "0"
+                },
+       credentials => #{
+                        username => <<"bcrypt1">>,
+                        password => <<"bcrypt">>
+                       },
+       config_params => #{
+              % Bad keys in query
+              query => <<"SELECT 1 AS unknown_field
+                            FROM users where username = ${username} LIMIT 1">>,
+              password_hash_algorithm => <<"bcrypt">>,
+              salt_position => <<"suffix">>
+             },
+       result => {error,not_authorized}
+      },
+
+     #{data => #{
+                 username => <<"bcrypt2">>,
+                 password_hash => "$2b$12$wtY3h20mUjjmeaClpqZVveDWGlHzCGsvuThMlneGHA7wVeFYyns2u",
+                 salt => "$2b$12$wtY3h20mUjjmeaClpqZVve",
+                 is_superuser => "0"
+                },
+       credentials => #{
+                        username => <<"bcrypt2">>,
+                        % Wrong password
+                        password => <<"wrongpass">>
+                       },
+       config_params => #{
+              password_hash_algorithm => <<"bcrypt">>,
+              salt_position => <<"suffix">>
+             },
+       result => {error,bad_username_or_password}
+      }
+    ].
+
+init_seeds() ->
+    ok = drop_seeds(),
+    {ok, _, _} = q("CREATE TABLE users(
+                       username varchar(255),
+                       password_hash varchar(255),
+                       salt varchar(255),
+                       is_superuser_str varchar(255),
+                       is_superuser_int smallint,
+                       is_superuser_bool boolean)"),
+
+    lists:foreach(
+      fun(#{data := Values}) ->
+              ok = create_user(Values)
+      end,
+      user_seeds()).
+
+create_user(Values) ->
+    Fields = [username, password_hash, salt, is_superuser_str, is_superuser_int, is_superuser_bool],
+
+    InsertQuery = "INSERT INTO users(username, password_hash, salt,"
+    "is_superuser_str, is_superuser_int, is_superuser_bool) "
+    "VALUES($1, $2, $3, $4, $5, $6)",
+
+    Params = [maps:get(F, Values, null) || F <- Fields],
+    {ok, 1} = q(InsertQuery, Params),
+    ok.
+
+q(Sql) ->
+    emqx_resource:query(
+      ?PGSQL_RESOURCE,
+      {sql, Sql}).
+
+q(Sql, Params) ->
+    emqx_resource:query(
+      ?PGSQL_RESOURCE,
+      {sql, Sql, Params}).
+
+drop_seeds() ->
+    {ok, _, _} = q("DROP TABLE IF EXISTS users"),
+    ok.
+
+pgsql_server() ->
+    iolist_to_binary(
+      io_lib:format(
+        "~s:~b",
+        [?PGSQL_HOST, ?PGSQL_PORT])).
+
+pgsql_config() ->
+    #{auto_reconnect => true,
+      database => <<"mqtt">>,
+      username => <<"root">>,
+      password => <<"public">>,
+      pool_size => 8,
+      server => {?PGSQL_HOST, ?PGSQL_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_authn/test/emqx_authn_redis_SUITE.erl

@@ -26,7 +26,6 @@
 
 -define(REDIS_HOST, "redis").
 -define(REDIS_PORT, 6379).
--define(REDIS_PROBE_TIMEOUT, 1000).
 -define(REDIS_RESOURCE, <<"emqx_authn_redis_SUITE">>).
 
 
@@ -54,7 +53,7 @@ end_per_group(require_seeds, Config) ->
     Config.
 
 init_per_suite(Config) ->
-    case is_redis_available() of
+    case emqx_authn_test_lib:is_tcp_server_available(?REDIS_HOST, ?REDIS_PORT) of
         true ->
             ok = emqx_common_test_helpers:start_apps([emqx_authn]),
             ok = start_apps([emqx_resource, emqx_connector]),
@@ -374,15 +373,6 @@ redis_server() ->
         "~s:~b",
         [?REDIS_HOST, ?REDIS_PORT])).
 
-is_redis_available() ->
-    case gen_tcp:connect(?REDIS_HOST, ?REDIS_PORT, [], ?REDIS_PROBE_TIMEOUT) of
-        {ok, Socket} ->
-            gen_tcp:close(Socket),
-            true;
-        {error, _} ->
-            false
-    end.
-
 redis_config() ->
     #{auto_reconnect => true,
       database => 1,

apps/emqx_authn/test/emqx_authn_test_lib.erl

@@ -21,6 +21,8 @@
 -compile(nowarn_export_all).
 -compile(export_all).
 
+-define(DEFAULT_CHECK_AVAIL_TIMEOUT, 1000).
+
 authenticator_example(Id) ->
     #{Id := #{value := Example}} = emqx_authn_api:authenticator_examples(),
     Example.
@@ -54,3 +56,13 @@ delete_config(ID) ->
             [authentication],
             {delete_authenticator, ?GLOBAL, ID},
             #{rawconf_with_defaults => false}).
+
+is_tcp_server_available(Host, Port) ->
+    case gen_tcp:connect(Host, Port, [], ?DEFAULT_CHECK_AVAIL_TIMEOUT) of
+        {ok, Socket} ->
+            gen_tcp:close(Socket),
+            true;
+        {error, _} ->
+            false
+    end.
+