feat(authz api): support upload ssl cert file for api

apps/emqx_authz/src/emqx_authz.erl

@@ -309,7 +309,7 @@ check_sources(RawSources) ->
 
 find_source_by_type(Type) -> find_source_by_type(Type, lookup()).
 find_source_by_type(Type, Sources) -> find_source_by_type(Type, Sources, 1).
-find_source_by_type(_, [], _N) -> error(not_found_rule);
+find_source_by_type(_, [], _N) -> error(not_found_source);
 find_source_by_type(Type, [ Source = #{type := T} | Tail], N) ->
     case Type =:= T of
         true -> {N, Source};

apps/emqx_authz/src/emqx_authz_api_schema.erl

@@ -26,6 +26,9 @@ definitions() ->
                             type => object,
                             required => [status],
                             properties => #{
+                                id => #{
+                                    type => string
+                                },
                                 status => #{
                                     type => string,
                                     example => <<"healthy">>
@@ -41,6 +44,17 @@ definitions() ->
         oneOf => [  minirest:ref(<<"connector_redis">>)
                  ]
     },
+    SSL = #{
+      type => object,
+      required => [enable],
+      properties => #{
+         enable => #{type => boolean, example => true},
+         cacertfile => #{type => string},
+         keyfile => #{type => string},
+         certfile => #{type => string},
+         verify => #{type => boolean, example => false}
+      }
+    },
     ConnectorRedis = #{
         type => object,
         required => [type, enable, config, cmd],
@@ -65,7 +79,8 @@ definitions() ->
                                 pool_size => #{type => integer},
                                 auto_reconnect => #{type => boolean, example => true},
                                 password => #{type => string},
-                                database => #{type => string, example => mqtt}
+                                database => #{type => integer},
+                                ssl => minirest:ref(<<"ssl">>)
                              }
                             }
                          , #{type => object,
@@ -80,7 +95,8 @@ definitions() ->
                                 pool_size => #{type => integer},
                                 auto_reconnect => #{type => boolean, example => true},
                                 password => #{type => string},
-                                database => #{type => string, example => mqtt}
+                                database => #{type => integer},
+                                ssl => minirest:ref(<<"ssl">>)
                              }
                             }
                          , #{type => object,
@@ -94,7 +110,8 @@ definitions() ->
                                 pool_size => #{type => integer},
                                 auto_reconnect => #{type => boolean, example => true},
                                 password => #{type => string},
-                                database => #{type => string, example => mqtt}
+                                database => #{type => integer},
+                                ssl => minirest:ref(<<"ssl">>)
                              }
                             }
                          ],
@@ -108,5 +125,6 @@ definitions() ->
     },
     [ #{<<"returned_sources">> => RetruenedSources}
     , #{<<"sources">> => Sources}
+    , #{<<"ssl">> => SSL}
     , #{<<"connector_redis">> => ConnectorRedis}
     ].

apps/emqx_authz/src/emqx_authz_api_sources.erl

@@ -19,6 +19,7 @@
 -behavior(minirest_api).
 
 -include("emqx_authz.hrl").
+-include_lib("emqx/include/logger.hrl").
 
 -define(EXAMPLE_REDIS,
         #{type=> redis,
@@ -32,7 +33,7 @@
         maps:put(annotations, #{status => healthy}, ?EXAMPLE_REDIS)
         ).
 
--define(EXAMPLE_RETURNED_RULES,
+-define(EXAMPLE_RETURNED,
         #{sources => [?EXAMPLE_RETURNED_REDIS
                      ]
         }).
@@ -55,24 +56,6 @@ sources_api() ->
     Metadata = #{
         get => #{
             description => "List authorization sources",
-            parameters => [
-                #{
-                    name => page,
-                    in => query,
-                    schema => #{
-                       type => integer
-                    },
-                    required => false
-                },
-                #{
-                    name => limit,
-                    in => query,
-                    schema => #{
-                       type => integer
-                    },
-                    required => false
-                }
-            ],
             responses => #{
                 <<"200">> => #{
                     description => <<"OK">>,
