fix(emqx_packet): no crash if publish packet has no data

src/emqx_frame.erl

@@ -156,7 +156,6 @@ parse_remaining_len(<<0:1, Len:7, Rest/binary>>, Header, Multiplier, Value,
 
 parse_frame(Bin, Header, 0, Options) ->
     {ok, packet(Header), Bin, ?none(Options)};
-
 parse_frame(Bin, Header, Length, Options) ->
     case Bin of
         <<FrameBin:Length/binary, Rest/binary>> ->

src/emqx_packet.erl

@@ -251,13 +251,16 @@ set_props(Props, #mqtt_packet_auth{} = Pkt) ->
 %% @doc Check PubSub Packet.
 -spec(check(emqx_types:packet()|publish()|subscribe()|unsubscribe())
       -> ok | {error, emqx_types:reason_code()}).
-check(#mqtt_packet{variable = PubPkt}) when is_record(PubPkt, mqtt_packet_publish) ->
+check(#mqtt_packet{header = #mqtt_packet_header{type = ?PUBLISH},
+                   variable = PubPkt}) when not is_tuple(PubPkt) ->
+    %% publish without any data
+    %% disconnect instead of crash
+    {error, ?RC_PROTOCOL_ERROR};
+check(#mqtt_packet{variable = #mqtt_packet_publish{} = PubPkt}) ->
     check(PubPkt);
-
-check(#mqtt_packet{variable = SubPkt}) when is_record(SubPkt, mqtt_packet_subscribe) ->
+check(#mqtt_packet{variable = #mqtt_packet_subscribe{} = SubPkt}) ->
     check(SubPkt);
-
-check(#mqtt_packet{variable = UnsubPkt}) when is_record(UnsubPkt, mqtt_packet_unsubscribe) ->
+check(#mqtt_packet{variable = #mqtt_packet_unsubscribe{} = UnsubPkt}) ->
     check(UnsubPkt);
 
 %% A Topic Alias of 0 is not permitted.

test/emqx_packet_SUITE.erl

@@ -309,3 +309,7 @@ t_format(_) ->
     io:format("~s", [emqx_packet:format(?UNSUBACK_PACKET(90))]),
     io:format("~s", [emqx_packet:format(?DISCONNECT_PACKET(128))]).
 
+t_parse_empty_publish(_) ->
+    %% 52: 0011(type=PUBLISH) 0100 (QoS=2)
+    {ok, Packet, <<>>, {none, _}} = emqx_frame:parse(<<52, 0>>),
+    ?assertEqual({error, ?RC_PROTOCOL_ERROR}, emqx_packet:check(Packet)).