Merge pull request #11809 from keynslug/ft/EMQX-10808/file-secrets
feat(mqttbridge): support file-sourced secrets as passwords
This commit is contained in:
commit
7092c75597
|
@ -0,0 +1,85 @@
|
|||
%%--------------------------------------------------------------------
|
||||
%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
|
||||
%%
|
||||
%% 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
|
||||
%%
|
||||
%% Unless required by applicable law or agreed to in writing, software
|
||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
%% See the License for the specific language governing permissions and
|
||||
%% limitations under the License.
|
||||
%%--------------------------------------------------------------------
|
||||
|
||||
%% @doc HOCON schema that defines _secret_ concept.
|
||||
-module(emqx_schema_secret).
|
||||
|
||||
-include_lib("emqx/include/logger.hrl").
|
||||
-include_lib("typerefl/include/types.hrl").
|
||||
|
||||
-export([mk/1]).
|
||||
|
||||
%% HOCON Schema API
|
||||
-export([convert_secret/2]).
|
||||
|
||||
%% @doc Secret value.
|
||||
-type t() :: binary().
|
||||
|
||||
%% @doc Source of the secret value.
|
||||
%% * "file://...": file path to a file containing secret value.
|
||||
%% * other binaries: secret value itself.
|
||||
-type source() :: iodata().
|
||||
|
||||
-type secret() :: binary() | function().
|
||||
-reflect_type([secret/0]).
|
||||
|
||||
-define(SCHEMA, #{
|
||||
required => false,
|
||||
format => <<"password">>,
|
||||
sensitive => true,
|
||||
converter => fun ?MODULE:convert_secret/2
|
||||
}).
|
||||
|
||||
-dialyzer({nowarn_function, source/1}).
|
||||
|
||||
%%
|
||||
|
||||
-spec mk(#{atom() => _}) -> hocon_schema:field_schema().
|
||||
mk(Overrides = #{}) ->
|
||||
hoconsc:mk(secret(), maps:merge(?SCHEMA, Overrides)).
|
||||
|
||||
convert_secret(undefined, #{}) ->
|
||||
undefined;
|
||||
convert_secret(Secret, #{make_serializable := true}) ->
|
||||
unicode:characters_to_binary(source(Secret));
|
||||
convert_secret(Secret, #{}) when is_function(Secret, 0) ->
|
||||
Secret;
|
||||
convert_secret(Secret, #{}) when is_integer(Secret) ->
|
||||
wrap(integer_to_binary(Secret));
|
||||
convert_secret(Secret, #{}) ->
|
||||
try unicode:characters_to_binary(Secret) of
|
||||
String when is_binary(String) ->
|
||||
wrap(String);
|
||||
{error, _, _} ->
|
||||
throw(invalid_string)
|
||||
catch
|
||||
error:_ ->
|
||||
throw(invalid_type)
|
||||
end.
|
||||
|
||||
-spec wrap(source()) -> emqx_secret:t(t()).
|
||||
wrap(<<"file://", Filename/binary>>) ->
|
||||
emqx_secret:wrap_load({file, Filename});
|
||||
wrap(Secret) ->
|
||||
emqx_secret:wrap(Secret).
|
||||
|
||||
-spec source(emqx_secret:t(t())) -> source().
|
||||
source(Secret) when is_function(Secret) ->
|
||||
source(emqx_secret:term(Secret));
|
||||
source({file, Filename}) ->
|
||||
<<"file://", Filename/binary>>;
|
||||
source(Secret) ->
|
||||
Secret.
|
|
@ -19,23 +19,52 @@
|
|||
-module(emqx_secret).
|
||||
|
||||
%% API:
|
||||
-export([wrap/1, unwrap/1]).
|
||||
-export([wrap/1, wrap_load/1, unwrap/1, term/1]).
|
||||
|
||||
-export_type([t/1]).
|
||||
|
||||
-opaque t(T) :: T | fun(() -> t(T)).
|
||||
|
||||
%% Secret loader module.
|
||||
%% Any changes related to processing of secrets should be made there.
|
||||
-define(LOADER, emqx_secret_loader).
|
||||
|
||||
%%================================================================================
|
||||
%% API funcions
|
||||
%%================================================================================
|
||||
|
||||
%% @doc Wrap a term in a secret closure.
|
||||
%% This effectively hides the term from any term formatting / printing code.
|
||||
-spec wrap(T) -> t(T).
|
||||
wrap(Term) ->
|
||||
fun() ->
|
||||
Term
|
||||
end.
|
||||
|
||||
%% @doc Wrap a loader function call over a term in a secret closure.
|
||||
%% This is slightly more flexible form of `wrap/1` with the same basic purpose.
|
||||
-spec wrap_load(emqx_secret_loader:source()) -> t(_).
|
||||
wrap_load(Source) ->
|
||||
fun() ->
|
||||
apply(?LOADER, load, [Source])
|
||||
end.
|
||||
|
||||
%% @doc Unwrap a secret closure, revealing the secret.
|
||||
%% This is either `Term` or `Module:Function(Term)` depending on how it was wrapped.
|
||||
-spec unwrap(t(T)) -> T.
|
||||
unwrap(Term) when is_function(Term, 0) ->
|
||||
%% Handle potentially nested funs
|
||||
unwrap(Term());
|
||||
unwrap(Term) ->
|
||||
Term.
|
||||
|
||||
%% @doc Inspect the term wrapped in a secret closure.
|
||||
-spec term(t(_)) -> _Term.
|
||||
term(Wrap) when is_function(Wrap, 0) ->
|
||||
case erlang:fun_info(Wrap, module) of
|
||||
{module, ?MODULE} ->
|
||||
{env, Env} = erlang:fun_info(Wrap, env),
|
||||
lists:last(Env);
|
||||
_ ->
|
||||
error(badarg, [Wrap])
|
||||
end.
|
||||
|
|
|
@ -0,0 +1,42 @@
|
|||
%%--------------------------------------------------------------------
|
||||
%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
|
||||
%%
|
||||
%% 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
|
||||
%%
|
||||
%% Unless required by applicable law or agreed to in writing, software
|
||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
%% See the License for the specific language governing permissions and
|
||||
%% limitations under the License.
|
||||
%%--------------------------------------------------------------------
|
||||
|
||||
-module(emqx_secret_loader).
|
||||
|
||||
%% API
|
||||
-export([load/1]).
|
||||
-export([file/1]).
|
||||
|
||||
-export_type([source/0]).
|
||||
|
||||
-type source() :: {file, file:filename_all()}.
|
||||
|
||||
-spec load(source()) -> binary() | no_return().
|
||||
load({file, Filename}) ->
|
||||
file(Filename).
|
||||
|
||||
-spec file(file:filename_all()) -> binary() | no_return().
|
||||
file(Filename) ->
|
||||
case file:read_file(Filename) of
|
||||
{ok, Secret} ->
|
||||
string:trim(Secret, trailing);
|
||||
{error, Reason} ->
|
||||
throw(#{
|
||||
msg => failed_to_read_secret_file,
|
||||
path => Filename,
|
||||
reason => emqx_utils:explain_posix(Reason)
|
||||
})
|
||||
end.
|
|
@ -0,0 +1,76 @@
|
|||
%%--------------------------------------------------------------------
|
||||
%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
|
||||
%%
|
||||
%% 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
|
||||
%%
|
||||
%% Unless required by applicable law or agreed to in writing, software
|
||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
%% See the License for the specific language governing permissions and
|
||||
%% limitations under the License.
|
||||
%%--------------------------------------------------------------------
|
||||
|
||||
-module(emqx_secret_tests).
|
||||
|
||||
-include_lib("eunit/include/eunit.hrl").
|
||||
|
||||
wrap_unwrap_test() ->
|
||||
?assertEqual(
|
||||
42,
|
||||
emqx_secret:unwrap(emqx_secret:wrap(42))
|
||||
).
|
||||
|
||||
unwrap_immediate_test() ->
|
||||
?assertEqual(
|
||||
42,
|
||||
emqx_secret:unwrap(42)
|
||||
).
|
||||
|
||||
wrap_unwrap_load_test_() ->
|
||||
Secret = <<"foobaz">>,
|
||||
{
|
||||
setup,
|
||||
fun() -> write_temp_file(Secret) end,
|
||||
fun(Filename) -> file:delete(Filename) end,
|
||||
fun(Filename) ->
|
||||
?_assertEqual(
|
||||
Secret,
|
||||
emqx_secret:unwrap(emqx_secret:wrap_load({file, Filename}))
|
||||
)
|
||||
end
|
||||
}.
|
||||
|
||||
wrap_load_term_test() ->
|
||||
?assertEqual(
|
||||
{file, "no/such/file/i/swear"},
|
||||
emqx_secret:term(emqx_secret:wrap_load({file, "no/such/file/i/swear"}))
|
||||
).
|
||||
|
||||
wrap_unwrap_missing_file_test() ->
|
||||
?assertThrow(
|
||||
#{msg := failed_to_read_secret_file, reason := "No such file or directory"},
|
||||
emqx_secret:unwrap(emqx_secret:wrap_load({file, "no/such/file/i/swear"}))
|
||||
).
|
||||
|
||||
wrap_term_test() ->
|
||||
?assertEqual(
|
||||
42,
|
||||
emqx_secret:term(emqx_secret:wrap(42))
|
||||
).
|
||||
|
||||
external_fun_term_error_test() ->
|
||||
Term = {foo, bar},
|
||||
?assertError(
|
||||
badarg,
|
||||
emqx_secret:term(fun() -> Term end)
|
||||
).
|
||||
|
||||
write_temp_file(Bytes) ->
|
||||
Ts = erlang:system_time(millisecond),
|
||||
Filename = filename:join("/tmp", ?MODULE_STRING ++ integer_to_list(-Ts)),
|
||||
ok = file:write_file(Filename, Bytes),
|
||||
Filename.
|
|
@ -326,7 +326,7 @@ mk_client_opts(
|
|||
],
|
||||
Config
|
||||
),
|
||||
Options#{
|
||||
mk_client_opt_password(Options#{
|
||||
hosts => [HostPort],
|
||||
clientid => clientid(ResourceId, ClientScope, Config),
|
||||
connect_timeout => 30,
|
||||
|
@ -334,7 +334,13 @@ mk_client_opts(
|
|||
force_ping => true,
|
||||
ssl => EnableSsl,
|
||||
ssl_opts => maps:to_list(maps:remove(enable, Ssl))
|
||||
}.
|
||||
}).
|
||||
|
||||
mk_client_opt_password(Options = #{password := Secret}) ->
|
||||
%% TODO: Teach `emqtt` to accept 0-arity closures as passwords.
|
||||
Options#{password := emqx_secret:unwrap(Secret)};
|
||||
mk_client_opt_password(Options) ->
|
||||
Options.
|
||||
|
||||
ms_to_s(Ms) ->
|
||||
erlang:ceil(Ms / 1000).
|
||||
|
|
|
@ -99,13 +99,9 @@ fields("server_configs") ->
|
|||
}
|
||||
)},
|
||||
{password,
|
||||
mk(
|
||||
binary(),
|
||||
emqx_schema_secret:mk(
|
||||
#{
|
||||
format => <<"password">>,
|
||||
sensitive => true,
|
||||
desc => ?DESC("password"),
|
||||
converter => fun emqx_schema:password_converter/2
|
||||
desc => ?DESC("password")
|
||||
}
|
||||
)},
|
||||
{clean_start,
|
||||
|
|
|
@ -21,13 +21,15 @@
|
|||
-import(emqx_dashboard_api_test_helpers, [request/4, uri/1]).
|
||||
|
||||
-include("emqx/include/emqx.hrl").
|
||||
-include("emqx/include/emqx_hooks.hrl").
|
||||
-include("emqx/include/asserts.hrl").
|
||||
-include_lib("eunit/include/eunit.hrl").
|
||||
-include_lib("common_test/include/ct.hrl").
|
||||
-include_lib("snabbkaffe/include/snabbkaffe.hrl").
|
||||
|
||||
%% output functions
|
||||
-export([inspect/3]).
|
||||
|
||||
-define(BRIDGE_CONF_DEFAULT, <<"bridges: {}">>).
|
||||
-define(TYPE_MQTT, <<"mqtt">>).
|
||||
-define(BRIDGE_NAME_INGRESS, <<"ingress_mqtt_bridge">>).
|
||||
-define(BRIDGE_NAME_EGRESS, <<"egress_mqtt_bridge">>).
|
||||
|
@ -38,14 +40,18 @@
|
|||
-define(EGRESS_REMOTE_TOPIC, "egress_remote_topic").
|
||||
-define(EGRESS_LOCAL_TOPIC, "egress_local_topic").
|
||||
|
||||
-define(SERVER_CONF(Username), #{
|
||||
-define(SERVER_CONF, #{
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
<<"server">> => <<"127.0.0.1:1883">>,
|
||||
<<"username">> => Username,
|
||||
<<"password">> => <<"">>,
|
||||
<<"proto_ver">> => <<"v4">>,
|
||||
<<"ssl">> => #{<<"enable">> => false}
|
||||
}).
|
||||
|
||||
-define(SERVER_CONF(Username, Password), (?SERVER_CONF)#{
|
||||
<<"username">> => Username,
|
||||
<<"password">> => Password
|
||||
}).
|
||||
|
||||
-define(INGRESS_CONF, #{
|
||||
<<"remote">> => #{
|
||||
<<"topic">> => <<?INGRESS_REMOTE_TOPIC, "/#">>,
|
||||
|
@ -129,43 +135,32 @@ suite() ->
|
|||
[{timetrap, {seconds, 30}}].
|
||||
|
||||
init_per_suite(Config) ->
|
||||
_ = application:load(emqx_conf),
|
||||
ok = emqx_common_test_helpers:start_apps(
|
||||
Apps = emqx_cth_suite:start(
|
||||
[
|
||||
emqx_conf,
|
||||
emqx_bridge,
|
||||
emqx_rule_engine,
|
||||
emqx_bridge,
|
||||
emqx_bridge_mqtt,
|
||||
emqx_dashboard
|
||||
{emqx_dashboard,
|
||||
"dashboard {"
|
||||
"\n listeners.http { bind = 18083 }"
|
||||
"\n default_username = connector_admin"
|
||||
"\n default_password = public"
|
||||
"\n }"}
|
||||
],
|
||||
fun set_special_configs/1
|
||||
#{work_dir => emqx_cth_suite:work_dir(Config)}
|
||||
),
|
||||
ok = emqx_common_test_helpers:load_config(
|
||||
emqx_rule_engine_schema,
|
||||
<<"rule_engine {rules {}}">>
|
||||
),
|
||||
ok = emqx_common_test_helpers:load_config(emqx_bridge_schema, ?BRIDGE_CONF_DEFAULT),
|
||||
Config.
|
||||
[{suite_apps, Apps} | Config].
|
||||
|
||||
end_per_suite(_Config) ->
|
||||
emqx_common_test_helpers:stop_apps([
|
||||
emqx_dashboard,
|
||||
emqx_bridge_mqtt,
|
||||
emqx_bridge,
|
||||
emqx_rule_engine
|
||||
]),
|
||||
ok.
|
||||
|
||||
set_special_configs(emqx_dashboard) ->
|
||||
emqx_dashboard_api_test_helpers:set_default_config(<<"connector_admin">>);
|
||||
set_special_configs(_) ->
|
||||
ok.
|
||||
end_per_suite(Config) ->
|
||||
emqx_cth_suite:stop(?config(suite_apps, Config)).
|
||||
|
||||
init_per_testcase(_, Config) ->
|
||||
{ok, _} = emqx_cluster_rpc:start_link(node(), emqx_cluster_rpc, 1000),
|
||||
ok = snabbkaffe:start_trace(),
|
||||
Config.
|
||||
|
||||
end_per_testcase(_, _Config) ->
|
||||
ok = unhook_authenticate(),
|
||||
clear_resources(),
|
||||
snabbkaffe:stop(),
|
||||
ok.
|
||||
|
@ -187,14 +182,86 @@ clear_resources() ->
|
|||
%%------------------------------------------------------------------------------
|
||||
%% Testcases
|
||||
%%------------------------------------------------------------------------------
|
||||
|
||||
t_conf_bridge_authn_anonymous(_) ->
|
||||
ok = hook_authenticate(),
|
||||
{ok, 201, _Bridge} = request(
|
||||
post,
|
||||
uri(["bridges"]),
|
||||
?SERVER_CONF#{
|
||||
<<"name">> => <<"t_conf_bridge_anonymous">>,
|
||||
<<"ingress">> => ?INGRESS_CONF#{<<"pool_size">> => 1}
|
||||
}
|
||||
),
|
||||
?assertReceive(
|
||||
{authenticate, #{username := undefined, password := undefined}}
|
||||
).
|
||||
|
||||
t_conf_bridge_authn_password(_) ->
|
||||
Username1 = <<"user1">>,
|
||||
Password1 = <<"from-here">>,
|
||||
ok = hook_authenticate(),
|
||||
{ok, 201, _Bridge1} = request(
|
||||
post,
|
||||
uri(["bridges"]),
|
||||
?SERVER_CONF(Username1, Password1)#{
|
||||
<<"name">> => <<"t_conf_bridge_authn_password">>,
|
||||
<<"ingress">> => ?INGRESS_CONF#{<<"pool_size">> => 1}
|
||||
}
|
||||
),
|
||||
?assertReceive(
|
||||
{authenticate, #{username := Username1, password := Password1}}
|
||||
).
|
||||
|
||||
t_conf_bridge_authn_passfile(Config) ->
|
||||
DataDir = ?config(data_dir, Config),
|
||||
Username2 = <<"user2">>,
|
||||
PasswordFilename = filename:join(DataDir, "password"),
|
||||
Password2 = <<"from-there">>,
|
||||
ok = hook_authenticate(),
|
||||
{ok, 201, _Bridge2} = request(
|
||||
post,
|
||||
uri(["bridges"]),
|
||||
?SERVER_CONF(Username2, iolist_to_binary(["file://", PasswordFilename]))#{
|
||||
<<"name">> => <<"t_conf_bridge_authn_passfile">>,
|
||||
<<"ingress">> => ?INGRESS_CONF#{<<"pool_size">> => 1}
|
||||
}
|
||||
),
|
||||
?assertReceive(
|
||||
{authenticate, #{username := Username2, password := Password2}}
|
||||
),
|
||||
?assertMatch(
|
||||
{ok, 201, #{
|
||||
<<"status">> := <<"disconnected">>,
|
||||
<<"status_reason">> := <<"#{msg => failed_to_read_secret_file", _/bytes>>
|
||||
}},
|
||||
request_json(
|
||||
post,
|
||||
uri(["bridges"]),
|
||||
?SERVER_CONF(<<>>, <<"file://im/pretty/sure/theres/no/such/file">>)#{
|
||||
<<"name">> => <<"t_conf_bridge_authn_no_passfile">>
|
||||
}
|
||||
)
|
||||
).
|
||||
|
||||
hook_authenticate() ->
|
||||
emqx_hooks:add('client.authenticate', {?MODULE, authenticate, [self()]}, ?HP_HIGHEST).
|
||||
|
||||
unhook_authenticate() ->
|
||||
emqx_hooks:del('client.authenticate', {?MODULE, authenticate}).
|
||||
|
||||
authenticate(Credential, _, TestRunnerPid) ->
|
||||
_ = TestRunnerPid ! {authenticate, Credential},
|
||||
ignore.
|
||||
|
||||
%%------------------------------------------------------------------------------
|
||||
|
||||
t_mqtt_conn_bridge_ingress(_) ->
|
||||
User1 = <<"user1">>,
|
||||
%% create an MQTT bridge, using POST
|
||||
{ok, 201, Bridge} = request(
|
||||
post,
|
||||
uri(["bridges"]),
|
||||
ServerConf = ?SERVER_CONF(User1)#{
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
ServerConf = ?SERVER_CONF#{
|
||||
<<"name">> => ?BRIDGE_NAME_INGRESS,
|
||||
<<"ingress">> => ?INGRESS_CONF
|
||||
}
|
||||
|
@ -249,7 +316,6 @@ t_mqtt_conn_bridge_ingress(_) ->
|
|||
ok.
|
||||
|
||||
t_mqtt_conn_bridge_ingress_full_context(_Config) ->
|
||||
User1 = <<"user1">>,
|
||||
IngressConf =
|
||||
emqx_utils_maps:deep_merge(
|
||||
?INGRESS_CONF,
|
||||
|
@ -258,8 +324,7 @@ t_mqtt_conn_bridge_ingress_full_context(_Config) ->
|
|||
{ok, 201, _Bridge} = request(
|
||||
post,
|
||||
uri(["bridges"]),
|
||||
?SERVER_CONF(User1)#{
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
?SERVER_CONF#{
|
||||
<<"name">> => ?BRIDGE_NAME_INGRESS,
|
||||
<<"ingress">> => IngressConf
|
||||
}
|
||||
|
@ -297,8 +362,7 @@ t_mqtt_conn_bridge_ingress_shared_subscription(_) ->
|
|||
Ns = lists:seq(1, 10),
|
||||
BridgeName = atom_to_binary(?FUNCTION_NAME),
|
||||
BridgeID = create_bridge(
|
||||
?SERVER_CONF(<<>>)#{
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
?SERVER_CONF#{
|
||||
<<"name">> => BridgeName,
|
||||
<<"ingress">> => #{
|
||||
<<"pool_size">> => PoolSize,
|
||||
|
@ -337,8 +401,7 @@ t_mqtt_conn_bridge_ingress_shared_subscription(_) ->
|
|||
t_mqtt_egress_bridge_ignores_clean_start(_) ->
|
||||
BridgeName = atom_to_binary(?FUNCTION_NAME),
|
||||
BridgeID = create_bridge(
|
||||
?SERVER_CONF(<<"user1">>)#{
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
?SERVER_CONF#{
|
||||
<<"name">> => BridgeName,
|
||||
<<"egress">> => ?EGRESS_CONF,
|
||||
<<"clean_start">> => false
|
||||
|
@ -366,8 +429,7 @@ t_mqtt_egress_bridge_ignores_clean_start(_) ->
|
|||
t_mqtt_conn_bridge_ingress_downgrades_qos_2(_) ->
|
||||
BridgeName = atom_to_binary(?FUNCTION_NAME),
|
||||
BridgeID = create_bridge(
|
||||
?SERVER_CONF(<<"user1">>)#{
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
?SERVER_CONF#{
|
||||
<<"name">> => BridgeName,
|
||||
<<"ingress">> => emqx_utils_maps:deep_merge(
|
||||
?INGRESS_CONF,
|
||||
|
@ -392,9 +454,8 @@ t_mqtt_conn_bridge_ingress_downgrades_qos_2(_) ->
|
|||
ok.
|
||||
|
||||
t_mqtt_conn_bridge_ingress_no_payload_template(_) ->
|
||||
User1 = <<"user1">>,
|
||||
BridgeIDIngress = create_bridge(
|
||||
?SERVER_CONF(User1)#{
|
||||
?SERVER_CONF#{
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
<<"name">> => ?BRIDGE_NAME_INGRESS,
|
||||
<<"ingress">> => ?INGRESS_CONF_NO_PAYLOAD_TEMPLATE
|
||||
|
@ -428,10 +489,8 @@ t_mqtt_conn_bridge_ingress_no_payload_template(_) ->
|
|||
|
||||
t_mqtt_conn_bridge_egress(_) ->
|
||||
%% then we add a mqtt connector, using POST
|
||||
User1 = <<"user1">>,
|
||||
BridgeIDEgress = create_bridge(
|
||||
?SERVER_CONF(User1)#{
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
?SERVER_CONF#{
|
||||
<<"name">> => ?BRIDGE_NAME_EGRESS,
|
||||
<<"egress">> => ?EGRESS_CONF
|
||||
}
|
||||
|
@ -473,11 +532,8 @@ t_mqtt_conn_bridge_egress(_) ->
|
|||
|
||||
t_mqtt_conn_bridge_egress_no_payload_template(_) ->
|
||||
%% then we add a mqtt connector, using POST
|
||||
User1 = <<"user1">>,
|
||||
|
||||
BridgeIDEgress = create_bridge(
|
||||
?SERVER_CONF(User1)#{
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
?SERVER_CONF#{
|
||||
<<"name">> => ?BRIDGE_NAME_EGRESS,
|
||||
<<"egress">> => ?EGRESS_CONF_NO_PAYLOAD_TEMPLATE
|
||||
}
|
||||
|
@ -520,11 +576,9 @@ t_mqtt_conn_bridge_egress_no_payload_template(_) ->
|
|||
ok.
|
||||
|
||||
t_egress_custom_clientid_prefix(_Config) ->
|
||||
User1 = <<"user1">>,
|
||||
BridgeIDEgress = create_bridge(
|
||||
?SERVER_CONF(User1)#{
|
||||
?SERVER_CONF#{
|
||||
<<"clientid_prefix">> => <<"my-custom-prefix">>,
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
<<"name">> => ?BRIDGE_NAME_EGRESS,
|
||||
<<"egress">> => ?EGRESS_CONF
|
||||
}
|
||||
|
@ -545,17 +599,14 @@ t_egress_custom_clientid_prefix(_Config) ->
|
|||
ok.
|
||||
|
||||
t_mqtt_conn_bridge_ingress_and_egress(_) ->
|
||||
User1 = <<"user1">>,
|
||||
BridgeIDIngress = create_bridge(
|
||||
?SERVER_CONF(User1)#{
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
?SERVER_CONF#{
|
||||
<<"name">> => ?BRIDGE_NAME_INGRESS,
|
||||
<<"ingress">> => ?INGRESS_CONF
|
||||
}
|
||||
),
|
||||
BridgeIDEgress = create_bridge(
|
||||
?SERVER_CONF(User1)#{
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
?SERVER_CONF#{
|
||||
<<"name">> => ?BRIDGE_NAME_EGRESS,
|
||||
<<"egress">> => ?EGRESS_CONF
|
||||
}
|
||||
|
@ -627,8 +678,7 @@ t_mqtt_conn_bridge_ingress_and_egress(_) ->
|
|||
|
||||
t_ingress_mqtt_bridge_with_rules(_) ->
|
||||
BridgeIDIngress = create_bridge(
|
||||
?SERVER_CONF(<<"user1">>)#{
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
?SERVER_CONF#{
|
||||
<<"name">> => ?BRIDGE_NAME_INGRESS,
|
||||
<<"ingress">> => ?INGRESS_CONF
|
||||
}
|
||||
|
@ -712,8 +762,7 @@ t_ingress_mqtt_bridge_with_rules(_) ->
|
|||
|
||||
t_egress_mqtt_bridge_with_rules(_) ->
|
||||
BridgeIDEgress = create_bridge(
|
||||
?SERVER_CONF(<<"user1">>)#{
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
?SERVER_CONF#{
|
||||
<<"name">> => ?BRIDGE_NAME_EGRESS,
|
||||
<<"egress">> => ?EGRESS_CONF
|
||||
}
|
||||
|
@ -789,10 +838,8 @@ t_egress_mqtt_bridge_with_rules(_) ->
|
|||
|
||||
t_mqtt_conn_bridge_egress_reconnect(_) ->
|
||||
%% then we add a mqtt connector, using POST
|
||||
User1 = <<"user1">>,
|
||||
BridgeIDEgress = create_bridge(
|
||||
?SERVER_CONF(User1)#{
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
?SERVER_CONF#{
|
||||
<<"name">> => ?BRIDGE_NAME_EGRESS,
|
||||
<<"egress">> => ?EGRESS_CONF,
|
||||
<<"resource_opts">> => #{
|
||||
|
@ -897,10 +944,8 @@ t_mqtt_conn_bridge_egress_reconnect(_) ->
|
|||
ok.
|
||||
|
||||
t_mqtt_conn_bridge_egress_async_reconnect(_) ->
|
||||
User1 = <<"user1">>,
|
||||
BridgeIDEgress = create_bridge(
|
||||
?SERVER_CONF(User1)#{
|
||||
<<"type">> => ?TYPE_MQTT,
|
||||
?SERVER_CONF#{
|
||||
<<"name">> => ?BRIDGE_NAME_EGRESS,
|
||||
<<"egress">> => ?EGRESS_CONF,
|
||||
<<"resource_opts">> => #{
|
||||
|
@ -1018,5 +1063,9 @@ request_bridge_metrics(BridgeID) ->
|
|||
{ok, 200, BridgeMetrics} = request(get, uri(["bridges", BridgeID, "metrics"]), []),
|
||||
emqx_utils_json:decode(BridgeMetrics).
|
||||
|
||||
request_json(Method, Url, Body) ->
|
||||
{ok, Code, Response} = request(Method, Url, Body),
|
||||
{ok, Code, emqx_utils_json:decode(Response)}.
|
||||
|
||||
request(Method, Url, Body) ->
|
||||
request(<<"connector_admin">>, Method, Url, Body).
|
||||
|
|
|
@ -0,0 +1 @@
|
|||
from-there
|
|
@ -908,6 +908,9 @@ typename_to_spec("port_number()", _Mod) ->
|
|||
range("1..65535");
|
||||
typename_to_spec("secret_access_key()", _Mod) ->
|
||||
#{type => string, example => <<"TW8dPwmjpjJJuLW....">>};
|
||||
typename_to_spec("secret()", _Mod) ->
|
||||
%% TODO: ideally, this should be dispatched to the module that defines this type
|
||||
#{type => string, example => <<"R4ND0M/S∃CЯ∃T"/utf8>>};
|
||||
typename_to_spec(Name, Mod) ->
|
||||
try_convert_to_spec(Name, Mod, [
|
||||
fun try_remote_module_type/2,
|
||||
|
|
Loading…
Reference in New Issue