Obfuscate credentials in shovel worker states to avoid plaintext pass…

deps/rabbitmq_shovel/src/rabbit_shovel_dyn_worker_sup_sup.erl

@@ -11,6 +11,7 @@
 -export([start_link/0, init/1, adjust/2, stop_child/1, cleanup_specs/0]).
 
 -import(rabbit_misc, [pget/2]).
+-import(rabbit_data_coercion, [to_map/1, to_list/1]).
 
 -include("rabbit_shovel.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
@@ -42,7 +43,7 @@ start_child({VHost, ShovelName} = Name, Def) ->
     rabbit_log_shovel:debug("Starting a mirrored supervisor named '~s' in virtual host '~s'", [ShovelName, VHost]),
     Result = case mirrored_supervisor:start_child(
            ?SUPERVISOR,
-           {Name, {rabbit_shovel_dyn_worker_sup, start_link, [Name, Def]},
+           {Name, {rabbit_shovel_dyn_worker_sup, start_link, [Name, obfuscated_uris_parameters(Def)]},
             transient, ?WORKER_WAIT, worker, [rabbit_shovel_dyn_worker_sup]}) of
         {ok,                      _Pid}  -> ok;
         {error, {already_started, _Pid}} -> ok
@@ -51,6 +52,11 @@ start_child({VHost, ShovelName} = Name, Def) ->
     rabbit_shovel_locks:unlock(LockId),
     Result.
 
+obfuscated_uris_parameters(Def) when is_map(Def) ->
+    to_map(rabbit_shovel_parameters:obfuscate_uris_parameters(to_list(Def)));
+obfuscated_uris_parameters(Def) when is_list(Def) ->
+    rabbit_shovel_parameters:obfuscate_uris_parameters(Def).
+
 child_exists(Name) ->
     lists:any(fun ({N, _, _, _}) -> N =:= Name end,
               mirrored_supervisor:which_children(?SUPERVISOR)).

deps/rabbitmq_shovel/src/rabbit_shovel_parameters.erl

@@ -13,8 +13,9 @@
 
 -export([validate/5, notify/5, notify_clear/4]).
 -export([register/0, unregister/0, parse/3]).
+-export([obfuscate_uris_parameters/1]).
 
--import(rabbit_misc, [pget/2, pget/3]).
+-import(rabbit_misc, [pget/2, pget/3, pset/3]).
 
 -rabbit_boot_step({?MODULE,
                    [{description, "shovel parameters"},
@@ -82,6 +83,16 @@ validate_amqp091_src(Def) ->
              ok
      end].
 
+obfuscate_uris_parameters(Def) ->
+  SrcURIs  = get_uris(<<"src-uri">>, Def),
+  ObfuscatedSrcURIsDef = pset(<<"src-uri">>, obfuscate_uris(SrcURIs), Def),
+  DestURIs  = get_uris(<<"dest-uri">>, Def),
+  ObfuscatedDef = pset(<<"dest-uri">>, obfuscate_uris(DestURIs), ObfuscatedSrcURIsDef),
+  ObfuscatedDef.
+
+obfuscate_uris(URIs) ->
+  [credentials_obfuscation:encrypt(URI) || URI <- URIs].
+
 validate_amqp091_dest(Def) ->
     [case pget2(<<"dest-exchange">>, <<"dest-queue">>, Def) of
          zero -> ok;
@@ -279,7 +290,7 @@ parse_dest(VHostName, ClusterName, Def, SourceHeaders) ->
     end.
 
 parse_amqp10_dest({_VHost, _Name}, _ClusterName, Def, SourceHeaders) ->
-    Uris = get_uris(<<"dest-uri">>, Def),
+    Uris = get_unencrypted_uris(<<"dest-uri">>, Def),
     Address = pget(<<"dest-address">>, Def),
     Properties =
         rabbit_data_coercion:to_proplist(
@@ -305,7 +316,7 @@ parse_amqp10_dest({_VHost, _Name}, _ClusterName, Def, SourceHeaders) ->
      }.
 
 parse_amqp091_dest({VHost, Name}, ClusterName, Def, SourceHeaders) ->
-    DestURIs  = get_uris(<<"dest-uri">>,      Def),
+    DestURIs  = get_unencrypted_uris(<<"dest-uri">>,      Def),
     DestX     = pget(<<"dest-exchange">>,     Def, none),
     DestXKey  = pget(<<"dest-exchange-key">>, Def, none),
     DestQ     = pget(<<"dest-queue">>,        Def, none),
@@ -373,7 +384,7 @@ parse_amqp091_dest({VHost, Name}, ClusterName, Def, SourceHeaders) ->
                 }, Details).
 
 parse_amqp10_source(Def) ->
-    Uris = get_uris(<<"src-uri">>, Def),
+    Uris = get_unencrypted_uris(<<"src-uri">>, Def),
     Address = pget(<<"src-address">>, Def),
     DeleteAfter = pget(<<"src-delete-after">>, Def, <<"never">>),
     PrefetchCount = pget(<<"src-prefetch-count">>, Def, 1000),
@@ -386,7 +397,7 @@ parse_amqp10_source(Def) ->
        consumer_args => []}, Headers}.
 
 parse_amqp091_source(Def) ->
-    SrcURIs  = get_uris(<<"src-uri">>, Def),
+    SrcURIs  = get_unencrypted_uris(<<"src-uri">>, Def),
     SrcX     = pget(<<"src-exchange">>,Def, none),
     SrcXKey  = pget(<<"src-exchange-key">>, Def, <<>>), %% [1]
     SrcQ     = pget(<<"src-queue">>, Def, none),
@@ -430,6 +441,11 @@ get_uris(Key, Def) ->
            end,
     [binary_to_list(URI) || URI <- URIs].
 
+get_unencrypted_uris(Key, Def) ->
+    ObfuscatedURIs = pget(Key, Def),
+    URIs = [credentials_obfuscation:decrypt(ObfuscatedURI) || ObfuscatedURI <- ObfuscatedURIs],
+    [binary_to_list(URI) || URI <- URIs].
+
 translate_ack_mode(<<"on-confirm">>) -> on_confirm;
 translate_ack_mode(<<"on-publish">>) -> on_publish;
 translate_ack_mode(<<"no-ack">>)     -> no_ack.

deps/rabbitmq_shovel/test/parameters_SUITE.erl

@@ -43,9 +43,18 @@ groups() ->
 %% -------------------------------------------------------------------
 
 init_per_suite(Config) ->
+    {ok, _} = application:ensure_all_started(credentials_obfuscation),
+    Secret = crypto:strong_rand_bytes(128),
+    ok = credentials_obfuscation:set_secret(Secret),
     Config.
 
 end_per_suite(Config) ->
+    case application:stop(credentials_obfuscation) of
+      ok ->
+        ok;
+      {error, {not_started, credentials_obfuscation}} ->
+        ok
+    end,
     Config.
 
 init_per_group(_, Config) ->
@@ -54,9 +63,11 @@ init_per_group(_, Config) ->
 end_per_group(_, Config) ->
     Config.
 
-init_per_testcase(_Testcase, Config) -> Config.
+init_per_testcase(_Testcase, Config) ->
+  Config.
 
-end_per_testcase(_Testcase, Config) -> Config.
+end_per_testcase(_Testcase, Config) ->
+  Config.
 
 
 %% -------------------------------------------------------------------
@@ -140,8 +151,9 @@ parse_amqp091_empty_proplists(_Config) ->
 
 
 test_parse_amqp091(Params) ->
+    ObfuscatedParams = rabbit_shovel_parameters:obfuscate_uris_parameters(Params),
     {ok, Result} = rabbit_shovel_parameters:parse({"vhost", "name"},
-                                                  "my-cluster", Params),
+                                                  "my-cluster", ObfuscatedParams),
     #{ack_mode := on_publish,
       name := "name",
       reconnect_delay := 1001,
