andreacavalli/netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

MQTT: foolproof SUBSCRIBE QoS encoding (#10874)

Browse Source 
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.
This commit is contained in:
Paul Lysak 2020-12-18 11:40:15 +02:00 committed by Norman Maurer
parent 26fc8d4614
commit 6a057338ee
2 changed files with 51 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -287,6 +287,9 @@ public final class MqttEncoder extends MessageToMessageEncoder<MqttMessage> {
// Payload // Payload
for (MqttTopicSubscription topic : payload.topicSubscriptions()) { for (MqttTopicSubscription topic : payload.topicSubscriptions()) {
writeUnsafeUTF8String(buf, topic.topicName()); writeUnsafeUTF8String(buf, topic.topicName());
if (mqttVersion == MqttVersion.MQTT_3_1_1 || mqttVersion == MqttVersion.MQTT_3_1) {
buf.writeByte(topic.qualityOfService().value());
} else {
final MqttSubscriptionOption option = topic.option(); final MqttSubscriptionOption option = topic.option();
int optionEncoded = option.retainHandling().value() << 4; int optionEncoded = option.retainHandling().value() << 4;
@ -300,6 +303,7 @@ public final class MqttEncoder extends MessageToMessageEncoder<MqttMessage> {
buf.writeByte(optionEncoded); buf.writeByte(optionEncoded);
} }
}
return buf; return buf;
} finally { } finally {

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

@ -620,6 +620,43 @@ public class MqttCodecTest {
validateSubscribePayload(message.payload(), decodedMessage.payload()); 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>();
ArgumentCaptor<MqttSubscribeMessage> captor = ArgumentCaptor.forClass(MqttSubscribeMessage.class);
mqttDecoder.decode(ctx, byteBuf);
verify(ctx).fireChannelRead(captor.capture());
final MqttSubscribeMessage decodedMessage = captor.getValue();
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 @Test
public void testUnsubAckMessageForMqtt5() throws Exception { public void testUnsubAckMessageForMqtt5() throws Exception {
when(versionAttrMock.get()).thenReturn(MqttVersion.MQTT_5); when(versionAttrMock.get()).thenReturn(MqttVersion.MQTT_5);