Merge pull request #6674 from emqx/proxy-protocol-frame-log

proxy protocol frame log
revert `zone.external.max_topic_levels`

etc/zones.conf

@@ -60,8 +60,7 @@ zone.external.force_shutdown_policy = 10000|32MB
 ## Depth so big may lead to subscribing performance issues.
 ##
 ## Value: Number [0-65535]
-## Default 7
-zone.external.max_topic_levels = 7
+## zone.external.max_topic_levels = 7
 
 ## Maximum QoS allowed.
 ##

src/emqx.appup.src

@@ -126,9 +126,13 @@
      {load_module, emqx_frame, brutal_purge, soft_purge, []},
      {load_module, emqx_channel, brutal_purge, soft_purge, []},
      {load_module, emqx_connection, brutal_purge, soft_purge, []},
+     {load_module, emqx_ws_connection, brutal_purge, soft_purge, []},
      {load_module, emqx_cm, brutal_purge, soft_purge, []}
    ]},
    {<<"4.2.9">>, [
+     {load_module, emqx_frame, brutal_purge, soft_purge, []},
+     {load_module, emqx_connection, brutal_purge, soft_purge, []},
+     {load_module, emqx_ws_connection, brutal_purge, soft_purge, []},
      {load_module, emqx_zone, brutal_purge, soft_purge, []}
    ]},
    {<<".*">>, []}
@@ -258,9 +262,13 @@
      {load_module, emqx_frame, brutal_purge, soft_purge, []},
      {load_module, emqx_channel, brutal_purge, soft_purge, []},
      {load_module, emqx_connection, brutal_purge, soft_purge, []},
+     {load_module, emqx_ws_connection, brutal_purge, soft_purge, []},
      {load_module, emqx_cm, brutal_purge, soft_purge, []}
    ]},
    {<<"4.2.9">>, [
+     {load_module, emqx_frame, brutal_purge, soft_purge, []},
+     {load_module, emqx_connection, brutal_purge, soft_purge, []},
+     {load_module, emqx_ws_connection, brutal_purge, soft_purge, []},
      {load_module, emqx_zone, brutal_purge, soft_purge, []}
    ]},
    {<<".*">>, []}

src/emqx_connection.erl

@@ -589,6 +589,11 @@ parse_incoming(Data, Packets, State = #state{parse_state = ParseState}) ->
             NState = State#state{parse_state = NParseState},
             parse_incoming(Rest, [Packet|Packets], NState)
     catch
+        error:proxy_protocol_config_disabled ->
+            ?LOG(error,
+                 "~nMalformed packet, "
+                 "please check proxy_protocol config for specific listeners and zones~n"),
+            {[{frame_error, proxy_protocol_config_disabled} | Packets], State};
         error:Reason:Stk ->
             ?LOG(error, "~nParse failed for ~0p~n~0p~nFrame data:~0p",
                  [Reason, Stk, Data]),
@@ -812,4 +817,3 @@ stop(Reason, Reply, State) ->
 set_field(Name, Value, State) ->
     Pos = emqx_misc:index_of(Name, record_info(fields, state)),
     setelement(Pos+1, State, Value).
-

src/emqx_frame.erl

@@ -69,6 +69,13 @@
 
 -define(MULTIPLIER_MAX, 16#200000).
 
+%% proxy_protocol v1 header human readable
+-define(PPV1_PROXY, "PROXY ").
+-define(PPV1_PROXY_UNKNOWN, "PROXY UNKNOWN").
+%% proxy_protocol v2 header signature:
+%% 16#0D,16#0A, 16#0D,16#0A,16#00,16#0D,16#0A,16#51,16#55,16#49,16#54,16#0A
+-define(PPV2_HEADER_SIG, "\r\n\r\n\0\r\nQUIT\n").
+
 -dialyzer({no_match, [serialize_utf8_string/2]}).
 
 %%--------------------------------------------------------------------
@@ -98,6 +105,13 @@ parse(Bin) ->
 -spec(parse(binary(), parse_state()) -> parse_result()).
 parse(<<>>, {none, Options}) ->
     {more, {none, Options}};
+parse(<<?PPV1_PROXY, IPVer:5/binary, _Rest/binary>>, {none, _Options})
+  when IPVer =:= <<"TCP4 ">> orelse IPVer =:= <<"TCP6 ">> ->
+    error(proxy_protocol_config_disabled);
+parse(<<?PPV1_PROXY_UNKNOWN, _Rest/binary>>, {none, _Options}) ->
+    error(proxy_protocol_config_disabled);
+parse(<<?PPV2_HEADER_SIG, _Rest/binary>>, {none, _Options}) ->
+    error(proxy_protocol_config_disabled);
 parse(<<Type:4, Dup:1, QoS:2, Retain:1, Rest/binary>>,
       {none, Options = #{strict_mode := StrictMode}}) ->
     %% Validate header if strict mode.

src/emqx_ws_connection.erl

@@ -486,6 +486,12 @@ parse_incoming(Data, State = #state{parse_state = ParseState}) ->
             NState = State#state{parse_state = NParseState},
             parse_incoming(Rest, postpone({incoming, Packet}, NState))
     catch
+        error:proxy_protocol_config_disabled ->
+            ?LOG(error,
+                 "~nMalformed packet, "
+                 "please check proxy_protocol config for specific listeners and zones~n"),
+            FrameError = {frame_error, proxy_protocol_config_disabled},
+            postpone({incoming, FrameError} ,State);
         error:Reason:Stk ->
             ?LOG(error, "~nParse failed for ~0p~n~0p~nFrame data: ~0p",
                  [Reason, Stk, Data]),
@@ -689,4 +695,3 @@ trigger(Event) -> erlang:send(self(), Event).
 set_field(Name, Value, State) ->
     Pos = emqx_misc:index_of(Name, record_info(fields, state)),
     setelement(Pos+1, State, Value).
-