@@ -165,8 +177,9 @@ test_parse_amqp091(Params) ->
     ok.
 
 test_parse_amqp091_with_blank_proprties(Params) ->
+    ObfuscatedParams = rabbit_shovel_parameters:obfuscate_uris_parameters(Params),
     {ok, Result} = rabbit_shovel_parameters:parse({"vhost", "name"},
-                                                  "my-cluster", Params),
+                                                  "my-cluster", ObfuscatedParams),
     #{ack_mode := on_publish,
       name := "name",
       reconnect_delay := 1001,
@@ -229,7 +242,7 @@ parse_amqp10(_Config) ->
                                             <<"message-ann-value">>}]},
          {<<"dest-properties">>, [{<<"user_id">>, <<"some-user">>}]}
         ],
-
+    ObfuscatedParams = rabbit_shovel_parameters:obfuscate_uris_parameters(Params),
     ?assertMatch(
        {ok, #{name := "my_shovel",
               ack_mode := on_publish,
@@ -252,7 +265,7 @@ parse_amqp10(_Config) ->
                        }
              }},
         rabbit_shovel_parameters:parse({"vhost", "my_shovel"}, "my-cluster",
-                                       Params)),
+                                       ObfuscatedParams)),
     ok.
 
 parse_amqp10_minimal(_Config) ->
@@ -266,6 +279,7 @@ parse_amqp10_minimal(_Config) ->
          {<<"dest-uri">>, <<"amqp://remotehost:5672">>},
          {<<"dest-address">>, <<"a-dest-queue">>}
         ],
+    ObfuscatedParams = rabbit_shovel_parameters:obfuscate_uris_parameters(Params),
     ?assertMatch(
        {ok, #{name := "my_shovel",
               ack_mode := on_confirm,
@@ -281,7 +295,7 @@ parse_amqp10_minimal(_Config) ->
                        }
              }},
         rabbit_shovel_parameters:parse({"vhost", "my_shovel"}, "my-cluster",
-                                       Params)),
+                                       ObfuscatedParams)),
     ok.
 
 validate_amqp10(_Config) ->