Merge branch 'master' into rabbitmq-server-567

6 files changed, 169 insertions, 212 deletions

diff --git a/src/rabbit_control_main.erl b/src/rabbit_control_main.erl
index 1619f25494..babaf4fd4e 100644
--- a/src/rabbit_control_main.erl
+++ b/src/rabbit_control_main.erl
@@ -537,7 +537,7 @@ action(set_policy, Node, [Key, Pattern, Defn], Opts, Inform) ->
     Inform(Msg, [Key, Pattern, Defn, PriorityArg]),
     Res = rpc_call(
       Node, rabbit_policy, parse_set,
-      [VHostArg, list_to_binary(Key), Pattern, Defn, PriorityArg, ApplyToArg]),
+      [VHostArg, list_to_binary(Key), list_to_binary(Pattern), list_to_binary(Defn), list_to_binary(PriorityArg), ApplyToArg]),
     case Res of
         {error, Format, Args} when is_list(Format) andalso is_list(Args) ->
             {error_string, rabbit_misc:format(Format, Args)};
@@ -558,7 +558,7 @@ action(set_operator_policy, Node, [Key, Pattern, Defn], Opts, Inform) ->
     Inform(Msg, [Key, Pattern, Defn, PriorityArg]),
     Res = rpc_call(
       Node, rabbit_policy, parse_set_op,
-      [VHostArg, list_to_binary(Key), Pattern, Defn, PriorityArg, ApplyToArg]),
+      [VHostArg, list_to_binary(Key), list_to_binary(Pattern), list_to_binary(Defn), list_to_binary(PriorityArg), ApplyToArg]),
     case Res of
         {error, Format, Args} when is_list(Format) andalso is_list(Args) ->
             {error_string, rabbit_misc:format(Format, Args)};
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 1ec9a46880..51deed8597 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -98,11 +98,10 @@ init() ->
     ensure_mnesia_dir(),
     case is_virgin_node() of
         true  ->
-            rabbit_log:info("Database directory at ~s is empty. "
+            rabbit_log:info("Node database directory at ~s is empty. "
                             "Assuming we need to join an existing cluster or initialise from scratch...~n",
                             [dir()]),
-            rabbit_log:info("Using ~p as peer discovery backend~n",
-                            [rabbit_peer_discovery:backend()]),
+             rabbit_peer_discovery:log_configured_backend(),
             init_from_config();
         false ->
             NodeType = node_type(),
@@ -141,7 +140,9 @@ init_from_config() ->
                 e(invalid_cluster_nodes_conf)
         end,
     case DiscoveredNodes of
-        [] -> init_db_and_upgrade([node()], disc, false, _Retry = true);
+        [] ->
+            rabbit_log:info("Discovered no peer nodes to cluster with"),
+            init_db_and_upgrade([node()], disc, false, _Retry = true);
         _  ->
             rabbit_log:info("Discovered peer nodes: ~s~n",
                 [rabbit_peer_discovery:format_discovered_nodes(DiscoveredNodes)]),
@@ -158,8 +159,9 @@ auto_cluster(TryNodes, NodeType) ->
             rabbit_node_monitor:notify_joined_cluster();
         none ->
             rabbit_log:warning(
-              "Could not find any node for auto-clustering from: ~p~n"
-              "Starting blank node...~n", [TryNodes]),
+              "Could not successfully contact any node of: ~s (as in Erlang distribution). "
+               "Starting as a blank standalone node...~n",
+                [string:join(lists:map(fun atom_to_list/1, TryNodes), ",")]),
             init_db_and_upgrade([node()], disc, false, _Retry = true)
     end.
 
