Merge pull request #94 from rabbitmq/rabbitmq-erlang-client-123-crypted-password-in-state

Obfuscate credentials

deps/rabbitmq_federation/src/rabbit_federation_exchange_link.erl

@@ -75,7 +75,9 @@ init({Upstream, XName}) ->
     %% before we got here. So check if we still exist.
     case rabbit_exchange:lookup(XName) of
         {ok, X} ->
-            UParams = rabbit_federation_upstream:to_params(Upstream, X),
+            DeobfuscatedUpstream = rabbit_federation_util:deobfuscate_upstream(Upstream),
+            DeobfuscatedUParams = rabbit_federation_upstream:to_params(DeobfuscatedUpstream, X),
+            UParams = rabbit_federation_util:obfuscate_upstream_params(DeobfuscatedUParams),
             rabbit_federation_status:report(Upstream, UParams, XName, starting),
             join(rabbit_federation_exchanges),
             join({rabbit_federation_exchange, XName}),
@@ -527,7 +529,7 @@ ensure_internal_exchange(IntXNameBin,
                                 upstream_params = UParams,
                                 connection      = Conn,
                                 channel         = Ch}) ->
-    #upstream_params{params = Params} = UParams,
+    #upstream_params{params = Params} = rabbit_federation_util:deobfuscate_upstream_params(UParams),
     delete_upstream_exchange(Conn, IntXNameBin),
     Base = #'exchange.declare'{exchange    = IntXNameBin,
                                durable     = true,
@@ -550,7 +552,8 @@ check_internal_exchange(IntXNameBin,
                          #state{upstream        = #upstream{max_hops = MaxHops},
                                 upstream_params = UParams,
                                 downstream_exchange = XName}) ->
-    #upstream_params{params = Params} = UParams,
+    #upstream_params{params = Params} =
+        rabbit_federation_util:deobfuscate_upstream_params(UParams),
     Base = #'exchange.declare'{exchange    = IntXNameBin,
                                passive     = true,
                                durable     = true,

deps/rabbitmq_federation/src/rabbit_federation_link_sup.erl

@@ -77,7 +77,7 @@ restart(Sup, Upstream) ->
     ok.
 
 start(Sup, Upstream, XorQ) ->
-    {ok, _Pid} = supervisor2:start_child(Sup, spec(Upstream, XorQ)),
+    {ok, _Pid} = supervisor2:start_child(Sup, spec(rabbit_federation_util:obfuscate_upstream(Upstream), XorQ)),
     ok.
 
 stop(Sup, Upstream, XorQ) ->
@@ -102,7 +102,8 @@ init(XorQ) ->
     {ok, {{one_for_one, 1, ?MAX_WAIT}, specs(XorQ)}}.
 
 specs(XorQ) ->
