Merge pull request #12396 from HJianBo/import_user_2

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

apps/emqx_auth/src/emqx_authn/emqx_authn_chains.erl

@@ -310,7 +310,11 @@ reorder_authenticator(_ChainName, []) ->
 reorder_authenticator(ChainName, AuthenticatorIDs) ->
     call({reorder_authenticator, ChainName, AuthenticatorIDs}).
 
--spec import_users(chain_name(), authenticator_id(), {binary(), binary()}) ->
+-spec import_users(
+    chain_name(),
+    authenticator_id(),
+    {plain | hash, prepared_user_list | binary(), binary()}
+) ->
     ok | {error, term()}.
 import_users(ChainName, AuthenticatorID, Filename) ->
     call({import_users, ChainName, AuthenticatorID, Filename}).

apps/emqx_auth/src/emqx_authn/emqx_authn_provider.erl

@@ -53,11 +53,14 @@ when
 when
     State :: state().
 
--callback import_users({Filename, FileData}, State) ->
+-callback import_users({PasswordType, Filename, FileData}, State) ->
     ok
     | {error, term()}
 when
-    Filename :: binary(), FileData :: binary(), State :: state().
+    PasswordType :: plain | hash,
+    Filename :: prepared_user_list | binary(),
+    FileData :: binary(),
+    State :: state().
 
 -callback add_user(UserInfo, State) ->
     {ok, User}

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,116 @@ 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),
+
+    DoImport = fun(FileName, FileData) ->
         emqx_authn_api:with_chain(
             ListenerID,
             fun(ChainName) ->
-            case emqx_authn_chains:import_users(ChainName, AuthenticatorID, {FileName, FileData}) of
+                case
+                    emqx_authn_chains:import_users(
+                        ChainName, AuthenticatorID, {PasswordType, FileName, FileData}
+                    )
+                of
                     ok -> {204};
                     {error, Reason} -> emqx_authn_api:serialize_error(Reason)
                 end
             end
-    );
+        )
-listener_authenticator_import_users(post, #{bindings := #{listener_id := _, id := _}, body := _}) ->
+    end,
-    emqx_authn_api:serialize_error({missing_parameter, filename}).
+    case maps:get(<<"content-type">>, Headers, undefined) of
+        <<"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">> := <<"plain">>}}) ->
+    plain;
+password_type(_Req = #{query_string := #{<<"type">> := <<"hash">>}}) ->
+    hash;
+password_type(_) ->
+    hash.

apps/emqx_auth/test/data/user-credentials-plain.csv

@@ -0,0 +1,3 @@
+user_id,password,is_superuser
+myuser3,password3,true
+myuser4,password4,false

apps/emqx_auth/test/data/user-credentials-plain.json

@@ -0,0 +1,12 @@
+[
+    {
+        "user_id":"myuser1",
+        "password":"password1",
+        "is_superuser": true
+    },
+    {
+        "user_id":"myuser2",
+        "password":"password2",
+        "is_superuser": false
+    }
+]

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,67 @@ 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);
+        case length(Users) > 0 andalso do_import_users(Users) of
-        <<".csv">> ->
+            false ->
-            CSV = csv_data(FileData),
+                error(empty_users);
-            import_users_from_csv(CSV, State);
+            ok ->
-        <<>> ->
+                ok;
-            {error, unknown_file_format};
+            {error, Reason} ->
-        Extension ->
+                _ = do_clean_imported_users(Users),
-            {error, {unsupported_file_format, Extension}}
+                error(Reason)
+        end
+    catch
+        error:Reason1:Stk ->
+            ?SLOG(
+                warning,
+                #{
+                    msg => "import_users_failed",
+                    reason => Reason1,
+                    type => PasswordType,
+                    filename => Filename,
+                    stacktrace => Stk
+                }
+            ),
+            {error, Reason1}
     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 +338,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 +407,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,11 +446,91 @@ group_match_spec(UserGroup, QString) ->
             end)
     end.
 
