Disconnect at pub qos > server max qos

- "If the Server included a Maximum QoS in its CONNACK response
to a Client and it receives a PUBLISH packet with a QoS greater than this
then it uses DISCONNECT with Reason Code 0x9B (QoS not supported)"
- only affects mqtt v5, server max qos is 1

deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -263,7 +263,7 @@ process_connect(State0) ->
 
 -spec process_packet(mqtt_packet(), state()) ->
     {ok, state()} |
-    {stop, disconnect, state()} |
+    {stop, {disconnect, client_initiated | server_initiated}, state()} |
     {error, Reason :: term(), state()}.
 process_packet(Packet = #mqtt_packet{fixed = #mqtt_packet_fixed{type = Type}},
                State = #state{auth_state = #auth_state{}})
@@ -272,7 +272,7 @@ process_packet(Packet = #mqtt_packet{fixed = #mqtt_packet_fixed{type = Type}},
 
 -spec process_request(packet_type(), mqtt_packet(), state()) ->
     {ok, state()} |
-    {stop, disconnect, state()} |
+    {stop, {disconnect, client_initiated | server_initiated}, state()} |
     {error, Reason :: term(), state()}.
 process_request(?PUBACK,
                 #mqtt_packet{variable = #mqtt_packet_puback{packet_id = PacketId}},
@@ -293,6 +293,20 @@ process_request(?PUBACK,
             {ok, State}
     end;
 
+%% MQTT 5 spec 3.3.1.2 QoS
+%% If the Server included a Maximum QoS in its CONNACK response
+%% to a Client and it receives a PUBLISH packet with a QoS greater than this
+%% then it uses DISCONNECT with Reason Code 0x9B (QoS not supported).
+process_request(?PUBLISH,
+                #mqtt_packet{fixed = #mqtt_packet_fixed{qos = ?QOS_2}},
+                State = #state{cfg = #cfg{
+                                        proto_ver = ?MQTT_PROTO_V5,
+                                        client_id = ClientID}}) ->
+    ?LOG_INFO("Terminating MQTT connection. QoS not supported, client ID: ~s, "
+               "protocol version: ~p, QoS: ~p",
+               [ClientID, ?MQTT_PROTO_V5, ?QOS_2]),
+    send_disconnect(?RC_QOS_NOT_SUPPORTED, State),
+    {stop, {disconnect, server_initiated}, State};
 process_request(?PUBLISH,
                 #mqtt_packet{
                    fixed = #mqtt_packet_fixed{qos = Qos,
@@ -424,7 +438,7 @@ process_request(?PINGREQ, #mqtt_packet{}, State = #state{cfg = #cfg{client_id =
 
 process_request(?DISCONNECT, #mqtt_packet{}, State) ->
     ?LOG_DEBUG("Received a DISCONNECT"),
-    {stop, disconnect, State}.
+    {stop, {disconnect, client_initiated}, State}.
 
 check_protocol_version(ProtoVersion) ->
     case lists:member(ProtoVersion, proplists:get_keys(?PROTOCOL_NAMES)) of

deps/rabbitmq_mqtt/src/rabbit_mqtt_reader.erl

@@ -362,7 +362,10 @@ process_received_bytes(Bytes, State = #state{socket = Socket,
                         {error, Reason, ProcState1} ->
                             ?LOG_ERROR("MQTT protocol error on connection ~ts: ~tp", [ConnName, Reason]),
                             {stop, {shutdown, Reason}, pstate(State, ProcState1)};
-                        {stop, disconnect, ProcState1} ->
+                        {stop, {disconnect, server_initiated} = Reason, ProcState1} ->
+                            ?LOG_ERROR("MQTT protocol error on connection ~ts: ~tp", [ConnName, Reason]),
+                            {stop, {shutdown, Reason}, pstate(State, ProcState1)};
+                        {stop, {disconnect, client_initiated}, ProcState1} ->
                             {stop, normal, {_SendWill = false, pstate(State, ProcState1)}}
                     end
             end;

deps/rabbitmq_mqtt/test/v5_SUITE.erl

@@ -39,7 +39,8 @@ cluster_size_1_tests() ->
      client_set_max_packet_size_connack,
      client_set_max_packet_size_invalid,
      message_expiry_interval,
-     message_expiry_interval_will_message
+     message_expiry_interval_will_message,
+     client_publish_qos2
     ].
 
 cluster_size_3_tests() ->
@@ -205,6 +206,14 @@ message_expiry_interval_will_message(Config) ->
     assert_nothing_received(),
     ok = emqtt:disconnect(Sub2).
 
+client_publish_qos2(Config) ->
+    Topic = ClientId = atom_to_binary(?FUNCTION_NAME),
+    {C, Connect} = start_client(ClientId, Config, 0, []),
+    ok, {_, Props} = Connect(C),
+    ?assertEqual(1, maps:get('Maximum-QoS', Props)),
+    error, {_, Response} = emqtt:publish(C, Topic, #{}, <<"msg">>, [{qos, 2}]),
+    ?assertEqual({disconnected, 155, #{}}, Response).
+
 satisfy_bazel(_Config) ->
     ok.
 

deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_handler.erl

@@ -285,7 +285,10 @@ handle_data1(Data, State = #state{socket = Socket,
                                           proc_state = ProcState1});
                         {error, Reason, _} ->
                             stop_mqtt_protocol_error(State, Reason, ConnName);
-                        {stop, disconnect, ProcState1} ->
+                        {stop, {disconnect, server_initiated}, _} ->
+                            self() ! {stop, ?CLOSE_PROTOCOL_ERROR, server_initiated_disconnect},
+                            {[], State};
+                        {stop, {disconnect, client_initiated}, ProcState1} ->
                             stop({_SendWill = false, State#state{proc_state = ProcState1}})
                     end
             end;