Merge branch 'stable'

Conflicts:
	include/rabbit_cli.hrl
	src/rabbit.erl

4 files changed, 376 insertions, 5 deletions

diff --git a/test/dynamic_ha_SUITE.erl b/test/dynamic_ha_SUITE.erl
index bba7fad707..502e3a7e86 100644
--- a/test/dynamic_ha_SUITE.erl
+++ b/test/dynamic_ha_SUITE.erl
@@ -61,7 +61,8 @@ groups() ->
             ]},
           {cluster_size_3, [], [
               change_policy,
-              rapid_change
+              rapid_change,
+              nodes_policy_should_pick_master_from_its_params
               % FIXME: Re-enable those tests when the know issues are
               % fixed.
               %failing_random_policies,
@@ -258,6 +259,48 @@ promote_on_shutdown(Config) ->
                                  durable = true}),
     ok.
 
+nodes_policy_should_pick_master_from_its_params(Config) ->
+    [A | _] = rabbit_ct_broker_helpers:get_node_configs(Config,
+      nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, A),
+    ?assertEqual(true, apply_policy_to_declared_queue(Config, Ch, [A],
+        [all])),
+    %% --> Master: A
+    %%     Slaves: [B, C] or [C, B]
+    Info = find_queue(?QNAME, A),
+    SSPids = proplists:get_value(synchronised_slave_pids, Info),
+
+    %% Choose slave that isn't the first sync slave. Cover a bug that always
+    %% chose the first, even if it was not part of the policy
+    LastSlave = node(lists:last(SSPids)),
+    ?assertEqual(true, apply_policy_to_declared_queue(Config, Ch, [A],
+        [{nodes, [LastSlave]}])),
+    %% --> Master: B or C (depends on the order of current slaves)
+    %%     Slaves: []
+
+    %% Now choose a new master that isn't synchronised. The previous
+    %% policy made sure that the queue only runs on one node (the last
+    %% from the initial synchronised list). Thus, by taking the first
+    %% node from this list, we know it is not synchronised.
+    %%
+    %% Because the policy doesn't cover any synchronised slave, RabbitMQ
+    %% should instead use an existing synchronised slave as the new master,
+    %% even though that isn't in the policy.
+    ?assertEqual(true, apply_policy_to_declared_queue(Config, Ch, [A],
+        [{nodes, [LastSlave, A]}])),
+    %% --> Master: B or C (same as previous policy)
+    %%     Slaves: [A]
+
+    NewMaster = node(erlang:hd(SSPids)),
+    ?assertEqual(true, apply_policy_to_declared_queue(Config, Ch, [A],
+        [{nodes, [NewMaster]}])),
+    %% --> Master: B or C (the other one compared to previous policy)
+    %%     Slaves: []
+
+    amqp_channel:call(Ch, #'queue.delete'{queue = ?QNAME}),
+    _ = rabbit_ct_broker_helpers:clear_policy(Config, A, ?POLICY).
+
 random_policy(Config) ->
     run_proper(fun prop_random_policy/1, [Config]).
 
@@ -364,9 +407,8 @@ prop_random_policy(Config) ->
        Policies, non_empty(list(policy_gen(Nodes))),
        test_random_policy(Config, Nodes, Policies)).
 
-test_random_policy(Config, Nodes, Policies) ->
+apply_policy_to_declared_queue(Config, Ch, Nodes, Policies) ->
     [NodeA | _] = Nodes,
-    Ch = rabbit_ct_client_helpers:open_channel(Config, NodeA),
     amqp_channel:call(Ch, #'queue.declare'{queue = ?QNAME}),
     %% Add some load so mirrors can be busy synchronising
     rabbit_ct_client_helpers:publish(Ch, ?QNAME, 100000),
@@ -375,7 +417,12 @@ test_random_policy(Config, Nodes, Policies) ->
     %% Give it some time to generate all internal notifications
     timer:sleep(2000),
     %% Check the result
-    Result = wait_for_last_policy(?QNAME, NodeA, Policies, 30),
+    wait_for_last_policy(?QNAME, NodeA, Policies, 30).
+
+test_random_policy(Config, Nodes, Policies) ->
+    [NodeA | _] = Nodes,
+    Ch = rabbit_ct_client_helpers:open_channel(Config, NodeA),
+    Result = apply_policy_to_declared_queue(Config, Ch, Nodes, Policies),
     %% Cleanup
     amqp_channel:call(Ch, #'queue.delete'{queue = ?QNAME}),
     _ = rabbit_ct_broker_helpers:clear_policy(Config, NodeA, ?POLICY),
diff --git a/test/unit_SUITE.erl b/test/unit_SUITE.erl
index 5faeccdaab..b270a3a432 100644
--- a/test/unit_SUITE.erl
+++ b/test/unit_SUITE.erl
@@ -24,7 +24,8 @@
 
 all() ->
     [
-      {group, parallel_tests}
+      {group, parallel_tests},
+      {group, sequential_tests}
     ].
 
 groups() ->
@@ -41,6 +42,10 @@ groups() ->
             ]},
           content_framing,
           content_transcoding,
