merge default into bug20337

6 files changed, 89 insertions, 30 deletions

diff --git a/src/rabbit.erl b/src/rabbit.erl
index 0a0ca90a63..dd5fb89ce4 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -212,14 +212,13 @@
 -type(file_suffix() :: binary()).
 %% this really should be an abstract type
 -type(log_location() :: 'tty' | 'undefined' | file:filename()).
+-type(param() :: atom()).
 
 -spec(maybe_hipe_compile/0 :: () -> 'ok').
 -spec(prepare/0 :: () -> 'ok').
 -spec(start/0 :: () -> 'ok').
 -spec(stop/0 :: () -> 'ok').
 -spec(stop_and_halt/0 :: () -> no_return()).
--spec(rotate_logs/1 :: (file_suffix()) -> rabbit_types:ok_or_error(any())).
--spec(force_event_refresh/0 :: () -> 'ok').
 -spec(status/0 ::
         () -> [{pid, integer()} |
                {running_applications, [{atom(), string(), string()}]} |
@@ -228,12 +227,11 @@
                {memory, any()}]).
 -spec(is_running/0 :: () -> boolean()).
 -spec(is_running/1 :: (node()) -> boolean()).
--spec(environment/0 :: () -> [{atom() | term()}]).
--spec(log_location/1 :: ('sasl' | 'kernel') -> log_location()).
+-spec(environment/0 :: () -> [{param() | term()}]).
+-spec(rotate_logs/1 :: (file_suffix()) -> rabbit_types:ok_or_error(any())).
+-spec(force_event_refresh/0 :: () -> 'ok').
 
--spec(maybe_insert_default_data/0 :: () -> 'ok').
--spec(boot_delegate/0 :: () -> 'ok').
--spec(recover/0 :: () -> 'ok').
+-spec(log_location/1 :: ('sasl' | 'kernel') -> log_location()).
 
 -spec(start/2 :: ('normal',[]) ->
 		      {'error',
@@ -243,6 +241,10 @@
 		      {'ok',pid()}).
 -spec(stop/1 :: (_) -> 'ok').
 
+-spec(maybe_insert_default_data/0 :: () -> 'ok').
+-spec(boot_delegate/0 :: () -> 'ok').
+-spec(recover/0 :: () -> 'ok').
+
 -endif.
 
 %%----------------------------------------------------------------------------
@@ -712,6 +714,6 @@ config_files() ->
     case init:get_argument(config) of
         {ok, Files} -> [filename:absname(
                           filename:rootname(File, ".config") ++ ".config") ||
-                           File <- Files];
+                           [File] <- Files];
         error       -> []
     end.
diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl
index 6d8bed83b8..dca3bead75 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -742,13 +742,14 @@ gb_trees_foreach(Fun, Tree) ->
 %%     [{"-q",true},{"-p","/"}]}
 get_options(Defs, As) ->
     lists:foldl(fun(Def, {AsIn, RsIn}) ->
-                        {AsOut, Value} = case Def of
-                                             {flag, Key} ->
-                                                 get_flag(Key, AsIn);
-                                             {option, Key, Default} ->
-                                                 get_option(Key, Default, AsIn)
-                                         end,
-                        {AsOut, [{Key, Value} | RsIn]}
+                        {K, {AsOut, V}} =
+                            case Def of
+                                {flag, Key} ->
+                                    {Key, get_flag(Key, AsIn)};
+                                {option, Key, Default} ->
+                                    {Key, get_option(Key, Default, AsIn)}
+                            end,
+                        {AsOut, [{K, V} | RsIn]}
                 end, {As, []}, Defs).
 
 get_option(K, _Default, [K, V | As]) ->
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 60dd07708f..4d419fd9a9 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -732,18 +732,18 @@ reset(Force) ->
         false -> ok
     end,
     Node = node(),
+    Nodes = all_clustered_nodes() -- [Node],
     case Force of
         true  -> ok;
         false ->
             ensure_mnesia_dir(),
             start_mnesia(),
