refactor(emqx_frame): better error handling

apps/emqx/src/emqx_frame.erl

@@ -187,14 +187,14 @@ parse_remaining_len(
 ) ->
     parse_frame(Rest, Header, 0, Options);
 parse_remaining_len(
-    <<0:8, Rest/binary>>, Header = #mqtt_packet_header{type = ?PINGRESP}, 1, 0, Options
+    <<0:8, _Rest/binary>>, _Header = #mqtt_packet_header{type = ?PINGRESP}, 1, 0, _Options
 ) ->
-    parse_frame(Rest, Header, 0, Options);
+    ?PARSE_ERR(#{hint => unexpected_packet, header_type => 'PINGRESP'});
-%% Without this clause a crash may will happen when data incorrect, this was found by fuzzing test
+%% All other types of messages should not have a zero remaining length.
 parse_remaining_len(
-    <<0:8, _Rest/binary>>, _Header, 1, 0, _Options
+    <<0:8, _Rest/binary>>, Header, 1, 0, _Options
 ) ->
-    ?PARSE_ERR(invalid_remaining_len);
+    ?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);
@@ -266,42 +266,75 @@ packet(Header, Variable) ->
 packet(Header, Variable, Payload) ->
     #mqtt_packet{header = Header, variable = Variable, payload = Payload}.
 
+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
+) ->
+    {Properties, Rest3} = parse_properties(Rest2, ProtoVer, StrictMode),
+    {ClientId, Rest4} = parse_utf8_string_with_hint(Rest3, StrictMode, invalid_username),
+    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_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}
 ) ->
-    {ProtoName, Rest} = parse_utf8_string(FrameBin, StrictMode),
+    parse_connect(FrameBin, StrictMode);
-    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>>,
@@ -322,7 +355,7 @@ parse_packet(
     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};
@@ -445,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{
@@ -483,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);
@@ -496,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);
@@ -512,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);
@@ -609,10 +642,18 @@ parse_utf8_pair(Bin, _StrictMode) when
         total_bytes => byte_size(Bin)
     }).
 
-parse_utf8_string(Bin, _StrictMode, false) ->
+parse_utf8_string_with_hint(Bin, StrictMode, Hint) ->
-    {undefined, Bin};
+    try
-parse_utf8_string(Bin, StrictMode, true) ->
+        parse_utf8_string(Bin, StrictMode)
-    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};
@@ -629,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),
@@ -621,7 +626,7 @@ t_serialize_parse_auth_v5(_) ->
 
 t_parse_invalid_remaining_len(_) ->
     ?assertException(
-        throw, {frame_parse_error, invalid_remaining_len}, emqx_frame:parse(<<?CONNECT, 0>>)
+        throw, {frame_parse_error, #{hint := zero_remaining_len}}, emqx_frame:parse(<<?CONNECT, 0>>)
     ).
 
 t_parse_malformed_properties(_) ->
@@ -631,6 +636,20 @@ t_parse_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}).
 

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