+          encrypt_decrypt,
+          encrypt_decrypt_term,
+          decrypt_config,
+          rabbitmqctl_encode,
           pg_local,
           pmerge,
           plmerge,
@@ -63,12 +68,36 @@ groups() ->
           {vm_memory_monitor, [parallel], [
               parse_line_linux
             ]}
+        ]},
+      {sequential_tests, [], [
+          decrypt_start_app,
+          decrypt_start_app_file,
+          decrypt_start_app_undefined,
+          decrypt_start_app_wrong_passphrase
         ]}
     ].
 
 init_per_group(_, Config) -> Config.
 end_per_group(_, Config) -> Config.
 
+init_per_testcase(TC, Config) when TC =:= decrypt_start_app;
+                                   TC =:= decrypt_start_app_file;
+                                   TC =:= decrypt_start_app_undefined ->
+    application:load(rabbit),
+    Config;
+init_per_testcase(_, Config) ->
+    Config.
+
+end_per_testcase(TC, _Config) when TC =:= decrypt_start_app;
+                                   TC =:= decrypt_start_app_file;
+                                   TC =:= decrypt_start_app_undefined ->
+    application:unload(rabbit),
+    application:unload(rabbit_shovel_test);
+end_per_testcase(decrypt_config, _Config) ->
+    application:unload(rabbit);
+end_per_testcase(_TC, _Config) ->
+    ok.
+
 %% -------------------------------------------------------------------
 %% Argument parsing.
 %% -------------------------------------------------------------------
@@ -233,6 +262,254 @@ prepend_check(HeaderKey, HeaderTable, Headers) ->
         rabbit_misc:table_lookup(Invalid, HeaderKey),
     Headers1.
 
