Fix/fuzzing errors v50

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_ERR(#{hint => unexpected_packet, header_type => 'PINGRESP'});
+%% All other types of messages should not have a zero remaining length.
+parse_remaining_len(
+    <<0:8, _Rest/binary>>, Header, 1, 0, _Options
+) ->
+    ?PARSE_ERR(#{hint => zero_remaining_len, header_type => Header#mqtt_packet_header.type});
 %% 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);
@@ -255,20 +266,33 @@ packet(Header, Variable) ->
 packet(Header, Variable, Payload) ->
     #mqtt_packet{header = Header, variable = Variable, payload = Payload}.
 
-parse_packet(
-    #mqtt_packet_header{type = ?CONNECT},
-    FrameBin,
-    #{strict_mode := StrictMode}
+parse_connect(FrameBin, StrictMode) ->
+    {ProtoName, Rest} = parse_utf8_string_with_hint(FrameBin, StrictMode, invalid_proto_name),
+    case ProtoName of
+        <<"MQTT">> ->
+            ok;
+        <<"MQIsdp">> ->
+            ok;
+        _ ->
+            %% from spec: the server MAY send disconnect with reason code 0x84
+            %% we chose to close socket because the client is likely not talking MQTT anyway
+            ?PARSE_ERR(#{
+                hint => invalid_proto_name,
+                expected => <<"'MQTT' or 'MQIsdp'">>,
+                received => ProtoName
+            })
+    end,
+    parse_connect2(ProtoName, Rest, StrictMode).
+
+% Note: return malformed if reserved flag is not 0.
+parse_connect2(
+    ProtoName,
+    <<BridgeTag:4, ProtoVer:4, UsernameFlag:1, PasswordFlag:1, WillRetain:1, WillQoS:2, WillFlag:1,
+        CleanStart:1, 0:1, KeepAlive:16/big, Rest2/binary>>,
+    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),
+    {ClientId, Rest4} = parse_utf8_string_with_hint(Rest3, StrictMode, invalid_username),
     ConnPacket = #mqtt_packet_connect{
         proto_name = ProtoName,
         proto_ver = ProtoVer,
@@ -282,26 +306,56 @@ parse_packet(
         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};
+    {Username, Rest6} = parse_optional(
+        Rest5,
+        fun(Bin) ->
+            parse_utf8_string_with_hint(Bin, StrictMode, invalid_username)
+        end,
+        bool(UsernameFlag)
+    ),
+    {Password, Rest7} = parse_optional(
+        Rest6,
+        fun(Bin) ->
+            parse_utf8_string_with_hint(Bin, StrictMode, invalid_password)
+        end,
+        bool(PasswordFlag)
+    ),
+    case Rest7 of
+        <<>> ->
+            ConnPacket1#mqtt_packet_connect{username = Username, password = Password};
+        _ ->
+            ?PARSE_ERR(malformed_connect_payload)
+    end;
+parse_connect2(_ProtoName, _, _) ->
+    ?PARSE_ERR(malformed_connect_header).
+
+parse_packet(
+    #mqtt_packet_header{type = ?CONNECT},
+    FrameBin,
+    #{strict_mode := StrictMode}
+) ->
+    parse_connect(FrameBin, StrictMode);
 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,
     #{strict_mode := StrictMode, version := Ver}
 ) ->
-    {TopicName, Rest} = parse_utf8_string(Bin, StrictMode),
+    {TopicName, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_topic),
     {PacketId, Rest1} =
         case QoS of
             ?QOS_0 -> {undefined, Rest};
