Reject max-priority arguments >= 256

This is the value we advertise in the docs
and it should be enforced to avoid process explosion
e.g. when an overflow value is provided.

Part of #1590.

[#157380396]

Author: michaelklishin

src/rabbit_amqqueue.erl

@@ -575,7 +575,7 @@ declare_args() ->
      {<<"x-dead-letter-routing-key">>, fun check_dlxrk_arg/2},
      {<<"x-max-length">>,              fun check_non_neg_int_arg/2},
      {<<"x-max-length-bytes">>,        fun check_non_neg_int_arg/2},
-     {<<"x-max-priority">>,            fun check_non_neg_int_arg/2},
+     {<<"x-max-priority">>,            fun check_max_priority_arg/2},
      {<<"x-overflow">>,                fun check_overflow/2},
      {<<"x-queue-mode">>,              fun check_queue_mode/2}].
 
@@ -611,6 +611,13 @@ check_message_ttl_arg({Type, Val}, Args) ->
         Error -> Error
     end.
 
+check_max_priority_arg({Type, Val}, Args) ->
+    case check_non_neg_int_arg({Type, Val}, Args) of
+        ok when Val =< 255 -> ok;
+        ok                 -> {error, {max_value_exceeded, Val}};
+        Error              -> Error
+    end.
+
 %% Note that the validity of x-dead-letter-exchange is already verified
 %% by rabbit_channel's queue.declare handler.
 check_dlxname_arg({longstr, _}, _) -> ok;

src/rabbit_priority_queue.erl

@@ -128,11 +128,14 @@ collapse_recovery(QNames, DupNames, Recovery) ->
 priorities(#amqqueue{arguments = Args}) ->
     Ints = [long, short, signedint, byte, unsignedbyte, unsignedshort, unsignedint],
     case rabbit_misc:table_lookup(Args, <<"x-max-priority">>) of
-        {Type, Max} -> case lists:member(Type, Ints) of
-                           false -> none;
-                           true  -> lists:reverse(lists:seq(0, Max))
-                       end;
-        _           -> none
+        {Type, RequestedMax} ->
+            case lists:member(Type, Ints) of
+                false -> none;
+                true  ->
+                    Max = min(RequestedMax, ?MAX_SUPPORTED_PRIORITY),
+                    lists:reverse(lists:seq(0, Max))
+            end;
+        _                    -> none
     end.
 
 %%----------------------------------------------------------------------------

test/priority_queue_SUITE.erl

@@ -17,6 +17,7 @@
 -module(priority_queue_SUITE).
 
 -include_lib("common_test/include/ct.hrl").
+-include_lib("eunit/include/eunit.hrl").
 -include_lib("amqp_client/include/amqp_client.hrl").
 
 -compile(export_all).
@@ -46,7 +47,9 @@ groups() ->
                            simple_order,
                            straight_through,
                            invoke,
-                           gen_server2_stats
+                           gen_server2_stats,
+                           negative_max_priorities,
+                           max_priorities_above_hard_limit
                           ]},
      {cluster_size_3, [], [
                            mirror_queue_auto_ack,
@@ -192,6 +195,28 @@ straight_through(Config) ->
     rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
+max_priorities_above_hard_limit(Config) ->
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+    Q = <<"max_priorities_above_hard_limit">>,
+    ?assertExit(
+       {{shutdown, {server_initiated_close, 406, _}}, _},
+       %% Note that lower values (e.g. 300) will cause overflow the byte type here.
+       %% However, values >= 256 would still be rejected when used by
+       %% other clients
+       declare(Ch, Q, 3000)),
+    rabbit_ct_client_helpers:close_connection(Conn),
+    passed.
+
+negative_max_priorities(Config) ->
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+    Q = <<"negative_max_priorities">>,
+    ?assertExit(
+       {{shutdown, {server_initiated_close, 406, _}}, _},
+       declare(Ch, Q, -10)),
+    rabbit_ct_client_helpers:close_connection(Conn),
+    passed.
+
+
 invoke(Config) ->
     %% Synthetic test to check the invoke callback, as the bug tested here
     %% is only triggered with a race condition.
@@ -669,7 +694,7 @@ get_ok(Ch, Q, Ack, PBin) ->
     {#'basic.get_ok'{delivery_tag = DTag}, #amqp_msg{payload = PBin2}} =
         amqp_channel:call(Ch, #'basic.get'{queue  = Q,
                                            no_ack = Ack =:= no_ack}),
-    PBin = PBin2,
+    ?assertEqual(PBin, PBin2),
     maybe_ack(Ch, Ack, DTag).
 
 get_payload(Ch, Q, Ack, Ps) ->