feat(import_users): support user's password in plain text

apps/emqx_auth/src/emqx_authn/emqx_authn_user_import_api.erl

@@ -58,8 +58,8 @@ schema("/authentication/:id/import_users") ->
         post => #{
             tags => ?API_TAGS_GLOBAL,
             description => ?DESC(authentication_id_import_users_post),
-            parameters => [emqx_authn_api:param_auth_id()],
+            parameters => [emqx_authn_api:param_auth_id(), param_password_type()],
-            'requestBody' => emqx_dashboard_swagger:file_schema(filename),
+            'requestBody' => request_body_schema(),
             responses => #{
                 204 => <<"Users imported">>,
                 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>),
@@ -74,8 +74,12 @@ schema("/listeners/:listener_id/authentication/:id/import_users") ->
             tags => ?API_TAGS_SINGLE,
             deprecated => true,
             description => ?DESC(listeners_listener_id_authentication_id_import_users_post),
-            parameters => [emqx_authn_api:param_listener_id(), emqx_authn_api:param_auth_id()],
+            parameters => [
-            'requestBody' => emqx_dashboard_swagger:file_schema(filename),
+                emqx_authn_api:param_listener_id(),
+                emqx_authn_api:param_auth_id(),
+                param_password_type()
+            ],
+            'requestBody' => request_body_schema(),
             responses => #{
                 204 => <<"Users imported">>,
                 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>),
@@ -84,37 +88,114 @@ schema("/listeners/:listener_id/authentication/:id/import_users") ->
         }
     }.
 
+request_body_schema() ->
+    #{content := Content} = emqx_dashboard_swagger:file_schema(filename),
+    Content1 =
+        Content#{
+            <<"application/json">> => #{
+                schema => #{
+                    type => object,
+                    example => [
+                        #{<<"user_id">> => <<"user1">>, <<"password">> => <<"password1">>},
+                        #{<<"user_id">> => <<"user2">>, <<"password">> => <<"password2">>}
+                    ]
+                }
+            }
+        },
+    #{
+        content => Content1,
+        description => <<"Import body">>
+    }.
+
 authenticator_import_users(
     post,
-    #{
+    Req = #{
         bindings := #{id := AuthenticatorID},
-        body := #{<<"filename">> := #{type := _} = File}
+        headers := Headers,
+        body := Body
     }
 ) ->
-    [{FileName, FileData}] = maps:to_list(maps:without([type], File)),
+    PasswordType = password_type(Req),
-    case emqx_authn_chains:import_users(?GLOBAL, AuthenticatorID, {FileName, FileData}) of
+    Result =
+        case maps:get(<<"content-type">>, Headers, undefined) of
+            <<"application/json">> ->
+                emqx_authn_chains:import_users(
+                    ?GLOBAL, AuthenticatorID, {PasswordType, prepared_user_list, Body}
+                );
+            _ ->
+                case Body of
+                    #{<<"filename">> := #{type := _} = File} ->
+                        [{Name, Data}] = maps:to_list(maps:without([type], File)),
+                        emqx_authn_chains:import_users(
+                            ?GLOBAL, AuthenticatorID, {PasswordType, Name, Data}
+                        );
+                    _ ->
+                        {error, {missing_parameter, filename}}
+                end
+        end,
+    case Result of
         ok -> {204};
         {error, Reason} -> emqx_authn_api:serialize_error(Reason)
-    end;
+    end.
-authenticator_import_users(post, #{bindings := #{id := _}, body := _}) ->
-    emqx_authn_api:serialize_error({missing_parameter, filename}).
 
 listener_authenticator_import_users(
     post,
-    #{
+    Req = #{
         bindings := #{listener_id := ListenerID, id := AuthenticatorID},
-        body := #{<<"filename">> := #{type := _} = File}
+        headers := Headers,
+        body := Body
     }
 ) ->
