Prohibit bindings to topic exchange with empty binding key

deps/rabbit/src/rabbit_exchange_type_topic.erl

@@ -46,6 +46,8 @@ route(#exchange{name = XName}, Msg, Opts) ->
 validate(_X) ->
     ok.
 
+validate_binding(_X, #binding{key = <<>>}) ->
+    {error, {binding_invalid, "topic exchange binding key cannot be empty", []}};
 validate_binding(_X, _B) ->
     ok.
 

deps/rabbit/test/bindings_SUITE.erl

@@ -46,13 +46,15 @@ all_tests() ->
      unbind_from_volatile_queue,
      binding_args_direct_exchange,
      binding_args_fanout_exchange,
+     empty_binding_key_queue_to_topic_exchange,
 
      %% Exchange bindings
      bind_and_unbind_direct_exchange,
      bind_and_unbind_fanout_exchange,
      bind_and_delete_exchange_source,
      bind_and_delete_exchange_destination,
-     bind_to_unknown_exchange
+     bind_to_unknown_exchange,
+     empty_binding_key_exchange_to_topic_exchange
     ].
 
 %% -------------------------------------------------------------------
@@ -842,6 +844,38 @@ bind_and_delete_exchange_source(Config) ->
                  rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_binding, list, [<<"/">>])),
     ok.
 
+empty_binding_key_queue_to_topic_exchange(Config) ->
+    Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
+    Q = ?config(queue_name, Config),
+
+    #'queue.declare_ok'{} = declare(Ch, Q, []),
+    ?assertExit(
+       {{shutdown,
+         {server_initiated_close, 406,
+          <<"PRECONDITION_FAILED - topic exchange binding key cannot be empty">>}},
+        _},
+       amqp_channel:call(Ch, #'queue.bind'{exchange = <<"amq.topic">>,
+                                           queue = Q,
+                                           routing_key = <<"">>})),
+    ?assertEqual([], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_binding, list_explicit, [])).
+
+empty_binding_key_exchange_to_topic_exchange(Config) ->
+    Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
+    X = ?config(exchange_name, Config),
+
+    #'exchange.declare_ok'{} = amqp_channel:call(Ch, #'exchange.declare'{exchange = X}),
+    ?assertExit(
+       {{shutdown,
+         {server_initiated_close, 406,
+          <<"PRECONDITION_FAILED - topic exchange binding key cannot be empty">>}},
+        _},
+       amqp_channel:call(Ch, #'exchange.bind'{source = <<"amq.topic">>,
+                                              destination = X,
+                                              routing_key = <<"">>})),
+    ?assertEqual([], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_binding, list_explicit, [])).
+
 %% Internal
 
 delete_queues() ->

deps/rabbit/test/rabbit_db_topic_exchange_SUITE.erl

@@ -432,7 +432,7 @@ topic_match_empty_routing_key(Config) ->
         %% Empty routing key should only match '#'
         ?assertEqual(
            [QHash],
-           sort_dests(do_match(Config, XName, <<>>, #{}))),
+           do_match(Config, XName, <<>>, #{})),
         %% Non-empty routing key should also match '#'
         ?assertEqual(
            sort_dests([QExact, QHash]),
@@ -905,10 +905,12 @@ topic_gen() ->
          list_to_binary(lists:join(".", Words))).
 
 topic_filter_gen() ->
-    frequency([{3, ?LET(Pat, resize(5, list(mqtt_segment_gen())),
-                        mk_filter_binary_mqtt(Pat))},
-               {1, ?LET(Pat, resize(5, list(amqp_segment_gen())),
-                        mk_filter_binary_amqp(Pat))}]).
+    ?SUCHTHAT(Filter,
+              frequency([{3, ?LET(Pat, resize(5, list(mqtt_segment_gen())),
+                                  mk_filter_binary_mqtt(Pat))},
+                         {1, ?LET(Pat, resize(5, list(amqp_segment_gen())),
+                                  mk_filter_binary_amqp(Pat))}]),
+              Filter =/= <<>>).
 
 topic_word_gen() ->
     oneof([<<"a">>, <<"b">>, <<"c">>, <<"d">>,