diff options
| -rw-r--r-- | .hgignore | 2 | ||||
| -rw-r--r-- | ebin/rabbit_app.in | 2 | ||||
| -rw-r--r-- | include/rabbit.hrl | 5 | ||||
| -rw-r--r-- | src/rabbit.erl | 18 | ||||
| -rw-r--r-- | src/rabbit_heartbeat.erl | 4 | ||||
| -rw-r--r-- | src/rabbit_net.erl | 133 | ||||
| -rw-r--r-- | src/rabbit_networking.erl | 56 | ||||
| -rw-r--r-- | src/rabbit_reader.erl | 18 | ||||
| -rw-r--r-- | src/rabbit_writer.erl | 4 |
9 files changed, 214 insertions, 28 deletions
@@ -1,6 +1,8 @@ syntax: glob *.beam *~ +*.swp +*.patch erl_crash.dump syntax: regexp diff --git a/ebin/rabbit_app.in b/ebin/rabbit_app.in index 77f9d299b7..5dd701b028 100644 --- a/ebin/rabbit_app.in +++ b/ebin/rabbit_app.in @@ -18,4 +18,6 @@ {default_pass, <<"guest">>}, {default_vhost, <<"/">>}, {default_permissions, [<<".*">>, <<".*">>]}, + {ssl_listeners, []}, + {ssl_options, []}, {memory_alarms, auto}]}]}. diff --git a/include/rabbit.hrl b/include/rabbit.hrl index 8aba8a6fb6..c42df1eab3 100644 --- a/include/rabbit.hrl +++ b/include/rabbit.hrl @@ -64,6 +64,8 @@ -record(basic_message, {exchange_name, routing_key, content, persistent_key}). +-record(ssl_socket, {tcp, ssl}). + %%---------------------------------------------------------------------------- -ifdef(use_specs). @@ -72,7 +74,8 @@ -type(maybe(T) :: T | 'none'). -type(erlang_node() :: atom()). --type(socket() :: port()). +-type(ssl_socket() :: #ssl_socket{}). +-type(socket() :: port() | ssl_socket()). -type(thunk(T) :: fun(() -> T)). -type(info_key() :: atom()). -type(info() :: {info_key(), any()}). diff --git a/src/rabbit.erl b/src/rabbit.erl index 7f808bc9dd..c513ddd791 100644 --- a/src/rabbit.erl +++ b/src/rabbit.erl @@ -171,6 +171,24 @@ start(normal, []) -> ok = rabbit_networking:start_tcp_listener(Host, Port) end, TCPListeners) + end}, + {"SSL listeners", + fun () -> + {ok, SSLListeners} = application:get_env(ssl_listeners), + case length(SSLListeners) of + 0 -> + ok; + _Len -> + ok = rabbit_misc:start_applications([crypto, ssl]), + + {ok, SslOpts} = application:get_env(ssl_options), + + lists:foreach( + fun ({Host, Port}) -> + ok = rabbit_networking:start_ssl_listener(Host, Port, SslOpts) + end, + SSLListeners) + end end}] ++ ExtraSteps), diff --git a/src/rabbit_heartbeat.erl b/src/rabbit_heartbeat.erl index 0a68c9adad..ed0066fe07 100644 --- a/src/rabbit_heartbeat.erl +++ b/src/rabbit_heartbeat.erl @@ -53,7 +53,7 @@ start_heartbeat(Sock, TimeoutSec) -> spawn_link(fun () -> heartbeater(Sock, TimeoutSec * 1000 div 2, send_oct, 0, fun () -> - catch gen_tcp:send(Sock, rabbit_binary_generator:build_heartbeat_frame()), + catch rabbit_net:send(Sock, rabbit_binary_generator:build_heartbeat_frame()), continue end, erlang:monitor(process, Parent)) end), @@ -73,7 +73,7 @@ heartbeater(Sock, TimeoutMillisec, StatName, Threshold, Handler, MonitorRef) -> {'DOWN', MonitorRef, process, _Object, _Info} -> ok; Other -> exit({unexpected_message, Other}) after TimeoutMillisec -> - case inet:getstat(Sock, [StatName]) of + case rabbit_net:getstat(Sock, [StatName]) of {ok, [{StatName, NewStatVal}]} -> if NewStatVal =/= StatVal -> F({NewStatVal, 0}); diff --git a/src/rabbit_net.erl b/src/rabbit_net.erl new file mode 100644 index 0000000000..2f0b9ce09f --- /dev/null +++ b/src/rabbit_net.erl @@ -0,0 +1,133 @@ +%% 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 Developers of the Original Code are LShift Ltd, +%% Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, +%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd +%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial +%% Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift +%% Ltd. Portions created by Cohesive Financial Technologies LLC are +%% Copyright (C) 2007-2009 Cohesive Financial Technologies +%% LLC. Portions created by Rabbit Technologies Ltd are Copyright +%% (C) 2007-2009 Rabbit Technologies Ltd. +%% +%% All Rights Reserved. +%% +%% Contributor(s): ______________________________________. +%% + +-module(rabbit_net). +-include("rabbit.hrl"). +-include_lib("kernel/include/inet.hrl"). + +-export([async_recv/3, close/1, controlling_process/2, + getstat/2, peername/1, port_command/2, + send/2, sockname/1]). +%%--------------------------------------------------------------------------- + +-ifdef(use_specs). + +-type(stat_option() :: + 'recv_cnt' | 'recv_max' | 'recv_avg' | 'recv_oct' | 'recv_dvi' | + 'send_cnt' | 'send_max' | 'send_avg' | 'send_oct' | 'send_pend'). +-type(timeout() :: 'infinity' | -1 | non_neg_integer()). +-type(error() :: {'error', any()}). + +-spec(async_recv/3 :: (socket(), integer(), timeout()) -> {'ok', ref()}). +-spec(close/1 :: (socket()) -> 'ok' | error()). +-spec(controlling_process/2 :: (socket(), pid()) -> 'ok' | error()). +-spec(port_command/2 :: (socket(), iolist()) -> 'true'). +-spec(send/2 :: (socket(), iolist()) -> 'ok' | error()). +-spec(peername/1 :: (socket()) -> + {'ok', {ip_address(), non_neg_integer()}} | error()). +-spec(sockname/1 :: (socket()) -> + {'ok', {ip_address(), non_neg_integer()}} | error()). +-spec(getstat/2 :: (socket(), [stat_option()]) -> + {'ok', [{stat_option(), integer()}]} | error()). + +-endif. + +%%--------------------------------------------------------------------------- + + +async_recv(Sock, Length, Timeout) when is_record(Sock, ssl_socket) -> + Pid = self(), + Ref = make_ref(), + + spawn(fun() -> Pid ! {inet_async, Sock, Ref, + ssl:recv(Sock#ssl_socket.ssl, Length, Timeout)} + end), + + {ok, Ref}; + +async_recv(Sock, Length, infinity) when is_port(Sock) -> + async_recv(Sock, Length, -1); + +async_recv(Sock, Length, Timeout) when is_port(Sock) -> + prim_inet:async_recv(Sock, Length, Timeout). + +close(Sock) when is_record(Sock, ssl_socket) -> + ssl:close(Sock#ssl_socket.ssl); + +close(Sock) when is_port(Sock) -> + gen_tcp:close(Sock). + + +controlling_process(Sock, Pid) when is_record(Sock, ssl_socket) -> + ssl:controlling_process(Sock#ssl_socket.ssl, Pid); + +controlling_process(Sock, Pid) when is_port(Sock) -> + gen_tcp:controlling_process(Sock, Pid). + + +getstat(Sock, Stats) when is_record(Sock, ssl_socket) -> + inet:getstat(Sock#ssl_socket.tcp, Stats); + +getstat(Sock, Stats) when is_port(Sock) -> + inet:getstat(Sock, Stats). + + +peername(Sock) when is_record(Sock, ssl_socket) -> + ssl:peername(Sock#ssl_socket.ssl); + +peername(Sock) when is_port(Sock) -> + inet:peername(Sock). + + +port_command(Sock, Data) when is_record(Sock, ssl_socket) -> + case ssl:send(Sock#ssl_socket.ssl, Data) of + ok -> + self() ! {inet_reply, Sock, ok}, + true; + {error, Reason} -> + erlang:error(Reason) + end; + +port_command(Sock, Data) when is_port(Sock) -> + erlang:port_command(Sock, Data). + +send(Sock, Data) when is_record(Sock, ssl_socket) -> + ssl:send(Sock#ssl_socket.ssl, Data); + +send(Sock, Data) when is_port(Sock) -> + gen_tcp:send(Sock, Data). + + +sockname(Sock) when is_record(Sock, ssl_socket) -> + ssl:sockname(Sock#ssl_socket.ssl); + +sockname(Sock) when is_port(Sock) -> + inet:sockname(Sock). diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl index 2dbd5a5af2..30fb7b1f60 100644 --- a/src/rabbit_networking.erl +++ b/src/rabbit_networking.erl @@ -31,18 +31,28 @@ -module(rabbit_networking). --export([start/0, start_tcp_listener/2, stop_tcp_listener/2, - on_node_down/1, active_listeners/0, node_listeners/1, - connections/0, connection_info/1, connection_info/2, - connection_info_all/0, connection_info_all/1]). +-export([start/0, start_tcp_listener/2, start_ssl_listener/3, + stop_tcp_listener/2, on_node_down/1, active_listeners/0, + node_listeners/1, connections/0, connection_info/1, + connection_info/2, connection_info_all/0, + connection_info_all/1]). %%used by TCP-based transports, e.g. STOMP adapter -export([check_tcp_listener_address/3]). --export([tcp_listener_started/2, tcp_listener_stopped/2, start_client/1]). +-export([tcp_listener_started/2, ssl_connection_upgrade/2, + tcp_listener_stopped/2, start_client/1]). -include("rabbit.hrl"). -include_lib("kernel/include/inet.hrl"). +-define(RABBIT_TCP_OPTS, [ + binary, + {packet, raw}, % no packaging + {reuseaddr, true}, % allow rebind without waiting + %% {nodelay, true}, % TCP_NODELAY - disable Nagle's alg. + %% {delay_send, true}, + {exit_on_close, false} + ]). %%---------------------------------------------------------------------------- -ifdef(use_specs). @@ -52,6 +62,7 @@ -spec(start/0 :: () -> 'ok'). -spec(start_tcp_listener/2 :: (host(), ip_port()) -> 'ok'). +-spec(start_ssl_listener/3 :: (host(), ip_port(), [info()]) -> 'ok'). -spec(stop_tcp_listener/2 :: (host(), ip_port()) -> 'ok'). -spec(active_listeners/0 :: () -> [listener()]). -spec(node_listeners/1 :: (erlang_node()) -> [listener()]). @@ -96,21 +107,21 @@ check_tcp_listener_address(NamePrefix, Host, Port) -> {IPAddress, Name}. start_tcp_listener(Host, Port) -> + start_listener(Host, Port, {?MODULE, start_client, []}). + +start_ssl_listener(Host, Port, SslOpts) -> + start_listener(Host, Port, {?MODULE, ssl_connection_upgrade, [SslOpts]}). + +start_listener(Host, Port, OnConnect) -> {IPAddress, Name} = check_tcp_listener_address(rabbit_tcp_listener_sup, Host, Port), {ok,_} = supervisor:start_child( rabbit_sup, {Name, {tcp_listener_sup, start_link, - [IPAddress, Port, - [binary, - {packet, raw}, % no packaging - {reuseaddr, true}, % allow rebind without waiting - %% {nodelay, true}, % TCP_NODELAY - disable Nagle's alg. - %% {delay_send, true}, - {exit_on_close, false}], + [IPAddress, Port, ?RABBIT_TCP_OPTS , {?MODULE, tcp_listener_started, []}, {?MODULE, tcp_listener_stopped, []}, - {?MODULE, start_client, []}]}, + OnConnect]}, transient, infinity, supervisor, [tcp_listener_sup]}), ok. @@ -129,6 +140,7 @@ tcp_listener_started(IPAddress, Port) -> host = tcp_host(IPAddress), port = Port}). + tcp_listener_stopped(IPAddress, Port) -> ok = mnesia:dirty_delete_object( rabbit_listener, @@ -148,10 +160,26 @@ on_node_down(Node) -> start_client(Sock) -> {ok, Child} = supervisor:start_child(rabbit_tcp_client_sup, []), - ok = gen_tcp:controlling_process(Sock, Child), + ok = rabbit_net:controlling_process(Sock, Child), Child ! {go, Sock}, Child. +ssl_connection_upgrade(SslOpts, Sock) -> + {ok, {PeerAddress, PeerPort}} = rabbit_net:peername(Sock), + PeerIp = inet_parse:ntoa(PeerAddress), + + case ssl:ssl_accept(Sock, SslOpts) of + {ok, SslSock} -> + rabbit_log:info("Upgraded TCP connection from ~s:~p to SSL/TLS~n", + [PeerIp, PeerPort]), + RabbitSslSock = #ssl_socket{tcp=Sock, ssl=SslSock}, + start_client(RabbitSslSock); + {error, Reason} -> + rabbit_log:error("Failed to upgrade TCP connection from ~s:~p to SSL/TLS. Reason: ~p~n", + [PeerIp, PeerPort, Reason]), + {error, Reason} + end. + connections() -> [Pid || {_, Pid, _, _} <- supervisor:which_children( rabbit_tcp_client_sup)]. diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl index 9dbc49df47..a6094ebc58 100644 --- a/src/rabbit_reader.erl +++ b/src/rabbit_reader.erl @@ -200,7 +200,7 @@ inet_op(F) -> rabbit_misc:throw_on_error(inet_error, F). peername(Sock) -> try - {Address, Port} = inet_op(fun () -> inet:peername(Sock) end), + {Address, Port} = inet_op(fun () -> rabbit_net:peername(Sock) end), AddressS = inet_parse:ntoa(Address), {AddressS, Port} catch @@ -317,8 +317,8 @@ mainloop(Parent, Deb, State = #v1{sock= Sock, recv_ref = Ref}) -> end. switch_callback(OldState, NewCallback, Length) -> - Ref = inet_op(fun () -> prim_inet:async_recv( - OldState#v1.sock, Length, -1) end), + Ref = inet_op(fun () -> rabbit_net:async_recv( + OldState#v1.sock, Length, infinity) end), OldState#v1{callback = NewCallback, recv_ref = Ref}. @@ -525,7 +525,7 @@ handle_input(handshake, <<"AMQP",1,1,ProtocolMajor,ProtocolMinor>>, end; handle_input(handshake, Other, #v1{sock = Sock}) -> - ok = inet_op(fun () -> gen_tcp:send( + ok = inet_op(fun () -> rabbit_net:send( Sock, <<"AMQP",1,1, ?PROTOCOL_VERSION_MAJOR, ?PROTOCOL_VERSION_MINOR>>) end), @@ -661,23 +661,23 @@ infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items]. i(pid, #v1{}) -> self(); i(address, #v1{sock = Sock}) -> - {ok, {A, _}} = inet:sockname(Sock), + {ok, {A, _}} = rabbit_net:sockname(Sock), A; i(port, #v1{sock = Sock}) -> - {ok, {_, P}} = inet:sockname(Sock), + {ok, {_, P}} = rabbit_net:sockname(Sock), P; i(peer_address, #v1{sock = Sock}) -> - {ok, {A, _}} = inet:peername(Sock), + {ok, {A, _}} = rabbit_net:peername(Sock), A; i(peer_port, #v1{sock = Sock}) -> - {ok, {_, P}} = inet:peername(Sock), + {ok, {_, P}} = rabbit_net:peername(Sock), P; i(SockStat, #v1{sock = Sock}) when SockStat =:= recv_oct; SockStat =:= recv_cnt; SockStat =:= send_oct; SockStat =:= send_cnt; SockStat =:= send_pend -> - case inet:getstat(Sock, [SockStat]) of + case rabbit_net:getstat(Sock, [SockStat]) of {ok, [{SockStat, StatVal}]} -> StatVal; {error, einval} -> undefined; {error, Error} -> throw({cannot_get_socket_stats, Error}) diff --git a/src/rabbit_writer.erl b/src/rabbit_writer.erl index 9cf9f8aef9..5ca294b762 100644 --- a/src/rabbit_writer.erl +++ b/src/rabbit_writer.erl @@ -139,7 +139,7 @@ assemble_frames(Channel, MethodRecord, Content, FrameMax) -> tcp_send(Sock, Data) -> rabbit_misc:throw_on_error(inet_error, - fun () -> gen_tcp:send(Sock, Data) end). + fun () -> rabbit_net:send(Sock, Data) end). internal_send_command(Sock, Channel, MethodRecord) -> ok = tcp_send(Sock, assemble_frames(Channel, MethodRecord)). @@ -176,6 +176,6 @@ internal_send_command_async(Sock, Channel, MethodRecord, Content, FrameMax) -> ok. port_cmd(Sock, Data) -> - try erlang:port_command(Sock, Data) + try rabbit_net:port_command(Sock, Data) catch error:Error -> exit({writer, send_failed, Error}) end. |