diff --git a/src/rabbit_pbe.erl b/src/rabbit_pbe.erl
deleted file mode 100644
index f4998d4a13..0000000000
--- a/src/rabbit_pbe.erl
+++ /dev/null
@@ -1,194 +0,0 @@
-%% 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 http://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) 2007-2016 Pivotal Software, Inc.  All rights reserved.
-%%
-
--module(rabbit_pbe).
-
--export([supported_ciphers/0, supported_hashes/0, default_cipher/0, default_hash/0, default_iterations/0]).
--export([encrypt_term/5, decrypt_term/5]).
--export([encrypt/5, decrypt/5]).
-
-%% Supported ciphers and hashes
-
-supported_ciphers() ->
-    proplists:get_value(ciphers, crypto:supports())
-        -- [aes_ctr, aes_ecb, des_ecb, blowfish_ecb, rc4, aes_gcm].
-
-supported_hashes() ->
-    proplists:get_value(hashs, crypto:supports())
-        -- [md4, ripemd160].
-
-%% Default encryption parameters (keep those in sync with rabbit.app.src)
-default_cipher() ->
-    aes_cbc256.
-
-default_hash() ->
-    sha512.
-
-default_iterations() ->
-    1000.
-
-%% Encryption/decryption of arbitrary Erlang terms.
-
-encrypt_term(Cipher, Hash, Iterations, PassPhrase, Term) ->
-    encrypt(Cipher, Hash, Iterations, PassPhrase, term_to_binary(Term)).
-
-decrypt_term(Cipher, Hash, Iterations, PassPhrase, Base64Binary) ->
-    binary_to_term(decrypt(Cipher, Hash, Iterations, PassPhrase, Base64Binary)).
-
-%% The cipher for encryption is from the list of supported ciphers.
-%% The hash for generating the key from the passphrase is from the list
-%% of supported hashes. See crypto:supports/0 to obtain both lists.
-%% The key is generated by applying the hash N times with N >= 1.
-%%
-%% The encrypt/5 function returns a base64 binary and the decrypt/5
-%% function accepts that same base64 binary.
-
--spec encrypt(crypto:block_cipher(), crypto:hash_algorithms(),
-    pos_integer(), iodata(), binary()) -> binary().
-encrypt(Cipher, Hash, Iterations, PassPhrase, ClearText) ->
-    Salt = crypto:strong_rand_bytes(16),
-    Ivec = crypto:strong_rand_bytes(iv_length(Cipher)),
-    Key = make_key(Cipher, Hash, Iterations, PassPhrase, Salt),
-    Binary = crypto:block_encrypt(Cipher, Key, Ivec, pad(Cipher, ClearText)),
-    base64:encode(<< Salt/binary, Ivec/binary, Binary/binary >>).
-
--spec decrypt(crypto:block_cipher(), crypto:hash_algorithms(),
-    pos_integer(), iodata(), binary()) -> binary().
-decrypt(Cipher, Hash, Iterations, PassPhrase, Base64Binary) ->
-    IvLength = iv_length(Cipher),
-    << Salt:16/binary, Ivec:IvLength/binary, Binary/bits >> = base64:decode(Base64Binary),
-    Key = make_key(Cipher, Hash, Iterations, PassPhrase, Salt),
-    unpad(crypto:block_decrypt(Cipher, Key, Ivec, Binary)).
-
-%% Generate a key from a passphrase.
-
-make_key(Cipher, Hash, Iterations, PassPhrase, Salt) ->
-    Key = pbdkdf2(PassPhrase, Salt, Iterations, key_length(Cipher),
-        fun crypto:hmac/4, Hash, hash_length(Hash)),
-    if
-        Cipher =:= des3_cbc; Cipher =:= des3_cbf; Cipher =:= des3_cfb; Cipher =:= des_ede3 ->
-            << A:8/binary, B:8/binary, C:8/binary >> = Key,
-            [A, B, C];
-        true ->
-            Key
-    end.
-
-%% Functions to pad/unpad input to a multiplier of block size.
-
-pad(Cipher, Data) ->
-    BlockSize = block_size(Cipher),
-    N = BlockSize - (byte_size(Data) rem BlockSize),
-    Pad = list_to_binary(lists:duplicate(N, N)),
-    <<Data/binary, Pad/binary>>.
-
-unpad(Data) ->
-    N = binary:last(Data),
-    binary:part(Data, 0, byte_size(Data) - N).
-
-%% These functions are necessary because the current Erlang crypto interface
-%% is lacking interfaces to the following OpenSSL functions:
-%%
-%% * int EVP_MD_size(const EVP_MD *md);
-%% * int EVP_CIPHER_iv_length(const EVP_CIPHER *e);
-%% * int EVP_CIPHER_key_length(const EVP_CIPHER *e);
-%% * int EVP_CIPHER_block_size(const EVP_CIPHER *e);
-
-hash_length(md4) -> 16;
-hash_length(md5) -> 16;
-hash_length(sha) -> 20;
-hash_length(sha224) -> 28;
-hash_length(sha256) -> 32;
-hash_length(sha384) -> 48;
-hash_length(sha512) -> 64.
-
-iv_length(des_cbc) -> 8;
-iv_length(des_cfb) -> 8;
-iv_length(des3_cbc) -> 8;
-iv_length(des3_cbf) -> 8;
-iv_length(des3_cfb) -> 8;
-iv_length(des_ede3) -> 8;
-iv_length(blowfish_cbc) -> 8;
-iv_length(blowfish_cfb64) -> 8;
-iv_length(blowfish_ofb64) -> 8;
-iv_length(rc2_cbc) -> 8;
-iv_length(aes_cbc) -> 16;
-iv_length(aes_cbc128) -> 16;
-iv_length(aes_cfb8) -> 16;
-iv_length(aes_cfb128) -> 16;
-iv_length(aes_cbc256) -> 16;
-iv_length(aes_ige256) -> 32.
-
-key_length(des_cbc) -> 8;
-key_length(des_cfb) -> 8;
-key_length(des3_cbc) -> 24;
-key_length(des3_cbf) -> 24;
-key_length(des3_cfb) -> 24;
-key_length(des_ede3) -> 24;
-key_length(blowfish_cbc) -> 16;
-key_length(blowfish_cfb64) -> 16;
-key_length(blowfish_ofb64) -> 16;
-key_length(rc2_cbc) -> 16;
-key_length(aes_cbc) -> 16;
-key_length(aes_cbc128) -> 16;
-key_length(aes_cfb8) -> 16;
-key_length(aes_cfb128) -> 16;
-key_length(aes_cbc256) -> 32;
-key_length(aes_ige256) -> 16.
-
-block_size(aes_cbc256) -> 32;
-block_size(aes_cbc128) -> 32;
-block_size(aes_ige256) -> 32;
-block_size(aes_cbc) -> 32;
-block_size(_) -> 8.
-
-%% The following was taken from OTP's lib/public_key/src/pubkey_pbe.erl
-%%
-%% This is an undocumented interface to password-based encryption algorithms.
-%% These functions have been copied here to stay compatible with R16B03.
-
-%%--------------------------------------------------------------------
--spec pbdkdf2(string(), iodata(), integer(), integer(), fun(), atom(), integer())
-	     -> binary().
-%%
-%% Description: Implements password based decryption key derive function 2.
-%% Exported mainly for testing purposes.
-%%--------------------------------------------------------------------
-pbdkdf2(Password, Salt, Count, DerivedKeyLen, Prf, PrfHash, PrfOutputLen)->
-    NumBlocks = ceiling(DerivedKeyLen / PrfOutputLen),
-    NumLastBlockOctets = DerivedKeyLen - (NumBlocks - 1) * PrfOutputLen ,
-    blocks(NumBlocks, NumLastBlockOctets, 1, Password, Salt,
-	   Count, Prf, PrfHash, PrfOutputLen, <<>>).
-
-blocks(1, N, Index, Password, Salt, Count, Prf, PrfHash, PrfLen, Acc) ->
-    <<XorSum:N/binary, _/binary>> = xor_sum(Password, Salt, Count, Index, Prf, PrfHash, PrfLen),
-    <<Acc/binary, XorSum/binary>>;
-blocks(NumBlocks, N, Index, Password, Salt, Count, Prf, PrfHash, PrfLen, Acc) ->
-    XorSum = xor_sum(Password, Salt, Count, Index, Prf, PrfHash, PrfLen),
-    blocks(NumBlocks -1, N, Index +1, Password, Salt, Count, Prf, PrfHash,
-	   PrfLen, <<Acc/binary, XorSum/binary>>).
-
-xor_sum(Password, Salt, Count, Index, Prf, PrfHash, PrfLen) ->
-    Result = Prf(PrfHash, Password, [Salt,<<Index:32/unsigned-big-integer>>], PrfLen),
-    do_xor_sum(Prf, PrfHash, PrfLen, Result, Password, Count-1, Result).
-
-do_xor_sum(_, _, _, _, _, 0, Acc) ->
-    Acc;
-do_xor_sum(Prf, PrfHash, PrfLen, Prev, Password, Count, Acc) ->
-    Result = Prf(PrfHash, Password, Prev, PrfLen),
-    do_xor_sum(Prf, PrfHash, PrfLen, Result, Password, Count-1, crypto:exor(Acc, Result)).
-
-ceiling(Float) ->
-    erlang:round(Float + 0.5).
diff --git a/src/rabbit_peer_discovery.erl b/src/rabbit_peer_discovery.erl
index 965be3946d..c0b554e3b3 100644
--- a/src/rabbit_peer_discovery.erl
+++ b/src/rabbit_peer_discovery.erl
@@ -16,21 +16,53 @@
 
 -module(rabbit_peer_discovery).
 