@@ -411,7 +465,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{
@@ -422,7 +478,7 @@ parse_will_message(
     StrictMode
 ) ->
     {Props, Rest} = parse_properties(Bin, Ver, StrictMode),
-    {Topic, Rest1} = parse_utf8_string(Rest, StrictMode),
+    {Topic, Rest1} = parse_utf8_string_with_hint(Rest, StrictMode, invalid_topic),
     {Payload, Rest2} = parse_binary_data(Rest1),
     {
         Packet#mqtt_packet_connect{
@@ -437,7 +493,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};
@@ -458,10 +516,10 @@ parse_property(<<16#01, Val, Bin/binary>>, Props, StrictMode) ->
 parse_property(<<16#02, Val:32/big, Bin/binary>>, Props, StrictMode) ->
     parse_property(Bin, Props#{'Message-Expiry-Interval' => Val}, StrictMode);
 parse_property(<<16#03, Bin/binary>>, Props, StrictMode) ->
-    {Val, Rest} = parse_utf8_string(Bin, StrictMode),
+    {Val, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_content_type),
     parse_property(Rest, Props#{'Content-Type' => Val}, StrictMode);
 parse_property(<<16#08, Bin/binary>>, Props, StrictMode) ->
-    {Val, Rest} = parse_utf8_string(Bin, StrictMode),
+    {Val, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_response_topic),
     parse_property(Rest, Props#{'Response-Topic' => Val}, StrictMode);
 parse_property(<<16#09, Len:16/big, Val:Len/binary, Bin/binary>>, Props, StrictMode) ->
     parse_property(Bin, Props#{'Correlation-Data' => Val}, StrictMode);
@@ -471,12 +529,12 @@ parse_property(<<16#0B, Bin/binary>>, Props, StrictMode) ->
 parse_property(<<16#11, Val:32/big, Bin/binary>>, Props, StrictMode) ->
     parse_property(Bin, Props#{'Session-Expiry-Interval' => Val}, StrictMode);
 parse_property(<<16#12, Bin/binary>>, Props, StrictMode) ->
-    {Val, Rest} = parse_utf8_string(Bin, StrictMode),
+    {Val, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_assigned_client_id),
     parse_property(Rest, Props#{'Assigned-Client-Identifier' => Val}, StrictMode);
 parse_property(<<16#13, Val:16, Bin/binary>>, Props, StrictMode) ->
     parse_property(Bin, Props#{'Server-Keep-Alive' => Val}, StrictMode);
 parse_property(<<16#15, Bin/binary>>, Props, StrictMode) ->
-    {Val, Rest} = parse_utf8_string(Bin, StrictMode),
+    {Val, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_authn_method),
     parse_property(Rest, Props#{'Authentication-Method' => Val}, StrictMode);
 parse_property(<<16#16, Len:16/big, Val:Len/binary, Bin/binary>>, Props, StrictMode) ->
     parse_property(Bin, Props#{'Authentication-Data' => Val}, StrictMode);
@@ -487,13 +545,13 @@ parse_property(<<16#18, Val:32, Bin/binary>>, Props, StrictMode) ->
 parse_property(<<16#19, Val, Bin/binary>>, Props, StrictMode) ->
     parse_property(Bin, Props#{'Request-Response-Information' => Val}, StrictMode);
 parse_property(<<16#1A, Bin/binary>>, Props, StrictMode) ->
-    {Val, Rest} = parse_utf8_string(Bin, StrictMode),
+    {Val, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_response_info),
     parse_property(Rest, Props#{'Response-Information' => Val}, StrictMode);
 parse_property(<<16#1C, Bin/binary>>, Props, StrictMode) ->
-    {Val, Rest} = parse_utf8_string(Bin, StrictMode),
+    {Val, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_server_reference),
     parse_property(Rest, Props#{'Server-Reference' => Val}, StrictMode);
 parse_property(<<16#1F, Bin/binary>>, Props, StrictMode) ->
-    {Val, Rest} = parse_utf8_string(Bin, StrictMode),
+    {Val, Rest} = parse_utf8_string_with_hint(Bin, StrictMode, invalid_reason_string),
     parse_property(Rest, Props#{'Reason-String' => Val}, StrictMode);
 parse_property(<<16#21, Val:16/big, Bin/binary>>, Props, StrictMode) ->
     parse_property(Bin, Props#{'Receive-Maximum' => Val}, StrictMode);
@@ -584,10 +642,18 @@ parse_utf8_pair(Bin, _StrictMode) when
         total_bytes => byte_size(Bin)
     }).
 
-parse_utf8_string(Bin, _StrictMode, false) ->
-    {undefined, Bin};
-parse_utf8_string(Bin, StrictMode, true) ->
-    parse_utf8_string(Bin, StrictMode).
+parse_utf8_string_with_hint(Bin, StrictMode, Hint) ->
+    try
+        parse_utf8_string(Bin, StrictMode)
+    catch
+        throw:{?FRAME_PARSE_ERROR, Reason} when is_map(Reason) ->
+            ?PARSE_ERR(Reason#{hint => Hint})
+    end.
+
+parse_optional(Bin, F, true) ->
+    F(Bin);
+parse_optional(Bin, _F, false) ->
+    {undefined, Bin}.
 
 parse_utf8_string(<<Len:16/big, Str:Len/binary, Rest/binary>>, true) ->
     {validate_utf8(Str), Rest};
@@ -604,7 +670,7 @@ parse_utf8_string(<<Len:16/big, Rest/binary>>, _) when
 parse_utf8_string(Bin, _) when
     2 > byte_size(Bin)
 ->
-    ?PARSE_ERR(malformed_utf8_string_length).
+    ?PARSE_ERR(#{reason => malformed_utf8_string_length}).
 
 parse_binary_data(<<Len:16/big, Data:Len/binary, Rest/binary>>) ->
     {Data, Rest};

apps/emqx/test/emqx_frame_SUITE.erl

@@ -576,7 +576,12 @@ t_serialize_parse_pingreq(_) ->
 
 t_serialize_parse_pingresp(_) ->
     PingResp = ?PACKET(?PINGRESP),
-    ?assertEqual(PingResp, parse_serialize(PingResp)).
+    Packet = serialize_to_binary(PingResp),
+    ?assertException(
+        throw,
+        {frame_parse_error, #{hint := unexpected_packet, header_type := 'PINGRESP'}},
+        emqx_frame:parse(Packet)
+    ).
 
 t_parse_disconnect(_) ->
     Packet = ?DISCONNECT_PACKET(?RC_SUCCESS),
@@ -619,6 +624,32 @@ t_serialize_parse_auth_v5(_) ->
         })
     ).
 
+t_parse_invalid_remaining_len(_) ->
+    ?assertException(
+        throw, {frame_parse_error, #{hint := zero_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>>)
+    ).
+
+t_parse_malformed_connect(_) ->
+    ?assertException(
+        throw,
+        {frame_parse_error, malformed_connect_header},
+        emqx_frame:parse(<<16, 11, 0, 6, 77, 81, 73, 115, 110, 112, 3, 130, 1, 6>>)
+    ),
+    ?assertException(
+        throw,
+        {frame_parse_error, malformed_connect_payload},
+        emqx_frame:parse(
+            <<16, 21, 0, 6, 77, 81, 73, 115, 110, 112, 3, 130, 1, 6, 0, 0, 2, 67, 49.49>>
+        )
+    ).
+
 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-en.md.orig

@@ -0,0 +1,29 @@
+# v5.0.12
+
+## Enhancements
+
+- Disable global garbage collection by `node.global_gc_interval = disabled` [#9418](https://github.com/emqx/emqx/pull/9418)。
+
+- Improve the CLI to avoid waste atom table when typing erros [#9416](https://github.com/emqx/emqx/pull/9416).
+
+- Start building MacOS packages for Apple Silicon hadrdware [#9423](https://github.com/emqx/emqx/pull/9423).
+
+- Remove support for setting shared subscriptions using the non-standard `$queue` feature [#9412](https://github.com/emqx/emqx/pull/9412).
+  Shared subscriptions are now part of the MQTT spec. Use `$share` instead.
+
+- Refactor authn API by replacing `POST /authentication/{id}/move` with `PUT /authentication/{id}/position/{position}`. [#9419](https://github.com/emqx/emqx/pull/9419).
+  Same is done for `/listeners/{listener_id}/authentication/id/...`.
+
+- Return `204` instead of `200` for `PUT /authenticator/:id` [#9434](https://github.com/emqx/emqx/pull/9434/).
+
+## Bug fixes
+
+- Fix that the obsolete SSL files aren't deleted after the ExHook config update [#9432](https://github.com/emqx/emqx/pull/9432).
+
+- Fix doc and schema for `/trace` API [#9468](https://github.com/emqx/emqx/pull/9468).
+
+<<<<<<< HEAD
+- 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).
+>>>>>>> 030a07d8e (fix(frame): fix potential parse errors found by fuzzing test)

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)。

changes/v5.0.12-zh.md.orig

@@ -0,0 +1,28 @@
+# v5.0.12
+
+## 增强
+
+- 通过 `node.global_gc_interval = disabled` 来禁用全局垃圾回收 [#9418](https://github.com/emqx/emqx/pull/9418)。
+
+- 现在，`PUT /authenticator/:id` 将会返回 204 而不再是 200 [#9434](https://github.com/emqx/emqx/pull/9434/)。
+
+- 优化命令行实现, 避免输入错误指令时, 产生不必要的原子表消耗 [#9416](https://github.com/emqx/emqx/pull/9416)。
+
+- 支持在 Apple Silicon 架构下编译苹果系统的发行版本 [#9423](https://github.com/emqx/emqx/pull/9423)。
+
+- 删除了老的共享订阅支持方式， 不再使用 `$queue` 前缀 [#9412](https://github.com/emqx/emqx/pull/9412)。
+  共享订阅自 MQTT v5.0 开始已成为协议标准，可以使用 `$share` 前缀代替 `$queue`。
+
+- 重构认证 API，使用 `PUT /authentication/{id}/position/{position}` 代替了 `POST /authentication/{id}/move` [#9419](https://github.com/emqx/emqx/pull/9419)。
+
+## 修复
+
+- 修复 ExHook 更新 SSL 相关配置后，过时的 SSL 文件没有被删除的问题 [#9432](https://github.com/emqx/emqx/pull/9432)。
+
+- 修复 /trace API 的返回值格式和相关文档 [#9468](https://github.com/emqx/emqx/pull/9468)。
+
+<<<<<<< HEAD
+- 在遥测功能未开启时，通过 /telemetry/data 请求其数据，将会返回 404 [#9464](https://github.com/emqx/emqx/pull/9464)。
+=======
+- 修复了一些 MQTT 协议包的潜在解析错误 [#9477](https://github.com/emqx/emqx/pull/9477)。
+>>>>>>> 030a07d8e (fix(frame): fix potential parse errors found by fuzzing test)