Merge pull request #9875 from sstrigler/EMQX-7119-fix-upload-broken-plugin

EMQX 7119 fix upload broken plugin

apps/emqx_management/src/emqx_management.app.src

@@ -2,7 +2,7 @@
 {application, emqx_management, [
     {description, "EMQX Management API and CLI"},
     % strict semver, bump manually!
-    {vsn, "5.0.13"},
+    {vsn, "5.0.14"},
     {modules, []},
     {registered, [emqx_management_sup]},
     {applications, [kernel, stdlib, emqx_plugins, minirest, emqx]},

apps/emqx_management/src/emqx_mgmt_api_plugins.erl

@@ -17,7 +17,6 @@
 
 -behaviour(minirest_api).
 
--include_lib("kernel/include/file.hrl").
 -include_lib("typerefl/include/types.hrl").
 -include_lib("emqx/include/logger.hrl").
 %%-include_lib("emqx_plugins/include/emqx_plugins.hrl").
@@ -326,7 +325,8 @@ upload_install(post, #{body := #{<<"plugin">> := Plugin}}) when is_map(Plugin) -
     %% File bin is too large, we use rpc:multicall instead of cluster_rpc:multicall
     %% TODO what happens when a new node join in?
     %% emqx_plugins_monitor should copy plugins from other core node when boot-up.
-    case emqx_plugins:describe(string:trim(FileName, trailing, ".tar.gz")) of
+    NameVsn = string:trim(FileName, trailing, ".tar.gz"),
+    case emqx_plugins:describe(NameVsn) of
         {error, #{error := "bad_info_file", return := {enoent, _}}} ->
             case emqx_plugins:parse_name_vsn(FileName) of
                 {ok, AppName, _Vsn} ->
@@ -346,6 +346,7 @@ upload_install(post, #{body := #{<<"plugin">> := Plugin}}) when is_map(Plugin) -
                             }}
                     end;
                 {error, Reason} ->
+                    emqx_plugins:delete_package(NameVsn),
                     {400, #{
                         code => 'BAD_PLUGIN_INFO',
                         message => iolist_to_binary([Reason, ":", FileName])
@@ -367,9 +368,24 @@ upload_install(post, #{}) ->
 do_install_package(FileName, Bin) ->
     %% TODO: handle bad nodes
     {[_ | _] = Res, []} = emqx_mgmt_api_plugins_proto_v1:install_package(FileName, Bin),
-    %% TODO: handle non-OKs
+    case lists:filter(fun(R) -> R =/= ok end, Res) of
-    [] = lists:filter(fun(R) -> R =/= ok end, Res),
+        [] ->
-    {200}.
+            {200};
+        Filtered ->
+            %% crash if we have unexpected errors or results
+            [] = lists:filter(
+                fun
+                    ({error, {failed, _}}) -> true;
+                    ({error, _}) -> false
+                end,
+                Filtered
+            ),
+            {error, #{error := Reason}} = hd(Filtered),
+            {400, #{
+                code => 'BAD_PLUGIN_INFO',
+                message => iolist_to_binary([Reason, ":", FileName])
+            }}
+    end.
 
 plugin(get, #{bindings := #{name := Name}}) ->
     {Plugins, _} = emqx_mgmt_api_plugins_proto_v1:describe_package(Name),
@@ -408,7 +424,15 @@ install_package(FileName, Bin) ->
     File = filename:join(emqx_plugins:install_dir(), FileName),
     ok = file:write_file(File, Bin),
     PackageName = string:trim(FileName, trailing, ".tar.gz"),
-    emqx_plugins:ensure_installed(PackageName).
+    case emqx_plugins:ensure_installed(PackageName) of
+        {error, #{return := not_found}} = NotFound ->
+            NotFound;
+        {error, _Reason} = Error ->
+            _ = file:delete(File),
+            Error;
+        Result ->
+            Result
+    end.
 
 %% For RPC plugin get
 describe_package(Name) ->

apps/emqx_management/test/emqx_mgmt_api_plugins_SUITE.erl

@@ -20,7 +20,7 @@
 
 -include_lib("eunit/include/eunit.hrl").
 
--define(EMQX_PLUGIN_TEMPLATE_VSN, "5.0-rc.1").
+-define(EMQX_PLUGIN_TEMPLATE_VSN, "5.0.0").
 -define(PACKAGE_SUFFIX, ".tar.gz").
 
 all() ->
@@ -30,12 +30,11 @@ init_per_suite(Config) ->
     WorkDir = proplists:get_value(data_dir, Config),
     ok = filelib:ensure_dir(WorkDir),
     DemoShDir1 = string:replace(WorkDir, "emqx_mgmt_api_plugins", "emqx_plugins"),
-    DemoShDir = string:replace(DemoShDir1, "emqx_management", "emqx_plugins"),
+    DemoShDir = lists:flatten(string:replace(DemoShDir1, "emqx_management", "emqx_plugins")),
     OrigInstallDir = emqx_plugins:get_config(install_dir, undefined),
     ok = filelib:ensure_dir(DemoShDir),
     emqx_mgmt_api_test_util:init_suite([emqx_conf, emqx_plugins]),
     emqx_plugins:put_config(install_dir, DemoShDir),
-
     [{demo_sh_dir, DemoShDir}, {orig_install_dir, OrigInstallDir} | Config].
 
 end_per_suite(Config) ->
@@ -48,18 +47,20 @@ end_per_suite(Config) ->
     emqx_mgmt_api_test_util:end_suite([emqx_plugins, emqx_conf]),
     ok.
 
-todo_t_plugins(Config) ->
+t_plugins(Config) ->
     DemoShDir = proplists:get_value(demo_sh_dir, Config),
     PackagePath = get_demo_plugin_package(DemoShDir),
     ct:pal("package_location:~p install dir:~p", [PackagePath, emqx_plugins:install_dir()]),
     NameVsn = filename:basename(PackagePath, ?PACKAGE_SUFFIX),
+    ok = emqx_plugins:ensure_uninstalled(NameVsn),
     ok = emqx_plugins:delete_package(NameVsn),
     ok = install_plugin(PackagePath),
     {ok, StopRes} = describe_plugins(NameVsn),
+    Node = atom_to_binary(node()),
     ?assertMatch(
         #{
             <<"running_status">> := [
-                #{<<"node">> := <<"test@127.0.0.1">>, <<"status">> := <<"stopped">>}
+                #{<<"node">> := Node, <<"status">> := <<"stopped">>}
             ]
         },
         StopRes
@@ -70,7 +71,7 @@ todo_t_plugins(Config) ->
     ?assertMatch(
         #{
             <<"running_status">> := [
-                #{<<"node">> := <<"test@127.0.0.1">>, <<"status">> := <<"running">>}
+                #{<<"node">> := Node, <<"status">> := <<"running">>}
             ]
         },
         StartRes
@@ -80,7 +81,7 @@ todo_t_plugins(Config) ->
     ?assertMatch(
         #{
             <<"running_status">> := [
-                #{<<"node">> := <<"test@127.0.0.1">>, <<"status">> := <<"stopped">>}
+                #{<<"node">> := Node, <<"status">> := <<"stopped">>}
             ]
         },
         StopRes2
@@ -88,6 +89,28 @@ todo_t_plugins(Config) ->
     {ok, []} = uninstall_plugin(NameVsn),
     ok.
 
+t_bad_plugin(Config) ->
+    DemoShDir = proplists:get_value(demo_sh_dir, Config),
+    PackagePathOrig = get_demo_plugin_package(DemoShDir),
+    PackagePath = filename:join([
+        filename:dirname(PackagePathOrig),
+        "bad_plugin-1.0.0.tar.gz"
+    ]),
+    ct:pal("package_location:~p orig:~p", [PackagePath, PackagePathOrig]),
+    %% rename plugin tarball
+    file:copy(PackagePathOrig, PackagePath),
+    file:delete(PackagePathOrig),
+    {ok, {{"HTTP/1.1", 400, "Bad Request"}, _, _}} = install_plugin(PackagePath),
+    ?assertEqual(
+        {error, enoent},
+        file:delete(
+            filename:join([
+                emqx_plugins:install_dir(),
+                filename:basename(PackagePath)
+            ])
+        )
+    ).
+
 list_plugins() ->
     Path = emqx_mgmt_api_test_util:api_path(["plugins"]),
     case emqx_mgmt_api_test_util:request_api(get, Path) of

apps/emqx_plugins/src/emqx_plugins.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_plugins, [
     {description, "EMQX Plugin Management"},
-    {vsn, "0.1.0"},
+    {vsn, "0.1.1"},
     {modules, []},
     {mod, {emqx_plugins_app, []}},
     {applications, [kernel, stdlib, emqx]},

apps/emqx_plugins/src/emqx_plugins.erl

@@ -16,8 +16,13 @@
 
 -module(emqx_plugins).
 
--include_lib("emqx/include/emqx.hrl").
 -include_lib("emqx/include/logger.hrl").
+-include_lib("emqx/include/logger.hrl").
+-include("emqx_plugins.hrl").
+
+-ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
+-endif.
 
 -export([
     ensure_installed/1,
@@ -56,10 +61,6 @@
 -compile(nowarn_export_all).
 -endif.
 
--include_lib("emqx/include/emqx.hrl").
--include_lib("emqx/include/logger.hrl").
--include("emqx_plugins.hrl").
-
 %% "my_plugin-0.1.0"
 -type name_vsn() :: binary() | string().
 %% the parse result of the JSON info file
@@ -87,14 +88,15 @@ ensure_installed(NameVsn) ->
 
 do_ensure_installed(NameVsn) ->
     TarGz = pkg_file(NameVsn),
-    case erl_tar:extract(TarGz, [{cwd, install_dir()}, compressed]) of
+    case erl_tar:extract(TarGz, [compressed, memory]) of
-        ok ->
+        {ok, TarContent} ->
+            ok = write_tar_file_content(install_dir(), TarContent),
             case read_plugin(NameVsn, #{}) of
                 {ok, _} ->
                     ok;
                 {error, Reason} ->
                     ?SLOG(warning, Reason#{msg => "failed_to_read_after_install"}),
-                    _ = ensure_uninstalled(NameVsn),
+                    ok = delete_tar_file_content(install_dir(), TarContent),
                     {error, Reason}
             end;
         {error, {_, enoent}} ->
@@ -111,8 +113,68 @@ do_ensure_installed(NameVsn) ->
             }}
     end.
 
-%% @doc Ensure files and directories for the given plugin are delete.
+write_tar_file_content(BaseDir, TarContent) ->
-%% If a plugin is running, or enabled, error is returned.
+    lists:foreach(
+        fun({Name, Bin}) ->
+            Filename = filename:join(BaseDir, Name),
+            ok = filelib:ensure_dir(Filename),
+            ok = file:write_file(Filename, Bin)
+        end,
+        TarContent
+    ).
+
+delete_tar_file_content(BaseDir, TarContent) ->
+    lists:foreach(
+        fun({Name, _}) ->
+            Filename = filename:join(BaseDir, Name),
+            case filelib:is_file(Filename) of
+                true ->
+                    TopDirOrFile = top_dir(BaseDir, Filename),
+                    ok = file:del_dir_r(TopDirOrFile);
+                false ->
+                    %% probably already deleted
+                    ok
+            end
+        end,
+        TarContent
+    ).
+
+top_dir(BaseDir0, DirOrFile) ->
+    BaseDir = normalize_dir(BaseDir0),
+    case filename:dirname(DirOrFile) of
+        RockBottom when RockBottom =:= "/" orelse RockBottom =:= "." ->
+            throw({out_of_bounds, DirOrFile});
+        BaseDir ->
+            DirOrFile;
+        Parent ->
+            top_dir(BaseDir, Parent)
+    end.
+
+normalize_dir(Dir) ->
+    %% Get rid of possible trailing slash
+    filename:join([Dir, ""]).
+
+-ifdef(TEST).
+normalize_dir_test_() ->
+    [
+        ?_assertEqual("foo", normalize_dir("foo")),
+        ?_assertEqual("foo", normalize_dir("foo/")),
+        ?_assertEqual("/foo", normalize_dir("/foo")),
+        ?_assertEqual("/foo", normalize_dir("/foo/"))
+    ].
+
+top_dir_test_() ->
+    [
+        ?_assertEqual("base/foo", top_dir("base", filename:join(["base", "foo", "bar"]))),
+        ?_assertEqual("/base/foo", top_dir("/base", filename:join(["/", "base", "foo", "bar"]))),
+        ?_assertEqual("/base/foo", top_dir("/base/", filename:join(["/", "base", "foo", "bar"]))),
+        ?_assertThrow({out_of_bounds, _}, top_dir("/base", filename:join(["/", "base"]))),
+        ?_assertThrow({out_of_bounds, _}, top_dir("/base", filename:join(["/", "foo", "bar"])))
+    ].
+-endif.
+
+%% @doc Ensure files and directories for the given plugin are being deleted.
+%% If a plugin is running, or enabled, an error is returned.
 -spec ensure_uninstalled(name_vsn()) -> ok | {error, any()}.
 ensure_uninstalled(NameVsn) ->
     case read_plugin(NameVsn, #{}) of

apps/emqx_plugins/test/emqx_plugins_SUITE.erl

@@ -20,6 +20,7 @@
 -compile(nowarn_export_all).
 
 -include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
 
 -define(EMQX_PLUGIN_TEMPLATE_RELEASE_NAME, "emqx_plugin_template").
 -define(EMQX_PLUGIN_TEMPLATE_URL,
@@ -325,27 +326,60 @@ t_bad_tar_gz(Config) ->
     %% idempotent
     ok = emqx_plugins:delete_package("fake-vsn").
 
-%% create a corrupted .tar.gz
+%% create with incomplete info file
 %% failed install attempts should not leave behind extracted dir
 t_bad_tar_gz2({init, Config}) ->
-    Config;
-t_bad_tar_gz2({'end', _Config}) ->
-    ok;
-t_bad_tar_gz2(Config) ->
     WorkDir = proplists:get_value(data_dir, Config),
     NameVsn = "foo-0.2",
-    %% this an invalid info file content
+    %% this an invalid info file content (description missing)
     BadInfo = "name=foo, rel_vsn=\"0.2\", rel_apps=[foo]",
     ok = write_info_file(Config, NameVsn, BadInfo),
     TarGz = filename:join([WorkDir, NameVsn ++ ".tar.gz"]),
     ok = make_tar(WorkDir, NameVsn),
+    [{tar_gz, TarGz}, {name_vsn, NameVsn} | Config];
+t_bad_tar_gz2({'end', Config}) ->
+    NameVsn = ?config(name_vsn, Config),
+    ok = emqx_plugins:delete_package(NameVsn),
+    ok;
+t_bad_tar_gz2(Config) ->
+    TarGz = ?config(tar_gz, Config),
+    NameVsn = ?config(name_vsn, Config),
     ?assert(filelib:is_regular(TarGz)),
-    %% failed to install, it also cleans up the bad .tar.gz file
+    %% failed to install, it also cleans up the bad content of .tar.gz file
     ?assertMatch({error, _}, emqx_plugins:ensure_installed(NameVsn)),
+    ?assertEqual({error, enoent}, file:read_file_info(emqx_plugins:dir(NameVsn))),
+    %% but the tar.gz file is still around
+    ?assert(filelib:is_regular(TarGz)),
+    ok.
+
+%% test that we even cleanup content that doesn't match the expected name-vsn
+%% pattern
+t_tar_vsn_content_mismatch({init, Config}) ->
+    WorkDir = proplists:get_value(data_dir, Config),
+    NameVsn = "bad_tar-0.2",
+    %% this an invalid info file content
+    BadInfo = "name=foo, rel_vsn=\"0.2\", rel_apps=[\"foo-0.2\"], description=\"lorem ipsum\"",
+    ok = write_info_file(Config, "foo-0.2", BadInfo),
+    TarGz = filename:join([WorkDir, "bad_tar-0.2.tar.gz"]),
+    ok = make_tar(WorkDir, "foo-0.2", NameVsn),
+    file:delete(filename:join([WorkDir, "foo-0.2", "release.json"])),
+    [{tar_gz, TarGz}, {name_vsn, NameVsn} | Config];
+t_tar_vsn_content_mismatch({'end', Config}) ->
+    NameVsn = ?config(name_vsn, Config),
+    ok = emqx_plugins:delete_package(NameVsn),
+    ok;
+t_tar_vsn_content_mismatch(Config) ->
+    TarGz = ?config(tar_gz, Config),
+    NameVsn = ?config(name_vsn, Config),
+    ?assert(filelib:is_regular(TarGz)),
+    %% failed to install, it also cleans up content of the bad .tar.gz file even
+    %% if in other directory
+    ?assertMatch({error, _}, emqx_plugins:ensure_installed(NameVsn)),
+    ?assertEqual({error, enoent}, file:read_file_info(emqx_plugins:dir(NameVsn))),
+    ?assertEqual({error, enoent}, file:read_file_info(emqx_plugins:dir("foo-0.2"))),
     %% the tar.gz file is still around
     ?assert(filelib:is_regular(TarGz)),
-    ?assertEqual({error, enoent}, file:read_file_info(emqx_plugins:dir(NameVsn))),
+    ok.
-    ok = emqx_plugins:delete_package(NameVsn).
 
 t_bad_info_json({init, Config}) ->
     Config;
@@ -446,11 +480,14 @@ t_elixir_plugin(Config) ->
     ok.
 
 make_tar(Cwd, NameWithVsn) ->
+    make_tar(Cwd, NameWithVsn, NameWithVsn).
+
+make_tar(Cwd, NameWithVsn, TarfileVsn) ->
     {ok, OriginalCwd} = file:get_cwd(),
     ok = file:set_cwd(Cwd),
     try
         Files = filelib:wildcard(NameWithVsn ++ "/**"),
-        TarFile = NameWithVsn ++ ".tar.gz",
+        TarFile = TarfileVsn ++ ".tar.gz",
         ok = erl_tar:create(TarFile, Files, [compressed])
     after
         file:set_cwd(OriginalCwd)

changes/v5.0.17/fix-9875.en.md

@@ -0,0 +1 @@
+Return `400` if a broken plugin package is uploaded from HTTP API. Also cleanup if plugin is not accepted.

changes/v5.0.17/fix-9875.zh.md

@@ -0,0 +1 @@
+当通过 HTTP API 上传一个损坏的插件包时，返回 `400`，且删除该插件包。