feat(mongo): accept wrapped secrets as passwords

Also test authorization with mongo in bridge / auth test suites.

.ci/docker-compose-file/credentials.env

@@ -0,0 +1,7 @@
+MONGO_USERNAME=emqx
+MONGO_PASSWORD=passw0rd
+MONGO_AUTHSOURCE=admin
+
+# See "Environment Variables" @ https://hub.docker.com/_/mongo
+MONGO_INITDB_ROOT_USERNAME=${MONGO_USERNAME}
+MONGO_INITDB_ROOT_PASSWORD=${MONGO_PASSWORD}

.ci/docker-compose-file/docker-compose-mongo-single-tcp.yaml

@@ -9,6 +9,9 @@ services:
       - emqx_bridge
     ports:
       - "27017:27017"
+    env_file:
+      - .env
+      - credentials.env
     command:
       --ipv6
       --bind_ip_all

.ci/docker-compose-file/docker-compose.yaml

@@ -5,6 +5,7 @@ services:
     container_name: erlang
     image: ${DOCKER_CT_RUNNER_IMAGE:-ghcr.io/emqx/emqx-builder/5.2-3:1.14.5-25.3.2-2-ubuntu22.04}
     env_file:
+      - credentials.env
       - conf.env
     environment:
       GITHUB_ACTIONS: ${GITHUB_ACTIONS:-}

apps/emqx_auth_mongodb/test/emqx_authn_mongodb_SUITE.erl

@@ -278,6 +278,10 @@ raw_mongo_auth_config() ->
         <<"server">> => mongo_server(),
         <<"w_mode">> => <<"unsafe">>,
 
+        <<"auth_source">> => mongo_authsource(),
+        <<"username">> => mongo_username(),
+        <<"password">> => mongo_password(),
+
         <<"filter">> => #{<<"username">> => <<"${username}">>},
         <<"password_hash_field">> => <<"password_hash">>,
         <<"salt_field">> => <<"salt">>,
@@ -464,9 +468,21 @@ mongo_config() ->
         {database, <<"mqtt">>},
         {host, ?MONGO_HOST},
         {port, ?MONGO_DEFAULT_PORT},
+        {auth_source, mongo_authsource()},
+        {login, mongo_username()},
+        {password, mongo_password()},
         {register, ?MONGO_CLIENT}
     ].
 
+mongo_authsource() ->
+    iolist_to_binary(os:getenv("MONGO_AUTHSOURCE", "admin")).
+
+mongo_username() ->
+    iolist_to_binary(os:getenv("MONGO_USERNAME", "")).
+
+mongo_password() ->
+    iolist_to_binary(os:getenv("MONGO_PASSWORD", "")).
+
 start_apps(Apps) ->
     lists:foreach(fun application:ensure_all_started/1, Apps).
 

apps/emqx_auth_mongodb/test/emqx_authz_mongodb_SUITE.erl

@@ -397,6 +397,10 @@ raw_mongo_authz_config() ->
         <<"collection">> => <<"acl">>,
         <<"server">> => mongo_server(),
 
+        <<"auth_source">> => mongo_authsource(),
+        <<"username">> => mongo_username(),
+        <<"password">> => mongo_password(),
+
         <<"filter">> => #{<<"username">> => <<"${username}">>}
     }.
 
@@ -408,9 +412,21 @@ mongo_config() ->
         {database, <<"mqtt">>},
         {host, ?MONGO_HOST},
         {port, ?MONGO_DEFAULT_PORT},
+        {auth_source, mongo_authsource()},
+        {login, mongo_username()},
+        {password, mongo_password()},
         {register, ?MONGO_CLIENT}
     ].
 
+mongo_authsource() ->
+    iolist_to_binary(os:getenv("MONGO_AUTHSOURCE", "admin")).
+
+mongo_username() ->
+    iolist_to_binary(os:getenv("MONGO_USERNAME", "")).
+
+mongo_password() ->
+    iolist_to_binary(os:getenv("MONGO_PASSWORD", "")).
+
 start_apps(Apps) ->
     lists:foreach(fun application:ensure_all_started/1, Apps).
 

