MQTT: foolproof SUBSCRIBE QoS encoding (netty#10874)

Motivation:

If the MQTT client specifies Subscribe Options parameters only available in MQTT v5 and tries to encode the message as MQTT v3 then an invalid QoS value is encoded

Modification:

Check MQTT version when encoding SUBSCRIBE message options, if it's 3.1 or 3.1.1 - only encode QoS, skip other options.

Result:

MqttEncoder produces a valid SUBSCRIBE message even if the client has specified options not available in the current MQTT version.

codec-mqtt/src/main/java/io/netty/handler/codec/mqtt/MqttEncoder.java

@@ -287,18 +287,22 @@ private static ByteBuf encodeSubscribeMessage(
             // Payload
             for (MqttTopicSubscription topic : payload.topicSubscriptions()) {
                 writeUnsafeUTF8String(buf, topic.topicName());
-                final MqttSubscriptionOption option = topic.option();
+                if (mqttVersion == MqttVersion.MQTT_3_1_1 || mqttVersion == MqttVersion.MQTT_3_1) {
+                    buf.writeByte(topic.qualityOfService().value());
+                } else {
+                    final MqttSubscriptionOption option = topic.option();
+
+                    int optionEncoded = option.retainHandling().value() << 4;
+                    if (option.isRetainAsPublished()) {
+                        optionEncoded |= 0x08;
+                    }
+                    if (option.isNoLocal()) {
+                        optionEncoded |= 0x04;
+                    }
+                    optionEncoded |= option.qos().value();
 
-                int optionEncoded = option.retainHandling().value() << 4;
-                if (option.isRetainAsPublished()) {
-                    optionEncoded |= 0x08;
+                    buf.writeByte(optionEncoded);
                 }
-                if (option.isNoLocal()) {
-                    optionEncoded |= 0x04;
-                }
-                optionEncoded |= option.qos().value();
-
-                buf.writeByte(optionEncoded);
             }
 
             return buf;

codec-mqtt/src/test/java/io/netty/handler/codec/mqtt/MqttCodecTest.java

@@ -640,6 +640,43 @@ public void testSubscribeMessageForMqtt5() throws Exception {
         validateSubscribePayload(message.payload(), decodedMessage.payload());
     }
 
+    @Test
+    public void testSubscribeMessageMqtt5EncodeAsMqtt3() throws Exception {
+        when(versionAttrMock.get()).thenReturn(MqttVersion.MQTT_3_1_1);
+
+        //Set parameters only available in MQTT5 to see if they're dropped when encoding as MQTT3
+        MqttProperties props = new MqttProperties();
+        props.add(new MqttProperties.IntegerProperty(PAYLOAD_FORMAT_INDICATOR.value(), 6));
+        final MqttSubscribeMessage message = MqttMessageBuilders.subscribe()
+                .messageId((short) 1)
+                .properties(props)
+                .addSubscription("/topic", new MqttSubscriptionOption(AT_LEAST_ONCE,
+                        true,
+                        true,
+                        SEND_AT_SUBSCRIBE_IF_NOT_YET_EXISTS))
+                .build();
+        ByteBuf byteBuf = MqttEncoder.doEncode(ctx, message);
+
+        final List<Object> out = new LinkedList<Object>();
+
+        mqttDecoder.decode(ctx, byteBuf, out);
+
+        assertEquals("Expected one object but got " + out.size(), 1, out.size());
+        final MqttSubscribeMessage decodedMessage = (MqttSubscribeMessage) out.get(0);
+
+        final MqttSubscribeMessage expectedMessage = MqttMessageBuilders.subscribe()
+                .messageId((short) 1)
+                .addSubscription("/topic", MqttSubscriptionOption.onlyFromQos(AT_LEAST_ONCE))
+                .build();
+        validateFixedHeaders(expectedMessage.fixedHeader(), decodedMessage.fixedHeader());
+        final MqttMessageIdAndPropertiesVariableHeader expectedHeader =
+                (MqttMessageIdAndPropertiesVariableHeader) expectedMessage.variableHeader();
+        final MqttMessageIdAndPropertiesVariableHeader actualHeader =
+                (MqttMessageIdAndPropertiesVariableHeader) decodedMessage.variableHeader();
+        validatePacketIdAndPropertiesVariableHeader(expectedHeader, actualHeader);
+        validateSubscribePayload(expectedMessage.payload(), decodedMessage.payload());
+    }
+
     @Test
     public void testUnsubAckMessageForMqtt5() throws Exception {
         when(versionAttrMock.get()).thenReturn(MqttVersion.MQTT_5);