-            {Nodes, RunningNodes} =
+            RunningNodes =
                 try
                     %% Force=true here so that reset still works when clustered
                     %% with a node which is down
                     ok = init_db(read_cluster_nodes_config(), true),
-                    {all_clustered_nodes() -- [Node],
-                     running_clustered_nodes() -- [Node]}
+                    running_clustered_nodes() -- [Node]
                 after
                     stop_mnesia()
                 end,
@@ -751,6 +751,10 @@ reset(Force) ->
             rabbit_misc:ensure_ok(mnesia:delete_schema([Node]),
                                   cannot_delete_schema)
     end,
+    %% We need to make sure that we don't end up in a distributed
+    %% Erlang system with nodes while not being in an Mnesia cluster
+    %% with them. We don't handle that well.
+    [erlang:disconnect_node(N) || N <- Nodes],
     ok = delete_cluster_nodes_config(),
     %% remove persisted messages and any other garbage we find
     ok = rabbit_file:recursive_delete(filelib:wildcard(dir() ++ "/*")),
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 908a279c0c..01242e81b8 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -505,9 +505,11 @@ handle_frame(Type, Channel, Payload,
 process_frame(Frame, Channel, State) ->
     case get({channel, Channel}) of
         {ChPid, AState} ->
-            NewAState = process_channel_frame(Frame, Channel, ChPid, AState),
-            put({channel, Channel}, {ChPid, NewAState}),
-            post_process_frame(Frame, ChPid, State);
+            case process_channel_frame(Frame,  ChPid, AState) of
+                {ok, NewAState} -> put({channel, Channel}, {ChPid, NewAState}),
+                                   post_process_frame(Frame, ChPid, State);
+                {error, Reason} -> handle_exception(State, Channel, Reason)
+            end;
         undefined when ?IS_RUNNING(State) ->
             ok = create_channel(Channel, State),
             process_frame(Frame, Channel, State);
@@ -910,17 +912,15 @@ create_channel(Channel, State) ->
     put({channel, Channel}, {ChPid, AState}),
     ok.
 
-process_channel_frame(Frame, Channel, ChPid, AState) ->
+process_channel_frame(Frame, ChPid, AState) ->
     case rabbit_command_assembler:process(Frame, AState) of
-        {ok, NewAState}                  -> NewAState;
+        {ok, NewAState}                  -> {ok, NewAState};
         {ok, Method, NewAState}          -> rabbit_channel:do(ChPid, Method),
-                                            NewAState;
+                                            {ok, NewAState};
         {ok, Method, Content, NewAState} -> rabbit_channel:do_flow(
                                               ChPid, Method, Content),
-                                            NewAState;
-        {error, Reason}                  -> self() ! {channel_exit, Channel,
-                                                      Reason},
-                                            AState
+                                            {ok, NewAState};
+        {error, Reason}                  -> {error, Reason}
     end.
 
 handle_exception(State = #v1{connection_state = closed}, _Channel, _Reason) ->
diff --git a/src/rabbit_ssl.erl b/src/rabbit_ssl.erl
index 3025d981d4..22ff555ff0 100644
--- a/src/rabbit_ssl.erl
+++ b/src/rabbit_ssl.erl
@@ -21,7 +21,7 @@
 -include_lib("public_key/include/public_key.hrl").
 
 -export([peer_cert_issuer/1, peer_cert_subject/1, peer_cert_validity/1]).
--export([peer_cert_subject_items/2]).
+-export([peer_cert_subject_items/2, peer_cert_auth_name/1]).
 
 %%--------------------------------------------------------------------------
 
@@ -36,6 +36,8 @@
 -spec(peer_cert_validity/1      :: (certificate()) -> string()).
 -spec(peer_cert_subject_items/2  ::
         (certificate(), tuple()) -> [string()] | 'not_found').
+-spec(peer_cert_auth_name/1 ::
+        (certificate()) -> binary() | 'not_found' | 'unsafe').
 
 -endif.
 
@@ -76,6 +78,43 @@ peer_cert_validity(Cert) ->
                                                      format_asn1_value(End)])
               end, Cert).
 