apps/emqx_bridge_mongodb/src/emqx_bridge_mongodb.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_mongodb, [
     {description, "EMQX Enterprise MongoDB Bridge"},
-    {vsn, "0.2.1"},
+    {vsn, "0.2.2"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_mongodb/src/emqx_bridge_mongodb_connector.erl

@@ -6,9 +6,6 @@
 
 -behaviour(emqx_resource).
 
--include_lib("emqx_connector/include/emqx_connector_tables.hrl").
--include_lib("emqx_resource/include/emqx_resource.hrl").
--include_lib("typerefl/include/types.hrl").
 -include_lib("emqx/include/logger.hrl").
 -include_lib("snabbkaffe/include/snabbkaffe.hrl").
 

apps/emqx_bridge_mongodb/test/emqx_bridge_mongodb_SUITE.erl

@@ -11,6 +11,8 @@
 -include_lib("common_test/include/ct.hrl").
 -include_lib("snabbkaffe/include/snabbkaffe.hrl").
 
+-import(emqx_utils_conv, [bin/1]).
+
 %%------------------------------------------------------------------------------
 %% CT boilerplate
 %%------------------------------------------------------------------------------
@@ -96,14 +98,27 @@ init_per_group(Type = single, Config) ->
         true ->
             ok = start_apps(),
             emqx_mgmt_api_test_util:init_suite(),
-            {Name, MongoConfig} = mongo_config(MongoHost, MongoPort, Type, Config),
+            %% NOTE: `mongo-single` has auth enabled, see `credentials.env`.
+            AuthSource = bin(os:getenv("MONGO_AUTHSOURCE", "admin")),
+            Username = bin(os:getenv("MONGO_USERNAME", "")),
+            Password = bin(os:getenv("MONGO_PASSWORD", "")),
+            Passfile = filename:join(?config(priv_dir, Config), "passfile"),
+            ok = file:write_file(Passfile, Password),
+            NConfig = [
+                {mongo_authsource, AuthSource},
+                {mongo_username, Username},
+                {mongo_password, Password},
+                {mongo_passfile, Passfile}
+                | Config
+            ],
+            {Name, MongoConfig} = mongo_config(MongoHost, MongoPort, Type, NConfig),
             [
                 {mongo_host, MongoHost},
                 {mongo_port, MongoPort},
                 {mongo_config, MongoConfig},
                 {mongo_type, Type},
                 {mongo_name, Name}
-                | Config
+                | NConfig
             ];
         false ->
             {skip, no_mongo}
@@ -121,13 +136,13 @@ end_per_suite(_Config) ->
     ok.
 
 init_per_testcase(_Testcase, Config) ->
-    catch clear_db(Config),
+    clear_db(Config),
     delete_bridge(Config),
     snabbkaffe:start_trace(),
     Config.
 
 end_per_testcase(_Testcase, Config) ->
-    catch clear_db(Config),
+    clear_db(Config),
     delete_bridge(Config),
     snabbkaffe:stop(),
     ok.
@@ -175,19 +190,19 @@ mongo_config(MongoHost, MongoPort0, rs = Type, Config) ->
     Name = atom_to_binary(?MODULE),
     ConfigString =
         io_lib:format(
-            "bridges.mongodb_rs.~s {\n"
+            "bridges.mongodb_rs.~s {"
-            "  enable = true\n"
+            "\n   enable = true"
-            "  collection = mycol\n"
+            "\n   collection = mycol"
-            "  replica_set_name = rs0\n"
+            "\n   replica_set_name = rs0"
-            "  servers = [~p]\n"
+            "\n   servers = [~p]"
-            "  w_mode = safe\n"
+            "\n   w_mode = safe"
-            "  use_legacy_protocol = auto\n"
+            "\n   use_legacy_protocol = auto"
-            "  database = mqtt\n"
+            "\n   database = mqtt"
-            "  resource_opts = {\n"
+            "\n   resource_opts = {"
-            "    query_mode = ~s\n"
+            "\n     query_mode = ~s"
-            "    worker_pool_size = 1\n"
+            "\n     worker_pool_size = 1"
-            "  }\n"
+            "\n   }"
-            "}",
+            "\n }",
             [
                 Name,
                 Servers,
@@ -202,18 +217,18 @@ mongo_config(MongoHost, MongoPort0, sharded = Type, Config) ->
     Name = atom_to_binary(?MODULE),
     ConfigString =
         io_lib:format(
-            "bridges.mongodb_sharded.~s {\n"
+            "bridges.mongodb_sharded.~s {"
-            "  enable = true\n"
+            "\n   enable = true"
-            "  collection = mycol\n"
+            "\n   collection = mycol"
-            "  servers = [~p]\n"
+            "\n   servers = [~p]"
-            "  w_mode = safe\n"
+            "\n   w_mode = safe"
-            "  use_legacy_protocol = auto\n"
+            "\n   use_legacy_protocol = auto"
-            "  database = mqtt\n"
+            "\n   database = mqtt"
-            "  resource_opts = {\n"
+            "\n   resource_opts = {"
-            "    query_mode = ~s\n"
+            "\n     query_mode = ~s"
-            "    worker_pool_size = 1\n"
+            "\n     worker_pool_size = 1"
-            "  }\n"
+            "\n   }"
-            "}",
+            "\n }",
             [
                 Name,
                 Servers,
@@ -228,21 +243,27 @@ mongo_config(MongoHost, MongoPort0, single = Type, Config) ->
     Name = atom_to_binary(?MODULE),
     ConfigString =
         io_lib:format(
-            "bridges.mongodb_single.~s {\n"
+            "bridges.mongodb_single.~s {"
-            "  enable = true\n"
+            "\n   enable = true"
-            "  collection = mycol\n"
+            "\n   collection = mycol"
-            "  server = ~p\n"
+            "\n   server = ~p"
-            "  w_mode = safe\n"
+            "\n   w_mode = safe"
-            "  use_legacy_protocol = auto\n"
+            "\n   use_legacy_protocol = auto"
-            "  database = mqtt\n"
+            "\n   database = mqtt"
-            "  resource_opts = {\n"
+            "\n   auth_source = ~s"
-            "    query_mode = ~s\n"
+            "\n   username = ~s"
-            "    worker_pool_size = 1\n"
+            "\n   password = \"file://~s\""
-            "  }\n"
+            "\n   resource_opts = {"
-            "}",
+            "\n     query_mode = ~s"
+            "\n     worker_pool_size = 1"
+            "\n   }"
+            "\n }",
             [
                 Name,
                 Server,
+                ?config(mongo_authsource, Config),
+                ?config(mongo_username, Config),
+                ?config(mongo_passfile, Config),
                 QueryMode
             ]
         ),
@@ -284,8 +305,24 @@ clear_db(Config) ->
     Host = ?config(mongo_host, Config),
     Port = ?config(mongo_port, Config),
     Server = Host ++ ":" ++ integer_to_list(Port),
-    #{<<"database">> := Db, <<"collection">> := Collection} = ?config(mongo_config, Config),
+    #{
-    {ok, Client} = mongo_api:connect(Type, [Server], [], [{database, Db}, {w_mode, unsafe}]),
+        <<"database">> := Db,
+        <<"collection">> := Collection
+    } = ?config(mongo_config, Config),
+    WorkerOpts = [
+        {database, Db},
+        {w_mode, unsafe}
+        | lists:flatmap(
+            fun
+                ({mongo_authsource, AS}) -> [{auth_source, AS}];
+                ({mongo_username, User}) -> [{login, User}];
+                ({mongo_password, Pass}) -> [{password, Pass}];
+                (_) -> []
+            end,
+            Config
+        )
+    ],
+    {ok, Client} = mongo_api:connect(Type, [Server], [], WorkerOpts),
     {true, _} = mongo_api:delete(Client, Collection, _Selector = #{}),
     mongo_api:disconnect(Client).
 
@@ -386,13 +423,21 @@ t_setup_via_config_and_publish(Config) ->
     ok.
 
 t_setup_via_http_api_and_publish(Config) ->
-    Type = mongo_type_bin(?config(mongo_type, Config)),
+    Type = ?config(mongo_type, Config),
     Name = ?config(mongo_name, Config),
     MongoConfig0 = ?config(mongo_config, Config),
-    MongoConfig = MongoConfig0#{
+    MongoConfig1 = MongoConfig0#{
         <<"name">> => Name,
-        <<"type">> => Type
+        <<"type">> => mongo_type_bin(Type)
     },
+    MongoConfig =
+        case Type of
+            single ->
+                %% NOTE: using literal password with HTTP API requests.
+                MongoConfig1#{<<"password">> => ?config(mongo_password, Config)};
+            _ ->
+                MongoConfig1
+        end,
     ?assertMatch(
         {ok, _},
         create_bridge_http(MongoConfig)

apps/emqx_mongodb/src/emqx_mongodb.app.src

@@ -1,6 +1,6 @@
 {application, emqx_mongodb, [
     {description, "EMQX MongoDB Connector"},
-    {vsn, "0.1.2"},
+    {vsn, "0.1.3"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_mongodb/src/emqx_mongodb.erl

@@ -140,7 +140,7 @@ mongo_fields() ->
         {srv_record, fun srv_record/1},
         {pool_size, fun emqx_connector_schema_lib:pool_size/1},
         {username, fun emqx_connector_schema_lib:username/1},
-        {password, fun emqx_connector_schema_lib:password/1},
+        {password, emqx_connector_schema_lib:password_field()},
         {use_legacy_protocol,
             hoconsc:mk(hoconsc:enum([auto, true, false]), #{
                 default => auto,
@@ -428,8 +428,8 @@ init_worker_options([{auth_source, V} | R], Acc) ->
     init_worker_options(R, [{auth_source, V} | Acc]);
 init_worker_options([{username, V} | R], Acc) ->
     init_worker_options(R, [{login, V} | Acc]);
-init_worker_options([{password, V} | R], Acc) ->
+init_worker_options([{password, Secret} | R], Acc) ->
-    init_worker_options(R, [{password, emqx_secret:wrap(V)} | Acc]);
+    init_worker_options(R, [{password, Secret} | Acc]);
 init_worker_options([{w_mode, V} | R], Acc) ->
     init_worker_options(R, [{w_mode, V} | Acc]);
 init_worker_options([{r_mode, V} | R], Acc) ->

apps/emqx_mongodb/test/emqx_mongodb_SUITE.erl

@@ -20,6 +20,7 @@
 
 -include("emqx_connector.hrl").
 -include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
 -include_lib("emqx/include/emqx.hrl").
 -include_lib("stdlib/include/assert.hrl").
 
@@ -65,27 +66,36 @@ t_lifecycle(_Config) ->
         mongo_config()
     ).
 
+t_start_passfile(Config) ->
+    ResourceID = atom_to_binary(?FUNCTION_NAME),
+    PasswordFilename = filename:join(?config(priv_dir, Config), "passfile"),
+    ok = file:write_file(PasswordFilename, mongo_password()),
+    InitialConfig = emqx_utils_maps:deep_merge(mongo_config(), #{
+        <<"config">> => #{
+            <<"password">> => iolist_to_binary(["file://", PasswordFilename])
+        }
+    }),
+    ?assertMatch(
+        #{status := connected},
+        create_local_resource(ResourceID, check_config(InitialConfig))
+    ),
+    ?assertEqual(
+        ok,
+        emqx_resource:remove_local(ResourceID)
+    ).
+
 perform_lifecycle_check(ResourceId, InitialConfig) ->
-    {ok, #{config := CheckedConfig}} =
+    CheckedConfig = check_config(InitialConfig),
-        emqx_resource:check_config(?MONGO_RESOURCE_MOD, InitialConfig),
+    #{
-    {ok, #{
         state := #{pool_name := PoolName} = State,
         status := InitialStatus
-    }} =
+    } = create_local_resource(ResourceId, CheckedConfig),
-        emqx_resource:create_local(
-            ResourceId,
-            ?CONNECTOR_RESOURCE_GROUP,
-            ?MONGO_RESOURCE_MOD,
-            CheckedConfig,
-            #{}
-        ),
     ?assertEqual(InitialStatus, connected),
     % Instance should match the state and status of the just started resource
     {ok, ?CONNECTOR_RESOURCE_GROUP, #{
         state := State,
         status := InitialStatus
-    }} =
+    }} = emqx_resource:get_instance(ResourceId),
-        emqx_resource:get_instance(ResourceId),
     ?assertEqual({ok, connected}, emqx_resource:health_check(ResourceId)),
     % % Perform query as further check that the resource is working as expected
     ?assertMatch({ok, []}, emqx_resource:query(ResourceId, test_query_find())),
@@ -123,24 +133,52 @@ perform_lifecycle_check(ResourceId, InitialConfig) ->
 % %% Helpers
 % %%------------------------------------------------------------------------------
 
+check_config(Config) ->
+    {ok, #{config := CheckedConfig}} = emqx_resource:check_config(?MONGO_RESOURCE_MOD, Config),
+    CheckedConfig.
+
+create_local_resource(ResourceId, CheckedConfig) ->
+    {ok, Bridge} = emqx_resource:create_local(
+        ResourceId,
+        ?CONNECTOR_RESOURCE_GROUP,
+        ?MONGO_RESOURCE_MOD,
+        CheckedConfig,
+        #{}
+    ),
+    Bridge.
+
 mongo_config() ->
     RawConfig = list_to_binary(
         io_lib:format(
-            ""
+            "\n    mongo_type = single"
-            "\n"
+            "\n    database = mqtt"
-            "    mongo_type = single\n"
+            "\n    pool_size = 8"
-            "    database = mqtt\n"
+            "\n    server = \"~s:~b\""
-            "    pool_size = 8\n"
+            "\n    auth_source = ~p"
-            "    server = \"~s:~b\"\n"
+            "\n    username = ~p"
-            "    "
+            "\n    password = ~p"
-            "",
+            "\n",
-            [?MONGO_HOST, ?MONGO_DEFAULT_PORT]
+            [
+                ?MONGO_HOST,
+                ?MONGO_DEFAULT_PORT,
+                mongo_authsource(),
+                mongo_username(),
+                mongo_password()
+            ]
         )
     ),
-
     {ok, Config} = hocon:binary(RawConfig),
     #{<<"config">> => Config}.
 
+mongo_authsource() ->
+    os:getenv("MONGO_AUTHSOURCE", "admin").
+
+mongo_username() ->
+    os:getenv("MONGO_USERNAME", "").
+
+mongo_password() ->
+    os:getenv("MONGO_PASSWORD", "").
+
 test_query_find() ->
     {find, <<"foo">>, #{}, #{}}.