+encrypt_decrypt(_Config) ->
+    %% Take all available block ciphers.
+    Hashes = rabbit_pbe:supported_hashes(),
+    Ciphers = rabbit_pbe:supported_ciphers(),
+    %% For each cipher, try to encrypt and decrypt data sizes from 0 to 64 bytes
+    %% with a random passphrase.
+    _ = [begin
+        PassPhrase = crypto:strong_rand_bytes(16),
+        Iterations = rand:uniform(100),
+        Data = crypto:strong_rand_bytes(64),
+        [begin
+            Expected = binary:part(Data, 0, Len),
+            Enc = rabbit_pbe:encrypt(C, H, Iterations, PassPhrase, Expected),
+            Expected = iolist_to_binary(rabbit_pbe:decrypt(C, H, Iterations, PassPhrase, Enc))
+        end || Len <- lists:seq(0, byte_size(Data))]
+    end || H <- Hashes, C <- Ciphers],
+    ok.
+
+encrypt_decrypt_term(_Config) ->
+    %% Take all available block ciphers.
+    Hashes = rabbit_pbe:supported_hashes(),
+    Ciphers = rabbit_pbe:supported_ciphers(),
+    %% Different Erlang terms to try encrypting.
+    DataSet = [
+        10000,
+        [5672],
+        [{"127.0.0.1", 5672},
+         {"::1",       5672}],
+        [{connection, info}, {channel, info}],
+        [{cacertfile,           "/path/to/testca/cacert.pem"},
+         {certfile,             "/path/to/server/cert.pem"},
+         {keyfile,              "/path/to/server/key.pem"},
+         {verify,               verify_peer},
+         {fail_if_no_peer_cert, false}],
+        [<<".*">>, <<".*">>, <<".*">>]
+    ],
+    _ = [begin
+        PassPhrase = crypto:strong_rand_bytes(16),
+        Iterations = rand:uniform(100),
+        Enc = rabbit_pbe:encrypt_term(C, H, Iterations, PassPhrase, Data),
+        Data = rabbit_pbe:decrypt_term(C, H, Iterations, PassPhrase, Enc)
+    end || H <- Hashes, C <- Ciphers, Data <- DataSet],
+    ok.
+
+decrypt_config(_Config) ->
+    %% Take all available block ciphers.
+    Hashes = rabbit_pbe:supported_hashes(),
+    Ciphers = rabbit_pbe:supported_ciphers(),
+    Iterations = [1, 10, 100, 1000],
+    %% Loop through all hashes, ciphers and iterations.
+    _ = [begin
+        PassPhrase = crypto:strong_rand_bytes(16),
+        do_decrypt_config({C, H, I, PassPhrase})
+    end || H <- Hashes, C <- Ciphers, I <- Iterations],
+    ok.
+
+do_decrypt_config(Algo = {C, H, I, P}) ->
+    application:load(rabbit),
+    RabbitConfig = application:get_all_env(rabbit),
+    %% Encrypt a few values in configuration.
+    %% Common cases.
+    _ = [encrypt_value(Key, Algo) || Key <- [
+        tcp_listeners,
+        num_tcp_acceptors,
+        ssl_options,
+        vm_memory_high_watermark,
+        default_pass,
+        default_permissions,
+        cluster_nodes,
+        auth_mechanisms,
+        msg_store_credit_disc_bound]],
+    %% Special case: encrypt a value in a list.
+    {ok, [LoopbackUser]} = application:get_env(rabbit, loopback_users),
+    EncLoopbackUser = rabbit_pbe:encrypt_term(C, H, I, P, LoopbackUser),
+    application:set_env(rabbit, loopback_users, [{encrypted, EncLoopbackUser}]),
+    %% Special case: encrypt a value in a key/value list.
+    {ok, TCPOpts} = application:get_env(rabbit, tcp_listen_options),
+    {_, Backlog} = lists:keyfind(backlog, 1, TCPOpts),
+    {_, Linger} = lists:keyfind(linger, 1, TCPOpts),
+    EncBacklog = rabbit_pbe:encrypt_term(C, H, I, P, Backlog),
+    EncLinger = rabbit_pbe:encrypt_term(C, H, I, P, Linger),
+    TCPOpts1 = lists:keyreplace(backlog, 1, TCPOpts, {backlog, {encrypted, EncBacklog}}),
+    TCPOpts2 = lists:keyreplace(linger, 1, TCPOpts1, {linger, {encrypted, EncLinger}}),
+    application:set_env(rabbit, tcp_listen_options, TCPOpts2),
+    %% Decrypt configuration.
+    rabbit:decrypt_config([rabbit], Algo),
+    %% Check that configuration was decrypted properly.
+    RabbitConfig = application:get_all_env(rabbit),
+    application:unload(rabbit),
+    ok.
+
+encrypt_value(Key, {C, H, I, P}) ->
+    {ok, Value} = application:get_env(rabbit, Key),
+    EncValue = rabbit_pbe:encrypt_term(C, H, I, P, Value),
+    application:set_env(rabbit, Key, {encrypted, EncValue}).
+
+decrypt_start_app(Config) ->
+    do_decrypt_start_app(Config, "hello").
+
+decrypt_start_app_file(Config) ->
+    do_decrypt_start_app(Config, {file, ?config(data_dir, Config) ++ "/rabbit_shovel_test.passphrase"}).
+
+do_decrypt_start_app(Config, Passphrase) ->
+    %% Configure rabbit for decrypting configuration.
+    application:set_env(rabbit, decoder_config, [
+        {cipher, aes_cbc256},
+        {hash, sha512},
+        {iterations, 1000},
+        {passphrase, Passphrase}
+    ]),
+    %% Add the path to our test application.
+    code:add_path(?config(data_dir, Config) ++ "/lib/rabbit_shovel_test/ebin"),
+    %% Attempt to start our test application.
+    %%
+    %% We expect a failure *after* the decrypting has been done.
+    try
+        rabbit:start_apps([rabbit_shovel_test])
+    catch _:_ ->
+        ok
+    end,
+    %% Check if the values have been decrypted.
+    {ok, Shovels} = application:get_env(rabbit_shovel_test, shovels),
+    {_, FirstShovel} = lists:keyfind(my_first_shovel, 1, Shovels),
+    {_, Sources} = lists:keyfind(sources, 1, FirstShovel),
+    {_, Brokers} = lists:keyfind(brokers, 1, Sources),
+    ["amqp://fred:secret@host1.domain/my_vhost",
+     "amqp://john:secret@host2.domain/my_vhost"] = Brokers,
+    ok.
+
+decrypt_start_app_undefined(Config) ->
+    %% Configure rabbit for decrypting configuration.
+    application:set_env(rabbit, decoder_config, [
+        {cipher, aes_cbc256},
+        {hash, sha512},
+        {iterations, 1000}
+        %% No passphrase option!
+    ]),
+    %% Add the path to our test application.
+    code:add_path(?config(data_dir, Config) ++ "/lib/rabbit_shovel_test/ebin"),
+    %% Attempt to start our test application.
+    %%
+    %% We expect a failure during decryption because the passphrase is missing.
+    try
+        rabbit:start_apps([rabbit_shovel_test])
+    catch
+        exit:{bad_configuration,decoder_config} -> ok;
+        _:_ -> exit(unexpected_exception)
+    end.
+
+decrypt_start_app_wrong_passphrase(Config) ->
+    %% Configure rabbit for decrypting configuration.
+    application:set_env(rabbit, decoder_config, [
+        {cipher, aes_cbc256},
+        {hash, sha512},
+        {iterations, 1000},
+        {passphrase, "wrong passphrase"}
+    ]),
+    %% Add the path to our test application.
+    code:add_path(?config(data_dir, Config) ++ "/lib/rabbit_shovel_test/ebin"),
+    %% Attempt to start our test application.
+    %%
+    %% We expect a failure during decryption because the passphrase is wrong.
+    try
+        rabbit:start_apps([rabbit_shovel_test])
+    catch
+        exit:{decryption_error,_,_} -> ok;
+        _:_ -> exit(unexpected_exception)
+    end.
+
+rabbitmqctl_encode(_Config) ->
+    % list ciphers and hashes
+    {ok, _} = rabbit_control_pbe:encode(true, false, undefined, undefined, undefined, undefined, undefined),
+    {ok, _} = rabbit_control_pbe:encode(false, true, undefined, undefined, undefined, undefined, undefined),
+    % incorrect ciphers, hashes and iteration number
+    {error, _} = rabbit_control_pbe:encode(false, false, undefined, funny_cipher, undefined, undefined, undefined),
+    {error, _} = rabbit_control_pbe:encode(false, false, undefined, undefined, funny_hash, undefined, undefined),
+    {error, _} = rabbit_control_pbe:encode(false, false, undefined, undefined, undefined, -1, undefined),
+    {error, _} = rabbit_control_pbe:encode(false, false, undefined, undefined, undefined, 0, undefined),
+    % incorrect number of arguments
+    {error, _} = rabbit_control_pbe:encode(
+        false, false,
+        false, % encrypt
+        rabbit_pbe:default_cipher(), rabbit_pbe:default_hash(), rabbit_pbe:default_iterations(),
+        []
+    ),
+    {error, _} = rabbit_control_pbe:encode(
+        false, false,
+        false, % encrypt
+        rabbit_pbe:default_cipher(), rabbit_pbe:default_hash(), rabbit_pbe:default_iterations(),
+        [undefined]
+    ),
+    {error, _} = rabbit_control_pbe:encode(
+        false, false,
+        false, % encrypt
+        rabbit_pbe:default_cipher(), rabbit_pbe:default_hash(), rabbit_pbe:default_iterations(),
+        [undefined, undefined, undefined]
+    ),
+
+    % encrypt/decrypt
+    % string
+    rabbitmqctl_encode_encrypt_decrypt("foobar"),
+    % binary
+    rabbitmqctl_encode_encrypt_decrypt("<<\"foobar\">>"),
+    % tuple
+    rabbitmqctl_encode_encrypt_decrypt("{password,<<\"secret\">>}"),
+
+    ok.
+
+rabbitmqctl_encode_encrypt_decrypt(Secret) ->
+    PassPhrase = "passphrase",
+    {ok, Output} = rabbit_control_pbe:encode(
+        false, false,
+        false, % encrypt
+        rabbit_pbe:default_cipher(), rabbit_pbe:default_hash(), rabbit_pbe:default_iterations(),
+        [Secret, PassPhrase]
+    ),
+    {encrypted, Encrypted} = rabbit_control_pbe:evaluate_input_as_term(lists:flatten(Output)),
+
+    {ok, Result} = rabbit_control_pbe:encode(
+        false, false,
+        true, % decrypt
+        rabbit_pbe:default_cipher(), rabbit_pbe:default_hash(), rabbit_pbe:default_iterations(),
+        [lists:flatten(io_lib:format("~p", [Encrypted])), PassPhrase]
+    ),
+    Secret = lists:flatten(Result),
+    % decrypt with {encrypted, ...} form as input
+    {ok, Result} = rabbit_control_pbe:encode(
+        false, false,
+        true, % decrypt
+        rabbit_pbe:default_cipher(), rabbit_pbe:default_hash(), rabbit_pbe:default_iterations(),
+        [lists:flatten(io_lib:format("~p", [{encrypted, Encrypted}])), PassPhrase]
+    ),
+
+    % wrong passphrase
+    {error, _} = rabbit_control_pbe:encode(
+        false, false,
+        true, % decrypt
+        rabbit_pbe:default_cipher(), rabbit_pbe:default_hash(), rabbit_pbe:default_iterations(),
+        [lists:flatten(io_lib:format("~p", [Encrypted])), PassPhrase ++ " "]
+    ),
+    {error, _} = rabbit_control_pbe:encode(
+        false, false,
+        true, % decrypt
+        rabbit_pbe:default_cipher(), rabbit_pbe:default_hash(), rabbit_pbe:default_iterations(),
+        [lists:flatten(io_lib:format("~p", [{encrypted, Encrypted}])), PassPhrase ++ " "]
+    )
+    .
+
 %% -------------------------------------------------------------------
 %% pg_local.
 %% -------------------------------------------------------------------
