Add PROXY protocol support

deps/rabbitmq_web_stomp/src/rabbit_ws_handler.erl

@@ -45,7 +45,13 @@
 init(Req0, Opts) ->
     {PeerAddr, _PeerPort} = maps:get(peer, Req0),
     {_, KeepaliveSup} = lists:keyfind(keepalive_sup, 1, Opts),
-    {_, Sock} = lists:keyfind(socket, 1, Opts),
+    {_, Sock0} = lists:keyfind(socket, 1, Opts),
+    Sock = case maps:get(proxy_header, Req0, undefined) of
+        undefined ->
+            Sock0;
+        ProxyInfo ->
+            {rabbit_proxy_socket, Sock0, ProxyInfo}
+    end,
     Req = case cowboy_req:parse_header(<<"sec-websocket-protocol">>, Req0) of
         undefined  -> Req0;
         Protocols ->

deps/rabbitmq_web_stomp/src/rabbit_ws_listener.erl

@@ -33,7 +33,8 @@ init() ->
     TcpConf = get_tcp_conf(get_env(tcp_config, []), Port),
 
     WsFrame = get_env(ws_frame, text),
-    CowboyOpts = maps:from_list(get_env(cowboy_opts, [])),
+    CowboyOpts0 = maps:from_list(get_env(cowboy_opts, [])),
+    CowboyOpts = CowboyOpts0#{proxy_header => get_env(proxy_protocol, false)},
     CowboyWsOpts = maps:from_list(get_env(cowboy_ws_opts, [])),
 
     VhostRoutes = [

deps/rabbitmq_web_stomp/test/proxy_protocol_SUITE.erl

@@ -0,0 +1,111 @@
+%%   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 Management Console.
+%%
+%%   The Initial Developer of the Original Code is GoPivotal, Inc.
+%%   Copyright (c) 2007-2017 Pivotal Software, Inc.  All rights reserved.
+%%
+
+-module(proxy_protocol_SUITE).
+
+
+-compile(export_all).
+
+-include_lib("common_test/include/ct.hrl").
+
+-include_lib("eunit/include/eunit.hrl").
+
+suite() ->
+    [
+      %% If a test hangs, no need to wait for 30 minutes.
+      {timetrap, {minutes, 2}}
+    ].
+
+all() ->
+    [{group, http_tests},
+     {group, https_tests}].
+
+groups() ->
+    Tests = [
+        proxy_protocol
+    ],
+    [{https_tests, [], Tests},
+     {http_tests, [], Tests}].
+
+init_per_suite(Config) ->
+    rabbit_ct_helpers:log_environment(),
+    Config.
+
+end_per_suite(Config) ->
+    Config.
+
+init_per_group(Group, Config) ->
+    Protocol = case Group of
+        http_tests -> "ws";
+        https_tests -> "wss"
+    end,
+    Config1 = rabbit_ct_helpers:set_config(Config,
+                                           [{rmq_nodename_suffix, ?MODULE},
+                                            {protocol, Protocol},
+                                            {rabbitmq_ct_tls_verify, verify_none},
+                                            {rabbitmq_ct_tls_fail_if_no_peer_cert, false}]),
+
+    rabbit_ct_helpers:run_setup_steps(
+        Config1,
+        rabbit_ct_broker_helpers:setup_steps() ++ [
+            fun configure_proxy_protocol/1,
+            fun configure_ssl/1
+        ]).
+
+configure_proxy_protocol(Config) ->
+    rabbit_ws_test_util:update_app_env(Config, proxy_protocol, true),
+    Config.
+
+configure_ssl(Config) ->
+    ErlangConfig = proplists:get_value(erlang_node_config, Config, []),
+    RabbitAppConfig = proplists:get_value(rabbit, ErlangConfig, []),
+    RabbitSslConfig = proplists:get_value(ssl_options, RabbitAppConfig, []),
+    Port = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_web_stomp_tls),
+    rabbit_ws_test_util:update_app_env(Config, ssl_config, [{port, Port} | lists:keydelete(port, 1, RabbitSslConfig)]),
+    Config.
+
+end_per_group(_Group, Config) ->
+    rabbit_ct_helpers:run_teardown_steps(Config,
+      rabbit_ct_broker_helpers:teardown_steps()).
+
+init_per_testcase(Testcase, Config) ->
+    rabbit_ct_helpers:testcase_started(Config, Testcase).
+
+end_per_testcase(Testcase, Config) ->
+    rabbit_ct_helpers:testcase_finished(Config, Testcase).
+
+proxy_protocol(Config) ->
+    Port = list_to_integer(rabbit_ws_test_util:get_web_stomp_port_str(Config)),
+    PortStr = integer_to_list(Port),
+
+    Protocol = ?config(protocol, Config),
+    WS = rfc6455_client:new(Protocol ++ "://127.0.0.1:" ++ PortStr ++ "/ws", self(),
+        undefined, [], "PROXY TCP4 192.168.1.1 192.168.1.2 80 81\r\n"),
+    {ok, _} = rfc6455_client:open(WS),
+    Frame = stomp:marshal("CONNECT", [{"login","guest"}, {"passcode", "guest"}], <<>>),
+    rfc6455_client:send(WS, Frame),
+    {ok, _P} = rfc6455_client:recv(WS),
+    ConnectionName = rabbit_ct_broker_helpers:rpc(Config, 0,
+        ?MODULE, connection_name, []),
+    match = re:run(ConnectionName, <<"^192.168.1.1:80 ">>, [{capture, none}]),
+    {close, _} = rfc6455_client:close(WS),
+    ok.
+
+connection_name() ->
+    Connections = ets:tab2list(connection_created),
+    {_Key, Values} = lists:nth(1, Connections),
+    {_, Name} = lists:keyfind(name, 1, Values),
+    Name.