-    [{FileName, FileData}] = maps:to_list(maps:without([type], File)),
+    PasswordType = password_type(Req),
-    emqx_authn_api:with_chain(
+
-        ListenerID,
+    DoImport = fun(FileName, FileData) ->
-        fun(ChainName) ->
+        emqx_authn_api:with_chain(
-            case emqx_authn_chains:import_users(ChainName, AuthenticatorID, {FileName, FileData}) of
+            ListenerID,
-                ok -> {204};
+            fun(ChainName) ->
-                {error, Reason} -> emqx_authn_api:serialize_error(Reason)
+                case
+                    emqx_authn_chains:import_users(
+                        ChainName, AuthenticatorID, {PasswordType, FileName, FileData}
+                    )
+                of
+                    ok -> {204};
+                    {error, Reason} -> emqx_authn_api:serialize_error(Reason)
+                end
             end
-        end
+        )
-    );
+    end,
-listener_authenticator_import_users(post, #{bindings := #{listener_id := _, id := _}, body := _}) ->
+    case maps:get(<<"content-type">>, Headers, undefined) of
-    emqx_authn_api:serialize_error({missing_parameter, filename}).
+        <<"application/json">> ->
+            DoImport(prepared_user_list, Body);
+        _ ->
+            case Body of
+                #{<<"filename">> := #{type := _} = File} ->
+                    [{Name, Data}] = maps:to_list(maps:without([type], File)),
+                    DoImport(Name, Data);
+                _ ->
+                    emqx_authn_api:serialize_error({missing_parameter, filename})
+            end
+    end.
+
+%%--------------------------------------------------------------------
+%% helpers
+
+param_password_type() ->
+    {type,
+        hoconsc:mk(
+            binary(),
+            #{
+                in => query,
+                enum => [<<"plain">>, <<"hash">>],
+                required => true,
+                desc => <<
+                    "The import file template type, enum with `plain`,"
+                    "`hash`"
+                >>,
+                example => <<"hash">>
+            }
+        )}.
+
+password_type(_Req = #{query_string := #{<<"type">> := Type}}) ->
+    binary_to_existing_atom(Type);
+password_type(_) ->
+    hash.

apps/emqx_auth_mnesia/src/emqx_auth_mnesia.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_auth_mnesia, [
     {description, "EMQX Buitl-in Database Authentication and Authorization"},
-    {vsn, "0.1.2"},
+    {vsn, "0.1.3"},
     {registered, []},
     {mod, {emqx_auth_mnesia_app, []}},
     {applications, [

apps/emqx_auth_mnesia/src/emqx_authn_mnesia.erl

@@ -52,9 +52,7 @@
     do_destroy/1,
     do_add_user/1,
     do_delete_user/2,
-    do_update_user/3,
+    do_update_user/3
-    import/2,
-    import_csv/3
 ]).
 
 -export([mnesia/1, init_tables/0]).
@@ -173,20 +171,76 @@ do_destroy(UserGroup) ->
         mnesia:select(?TAB, group_match_spec(UserGroup), write)
     ).
 
-import_users({Filename0, FileData}, State) ->
+import_users({PasswordType, Filename, FileData}, State) ->
-    Filename = to_binary(Filename0),
+    Convertor = convertor(PasswordType, State),
-    case filename:extension(Filename) of
+    try
-        <<".json">> ->
+        {_NewUsersCnt, Users} = parse_import_users(Filename, FileData, Convertor),
-            import_users_from_json(FileData, State);
+        try
-        <<".csv">> ->
+            case do_import_users(Users) of
-            CSV = csv_data(FileData),
+                ok -> ok;
-            import_users_from_csv(CSV, State);
+                {error, Reason} -> error(Reason)
-        <<>> ->
+            end
-            {error, unknown_file_format};
+        catch
-        Extension ->
+            error:Reason1:Stk ->
-            {error, {unsupported_file_format, Extension}}
+                ?SLOG(
+                    warning,
+                    #{
+                        msg => "import_users_failed",
+                        type => PasswordType,
+                        filename => Filename,
+                        stacktrace => Stk
+                    }
+                ),
+                _ = do_clean_imported_users(Users),
+                {error, Reason1}
+        end
+    catch
+        error:Reason2:Stk2 ->
+            ?SLOG(
+                warning,
+                #{
+                    msg => "import_users_failed",
+                    type => PasswordType,
+                    filename => Filename,
+                    stacktrace => Stk2
+                }
+            ),
+            {error, Reason2}
     end.
 