-csv_data(Data) ->
+%%--------------------------------------------------------------------
-    Lines = binary:split(Data, [<<"\r">>, <<"\n">>], [global, trim_all]),
+%% parse import file/data
-    {csv_data, Lines}.
 
-csv_read_line({csv_data, [Line | Lines]}) ->
+parse_import_users(Filename, FileData, Convertor) ->
-    {ok, Line, {csv_data, Lines}};
+    Eval = fun _Eval(F) ->
-csv_read_line({csv_data, []}) ->
+        case F() of
-    eof.
+            [] -> [];
+            [User | F1] -> [Convertor(User) | _Eval(F1)]
+        end
+    end,
+    ReaderFn = reader_fn(Filename, FileData),
+    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, List) when is_list(List) ->
+    %% Example: [#{<<"user_id">> => <<>>, ...}]
+    emqx_utils_stream:list(List);
+reader_fn(Filename0, Data) ->
+    case filename:extension(to_binary(Filename0)) of
+        <<".json">> ->
+            %% Example: data/user-credentials.json
+            case emqx_utils_json:safe_decode(Data, [return_maps]) of
+                {ok, List} when is_list(List) ->
+                    emqx_utils_stream:list(List);
+                {ok, _} ->
+                    error(unknown_file_format);
+                {error, Reason} ->
+                    error(Reason)
+            end;
+        <<".csv">> ->
+            %% Example: data/user-credentials.csv
+            emqx_utils_stream:csv(Data);
+        <<>> ->
+            error(unknown_file_format);
+        Extension ->
+            error({unsupported_file_format, Extension})
+    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.

apps/emqx_auth_mnesia/test/emqx_authn_api_mnesia_SUITE.erl

@@ -336,7 +336,13 @@ test_authenticator_import_users(PathPrefix) ->
     {ok, CSVData} = file:read_file(CSVFileName),
     {ok, 204, _} = multipart_formdata_request(ImportUri, [], [
         {filename, "user-credentials.csv", CSVData}
-    ]).
+    ]),
+
+    %% test application/json
+    {ok, 204, _} = request(post, ImportUri ++ "?type=hash", emqx_utils_json:decode(JSONData)),
+    {ok, JSONData1} = file:read_file(filename:join([Dir, <<"data/user-credentials-plain.json">>])),
+    {ok, 204, _} = request(post, ImportUri ++ "?type=plain", emqx_utils_json:decode(JSONData1)),
+    ok.
 
 %%------------------------------------------------------------------------------
 %% Helpers

apps/emqx_auth_mnesia/test/emqx_authn_mnesia_SUITE.erl

@@ -216,7 +216,7 @@ t_import_users(_) ->
     ?assertMatch(
         {error, {unsupported_file_format, _}},
         emqx_authn_mnesia:import_users(
-            {<<"/file/with/unknown.extension">>, <<>>},
+            {hash, <<"/file/with/unknown.extension">>, <<>>},
             State
         )
     ),
@@ -224,7 +224,7 @@ t_import_users(_) ->
     ?assertEqual(
         {error, unknown_file_format},
         emqx_authn_mnesia:import_users(
-            {<<"/file/with/no/extension">>, <<>>},
+            {hash, <<"/file/with/no/extension">>, <<>>},
             State
         )
     ),
@@ -251,6 +251,93 @@ t_import_users(_) ->
             sample_filename_and_data(<<"user-credentials-malformed.csv">>),
             State
         )