-    [spec(Upstream, XorQ) || Upstream <- rabbit_federation_upstream:for(XorQ)].
+    [spec(rabbit_federation_util:obfuscate_upstream(Upstream), XorQ)
+     || Upstream <- rabbit_federation_upstream:for(XorQ)].
 
 spec(U = #upstream{reconnect_delay = Delay}, #exchange{name = XName}) ->
     {U, {rabbit_federation_exchange_link, start_link, [{U, XName}]},

deps/rabbitmq_federation/src/rabbit_federation_link_util.erl

@@ -36,8 +36,12 @@
 
 %%----------------------------------------------------------------------------
 
-start_conn_ch(Fun, Upstream, UParams,
+start_conn_ch(Fun, OUpstream, OUParams,
               XorQName = #resource{virtual_host = DownVHost}, State) ->
+
+    Upstream = rabbit_federation_util:deobfuscate_upstream(OUpstream),
+    UParams = rabbit_federation_util:deobfuscate_upstream_params(OUParams),
+
     ConnName = get_connection_name(Upstream, UParams),
     case open_monitor(#amqp_params_direct{virtual_host = DownVHost}, ConnName) of
         {ok, DConn, DCh} ->
@@ -65,7 +69,7 @@ start_conn_ch(Fun, Upstream, UParams,
                              UParams)]),
                         Name = pget(name, amqp_connection:info(DConn, [name])),
                         rabbit_federation_status:report(
-                          Upstream, UParams, XorQName, {running, Name}),
+                          OUpstream, OUParams, XorQName, {running, Name}),
                         R
                     catch exit:E ->
                             %% terminate/2 will not get this, as we
@@ -73,16 +77,16 @@ start_conn_ch(Fun, Upstream, UParams,
                             ensure_connection_closed(DConn),
                             ensure_connection_closed(Conn),
                             connection_error(remote_start, E,
-                                             Upstream, UParams, XorQName, State)
+                                             OUpstream, OUParams, XorQName, State)
                     end;
                 E ->
                     ensure_connection_closed(DConn),
                     connection_error(remote_start, E,
-                                     Upstream, UParams, XorQName, State)
+                                     OUpstream, OUParams, XorQName, State)
             end;
         E ->
             connection_error(local_start, E,
-                             Upstream, UParams, XorQName, State)
+                             OUpstream, OUParams, XorQName, State)
     end.
 
 get_connection_name(#upstream{name = UpstreamName},

deps/rabbitmq_federation/src/rabbit_federation_queue_link.erl

@@ -68,7 +68,9 @@ init({Upstream, Queue}) when ?is_amqqueue(Queue) ->
     QName = amqqueue:get_name(Queue),
     case rabbit_amqqueue:lookup(QName) of
         {ok, Q} ->
-            UParams = rabbit_federation_upstream:to_params(Upstream, Queue),
+            DeobfuscatedUpstream = rabbit_federation_util:deobfuscate_upstream(Upstream),
+            DeobfuscatedUParams = rabbit_federation_upstream:to_params(DeobfuscatedUpstream, Queue),
+            UParams = rabbit_federation_util:obfuscate_upstream_params(DeobfuscatedUParams),
             rabbit_federation_status:report(Upstream, UParams, QName, starting),
             join(rabbit_federation_queues),
             join({rabbit_federation_queue, QName}),

deps/rabbitmq_federation/src/rabbit_federation_status.erl

@@ -74,7 +74,8 @@ handle_call({remove, Upstream, XorQName}, _From, State) ->
 
 handle_call({lookup, Id}, _From, State) ->
     Link = case ets:match_object(?ETS_NAME, match_id(Id)) of
-               [Entry] -> [{key, Entry#entry.key},
+               [Entry] ->
+                   [{key, Entry#entry.key},
                            {uri, Entry#entry.uri},
                            {status, Entry#entry.status},
                            {timestamp, Entry#entry.timestamp},

deps/rabbitmq_federation/src/rabbit_federation_util.erl

@@ -22,6 +22,7 @@
 
 -export([should_forward/3, find_upstreams/2, already_seen/2]).
 -export([validate_arg/3, fail/2, name/1, vhost/1, r/1, pgname/1]).
+-export([obfuscate_upstream/1, deobfuscate_upstream/1, obfuscate_upstream_params/1, deobfuscate_upstream_params/1]).
 
 -import(rabbit_misc, [pget_or_die/2, pget/3]).
 
@@ -74,3 +75,31 @@ pgname(Name) ->
         {ok, false} -> Name;
         {ok, true}  -> {rabbit_nodes:cluster_name(), Name}
     end.
+
+obfuscate_upstream(#upstream{uris = Uris} = Upstream) ->
+    Upstream#upstream{uris = [credentials_obfuscation:encrypt(Uri) || Uri <- Uris]}.
+
+obfuscate_upstream_params(#upstream_params{uri = Uri, params = #amqp_params_network{password = Password} = Params} = UParams) ->
+    UParams#upstream_params{
+        uri = credentials_obfuscation:encrypt(Uri),
+        params = Params#amqp_params_network{password = credentials_obfuscation:encrypt(Password)}
+    };
+obfuscate_upstream_params(#upstream_params{uri = Uri, params = #amqp_params_direct{password = Password} = Params} = UParams) ->
+    UParams#upstream_params{
+        uri = credentials_obfuscation:encrypt(Uri),
+        params = Params#amqp_params_direct{password = credentials_obfuscation:encrypt(Password)}
+    }.
+
+deobfuscate_upstream(#upstream{uris = EncryptedUris} = Upstream) ->
+    Upstream#upstream{uris = [credentials_obfuscation:decrypt(EncryptedUri) || EncryptedUri <- EncryptedUris]}.
+
+deobfuscate_upstream_params(#upstream_params{uri = EncryptedUri, params = #amqp_params_network{password = EncryptedPassword} = Params} = UParams) ->
+    UParams#upstream_params{
+        uri = credentials_obfuscation:decrypt(EncryptedUri),
+        params = Params#amqp_params_network{password = credentials_obfuscation:decrypt(EncryptedPassword)}
+    };
+deobfuscate_upstream_params(#upstream_params{uri = EncryptedUri, params = #amqp_params_direct{password = EncryptedPassword} = Params} = UParams) ->
+    UParams#upstream_params{
+        uri = credentials_obfuscation:decrypt(EncryptedUri),
+        params = Params#amqp_params_direct{password = credentials_obfuscation:decrypt(EncryptedPassword)}
+    }.

deps/rabbitmq_federation/test/unit_SUITE.erl

@@ -0,0 +1,55 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License at
+%% https://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+%% License for the specific language governing rights and limitations
+%% under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2019 Pivotal Software, Inc.  All rights reserved.
+%%
+
+-module(unit_SUITE).
+-include_lib("common_test/include/ct.hrl").
+-include_lib("amqp_client/include/amqp_client.hrl").
+-include("rabbit_federation.hrl").
+
+-compile(export_all).
+ 
+all() -> [obfuscate_upstream, obfuscate_upstream_params_network, obfuscate_upstream_params_direct].
+ 
+init_per_suite(Config) ->
+    application:ensure_all_started(credentials_obfuscation),
+    Config.
+
+end_per_suite(Config) ->
+    Config.
+
+obfuscate_upstream(_Config) ->
+    Upstream = #upstream{uris = [<<"amqp://guest:password@localhost">>]},
+    ObfuscatedUpstream = rabbit_federation_util:obfuscate_upstream(Upstream),
+    Upstream = rabbit_federation_util:deobfuscate_upstream(ObfuscatedUpstream),
+    ok.
+
+obfuscate_upstream_params_network(_Config) ->
+    UpstreamParams = #upstream_params{
+        uri = <<"amqp://guest:password@localhost">>, 
+        params = #amqp_params_network{password = <<"password">>}
+    },
+    ObfuscatedUpstreamParams = rabbit_federation_util:obfuscate_upstream_params(UpstreamParams),
+    UpstreamParams = rabbit_federation_util:deobfuscate_upstream_params(ObfuscatedUpstreamParams),
+    ok.
+
+ obfuscate_upstream_params_direct(_Config) ->
+    UpstreamParams = #upstream_params{
+        uri = <<"amqp://guest:password@localhost">>, 
+        params = #amqp_params_direct{password = <<"password">>}
+    },
+    ObfuscatedUpstreamParams = rabbit_federation_util:obfuscate_upstream_params(UpstreamParams),
+    UpstreamParams = rabbit_federation_util:deobfuscate_upstream_params(ObfuscatedUpstreamParams),
+    ok.