+%%
 %% API
--export([discover_cluster_nodes/0, backend/0,
-         normalize/1, format_discovered_nodes/1]).
+%%
+
+-export([discover_cluster_nodes/0, backend/0, node_type/0,
+         normalize/1, format_discovered_nodes/1, log_configured_backend/0]).
+-export([append_node_prefix/1, node_prefix/0]).
 
+-define(DEFAULT_BACKEND,   rabbit_peer_discovery_classic_config).
+%% what node type is used by default for this node when joining
+%% a new cluster as a virgin node
+-define(DEFAULT_NODE_TYPE, disc).
+%% default node prefix to attach to discovered hostnames
+-define(DEFAULT_PREFIX, "rabbit").
+-define(NODENAME_PART_SEPARATOR, "@").
 
 
 -spec backend() -> atom().
 
 backend() ->
-  case application:get_env(rabbit, peer_discovery_backend) of
-    {ok, Backend} when is_atom(Backend) -> Backend;
-    undefined                           -> rabbit_peer_discovery_classic_config
+  case application:get_env(rabbit, autocluster) of
+    {ok, Proplist} ->
+      proplists:get_value(peer_discovery_backend, Proplist, ?DEFAULT_BACKEND);
+    undefined      ->
+      ?DEFAULT_BACKEND
   end.
 
 