+    ),
+
+    ?assertEqual(
+        {error, empty_users},
+        emqx_authn_mnesia:import_users(
+            {hash, <<"empty_users.json">>, <<"[]">>},
+            State
+        )
+    ),
+
+    ?assertEqual(
+        {error, empty_users},
+        emqx_authn_mnesia:import_users(
+            {hash, <<"empty_users.csv">>, <<>>},
+            State
+        )
+    ),
+
+    ?assertEqual(
+        {error, empty_users},
+        emqx_authn_mnesia:import_users(
+            {hash, prepared_user_list, []},
+            State
+        )
+    ).
+
+t_import_users_plain(_) ->
+    Config0 = config(),
+    Config = Config0#{password_hash_algorithm => #{name => sha256, salt_position => suffix}},
+    {ok, State} = emqx_authn_mnesia:create(?AUTHN_ID, Config),
+
+    ?assertEqual(
+        ok,
+        emqx_authn_mnesia:import_users(
+            sample_filename_and_data(plain, <<"user-credentials-plain.json">>),
+            State
+        )
+    ),
+
+    ?assertEqual(
+        ok,
+        emqx_authn_mnesia:import_users(
+            sample_filename_and_data(plain, <<"user-credentials-plain.csv">>),
+            State
+        )
+    ).
+
+t_import_users_prepared_list(_) ->
+    Config0 = config(),
+    Config = Config0#{password_hash_algorithm => #{name => sha256, salt_position => suffix}},
+    {ok, State} = emqx_authn_mnesia:create(?AUTHN_ID, Config),
+
+    Users1 = [
+        #{<<"user_id">> => <<"u1">>, <<"password">> => <<"p1">>, <<"is_superuser">> => true},
+        #{<<"user_id">> => <<"u2">>, <<"password">> => <<"p2">>, <<"is_superuser">> => true}
+    ],
+    Users2 = [
+        #{
+            <<"user_id">> => <<"u3">>,
+            <<"password_hash">> =>
+                <<"c5e46903df45e5dc096dc74657610dbee8deaacae656df88a1788f1847390242">>,
+            <<"salt">> => <<"e378187547bf2d6f0545a3f441aa4d8a">>,
+            <<"is_superuser">> => true
+        },
+        #{
+            <<"user_id">> => <<"u4">>,
+            <<"password_hash">> =>
+                <<"f4d17f300b11e522fd33f497c11b126ef1ea5149c74d2220f9a16dc876d4567b">>,
+            <<"salt">> => <<"6d3f9bd5b54d94b98adbcfe10b6d181f">>,
+            <<"is_superuser">> => true
+        }
+    ],
+
+    ?assertEqual(
+        ok,
+        emqx_authn_mnesia:import_users(
+            {plain, prepared_user_list, Users1},
+            State
+        )
+    ),
+
+    ?assertEqual(
+        ok,
+        emqx_authn_mnesia:import_users(
+            {hash, prepared_user_list, Users2},
+            State
+        )
     ).
 
 %%------------------------------------------------------------------------------
@@ -262,9 +349,12 @@ sample_filename(Name) ->
     filename:join([Dir, <<"data">>, Name]).
 
 sample_filename_and_data(Name) ->
+    sample_filename_and_data(hash, Name).
+
+sample_filename_and_data(Type, Name) ->
     Filename = sample_filename(Name),
     {ok, Data} = file:read_file(Filename),
-    {Filename, Data}.
+    {Type, Filename, Data}.
 
 config() ->
     #{

apps/emqx_gateway/src/emqx_gateway_api_authn_user_import.erl

@@ -81,7 +81,7 @@ import_users(post, #{
                 [{FileName, FileData}] = maps:to_list(maps:without([type], File)),
                 case
                     emqx_authn_chains:import_users(
-                        ChainName, AuthId, {FileName, FileData}
+                        ChainName, AuthId, {hash, FileName, FileData}
                     )
                 of
                     ok -> {204};
@@ -105,7 +105,7 @@ import_listener_users(post, #{
                     [{FileName, FileData}] = maps:to_list(maps:without([type], File)),
                     case
                         emqx_authn_chains:import_users(
-                            ChainName, AuthId, {FileName, FileData}
+                            ChainName, AuthId, {hash, FileName, FileData}
                         )
                     of
                         ok -> {204};

apps/emqx_utils/src/emqx_utils_stream.erl

@@ -37,6 +37,11 @@
     ets/1
 ]).
 
+%% Streams from .csv data
+-export([
+    csv/1
+]).
+
 -export_type([stream/1]).
 
 %% @doc A stream is essentially a lazy list.