+%% Extract a username from the certificate
+peer_cert_auth_name(Cert) ->
+    {ok, Mode} = application:get_env(rabbit, ssl_cert_login_from),
+    peer_cert_auth_name(Mode, Cert).
+
+peer_cert_auth_name(distinguished_name, Cert) ->
+    case auth_config_sane() of
+        true  -> iolist_to_binary(peer_cert_subject(Cert));
+        false -> unsafe
+    end;
+
+peer_cert_auth_name(common_name, Cert) ->
+    %% If there is more than one CN then we join them with "," in a
+    %% vaguely DN-like way. But this is more just so we do something
+    %% more intelligent than crashing, if you actually want to escape
+    %% things properly etc, use DN mode.
+    case auth_config_sane() of
+        true  -> case peer_cert_subject_items(Cert, ?'id-at-commonName') of
+                     not_found -> not_found;
+                     CNs       -> list_to_binary(string:join(CNs, ","))
+                 end;
+        false -> unsafe
+    end.
+
+auth_config_sane() ->
+    {ok, Opts} = application:get_env(rabbit, ssl_options),
+    case {proplists:get_value(fail_if_no_peer_cert, Opts),
+          proplists:get_value(verify, Opts)} of
+        {true, verify_peer} ->
+            true;
+        {F, V} ->
+            rabbit_log:warning("SSL certificate authentication disabled, "
+                               "fail_if_no_peer_cert=~p; "
+                               "verify=~p~n", [F, V]),
+            false
+    end.
+
 %%--------------------------------------------------------------------------
 
 cert_info(F, Cert) ->
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 165bdbe246..863292144e 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -71,10 +71,13 @@ maybe_run_cluster_dependent_tests() ->
 run_cluster_dependent_tests(SecondaryNode) ->
     SecondaryNodeS = atom_to_list(SecondaryNode),
 
+    cover:stop(SecondaryNode),
     ok = control_action(stop_app, []),
     ok = control_action(reset, []),
     ok = control_action(cluster, [SecondaryNodeS]),
     ok = control_action(start_app, []),
+    cover:start(SecondaryNode),
+    ok = control_action(start_app, SecondaryNode, [], []),
 
     io:format("Running cluster dependent tests with node ~p~n", [SecondaryNode]),
     passed = test_delegates_async(SecondaryNode),
@@ -960,7 +963,9 @@ test_cluster_management2(SecondaryNode) ->
     ok = control_action(cluster, [SecondaryNodeS, NodeS]),
     ok = control_action(start_app, []),
     ok = control_action(stop_app, []),
+    cover:stop(SecondaryNode),
     ok = control_action(reset, []),
+    cover:start(SecondaryNode),
 
     %% attempt to leave cluster when no other node is alive
     ok = control_action(cluster, [SecondaryNodeS, NodeS]),
@@ -977,7 +982,15 @@ test_cluster_management2(SecondaryNode) ->
     %% leave system clustered, with the secondary node as a ram node
     ok = control_action(force_reset, []),
     ok = control_action(start_app, []),
-    ok = control_action(force_reset, SecondaryNode, [], []),
+    %% Yes, this is rather ugly. But since we're a clustered Mnesia
+    %% node and we're telling another clustered node to reset itself,
+    %% we will get disconnected half way through causing a
+    %% badrpc. This never happens in real life since rabbitmqctl is
+    %% not a clustered Mnesia node.
+    cover:stop(SecondaryNode),
+    {badrpc, nodedown} = control_action(force_reset, SecondaryNode, [], []),
+    pong = net_adm:ping(SecondaryNode),
+    cover:start(SecondaryNode),
     ok = control_action(cluster, SecondaryNode, [NodeS], []),
     ok = control_action(start_app, SecondaryNode, [], []),