@@ -90,7 +73,7 @@ sources_api() ->
                             examples => #{
                                 sources => #{
                                     summary => <<"Sources">>,
-                                    value => jsx:encode(?EXAMPLE_RETURNED_RULES)
+                                    value => jsx:encode(?EXAMPLE_RETURNED)
                                 }
                             }
                          }
@@ -287,53 +270,38 @@ move_source_api() ->
     },
     {"/authorization/sources/:type/move", Metadata, move_source}.
 
-sources(get, #{query_string := Query}) ->
+sources(get, _) ->
-    Sources = lists:foldl(fun (#{type := _Type, enable := true, config := #{server := Server} = Config, annotations := #{id := Id}} = Source, AccIn) ->
+    Sources = lists:foldl(fun (#{type := _Type, enable := true, config := Config, annotations := #{id := Id}} = Source, AccIn) ->
-                                NSource = case emqx_resource:health_check(Id) of
+                                  NSource0 = case maps:get(server, Config, undefined) of
-                                    ok ->
+                                                 undefined -> Source;
-                                        Source#{config => Config#{server => emqx_connector_schema_lib:ip_port_to_string(Server)},
+                                                 Server ->
-                                                annotations => #{id => Id,
+                                                     Source#{config => Config#{server => emqx_connector_schema_lib:ip_port_to_string(Server)}}
-                                                                 status => healthy}};
-                                    _ ->
-                                        Source#{config => Config#{server => emqx_connector_schema_lib:ip_port_to_string(Server)},
-                                                annotations => #{id => Id,
-                                                                 status => unhealthy}}
                                              end,
-                                lists:append(AccIn, [NSource]);
+                                  NSource1 = case maps:get(servers, Config, undefined) of
-                            (#{type := _Type, enable := true, annotations := #{id := Id}} = Source, AccIn) ->
+                                                 undefined -> NSource0;
-                                NSource = case emqx_resource:health_check(Id) of
+                                                 Servers ->
-                                    ok ->
+                                                     NSource0#{config => Config#{servers => [emqx_connector_schema_lib:ip_port_to_string(Server) || Server <- Servers]}}
-                                        Source#{annotations => #{status => healthy}};
-                                    _ ->
-                                        Source#{annotations => #{status => unhealthy}}
                                              end,
-                                lists:append(AccIn, [NSource]);
+                                  NSource2 = case emqx_resource:health_check(Id) of
+                                                 ok ->
+                                                     NSource1#{annotations => #{status => healthy}};
+                                                 _ ->
+                                                     NSource1#{annotations => #{status => unhealthy}}
+                                             end,
+                                  lists:append(AccIn, [NSource2]);
                               (Source, AccIn) ->
-                                lists:append(AccIn, [Source])
+                                  lists:append(AccIn, [Source#{annotations => #{status => healthy}}])
                         end, [], emqx_authz:lookup()),
-    case maps:is_key(<<"page">>, Query) andalso maps:is_key(<<"limit">>, Query) of
+    {200, #{sources => Sources}};
-        true ->
+sources(post, #{body := Body}) ->
-            Page = maps:get(<<"page">>, Query),
+    case emqx_authz:update(head, [save_cert(Body)]) of
-            Limit = maps:get(<<"limit">>, Query),
-            Index = (binary_to_integer(Page) - 1) * binary_to_integer(Limit),
-            {_, Sources1} = lists:split(Index, Sources),
-            case binary_to_integer(Limit) < length(Sources1) of
-                true ->
-                    {Sources2, _} = lists:split(binary_to_integer(Limit), Sources1),
-                    {200, #{sources => Sources2}};
-                false -> {200, #{sources => Sources1}}
-            end;
-        false -> {200, #{sources => Sources}}
-    end;
-sources(post, #{body := RawConfig}) ->
-    case emqx_authz:update(head, [RawConfig]) of
         {ok, _} -> {204};
         {error, Reason} ->
             {400, #{code => <<"BAD_REQUEST">>,
                     messgae => atom_to_binary(Reason)}}
     end;
-sources(put, #{body := RawConfig}) ->
+sources(put, #{body := Body}) ->
-    case emqx_authz:update(replace, RawConfig) of
+    case emqx_authz:update(replace, save_cert(Body)) of
         {ok, _} -> {204};
         {error, Reason} ->
             {400, #{code => <<"BAD_REQUEST">>,
@@ -345,27 +313,28 @@ source(get, #{bindings := #{type := Type}}) ->
         {error, Reason} -> {404, #{messgae => atom_to_binary(Reason)}};
         #{enable := false} = Source -> {200, Source};
         #{type := file} = Source -> {200, Source};
-        #{config := #{server := Server,
+        #{config := Config, annotations := #{id := Id}} = Source ->
-                      annotations := #{id := Id}
+            NSource0 = case maps:get(server, Config, undefined) of
-                     } = Config} = Source ->
+                           undefined -> Source;
-            case emqx_resource:health_check(Id) of
+                           Server ->
+                               Source#{config => Config#{server => emqx_connector_schema_lib:ip_port_to_string(Server)}}
+                       end,
+            NSource1 = case maps:get(servers, Config, undefined) of
+                           undefined -> NSource0;
+                           Servers ->
+                               NSource0#{config => Config#{servers => [emqx_connector_schema_lib:ip_port_to_string(Server) || Server <- Servers]}}
+                       end,
+            NSource2 = case emqx_resource:health_check(Id) of
                 ok ->
-                    {200, Source#{config => Config#{server => emqx_connector_schema_lib:ip_port_to_string(Server)},
+                    NSource1#{annotations => #{status => healthy}};
-                                  annotations => #{status => healthy}}};
                 _ ->
-                    {200, Source#{config => Config#{server => emqx_connector_schema_lib:ip_port_to_string(Server)},
+                    NSource1#{annotations => #{status => unhealthy}}
-                                  annotations => #{status => unhealthy}}}
+            end,
+            {200, NSource2}
     end;
-        #{config := #{annotations := #{id := Id}}} = Source ->
+source(put, #{bindings := #{type := Type}, body := Body}) ->
-            case emqx_resource:health_check(Id) of
+
-                ok ->
+    case emqx_authz:update({replace_once, Type}, save_cert(Body)) of
-                    {200, Source#{annotations => #{status => healthy}}};
-                _ ->
-                    {200, Source#{annotations => #{status => unhealthy}}}
-            end
-    end;
-source(put, #{bindings := #{type := Type}, body := RawConfig}) ->
-    case emqx_authz:update({replace_once, Type}, RawConfig) of
         {ok, _} -> {204};
         {error, not_found_source} ->
             {404, #{code => <<"NOT_FOUND">>,
@@ -391,3 +360,39 @@ move_source(post, #{bindings := #{type := Type}, body := #{<<"position">> := Pos
             {400, #{code => <<"BAD_REQUEST">>,
                     messgae => atom_to_binary(Reason)}}
     end.
+
+save_cert(#{<<"config">> := #{<<"ssl">> := #{<<"enable">> := true} = SSL} = Config} = Body) ->
+    CertPath = filename:join([emqx:get_config([node, data_dir]), "certs"]),
+    CaCert = case maps:is_key(<<"cacertfile">>, SSL) of
+                 true ->
+                     write_file(filename:join([CertPath, "cacert-" ++ emqx_rule_id:gen() ++".pem"]),
+                                maps:get(<<"cacertfile">>, SSL));
+                 false -> ""
+             end,
+    Cert = case maps:is_key(<<"certfile">>, SSL) of
+                 true ->
+                     write_file(filename:join([CertPath, "cert-" ++ emqx_rule_id:gen() ++".pem"]),
+                                maps:get(<<"certfile">>, SSL));
+                 false -> ""
+             end,
+    Key = case maps:is_key(<<"keyfile">>, SSL) of
+                 true ->
+                     write_file(filename:join([CertPath, "key-" ++ emqx_rule_id:gen() ++".pem"]),
+                                maps:get(<<"keyfile">>, SSL));
+                 false -> ""
+             end,
+    Body#{<<"config">> := Config#{<<"ssl">> => SSL#{<<"cacertfile">> => CaCert,
+                                                    <<"certfile">> => Cert,
+                                                    <<"keyfile">> => Key}
+                           }
+         };
+save_cert(Body) -> Body.
+
+write_file(Filename, Bytes) ->
+    ok = filelib:ensure_dir(Filename),
+    case file:write_file(Filename, Bytes) of
+       ok -> Filename;
+       {error, Reason} ->
+           ?LOG(error, "Write File ~p Error: ~p", [Filename, Reason]),
+           error(Reason)
+    end.

apps/emqx_authz/test/emqx_authz_api_sources_SUITE.erl

@@ -48,8 +48,10 @@
 -define(SOURCE2, #{<<"type">> => <<"mongo">>,
                    <<"enable">> => true,
                    <<"config">> => #{
-                          <<"mongo_type">> => <<"single">>,
+                          <<"mongo_type">> => <<"sharded">>,
-                          <<"server">> => <<"127.0.0.1:27017">>,
+                          <<"servers">> => [<<"127.0.0.1:27017">>,
+                                            <<"192.168.0.1:27017">>
+                                           ],
                           <<"pool_size">> => 1,
                           <<"database">> => <<"mqtt">>,
                           <<"ssl">> => #{<<"enable">> => false}},
@@ -59,7 +61,7 @@
 -define(SOURCE3, #{<<"type">> => <<"mysql">>,
                    <<"enable">> => true,
                    <<"config">> => #{
-                       <<"server">> => <<"127.0.0.1:27017">>,
+                       <<"server">> => <<"127.0.0.1:3306">>,
                        <<"pool_size">> => 1,
                        <<"database">> => <<"mqtt">>,
                        <<"username">> => <<"xx">>,
@@ -71,7 +73,7 @@
 -define(SOURCE4, #{<<"type">> => <<"pgsql">>,
                    <<"enable">> => true,
                    <<"config">> => #{
-                       <<"server">> => <<"127.0.0.1:27017">>,
+                       <<"server">> => <<"127.0.0.1:5432">>,
                        <<"pool_size">> => 1,
                        <<"database">> => <<"mqtt">>,
                        <<"username">> => <<"xx">>,
@@ -83,12 +85,15 @@
 -define(SOURCE5, #{<<"type">> => <<"redis">>,
                    <<"enable">> => true,
                    <<"config">> => #{
-                       <<"server">> => <<"127.0.0.1:27017">>,
+                       <<"servers">> => [<<"127.0.0.1:6379">>,
+                                         <<"127.0.0.1:6380">>
+                                        ],
                        <<"pool_size">> => 1,
                        <<"database">> => 0,
                        <<"password">> => <<"ee">>,
                        <<"auto_reconnect">> => true,
-                       <<"ssl">> => #{<<"enable">> => false}},
+                       <<"ssl">> => #{<<"enable">> => false}
+                      },
                    <<"cmd">> => <<"HGETALL mqtt_authz:%u">>
                   }).
 
@@ -144,6 +149,26 @@ set_special_configs(emqx_authz) ->
 set_special_configs(_App) ->
     ok.
 
+init_per_testcase(t_api, Config) ->
+    meck:new(emqx_rule_id, [non_strict, passthrough, no_history, no_link]),
+    meck:expect(emqx_rule_id, gen, fun() -> "fake" end),
+
+    meck:new(emqx, [non_strict, passthrough, no_history, no_link]),
+    meck:expect(emqx, get_config, fun([node, data_dir]) ->
+                                          % emqx_ct_helpers:deps_path(emqx_authz, "test");
+                                          {data_dir, Data} = lists:keyfind(data_dir, 1, Config),
+                                          Data;
+                                     (C) -> meck:passthrough([C])
+                                  end),
+    Config;
+init_per_testcase(_, Config) -> Config.
+
+end_per_testcase(t_api, _Config) ->
+    meck:unload(emqx_rule_id),
+    meck:unload(emqx),
+    ok;
+end_per_testcase(_, _Config) -> ok.
+
 %%------------------------------------------------------------------------------
 %% Testcases
 %%------------------------------------------------------------------------------
@@ -158,13 +183,6 @@ t_api(_) ->
     {ok, 200, Result2} = request(get, uri(["authorization", "sources"]), []),
     ?assertEqual(20, length(get_sources(Result2))),
 
-    lists:foreach(fun(Page) ->
-                          Query = "?page=" ++ integer_to_list(Page) ++ "&&limit=10",
-                          Url = uri(["authorization/sources" ++ Query]),
-                          {ok, 200, Result} = request(get, Url, []),
-                          ?assertEqual(10, length(get_sources(Result)))
-                  end, lists:seq(1, 2)),
-
     {ok, 204, _} = request(put, uri(["authorization", "sources"]), [?SOURCE1, ?SOURCE2, ?SOURCE3, ?SOURCE4]),
 
     {ok, 200, Result3} = request(get, uri(["authorization", "sources"]), []),
@@ -176,15 +194,31 @@ t_api(_) ->
                  ], Sources),
 
     {ok, 204, _} = request(put, uri(["authorization", "sources", "http"]),  ?SOURCE1#{<<"enable">> := false}),
-
     {ok, 200, Result4} = request(get, uri(["authorization", "sources", "http"]), []),
     ?assertMatch(#{<<"type">> := <<"http">>, <<"enable">> := false}, jsx:decode(Result4)),
 
+    #{<<"config">> := Config} = ?SOURCE2,
+    {ok, 204, _} = request(put, uri(["authorization", "sources", "mongo"]),
+                           ?SOURCE2#{<<"config">> := Config#{<<"ssl">> := #{
+                                                                 <<"enable">> => true,
+                                                                 <<"cacertfile">> => <<"fake cacert file">>,
+                                                                 <<"certfile">> => <<"fake cert file">>,
+                                                                 <<"keyfile">> => <<"fake key file">>,
+                                                                 <<"verify">> => false
+                                                                }}}),
+    {ok, 200, Result5} = request(get, uri(["authorization", "sources", "mongo"]), []),
+    ?assertMatch(#{<<"type">> := <<"mongo">>,
+                   <<"config">> := #{<<"ssl">> := #{<<"enable">> := true}}
+                  }, jsx:decode(Result5)),
+    ?assert(filelib:is_file(filename:join([emqx:get_config([node, data_dir]), "certs", "cacert-fake.pem"]))),
+    ?assert(filelib:is_file(filename:join([emqx:get_config([node, data_dir]), "certs", "cert-fake.pem"]))),
+    ?assert(filelib:is_file(filename:join([emqx:get_config([node, data_dir]), "certs", "key-fake.pem"]))),
+
     lists:foreach(fun(#{<<"type">> := Type}) ->
                     {ok, 204, _} = request(delete, uri(["authorization", "sources", binary_to_list(Type)]), [])
                   end, Sources),
-    {ok, 200, Result5} = request(get, uri(["authorization", "sources"]), []),
+    {ok, 200, Result6} = request(get, uri(["authorization", "sources"]), []),
-    ?assertEqual([], get_sources(Result5)),
+    ?assertEqual([], get_sources(Result6)),
     ok.
 
 t_move_source(_) ->