diff --git a/test/unit_SUITE_data/lib/rabbit_shovel_test/ebin/rabbit_shovel_test.app b/test/unit_SUITE_data/lib/rabbit_shovel_test/ebin/rabbit_shovel_test.app
new file mode 100644
index 0000000000..a8481c9aa4
--- /dev/null
+++ b/test/unit_SUITE_data/lib/rabbit_shovel_test/ebin/rabbit_shovel_test.app
@@ -0,0 +1,46 @@
+{application, rabbit_shovel_test,
+ [{description, "Test .app file for tests for encrypting configuration"},
+  {vsn, ""},
+  {modules, []},
+  {env, [ {shovels, [ {my_first_shovel,
+                    [ {sources,
+                        [ {brokers, [ {encrypted, <<"CfJXuka/uJYsqAtiJnwKpSY4moMPcOBh4sO8XDcdmhXbVYGKCDLKEilWPMfvOAQ2lN1BQneGn6bvDZi2+gDu6iHVKfafQAZSv8zcsVB3uYdBXFzqTCWO8TAsgG6LUMPT">>}
+                                    , {encrypted, <<"dBO6n+G1OiBwZeLXhvmNYeTE57nhBOmicUBF34zo4nQjerzQaNoEk8GA2Ts5PzMhYeO6U6Y9eEmheqIr9Gzh2duLZic65ZMQtIKNpWcZJllEhGpk7aV1COr23Yur9fWG">>}
+                                    ]}
+                        , {declarations, [ {'exchange.declare',
+                                              [ {exchange, <<"my_fanout">>}
+                                              , {type, <<"fanout">>}
+                                              , durable
+                                              ]}
+                                         , {'queue.declare',
+                                              [{arguments,
+                                                 [{<<"x-message-ttl">>, long, 60000}]}]}
+                                         , {'queue.bind',
+                                              [ {exchange, <<"my_direct">>}
+                                              , {queue,    <<>>}
+                                              ]}
+                                         ]}
+                        ]}
+                    , {destinations,
+                        [ {broker, "amqp://"}
+                        , {declarations, [ {'exchange.declare',
+                                              [ {exchange, <<"my_direct">>}
+                                              , {type, <<"direct">>}
+                                              , durable
+                                              ]}
+                                         ]}
+                        ]}
+                    , {queue, <<>>}
+                    , {prefetch_count, 10}
+                    , {ack_mode, on_confirm}
+                    , {publish_properties, [ {delivery_mode, 2} ]}
+                    , {add_forward_headers, true}
+                    , {publish_fields, [ {exchange, <<"my_direct">>}
+                                       , {routing_key, <<"from_shovel">>}
+                                       ]}
+                    , {reconnect_delay, 5}
+                    ]}
+                ]}
+    ]},
+
+  {applications, [kernel, stdlib]}]}.
diff --git a/test/unit_SUITE_data/rabbit_shovel_test.passphrase b/test/unit_SUITE_data/rabbit_shovel_test.passphrase
new file mode 100644
index 0000000000..ce01362503
--- /dev/null
+++ b/test/unit_SUITE_data/rabbit_shovel_test.passphrase
@@ -0,0 +1 @@
+hello