+do_import_users(Users) ->
+    trans(
+        fun() ->
+            lists:foreach(
+                fun(
+                    #{
+                        <<"user_group">> := UserGroup,
+                        <<"user_id">> := UserID,
+                        <<"password_hash">> := PasswordHash,
+                        <<"salt">> := Salt,
+                        <<"is_superuser">> := IsSuperuser
+                    }
+                ) ->
+                    insert_user(UserGroup, UserID, PasswordHash, Salt, IsSuperuser)
+                end,
+                Users
+            )
+        end
+    ).
+
+do_clean_imported_users(Users) ->
+    lists:foreach(
+        fun(
+            #{
+                <<"user_group">> := UserGroup,
+                <<"user_id">> := UserID
+            }
+        ) ->
+            mria:dirty_delete(?TAB, {UserGroup, UserID})
+        end,
+        Users
+    ).
+
 add_user(
     UserInfo,
     State
@@ -293,93 +347,6 @@ run_fuzzy_filter(
 %% Internal functions
 %%------------------------------------------------------------------------------
 
-%% Example: data/user-credentials.json
-import_users_from_json(Bin, #{user_group := UserGroup}) ->
-    case emqx_utils_json:safe_decode(Bin, [return_maps]) of
-        {ok, List} ->
-            trans(fun ?MODULE:import/2, [UserGroup, List]);
-        {error, Reason} ->
-            {error, Reason}
-    end.
-
-%% Example: data/user-credentials.csv
-import_users_from_csv(CSV, #{user_group := UserGroup}) ->
-    case get_csv_header(CSV) of
-        {ok, Seq, NewCSV} ->
-            trans(fun ?MODULE:import_csv/3, [UserGroup, NewCSV, Seq]);
-        {error, Reason} ->
-            {error, Reason}
-    end.
-
-import(_UserGroup, []) ->
-    ok;
-import(UserGroup, [
-    #{
-        <<"user_id">> := UserID,
-        <<"password_hash">> := PasswordHash
-    } = UserInfo
-    | More
-]) when
-    is_binary(UserID) andalso is_binary(PasswordHash)
-->
-    Salt = maps:get(<<"salt">>, UserInfo, <<>>),
-    IsSuperuser = maps:get(<<"is_superuser">>, UserInfo, false),
-    insert_user(UserGroup, UserID, PasswordHash, Salt, IsSuperuser),
-    import(UserGroup, More);
-import(_UserGroup, [_ | _More]) ->
-    {error, bad_format}.
-
-%% Importing 5w users needs 1.7 seconds
-import_csv(UserGroup, CSV, Seq) ->
-    case csv_read_line(CSV) of
-        {ok, Line, NewCSV} ->
-            Fields = binary:split(Line, [<<",">>, <<" ">>, <<"\n">>], [global, trim_all]),
-            case get_user_info_by_seq(Fields, Seq) of
-                {ok,
-                    #{
-                        user_id := UserID,
-                        password_hash := PasswordHash
-                    } = UserInfo} ->
-                    Salt = maps:get(salt, UserInfo, <<>>),
-                    IsSuperuser = maps:get(is_superuser, UserInfo, false),
-                    insert_user(UserGroup, UserID, PasswordHash, Salt, IsSuperuser),
-                    import_csv(UserGroup, NewCSV, Seq);
-                {error, Reason} ->
-                    {error, Reason}
-            end;
-        eof ->
-            ok
-    end.
-
-get_csv_header(CSV) ->
-    case csv_read_line(CSV) of
-        {ok, Line, NewCSV} ->
-            Seq = binary:split(Line, [<<",">>, <<" ">>, <<"\n">>], [global, trim_all]),
-            {ok, Seq, NewCSV};
-        eof ->
-            {error, empty_file}
-    end.
-
-get_user_info_by_seq(Fields, Seq) ->
-    get_user_info_by_seq(Fields, Seq, #{}).
-
-get_user_info_by_seq([], [], #{user_id := _, password_hash := _} = Acc) ->
-    {ok, Acc};
-get_user_info_by_seq(_, [], _) ->
-    {error, bad_format};
-get_user_info_by_seq([UserID | More1], [<<"user_id">> | More2], Acc) ->
-    get_user_info_by_seq(More1, More2, Acc#{user_id => UserID});
-get_user_info_by_seq([PasswordHash | More1], [<<"password_hash">> | More2], Acc) ->
-    get_user_info_by_seq(More1, More2, Acc#{password_hash => PasswordHash});
-get_user_info_by_seq([Salt | More1], [<<"salt">> | More2], Acc) ->
-    get_user_info_by_seq(More1, More2, Acc#{salt => Salt});
-get_user_info_by_seq([<<"true">> | More1], [<<"is_superuser">> | More2], Acc) ->
-    get_user_info_by_seq(More1, More2, Acc#{is_superuser => true});
-get_user_info_by_seq([<<"false">> | More1], [<<"is_superuser">> | More2], Acc) ->
-    get_user_info_by_seq(More1, More2, Acc#{is_superuser => false});
-get_user_info_by_seq(_, _, _) ->
-    {error, bad_format}.
-
 insert_user(UserGroup, UserID, PasswordHash, Salt, IsSuperuser) ->
     UserInfoRecord = user_info_record(UserGroup, UserID, PasswordHash, Salt, IsSuperuser),
     insert_user(UserInfoRecord).
@@ -449,6 +416,12 @@ trans(Fun, Args) ->
         {aborted, Reason} -> {error, Reason}
     end.
 
+trans(Fun) ->
+    case mria:transaction(?AUTHN_SHARD, Fun) of
+        {atomic, Res} -> Res;
+        {aborted, Reason} -> {error, Reason}
+    end.
+
 to_binary(B) when is_binary(B) ->
     B;
 to_binary(L) when is_list(L) ->
@@ -482,10 +455,144 @@ group_match_spec(UserGroup, QString) ->
             end)
     end.
 
+%%--------------------------------------------------------------------
+%% parse import file/data
+
+parse_import_users(Filename, FileData, Convertor) ->
+    Eval = fun _Eval(F) ->
+        case F() of
+            eof -> [];
+            {User, F1} -> [User | _Eval(F1)]
+        end
+    end,
+    ReaderFn = reader_fn(Filename, FileData, Convertor),
+    Users = lists:reverse(Eval(ReaderFn)),
+    NewUsersCount =
+        lists:foldl(
+            fun(
+                #{
+                    <<"user_group">> := UserGroup,
+                    <<"user_id">> := UserID
+                },
+                Acc
+            ) ->
+                case ets:member(?TAB, {UserGroup, UserID}) of
+                    true ->
+                        Acc;
+                    false ->
+                        Acc + 1
+                end
+            end,
+            0,
+            Users
+        ),
+    {NewUsersCount, Users}.
+
+reader_fn(prepared_user_list, Data, Convertor) when is_list(Data) ->
+    reader(prepared_user_list, Data, Convertor);
+reader_fn(Filename0, Data, Convertor) ->
+    Filename = to_binary(Filename0),
+    case filename:extension(Filename) of
+        <<".json">> ->
+            reader(json, Data, Convertor);
+        <<".csv">> ->
+            reader(csv, Data, Convertor);
+        <<>> ->
+            error(unknown_file_format);
+        Extension ->
+            error({unsupported_file_format, Extension})
+    end.
+
+%% Example: data/user-credentials.json
+reader(json, Data, Convertor) when is_binary(Data) ->
+    case emqx_utils_json:safe_decode(Data, [return_maps]) of
+        {ok, List} ->
+            reader(prepared_user_list, List, Convertor);
+        {error, Reason} ->
+            error(Reason)
+    end;
+%% Example: data/user-credentials.csv
+reader(csv, Data, Convertor) when is_binary(Data) ->
+    CSVData = csv_data(Data),
+    case get_csv_header(CSVData) of
+        {ok, Headers, CSVLines} ->
+            Reader =
+                fun _Iter(Lines) ->
+                    case csv_read_line(Lines) of
+                        {ok, Line, Rest} ->
+                            %% XXX: not support ' ' for a field?
+                            Fields = binary:split(Line, [<<",">>, <<" ">>, <<"\n">>], [
+                                global, trim_all
+                            ]),
+                            case length(Fields) == length(Headers) of
+                                true ->
+                                    User = maps:from_list(lists:zip(Headers, Fields)),
+                                    {Convertor(User), fun() -> _Iter(Rest) end};
+                                false ->
+                                    error(bad_format)
+                            end;
+                        eof ->
+                            eof
+                    end
+                end,
+            fun() -> Reader(CSVLines) end;
+        {error, Reason} ->
+            error(Reason)
+    end;
+%% Example: [#{<<"user_id">> => <<>>, ...}]
+reader(prepared_user_list, Data, Convertor) when is_list(Data) ->
+    Reader =
+        fun
+            _Iter([]) -> eof;
+            _Iter([User | Rest]) -> {Convertor(User), fun() -> _Iter(Rest) end}
+        end,
+    fun() -> Reader(Data) end.
+
+convertor(PasswordType, State) ->
+    fun(User) ->
+        convert_user(User, PasswordType, State)
+    end.
+
+convert_user(
+    User = #{<<"user_id">> := UserId},
+    PasswordType,
+    #{user_group := UserGroup, password_hash_algorithm := Algorithm}
+) ->
+    {PasswordHash, Salt} = find_password_hash(PasswordType, User, Algorithm),
+    #{
+        <<"user_id">> => UserId,
+        <<"password_hash">> => PasswordHash,
+        <<"salt">> => Salt,
+        <<"is_superuser">> => is_superuser(User),
+        <<"user_group">> => UserGroup
+    };
+convert_user(_, _, _) ->
+    error(bad_format).
+
+find_password_hash(hash, User = #{<<"password_hash">> := PasswordHash}, _) ->
+    {PasswordHash, maps:get(<<"salt">>, User, <<>>)};
+find_password_hash(plain, #{<<"password">> := Password}, Algorithm) ->
+    emqx_authn_password_hashing:hash(Algorithm, Password);
+find_password_hash(_, _, _) ->
+    error(bad_format).
+
+is_superuser(#{<<"is_superuser">> := <<"true">>}) -> true;
+is_superuser(#{<<"is_superuser">> := true}) -> true;
+is_superuser(_) -> false.
+
 csv_data(Data) ->
     Lines = binary:split(Data, [<<"\r">>, <<"\n">>], [global, trim_all]),
     {csv_data, Lines}.
 
+get_csv_header(CSV) ->
+    case csv_read_line(CSV) of
+        {ok, Line, NewCSV} ->
+            Seq = binary:split(Line, [<<",">>, <<" ">>, <<"\n">>], [global, trim_all]),
+            {ok, Seq, NewCSV};
+        eof ->
+            {error, empty_file}
+    end.
+
 csv_read_line({csv_data, [Line | Lines]}) ->
     {ok, Line, {csv_data, Lines}};
 csv_read_line({csv_data, []}) ->