@@ -45,6 +50,8 @@
 
 -dialyzer(no_improper_lists).
 
+-elvis([{elvis_style, nesting_level, disable}]).
+
 %%
 
 %% @doc Make a stream that produces no values.
@@ -157,3 +164,36 @@ ets(Cont, ContF) ->
                 []
         end
     end.
+
+%% @doc Make a stream out of a .csv binary, where the .csv binary is loaded in all at once.
+%% The .csv binary is assumed to be in UTF-8 encoding and to have a header row.
+-spec csv(binary()) -> stream(map()).
+csv(Bin) when is_binary(Bin) ->
+    Reader = fun _Iter(Headers, Lines) ->
+        case csv_read_line(Lines) of
+            {Fields, Rest} ->
+                case length(Fields) == length(Headers) of
+                    true ->
+                        User = maps:from_list(lists:zip(Headers, Fields)),
+                        [User | fun() -> _Iter(Headers, Rest) end];
+                    false ->
+                        error(bad_format)
+                end;
+            eof ->
+                []
+        end
+    end,
+    HeadersAndLines = binary:split(Bin, [<<"\r">>, <<"\n">>], [global, trim_all]),
+    case csv_read_line(HeadersAndLines) of
+        {CSVHeaders, CSVLines} ->
+            fun() -> Reader(CSVHeaders, CSVLines) end;
+        eof ->
+            empty()
+    end.
+
+csv_read_line([Line | Lines]) ->
+    %% XXX: not support ' ' for the field value
+    Fields = binary:split(Line, [<<",">>, <<" ">>, <<"\n">>], [global, trim_all]),
+    {Fields, Lines};
+csv_read_line([]) ->
+    eof.

apps/emqx_utils/test/emqx_utils_stream_tests.erl

@@ -82,3 +82,34 @@ mqueue_test() ->
         [1, 42, 2],
         emqx_utils_stream:consume(emqx_utils_stream:mqueue(400))
     ).
+
+csv_test() ->
+    Data1 = <<"h1,h2,h3\r\nvv1,vv2,vv3\r\nvv4,vv5,vv6">>,
+    ?assertEqual(
+        [
+            #{<<"h1">> => <<"vv1">>, <<"h2">> => <<"vv2">>, <<"h3">> => <<"vv3">>},
+            #{<<"h1">> => <<"vv4">>, <<"h2">> => <<"vv5">>, <<"h3">> => <<"vv6">>}
+        ],
+        emqx_utils_stream:consume(emqx_utils_stream:csv(Data1))
+    ),
+
+    Data2 = <<"h1, h2, h3\nvv1, vv2, vv3\nvv4,vv5,vv6\n">>,
+    ?assertEqual(
+        [
+            #{<<"h1">> => <<"vv1">>, <<"h2">> => <<"vv2">>, <<"h3">> => <<"vv3">>},
+            #{<<"h1">> => <<"vv4">>, <<"h2">> => <<"vv5">>, <<"h3">> => <<"vv6">>}
+        ],
+        emqx_utils_stream:consume(emqx_utils_stream:csv(Data2))
+    ),
+
+    ?assertEqual(
+        [],
+        emqx_utils_stream:consume(emqx_utils_stream:csv(<<"">>))
+    ),
+
+    BadData = <<"h1,h2,h3\r\nv1,v2,v3\r\nv4,v5">>,
+    ?assertException(
+        error,
+        bad_format,
+        emqx_utils_stream:consume(emqx_utils_stream:csv(BadData))
+    ).

changes/ce/feat-12396.en.md

@@ -0,0 +1,4 @@
+Enhanced the `authentication/:id/import_users` interface for a more user-friendly user import feature:
+
+- Add new parameter `?type=plain` to support importing users with plaintext passwords in addition to the current solution which only supports password hash.
+- Support `content-type: application/json` to accept HTTP Body in JSON format in extension to the current solution which only supports `multipart/form-data` for csv format.