fix(frame): fix potential parse errors found by fuzzing test

apps/emqx/src/emqx_frame.erl

@@ -182,8 +182,19 @@ parse_remaining_len(
     Packet = packet(Header, #mqtt_packet_disconnect{reason_code = ?RC_SUCCESS}),
     {ok, Packet, Rest, ?NONE(Options)};
 %% Match PINGREQ.
-parse_remaining_len(<<0:8, Rest/binary>>, Header, 1, 0, Options) ->
+parse_remaining_len(
+    <<0:8, Rest/binary>>, Header = #mqtt_packet_header{type = ?PINGREQ}, 1, 0, Options
+) ->
     parse_frame(Rest, Header, 0, Options);
+parse_remaining_len(
+    <<0:8, Rest/binary>>, Header = #mqtt_packet_header{type = ?PINGRESP}, 1, 0, Options
+) ->
+    parse_frame(Rest, Header, 0, Options);
+%% Without this clause a crash may will happen when data incorrect, this was found by fuzzing test
+parse_remaining_len(
+    <<0:8, _Rest/binary>>, _Header, 1, 0, _Options
+) ->
+    ?PARSE_ERR(invalid_remaining_len);
 %% Match PUBACK, PUBREC, PUBREL, PUBCOMP, UNSUBACK...
 parse_remaining_len(<<0:1, 2:7, Rest/binary>>, Header, 1, 0, Options) ->
     parse_frame(Rest, Header, 2, Options);
@@ -261,41 +272,51 @@ parse_packet(
     #{strict_mode := StrictMode}
 ) ->
     {ProtoName, Rest} = parse_utf8_string(FrameBin, StrictMode),
-    <<BridgeTag:4, ProtoVer:4, Rest1/binary>> = Rest,
-    % Note: Crash when reserved flag doesn't equal to 0, there is no strict
-    % compliance with the MQTT5.0.
-    <<UsernameFlag:1, PasswordFlag:1, WillRetain:1, WillQoS:2, WillFlag:1, CleanStart:1, 0:1,
-        KeepAlive:16/big, Rest2/binary>> = Rest1,
-
-    {Properties, Rest3} = parse_properties(Rest2, ProtoVer, StrictMode),
-    {ClientId, Rest4} = parse_utf8_string(Rest3, StrictMode),
-    ConnPacket = #mqtt_packet_connect{
-        proto_name = ProtoName,
-        proto_ver = ProtoVer,
-        is_bridge = (BridgeTag =:= 8),
-        clean_start = bool(CleanStart),
-        will_flag = bool(WillFlag),
-        will_qos = WillQoS,
-        will_retain = bool(WillRetain),
-        keepalive = KeepAlive,
-        properties = Properties,
-        clientid = ClientId
-    },
-    {ConnPacket1, Rest5} = parse_will_message(ConnPacket, Rest4, StrictMode),
-    {Username, Rest6} = parse_utf8_string(Rest5, StrictMode, bool(UsernameFlag)),
-    {Password, <<>>} = parse_utf8_string(Rest6, StrictMode, bool(PasswordFlag)),
-    ConnPacket1#mqtt_packet_connect{username = Username, password = Password};
+    case Rest of
+        % Note: Crash when reserved flag doesn't equal to 0, there is no strict
+        % compliance with the MQTT5.0.
+        <<BridgeTag:4, ProtoVer:4, UsernameFlag:1, PasswordFlag:1, WillRetain:1, WillQoS:2,
+            WillFlag:1, CleanStart:1, 0:1, KeepAlive:16/big, Rest2/binary>> ->
+            {Properties, Rest3} = parse_properties(Rest2, ProtoVer, StrictMode),
+            {ClientId, Rest4} = parse_utf8_string(Rest3, StrictMode),
+            ConnPacket = #mqtt_packet_connect{
+                proto_name = ProtoName,
+                proto_ver = ProtoVer,
+                is_bridge = (BridgeTag =:= 8),
+                clean_start = bool(CleanStart),
+                will_flag = bool(WillFlag),
+                will_qos = WillQoS,
+                will_retain = bool(WillRetain),
+                keepalive = KeepAlive,
+                properties = Properties,
+                clientid = ClientId
+            },
+            {ConnPacket1, Rest5} = parse_will_message(ConnPacket, Rest4, StrictMode),
+            {Username, Rest6} = parse_utf8_string(Rest5, StrictMode, bool(UsernameFlag)),
+            case parse_utf8_string(Rest6, StrictMode, bool(PasswordFlag)) of
+                {Password, <<>>} ->
+                    ConnPacket1#mqtt_packet_connect{username = Username, password = Password};
+                _ ->
+                    ?PARSE_ERR(malformed_connect_payload)
+            end;
+        _ ->
+            ?PARSE_ERR(malformed_connect_header)
+    end;
 parse_packet(
     #mqtt_packet_header{type = ?CONNACK},
     <<AckFlags:8, ReasonCode:8, Rest/binary>>,
     #{version := Ver, strict_mode := StrictMode}
 ) ->