deps/rabbitmq_web_stomp/test/src/rfc6455_client.erl

@@ -16,19 +16,22 @@
 
 -module(rfc6455_client).
 
--export([new/2, new/3, new/4, open/1, recv/1, send/2, close/1, close/2]).
+-export([new/2, new/3, new/4, new/5, open/1, recv/1, send/2, close/1, close/2]).
 
 -record(state, {host, port, addr, path, ppid, socket, data, phase, transport}).
 
 %% --------------------------------------------------------------------------
 
 new(WsUrl, PPid) ->
-    new(WsUrl, PPid, undefined, []).
+    new(WsUrl, PPid, undefined, [], <<>>).
 
 new(WsUrl, PPid, AuthInfo) ->
-    new(WsUrl, PPid, AuthInfo, []).
+    new(WsUrl, PPid, AuthInfo, [], <<>>).
 
 new(WsUrl, PPid, AuthInfo, Protocols) ->
+    new(WsUrl, PPid, AuthInfo, Protocols, <<>>).
+
+new(WsUrl, PPid, AuthInfo, Protocols, TcpPreface) ->
     crypto:start(),
     application:ensure_all_started(ssl),
     {Transport, Url} = case WsUrl of
@@ -52,7 +55,7 @@ new(WsUrl, PPid, AuthInfo, Protocols) ->
                    ppid = PPid,
                    transport = Transport},
     spawn_link(fun () ->
-                  start_conn(State, AuthInfo, Protocols)
+                  start_conn(State, AuthInfo, Protocols, TcpPreface)
           end).
 
 open(WS) ->
@@ -90,10 +93,22 @@ close(WS, WsReason) ->
 
 %% --------------------------------------------------------------------------
 
-start_conn(State = #state{transport = Transport}, AuthInfo, Protocols) ->
-    {ok, Socket} = Transport:connect(State#state.host, State#state.port,
-                                   [binary,
-                                    {packet, 0}]),
+start_conn(State = #state{transport = Transport}, AuthInfo, Protocols, TcpPreface) ->
+    {ok, Socket} = case TcpPreface of
+        <<>> ->
+            Transport:connect(State#state.host, State#state.port,
+                              [binary,
+                               {packet, 0}]);
+        _ ->
+            {ok, Socket0} = gen_tcp:connect(State#state.host, State#state.port,
+                                            [binary,
+                                             {packet, 0}]),
+            gen_tcp:send(Socket0, TcpPreface),
+            case Transport of
+                gen_tcp -> {ok, Socket0};
+                ssl -> Transport:connect(Socket0, [])
+            end
+    end,
 
     AuthHd = case AuthInfo of
         undefined -> "";