Merge pull request #6979 from thalesmg/bugfix-find-alias-clause

fix(channel): wrong case clause when alias is inexistent

CHANGES-4.3.md

@@ -19,6 +19,8 @@ File format:
 
 ### Bug fixes
 
+* Fix case where publishing to a non-existent topic alias would crash the connection [#6979]
+
 ## v4.3.12
 ### Important changes
 

src/emqx.app.src

@@ -1,7 +1,12 @@
 {application, emqx,
  [{id, "emqx"},
   {description, "EMQ X"},
-  {vsn, "4.3.13"}, % strict semver, bump manually!
+  %% Note: this version is not the same as the release version!  This
+  %% is simply the emqx `application' version, which is separate from
+  %% the emqx `release' version, which in turn is comprised of several
+  %% apps, one of which is this.  See `emqx_release.hrl' for more
+  %% info.
+  {vsn, "4.3.14"}, % strict semver, bump manually!
   {modules, []},
   {registered, []},
   {applications, [kernel,stdlib,gproc,gen_rpc,esockd,cowboy,sasl,os_mon]},

src/emqx.appup.src

@@ -1,6 +1,7 @@
 %% -*- mode: erlang -*-
 {VSN,
-  [{"4.3.12",
+  [{"4.3.13",[{load_module,emqx_channel,brutal_purge,soft_purge,[]}]},
+   {"4.3.12",
     [{load_module,emqx_vm_mon,brutal_purge,soft_purge,[]},
      {load_module,emqx_ctl,brutal_purge,soft_purge,[]},
      {load_module,emqx_metrics,brutal_purge,soft_purge,[]},
@@ -335,7 +336,8 @@
      {load_module,emqx_message,brutal_purge,soft_purge,[]},
      {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
-  [{"4.3.12",
+  [{"4.3.13",[{load_module,emqx_channel,brutal_purge,soft_purge,[]}]},
+   {"4.3.12",
     [{load_module,emqx_vm_mon,brutal_purge,soft_purge,[]},
      {load_module,emqx_ctl,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},

src/emqx_channel.erl

@@ -1341,7 +1341,7 @@ process_alias(Packet = #mqtt_packet{
         {ok, Topic} ->
             NPublish = Publish#mqtt_packet_publish{topic_name = Topic},
             {ok, Packet#mqtt_packet{variable = NPublish}, Channel};
-        false -> {error, ?RC_PROTOCOL_ERROR}
+        error -> {error, ?RC_PROTOCOL_ERROR}
     end;
 
 process_alias(#mqtt_packet{
@@ -1685,7 +1685,7 @@ run_hooks(Name, Args, Acc) ->
 
 -compile({inline, [find_alias/3, save_alias/4]}).
 
-find_alias(_, _, undefined) -> false;
+find_alias(_, _, undefined) -> error;
 find_alias(inbound, AliasId, _TopicAliases = #{inbound := Aliases}) ->
     maps:find(AliasId, Aliases);
 find_alias(outbound, Topic, _TopicAliases = #{outbound := Aliases}) ->
@@ -1739,4 +1739,3 @@ flag(false) -> 0.
 set_field(Name, Value, Channel) ->
     Pos = emqx_misc:index_of(Name, record_info(fields, channel)),
     setelement(Pos+1, Channel, Value).
-

test/emqx_channel_SUITE.erl

@@ -674,6 +674,13 @@ t_process_alias(_) ->
     {ok, #mqtt_packet{variable = #mqtt_packet_publish{topic_name = <<"t">>}}, _Chan} =
         emqx_channel:process_alias(#mqtt_packet{variable = Publish}, Channel).
 
+t_process_alias_inexistent_alias(_) ->
+    Publish = #mqtt_packet_publish{topic_name = <<>>, properties = #{'Topic-Alias' => 1}},
+    Channel = channel(),
+    ?assertEqual(
+      {error, ?RC_PROTOCOL_ERROR},
+      emqx_channel:process_alias(#mqtt_packet{variable = Publish}, Channel)).
+
 t_packing_alias(_) ->
     Packet1 = #mqtt_packet{variable = #mqtt_packet_publish{
                                          topic_name = <<"x">>,
@@ -710,6 +717,20 @@ t_packing_alias(_) ->
                    #mqtt_packet{variable = #mqtt_packet_publish{topic_name = <<"z">>}},
                    channel())).
 
+t_packing_alias_inexistent_alias(_) ->
+    Publish = #mqtt_packet_publish{topic_name = <<>>, properties = #{'Topic-Alias' => 1}},
+    Channel = channel(),
+    Packet = #mqtt_packet{variable = Publish},
+    ExpectedChannel = emqx_channel:set_field(
+                        topic_aliases,
+                        #{ inbound => #{}
+                         , outbound => #{<<>> => 1}
+                         },
+                        Channel),
+    ?assertEqual(
+      {Packet, ExpectedChannel},
+      emqx_channel:packing_alias(Packet, Channel)).
+
 t_check_pub_acl(_) ->
     ok = meck:expect(emqx_zone, enable_acl, fun(_) -> true end),
     Publish = ?PUBLISH_PACKET(?QOS_0, <<"t">>, 1, <<"payload">>),