-    {Properties, <<>>} = parse_properties(Rest, Ver, StrictMode),
-    #mqtt_packet_connack{
-        ack_flags = AckFlags,
-        reason_code = ReasonCode,
-        properties = Properties
-    };
+    case parse_properties(Rest, Ver, StrictMode) of
+        {Properties, <<>>} ->
+            #mqtt_packet_connack{
+                ack_flags = AckFlags,
+                reason_code = ReasonCode,
+                properties = Properties
+            };
+        _ ->
+            ?PARSE_ERR(malformed_properties)
+    end;
 parse_packet(
     #mqtt_packet_header{type = ?PUBLISH, qos = QoS},
     Bin,
@@ -411,7 +432,9 @@ parse_packet(
     #{strict_mode := StrictMode, version := ?MQTT_PROTO_V5}
 ) ->
     {Properties, <<>>} = parse_properties(Rest, ?MQTT_PROTO_V5, StrictMode),
-    #mqtt_packet_auth{reason_code = ReasonCode, properties = Properties}.
+    #mqtt_packet_auth{reason_code = ReasonCode, properties = Properties};
+parse_packet(_Header, _FrameBin, _Options) ->
+    ?PARSE_ERR(malformed_packet).
 
 parse_will_message(
     Packet = #mqtt_packet_connect{
@@ -437,7 +460,9 @@ parse_will_message(Packet, Bin, _StrictMode) ->
 
 -compile({inline, [parse_packet_id/1]}).
 parse_packet_id(<<PacketId:16/big, Rest/binary>>) ->
-    {PacketId, Rest}.
+    {PacketId, Rest};
+parse_packet_id(_) ->
+    ?PARSE_ERR(invalid_packet_id).
 
 parse_properties(Bin, Ver, _StrictMode) when Ver =/= ?MQTT_PROTO_V5 ->
     {#{}, Bin};

apps/emqx/test/emqx_frame_SUITE.erl

@@ -619,6 +619,18 @@ t_serialize_parse_auth_v5(_) ->
         })
     ).
 
+t_parse_invalid_remaining_len(_) ->
+    ?assertException(
+        throw, {frame_parse_error, invalid_remaining_len}, emqx_frame:parse(<<?CONNECT, 0>>)
+    ).
+
+t_parse_malformed_properties(_) ->
+    ?assertException(
+        throw,
+        {frame_parse_error, malformed_properties},
+        emqx_frame:parse(<<2:4, 0:4, 3:8, 1:8, 0:8, 0:8>>)
+    ).
+
 parse_serialize(Packet) ->
     parse_serialize(Packet, #{strict_mode => true}).
 

apps/emqx/test/emqx_packet_SUITE.erl

@@ -212,7 +212,9 @@ t_check_publish(_) ->
         ?PUBLISH_PACKET(1, <<"topic">>, 1, #{'Topic-Alias' => 0}, <<"payload">>)
     ),
     %% TODO::
-    %% {error, ?RC_PROTOCOL_ERROR} = emqx_packet:check(?PUBLISH_PACKET(1, <<"topic">>, 1, #{'Subscription-Identifier' => 10}, <<"payload">>)),
+    %% {error, ?RC_PROTOCOL_ERROR} = emqx_packet:check(
+    %%    ?PUBLISH_PACKET(1, <<"topic">>, 1, #{'Subscription-Identifier' => 10}, <<"payload">>)
+    %%),
     ok = emqx_packet:check(
         ?PUBLISH_PACKET(1, <<"topic">>, 1, #{'Subscription-Identifier' => 10}, <<"payload">>)
     ),
@@ -414,5 +416,5 @@ t_format(_) ->
 
 t_parse_empty_publish(_) ->
     %% 52: 0011(type=PUBLISH) 0100 (QoS=2)
-    {ok, Packet, <<>>, {none, _}} = emqx_frame:parse(<<52, 0>>),
+    Packet = #mqtt_packet_publish{topic_name = <<>>},
     ?assertEqual({error, ?RC_PROTOCOL_ERROR}, emqx_packet:check(Packet)).

apps/emqx/test/emqx_ws_connection_SUITE.erl

@@ -535,7 +535,7 @@ t_parse_incoming(_) ->
 
 t_parse_incoming_frame_error(_) ->
     {Packets, _St} = ?ws_conn:parse_incoming(<<3, 2, 1, 0>>, [], st()),
-    FrameError = {frame_error, function_clause},
+    FrameError = {frame_error, malformed_packet},
     [{incoming, FrameError}] = Packets.
 
 t_handle_incomming_frame_error(_) ->

changes/v5.0.12-en.md

@@ -23,3 +23,5 @@
 - Fix doc and schema for `/trace` API [#9468](https://github.com/emqx/emqx/pull/9468).
 
 - Return `404` for `/telemetry/data` in case it's disabled [#9464](https://github.com/emqx/emqx/pull/9464).
+
+- Fix some potential MQTT packet parse errors [#9477](https://github.com/emqx/emqx/pull/9477).

changes/v5.0.12-zh.md

@@ -22,3 +22,5 @@
 - 修复 /trace API 的返回值格式和相关文档 [#9468](https://github.com/emqx/emqx/pull/9468)。
 
 - 在遥测功能未开启时，通过 /telemetry/data 请求其数据，将会返回 404 [#9464](https://github.com/emqx/emqx/pull/9464)。
+
+- 修复了一些 MQTT 协议包的潜在解析错误 [#9477](https://github.com/emqx/emqx/pull/9477)。