+
+-spec node_type() -> rabbit_types:node_type().
+
+node_type() ->
+  case application:get_env(rabbit, autocluster) of
+    {ok, Proplist} ->
+      proplists:get_value(node_type, Proplist, ?DEFAULT_NODE_TYPE);
+    undefined      ->
+      ?DEFAULT_NODE_TYPE
+  end.
+
+
+
+-spec log_configured_backend() -> ok.
+
+log_configured_backend() ->
+  rabbit_log:info("Configured peer discovery backend: ~s~n", [backend()]).
+
+
 -spec discover_cluster_nodes() -> {ok, Nodes :: list()} |
                                   {ok, {Nodes :: list(), NodeType :: rabbit_types:node_type()}} |
                                   {error, Reason :: string()}.
@@ -40,11 +72,17 @@ discover_cluster_nodes() ->
     normalize(Backend:list_nodes()).
 
 
--spec normalize({ok, Nodes :: list()} |
+-spec normalize(Nodes :: list() |
+                {Nodes :: list(), NodeType :: rabbit_types:node_type()} |
+                {ok, Nodes :: list()} |
                 {ok, {Nodes :: list(), NodeType :: rabbit_types:node_type()}} |
                 {error, Reason :: string()}) -> {ok, {Nodes :: list(), NodeType :: rabbit_types:node_type()}} |
                                                 {error, Reason :: string()}.
 
+normalize(Nodes) when is_list(Nodes) ->
+  {ok, {Nodes, disc}};
+normalize({Nodes, NodeType}) when is_list(Nodes) andalso is_atom(NodeType) ->
+  {ok, {Nodes, NodeType}};
 normalize({ok, Nodes}) when is_list(Nodes) ->
   {ok, {Nodes, disc}};
 normalize({ok, {Nodes, NodeType}}) when is_list(Nodes) andalso is_atom(NodeType) ->
@@ -57,3 +95,27 @@ normalize({error, Reason}) ->
 
 format_discovered_nodes(Nodes) ->
   string:join(lists:map(fun (Val) -> hd(io_lib:format("~s", [Val])) end, Nodes), ", ").
+
+
+
+-spec node_prefix() -> string().
+
+node_prefix() ->
+    case string:tokens(atom_to_list(node()), ?NODENAME_PART_SEPARATOR) of
+        [Prefix, _] -> Prefix;
+        [_]         -> ?DEFAULT_PREFIX
+    end.
+
+
+
+-spec append_node_prefix(Value :: binary() | list()) -> atom().
+
+append_node_prefix(Value) ->
+    Val = rabbit_data_coercion:to_list(Value),
+    Hostname = case string:tokens(Val, ?NODENAME_PART_SEPARATOR) of
+                   [_ExistingPrefix, Val] ->
+                       Val;
+                   [Val]                  ->
+                       Val
+               end,
+    string:join([node_prefix(), Hostname], ?NODENAME_PART_SEPARATOR).
diff --git a/src/rabbit_peer_discovery_dns.erl b/src/rabbit_peer_discovery_dns.erl
new file mode 100644
index 0000000000..554c914466
--- /dev/null
+++ b/src/rabbit_peer_discovery_dns.erl
@@ -0,0 +1,88 @@
+%% 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
+%% http://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) 2007-2016 Pivotal Software, Inc.  All rights reserved.
+%%
+
+-module(rabbit_peer_discovery_dns).
+-behaviour(rabbit_peer_discovery_backend).
+
+-include("rabbit.hrl").
+
+-export([list_nodes/0, register/0, unregister/0]).
+%% for tests
+-export([discover_nodes/2, discover_hostnames/2]).
+
+%%
+%% API
+%%
+
+-spec list_nodes() -> {ok, Nodes :: list()} | {error, Reason :: string()}.
+
+list_nodes() ->
+    case application:get_env(rabbit, autocluster) of
+      undefined         ->
+        {[], disc};
+      {ok, Autocluster} ->
+        case proplists:get_value(peer_discovery_dns, Autocluster) of
+            undefined ->
+              rabbit_log:warning("Peer discovery backend is set to ~s "
+                                 "but final config does not contain rabbit.autocluster.peer_discovery_dns. "
+                                 "Cannot discover any nodes because seed hostname is not configured!",
+                                 [?MODULE]),
+              {[], disc};
+            Proplist  ->
+              Hostname = rabbit_data_coercion:to_list(proplists:get_value(hostname, Proplist)),
+
+              {discover_nodes(Hostname, net_kernel:longnames()), rabbit_peer_discovery:node_type()}
+        end
+    end.
+
+-spec register() -> ok.
+
+register() ->
+    ok.
+
+-spec unregister() -> ok.
+
+unregister() ->
+    ok.
+
+
+%%
+%% Implementation
+%%
+
+discover_nodes(SeedHostname, LongNamesUsed) ->
+    [list_to_atom(rabbit_peer_discovery:append_node_prefix(H)) ||
+        H <- discover_hostnames(SeedHostname, LongNamesUsed)].
+
+discover_hostnames(SeedHostname, LongNamesUsed) ->
+    %% TODO: IPv6 support
+    IPs   = inet_res:lookup(SeedHostname, in, a),
+    rabbit_log:info("Addresses discovered via A records of ~s: ~s",
+      [SeedHostname, string:join([inet_parse:ntoa(IP) || IP <- IPs], ", ")]),
+    Hosts = [extract_host(inet_res:gethostbyaddr(A), LongNamesUsed, A) ||
+                A <- IPs],
+    lists:filter(fun(E) -> E =/= error end, Hosts).
+
+%% long node names are used
+extract_host({ok, {hostent, FQDN, _, _, _, _}}, true, _Address) ->
+  FQDN;
+%% short node names are used
+extract_host({ok, {hostent, FQDN, _, _, _, _}}, false, _Address) ->
+  lists:nth(1, string:tokens(FQDN, "."));
+extract_host({error, Error}, _, Address) ->
+  rabbit_log:error("Reverse DNS lookup for address ~s failed: ~p",
+                   [inet_parse:ntoa(Address), Error]),
+  error.
diff --git a/src/rabbit_policy.erl b/src/rabbit_policy.erl
index cfbf116cbd..437842b8dd 100644
--- a/src/rabbit_policy.erl
+++ b/src/rabbit_policy.erl
@@ -205,18 +205,17 @@ parse_set(VHost, Name, Pattern, Definition, Priority, ApplyTo) ->
     parse_set(<<"policy">>, VHost, Name, Pattern, Definition, Priority, ApplyTo).
 
 parse_set(Type, VHost, Name, Pattern, Definition, Priority, ApplyTo) ->
-    try list_to_integer(Priority) of
+    try rabbit_data_coercion:to_integer(Priority) of
         Num -> parse_set0(Type, VHost, Name, Pattern, Definition, Num, ApplyTo)
     catch
         error:badarg -> {error, "~p priority must be a number", [Priority]}
     end.
 
 parse_set0(Type, VHost, Name, Pattern, Defn, Priority, ApplyTo) ->
-    Definition = rabbit_data_coercion:to_binary(Defn),
-    case rabbit_json:try_decode(Definition) of
+    case rabbit_json:try_decode(Defn) of
         {ok, Term} ->
             set0(Type, VHost, Name,
-                 [{<<"pattern">>,    list_to_binary(Pattern)},
+                 [{<<"pattern">>,    Pattern},
                   {<<"definition">>, maps:to_list(Term)},
                   {<<"priority">>,   Priority},
                   {<<"apply-to">>,   ApplyTo}]);