1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17 package io.netty.handler.codec.mqtt;
18
19 import io.netty.buffer.ByteBuf;
20 import io.netty.channel.ChannelHandlerContext;
21 import io.netty.handler.codec.DecoderException;
22 import io.netty.handler.codec.ReplayingDecoder;
23 import io.netty.handler.codec.TooLongFrameException;
24 import io.netty.handler.codec.mqtt.MqttDecoder.DecoderState;
25 import io.netty.handler.codec.mqtt.MqttProperties.IntegerProperty;
26 import io.netty.util.CharsetUtil;
27 import io.netty.util.Signal;
28 import io.netty.util.internal.ObjectUtil;
29
30 import java.nio.ByteBuffer;
31 import java.nio.charset.CharacterCodingException;
32 import java.nio.charset.CharsetDecoder;
33 import java.nio.charset.CodingErrorAction;
34 import java.util.ArrayList;
35 import java.util.List;
36
37 import static io.netty.handler.codec.mqtt.MqttCodecUtil.isValidClientId;
38 import static io.netty.handler.codec.mqtt.MqttCodecUtil.isValidMessageId;
39 import static io.netty.handler.codec.mqtt.MqttCodecUtil.isValidPublishTopicName;
40 import static io.netty.handler.codec.mqtt.MqttCodecUtil.resetUnusedFields;
41 import static io.netty.handler.codec.mqtt.MqttCodecUtil.validateFixedHeader;
42 import static io.netty.handler.codec.mqtt.MqttConstant.DEFAULT_MAX_BYTES_IN_MESSAGE;
43 import static io.netty.handler.codec.mqtt.MqttConstant.DEFAULT_MAX_CLIENT_ID_LENGTH;
44 import static io.netty.handler.codec.mqtt.MqttSubscriptionOption.RetainedHandlingPolicy;
45
46
47
48
49
50
51
52
53
54 public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
55
56
57
58
59
60
61 enum DecoderState {
62 READ_FIXED_HEADER,
63 READ_VARIABLE_HEADER,
64 READ_PAYLOAD,
65 BAD_MESSAGE,
66 }
67
68 private MqttFixedHeader mqttFixedHeader;
69 private Object variableHeader;
70 private int bytesRemainingInVariablePart;
71
72 private final int maxBytesInMessage;
73 private final int maxClientIdLength;
74 private final boolean strictUtf8Validation;
75
76
77
78 private CharsetDecoder utf8Decoder;
79 private int maxAllowedRemainingBytes;
80
81 public MqttDecoder() {
82 this(DEFAULT_MAX_BYTES_IN_MESSAGE, DEFAULT_MAX_CLIENT_ID_LENGTH, true);
83 }
84
85 public MqttDecoder(int maxBytesInMessage) {
86 this(maxBytesInMessage, DEFAULT_MAX_CLIENT_ID_LENGTH, true);
87 }
88
89 public MqttDecoder(int maxBytesInMessage, int maxClientIdLength) {
90 this(maxBytesInMessage, maxClientIdLength, true);
91 }
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106 public MqttDecoder(int maxBytesInMessage, int maxClientIdLength, boolean strictUtf8Validation) {
107 super(DecoderState.READ_FIXED_HEADER);
108 this.maxBytesInMessage = ObjectUtil.checkPositive(maxBytesInMessage, "maxBytesInMessage");
109 this.maxClientIdLength = ObjectUtil.checkPositive(maxClientIdLength, "maxClientIdLength");
110 this.strictUtf8Validation = strictUtf8Validation;
111 }
112
113 @Override
114 protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
115 switch (state()) {
116 case READ_FIXED_HEADER: try {
117 maxAllowedRemainingBytes = maxBytesInMessage;
118 int startOfFixed = buffer.readerIndex();
119 mqttFixedHeader = decodeFixedHeader(ctx, buffer, maxAllowedRemainingBytes);
120 maxAllowedRemainingBytes -= buffer.readerIndex() - startOfFixed;
121 bytesRemainingInVariablePart = mqttFixedHeader.remainingLength();
122 checkpoint(DecoderState.READ_VARIABLE_HEADER);
123
124 } catch (Exception cause) {
125 out.add(invalidMessage(cause));
126 return;
127 }
128
129 case READ_VARIABLE_HEADER: try {
130 int bytesRemainingBeforeVariableHeader = bytesRemainingInVariablePart;
131 boolean bailOut = false;
132 try {
133 variableHeader = decodeVariableHeader(ctx, buffer, mqttFixedHeader, maxAllowedRemainingBytes);
134 } catch (Signal signal) {
135 if (bytesRemainingBeforeVariableHeader > maxBytesInMessage) {
136
137
138
139 bailOut = true;
140 } else {
141
142 throw signal;
143 }
144 }
145 if (bailOut || bytesRemainingBeforeVariableHeader > maxBytesInMessage) {
146 buffer.skipBytes(actualReadableBytes());
147 throw new TooLongFrameException("message length exceeds " + maxBytesInMessage + ": "
148 + bytesRemainingBeforeVariableHeader);
149 }
150 checkpoint(DecoderState.READ_PAYLOAD);
151
152 } catch (Exception cause) {
153 out.add(invalidMessage(cause));
154 return;
155 }
156
157 case READ_PAYLOAD: try {
158 final Object decodedPayload =
159 decodePayload(
160 buffer,
161 mqttFixedHeader.messageType(),
162 maxClientIdLength,
163 variableHeader,
164 maxAllowedRemainingBytes);
165 checkpoint(DecoderState.READ_FIXED_HEADER);
166 MqttMessage message = MqttMessageFactory.newMessage(
167 mqttFixedHeader, variableHeader, decodedPayload);
168 mqttFixedHeader = null;
169 variableHeader = null;
170 out.add(message);
171 break;
172 } catch (Exception cause) {
173 out.add(invalidMessage(cause));
174 return;
175 }
176
177 case BAD_MESSAGE:
178
179 buffer.skipBytes(actualReadableBytes());
180 break;
181
182 default:
183
184 throw new Error();
185 }
186 }
187
188 private MqttMessage invalidMessage(Throwable cause) {
189 checkpoint(DecoderState.BAD_MESSAGE);
190 return MqttMessageFactory.newInvalidMessage(mqttFixedHeader, variableHeader, cause);
191 }
192
193 private void checkMaxMessageLengthRemaining(int maxAllowedRemainingBytes) {
194 if (maxAllowedRemainingBytes < 0) {
195 throw new TooLongFrameException("message length exceeds " + maxBytesInMessage + " bytes");
196 }
197 }
198
199
200
201
202
203
204
205
206
207
208
209 private MqttFixedHeader decodeFixedHeader(ChannelHandlerContext ctx, ByteBuf buffer,
210 int maxAllowedRemainingBytes) {
211 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes - 1);
212 short b1 = buffer.readUnsignedByte();
213 maxAllowedRemainingBytes--;
214
215 MqttMessageType messageType = MqttMessageType.valueOf(b1 >> 4);
216 boolean dupFlag = (b1 & 0x08) == 0x08;
217 int qosLevel = (b1 & 0x06) >> 1;
218 boolean retain = (b1 & 0x01) != 0;
219
220 switch (messageType) {
221 case PUBLISH:
222 if (qosLevel == 3) {
223 throw new DecoderException("Illegal QOS Level in fixed header of PUBLISH message ("
224 + qosLevel + ')');
225 }
226 break;
227
228 case PUBREL:
229 case SUBSCRIBE:
230 case UNSUBSCRIBE:
231 if (dupFlag) {
232 throw new DecoderException("Illegal BIT 3 in fixed header of " + messageType
233 + " message, must be 0, found 1");
234 }
235 if (qosLevel != 1) {
236 throw new DecoderException("Illegal QOS Level in fixed header of " + messageType
237 + " message, must be 1, found " + qosLevel);
238 }
239 if (retain) {
240 throw new DecoderException("Illegal BIT 0 in fixed header of " + messageType
241 + " message, must be 0, found 1");
242 }
243 break;
244
245 case AUTH:
246 case CONNACK:
247 case CONNECT:
248 case DISCONNECT:
249 case PINGREQ:
250 case PINGRESP:
251 case PUBACK:
252 case PUBCOMP:
253 case PUBREC:
254 case SUBACK:
255 case UNSUBACK:
256 if (dupFlag) {
257 throw new DecoderException("Illegal BIT 3 in fixed header of " + messageType
258 + " message, must be 0, found 1");
259 }
260 if (qosLevel != 0) {
261 throw new DecoderException("Illegal BIT 2 or 1 in fixed header of " + messageType
262 + " message, must be 0, found " + qosLevel);
263 }
264 if (retain) {
265 throw new DecoderException("Illegal BIT 0 in fixed header of " + messageType
266 + " message, must be 0, found 1");
267 }
268 break;
269 default:
270 throw new DecoderException("Unknown message type, do not know how to validate fixed header");
271 }
272
273 int remainingLength = parseRemainingLength(buffer, messageType, maxAllowedRemainingBytes);
274 MqttFixedHeader decodedFixedHeader =
275 new MqttFixedHeader(messageType, dupFlag, MqttQoS.valueOf(qosLevel), retain, remainingLength);
276 return validateFixedHeader(ctx, resetUnusedFields(decodedFixedHeader));
277 }
278
279 private int parseRemainingLength(ByteBuf buffer, MqttMessageType messageType, int maxAllowedRemainingBytes) {
280 int remainingLength = 0;
281 int multiplier = 1;
282
283 for (int i = 0; i < 4; i++) {
284 checkMaxMessageLengthRemaining(--maxAllowedRemainingBytes);
285 short digit = buffer.readUnsignedByte();
286 remainingLength += (digit & 127) * multiplier;
287
288 if ((digit & 128) == 0) {
289 return remainingLength;
290 }
291
292 multiplier *= 128;
293 }
294
295
296 throw new DecoderException("remaining length exceeds 4 digits (" + messageType + ')');
297 }
298
299
300
301
302
303
304
305
306
307 private Object decodeVariableHeader(ChannelHandlerContext ctx, ByteBuf buffer, MqttFixedHeader mqttFixedHeader,
308 int maxAllowedRemainingBytes) {
309 switch (mqttFixedHeader.messageType()) {
310 case CONNECT:
311 return decodeConnectionVariableHeader(ctx, buffer, maxAllowedRemainingBytes);
312
313 case CONNACK:
314 return decodeConnAckVariableHeader(ctx, buffer, maxAllowedRemainingBytes);
315
316 case UNSUBSCRIBE:
317 case SUBSCRIBE:
318 case SUBACK:
319 case UNSUBACK:
320 return decodeMessageIdAndPropertiesVariableHeader(ctx, buffer, maxAllowedRemainingBytes);
321
322 case PUBACK:
323 case PUBREC:
324 case PUBCOMP:
325 case PUBREL:
326 return decodePubReplyMessage(buffer, maxAllowedRemainingBytes);
327
328 case PUBLISH:
329 return decodePublishVariableHeader(ctx, buffer, mqttFixedHeader, maxAllowedRemainingBytes);
330
331 case DISCONNECT:
332 case AUTH:
333 return decodeReasonCodeAndPropertiesVariableHeader(buffer, maxAllowedRemainingBytes);
334
335 case PINGREQ:
336 case PINGRESP:
337
338 return null;
339 default:
340
341 throw new DecoderException("Unknown message type: " + mqttFixedHeader.messageType());
342 }
343 }
344
345 private MqttConnectVariableHeader decodeConnectionVariableHeader(
346 ChannelHandlerContext ctx, ByteBuf buffer, final int maxAllowedRemainingBytes) {
347 final Result<String> protoString = decodeString(buffer, maxAllowedRemainingBytes);
348 int numberOfBytesConsumed = protoString.numberOfBytesConsumed;
349
350 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes - numberOfBytesConsumed - 1);
351 final byte protocolLevel = buffer.readByte();
352 numberOfBytesConsumed += 1;
353
354 MqttVersion version = MqttVersion.fromProtocolNameAndLevel(protoString.value, protocolLevel);
355 MqttCodecUtil.setMqttVersion(ctx, version);
356
357 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes - numberOfBytesConsumed - 1);
358 final int b1 = buffer.readUnsignedByte();
359 numberOfBytesConsumed += 1;
360
361 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes - numberOfBytesConsumed - 2);
362 final int keepAlive = decodeMsbLsb(buffer);
363 numberOfBytesConsumed += 2;
364
365 final boolean hasUserName = (b1 & 0x80) == 0x80;
366 final boolean hasPassword = (b1 & 0x40) == 0x40;
367 final boolean willRetain = (b1 & 0x20) == 0x20;
368 final int willQos = (b1 & 0x18) >> 3;
369 final boolean willFlag = (b1 & 0x04) == 0x04;
370 final boolean cleanSession = (b1 & 0x02) == 0x02;
371 if (version == MqttVersion.MQTT_3_1_1 || version == MqttVersion.MQTT_5) {
372 final boolean zeroReservedFlag = (b1 & 0x01) == 0x0;
373 if (!zeroReservedFlag) {
374
375
376
377 throw new DecoderException("non-zero reserved flag");
378 }
379 }
380
381 final MqttProperties properties;
382 if (version == MqttVersion.MQTT_5) {
383 final Result<MqttProperties> propertiesResult = decodeProperties(buffer,
384 maxAllowedRemainingBytes - numberOfBytesConsumed);
385 properties = propertiesResult.value;
386 numberOfBytesConsumed += propertiesResult.numberOfBytesConsumed;
387 } else {
388 properties = MqttProperties.NO_PROPERTIES;
389 }
390
391 bytesRemainingInVariablePart -= numberOfBytesConsumed;
392 this.maxAllowedRemainingBytes -= numberOfBytesConsumed;
393 return new MqttConnectVariableHeader(
394 version.protocolName(),
395 version.protocolLevel(),
396 hasUserName,
397 hasPassword,
398 willRetain,
399 willQos,
400 willFlag,
401 cleanSession,
402 keepAlive,
403 properties);
404 }
405
406 private MqttConnAckVariableHeader decodeConnAckVariableHeader(
407 ChannelHandlerContext ctx,
408 ByteBuf buffer, int maxAllowedRemainingBytes) {
409 final MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
410 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes -= 2);
411 final boolean sessionPresent = (buffer.readUnsignedByte() & 0x01) == 0x01;
412 byte returnCode = buffer.readByte();
413
414 final MqttProperties properties;
415 if (mqttVersion == MqttVersion.MQTT_5) {
416 final Result<MqttProperties> propertiesResult = decodeProperties(buffer, maxAllowedRemainingBytes);
417 properties = propertiesResult.value;
418 bytesRemainingInVariablePart -= 2 + propertiesResult.numberOfBytesConsumed;
419 maxAllowedRemainingBytes -= propertiesResult.numberOfBytesConsumed;
420 } else {
421 properties = MqttProperties.NO_PROPERTIES;
422 bytesRemainingInVariablePart -= 2;
423 }
424 this.maxAllowedRemainingBytes = maxAllowedRemainingBytes;
425
426 return new MqttConnAckVariableHeader(MqttConnectReturnCode.valueOf(returnCode), sessionPresent, properties);
427 }
428
429 private MqttMessageIdAndPropertiesVariableHeader decodeMessageIdAndPropertiesVariableHeader(
430 ChannelHandlerContext ctx,
431 ByteBuf buffer, int maxAllowedRemainingBytes) {
432 final MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
433 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes -= 2);
434 final int packetId = decodeMessageId(buffer);
435
436 if (mqttVersion == MqttVersion.MQTT_5) {
437 final Result<MqttProperties> properties = decodeProperties(buffer, maxAllowedRemainingBytes);
438 bytesRemainingInVariablePart -= 2 + properties.numberOfBytesConsumed;
439 this.maxAllowedRemainingBytes = maxAllowedRemainingBytes - properties.numberOfBytesConsumed;
440 return new MqttMessageIdAndPropertiesVariableHeader(packetId, properties.value);
441 } else {
442 bytesRemainingInVariablePart -= 2;
443 this.maxAllowedRemainingBytes = maxAllowedRemainingBytes;
444 return new MqttMessageIdAndPropertiesVariableHeader(packetId,
445 MqttProperties.NO_PROPERTIES);
446 }
447 }
448
449 private MqttPubReplyMessageVariableHeader decodePubReplyMessage(ByteBuf buffer, int maxAllowedRemainingBytes) {
450 final int packetIdNumberOfBytesConsumed = 2;
451 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes -= packetIdNumberOfBytesConsumed);
452 final int packetId = decodeMessageId(buffer);
453
454 if (bytesRemainingInVariablePart > 3) {
455 checkMaxMessageLengthRemaining(--maxAllowedRemainingBytes);
456 final byte reasonCode = buffer.readByte();
457 final Result<MqttProperties> properties = decodeProperties(buffer, maxAllowedRemainingBytes);
458 bytesRemainingInVariablePart -= packetIdNumberOfBytesConsumed + 1 + properties.numberOfBytesConsumed;
459 this.maxAllowedRemainingBytes = maxAllowedRemainingBytes - properties.numberOfBytesConsumed;
460 return new MqttPubReplyMessageVariableHeader(packetId,
461 reasonCode,
462 properties.value);
463 } else if (bytesRemainingInVariablePart > 2) {
464 checkMaxMessageLengthRemaining(--maxAllowedRemainingBytes);
465 final byte reasonCode = buffer.readByte();
466 bytesRemainingInVariablePart -= packetIdNumberOfBytesConsumed + 1;
467 this.maxAllowedRemainingBytes = maxAllowedRemainingBytes;
468 return new MqttPubReplyMessageVariableHeader(packetId,
469 reasonCode,
470 MqttProperties.NO_PROPERTIES);
471 } else {
472 bytesRemainingInVariablePart -= packetIdNumberOfBytesConsumed;
473 this.maxAllowedRemainingBytes = maxAllowedRemainingBytes;
474 return new MqttPubReplyMessageVariableHeader(packetId,
475 (byte) 0,
476 MqttProperties.NO_PROPERTIES);
477 }
478 }
479
480 private MqttReasonCodeAndPropertiesVariableHeader decodeReasonCodeAndPropertiesVariableHeader(
481 ByteBuf buffer, int maxAllowedRemainingBytes) {
482 final byte reasonCode;
483 final MqttProperties properties;
484 if (bytesRemainingInVariablePart > 1) {
485 checkMaxMessageLengthRemaining(--maxAllowedRemainingBytes);
486 reasonCode = buffer.readByte();
487 final Result<MqttProperties> propertiesResult = decodeProperties(buffer, maxAllowedRemainingBytes);
488 properties = propertiesResult.value;
489 bytesRemainingInVariablePart -= 1 + propertiesResult.numberOfBytesConsumed;
490 this.maxAllowedRemainingBytes = maxAllowedRemainingBytes - propertiesResult.numberOfBytesConsumed;
491 } else if (bytesRemainingInVariablePart > 0) {
492 checkMaxMessageLengthRemaining(--maxAllowedRemainingBytes);
493 reasonCode = buffer.readByte();
494 properties = MqttProperties.NO_PROPERTIES;
495 --bytesRemainingInVariablePart;
496 this.maxAllowedRemainingBytes = maxAllowedRemainingBytes;
497 } else {
498 reasonCode = 0;
499 properties = MqttProperties.NO_PROPERTIES;
500 }
501
502 return new MqttReasonCodeAndPropertiesVariableHeader(reasonCode, properties);
503 }
504
505 private MqttPublishVariableHeader decodePublishVariableHeader(
506 ChannelHandlerContext ctx,
507 ByteBuf buffer,
508 MqttFixedHeader mqttFixedHeader, int maxAllowedRemainingBytes) {
509 final MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
510 final Result<String> decodedTopic = decodeString(buffer, maxAllowedRemainingBytes);
511 if (!isValidPublishTopicName(decodedTopic.value)) {
512 throw new DecoderException("invalid publish topic name: " + decodedTopic.value + " (contains wildcards)");
513 }
514 int numberOfBytesConsumed = decodedTopic.numberOfBytesConsumed;
515 maxAllowedRemainingBytes -= decodedTopic.numberOfBytesConsumed;
516
517 int messageId = -1;
518 if (mqttFixedHeader.qosLevel().value() > 0) {
519 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes -= 2);
520 messageId = decodeMessageId(buffer);
521 numberOfBytesConsumed += 2;
522 }
523
524 final MqttProperties properties;
525 if (mqttVersion == MqttVersion.MQTT_5) {
526 final Result<MqttProperties> propertiesResult = decodeProperties(buffer, maxAllowedRemainingBytes);
527 properties = propertiesResult.value;
528 numberOfBytesConsumed += propertiesResult.numberOfBytesConsumed;
529 maxAllowedRemainingBytes -= propertiesResult.numberOfBytesConsumed;
530 } else {
531 properties = MqttProperties.NO_PROPERTIES;
532 }
533
534 bytesRemainingInVariablePart -= numberOfBytesConsumed;
535 this.maxAllowedRemainingBytes = maxAllowedRemainingBytes;
536 return new MqttPublishVariableHeader(decodedTopic.value, messageId, properties);
537 }
538
539
540
541
542 private static int decodeMessageId(ByteBuf buffer) {
543 final int messageId = decodeMsbLsb(buffer);
544 if (!isValidMessageId(messageId)) {
545 throw new DecoderException("invalid messageId: " + messageId);
546 }
547 return messageId;
548 }
549
550
551
552
553
554
555
556
557
558 private Object decodePayload(
559 ByteBuf buffer,
560 MqttMessageType messageType,
561 int maxClientIdLength,
562 Object variableHeader,
563 int maxAllowedRemainingBytes) {
564 switch (messageType) {
565 case CONNECT:
566 return decodeConnectionPayload(buffer, maxClientIdLength, (MqttConnectVariableHeader) variableHeader,
567 maxAllowedRemainingBytes);
568
569 case SUBSCRIBE:
570 return decodeSubscribePayload(buffer, maxAllowedRemainingBytes);
571
572 case SUBACK:
573 return decodeSubackPayload(buffer, maxAllowedRemainingBytes);
574
575 case UNSUBSCRIBE:
576 return decodeUnsubscribePayload(buffer, maxAllowedRemainingBytes);
577
578 case UNSUBACK:
579 return decodeUnsubAckPayload(buffer, maxAllowedRemainingBytes);
580
581 case PUBLISH:
582 return decodePublishPayload(buffer, maxAllowedRemainingBytes);
583
584 default:
585
586
587
588
589 validateNoBytesRemain(0);
590 return null;
591 }
592 }
593
594 private MqttConnectPayload decodeConnectionPayload(
595 ByteBuf buffer,
596 int maxClientIdLength,
597 MqttConnectVariableHeader mqttConnectVariableHeader,
598 int maxAllowedRemainingBytes) {
599 final Result<String> decodedClientId = decodeString(buffer, maxAllowedRemainingBytes);
600 final String decodedClientIdValue = decodedClientId.value;
601 final MqttVersion mqttVersion = MqttVersion.fromProtocolNameAndLevel(mqttConnectVariableHeader.name(),
602 (byte) mqttConnectVariableHeader.version());
603 if (!isValidClientId(mqttVersion, maxClientIdLength, decodedClientIdValue, !strictUtf8Validation)) {
604 throw new MqttIdentifierRejectedException("invalid clientIdentifier");
605 }
606 int numberOfBytesConsumed = decodedClientId.numberOfBytesConsumed;
607 maxAllowedRemainingBytes -= decodedClientId.numberOfBytesConsumed;
608
609 Result<String> decodedWillTopic = null;
610 byte[] decodedWillMessage = null;
611
612 final MqttProperties willProperties;
613 if (mqttConnectVariableHeader.isWillFlag()) {
614 if (mqttVersion == MqttVersion.MQTT_5) {
615 final Result<MqttProperties> propertiesResult = decodeProperties(buffer, maxAllowedRemainingBytes);
616 willProperties = propertiesResult.value;
617 numberOfBytesConsumed += propertiesResult.numberOfBytesConsumed;
618 maxAllowedRemainingBytes -= propertiesResult.numberOfBytesConsumed;
619 } else {
620 willProperties = MqttProperties.NO_PROPERTIES;
621 }
622 decodedWillTopic = decodeString(buffer, 0, 32767, maxAllowedRemainingBytes);
623 numberOfBytesConsumed += decodedWillTopic.numberOfBytesConsumed;
624 maxAllowedRemainingBytes -= decodedWillTopic.numberOfBytesConsumed;
625 decodedWillMessage = decodeByteArray(buffer, maxAllowedRemainingBytes);
626 numberOfBytesConsumed += decodedWillMessage.length + 2;
627 maxAllowedRemainingBytes -= decodedWillMessage.length + 2;
628 } else {
629 willProperties = MqttProperties.NO_PROPERTIES;
630 }
631 Result<String> decodedUserName = null;
632 byte[] decodedPassword = null;
633 if (mqttConnectVariableHeader.hasUserName()) {
634 decodedUserName = decodeString(buffer, maxAllowedRemainingBytes);
635 numberOfBytesConsumed += decodedUserName.numberOfBytesConsumed;
636 maxAllowedRemainingBytes -= decodedUserName.numberOfBytesConsumed;
637 }
638 if (mqttConnectVariableHeader.hasPassword()) {
639 decodedPassword = decodeByteArray(buffer, maxAllowedRemainingBytes);
640 numberOfBytesConsumed += decodedPassword.length + 2;
641 maxAllowedRemainingBytes -= decodedPassword.length + 2;
642 }
643
644 validateNoBytesRemain(numberOfBytesConsumed);
645 this.maxAllowedRemainingBytes = maxAllowedRemainingBytes;
646 return new MqttConnectPayload(
647 decodedClientId.value,
648 willProperties,
649 decodedWillTopic != null ? decodedWillTopic.value : null,
650 decodedWillMessage,
651 decodedUserName != null ? decodedUserName.value : null,
652 decodedPassword);
653 }
654
655 private MqttSubscribePayload decodeSubscribePayload(
656 ByteBuf buffer, int maxAllowedRemainingBytes) {
657 final List<MqttTopicSubscription> subscribeTopics = new ArrayList<MqttTopicSubscription>();
658 int numberOfBytesConsumed = 0;
659 while (numberOfBytesConsumed < bytesRemainingInVariablePart) {
660 final Result<String> decodedTopicName = decodeString(buffer, maxAllowedRemainingBytes);
661 numberOfBytesConsumed += decodedTopicName.numberOfBytesConsumed;
662 maxAllowedRemainingBytes -= decodedTopicName.numberOfBytesConsumed;
663
664 checkMaxMessageLengthRemaining(--maxAllowedRemainingBytes);
665 final short optionByte = buffer.readUnsignedByte();
666
667 MqttQoS qos = MqttQoS.valueOf(optionByte & 0x03);
668 boolean noLocal = ((optionByte & 0x04) >> 2) == 1;
669 boolean retainAsPublished = ((optionByte & 0x08) >> 3) == 1;
670 RetainedHandlingPolicy retainHandling = RetainedHandlingPolicy.valueOf((optionByte & 0x30) >> 4);
671
672 final MqttSubscriptionOption subscriptionOption = new MqttSubscriptionOption(qos,
673 noLocal,
674 retainAsPublished,
675 retainHandling);
676
677 numberOfBytesConsumed++;
678 subscribeTopics.add(new MqttTopicSubscription(decodedTopicName.value, subscriptionOption));
679 }
680 validateNoBytesRemain(numberOfBytesConsumed);
681 this.maxAllowedRemainingBytes = maxAllowedRemainingBytes;
682 return new MqttSubscribePayload(subscribeTopics);
683 }
684
685 private MqttSubAckPayload decodeSubackPayload(
686 ByteBuf buffer, int maxAllowedRemainingBytes) {
687 int bytesRemainingInVariablePart = this.bytesRemainingInVariablePart;
688 final List<Integer> grantedQos = new ArrayList<Integer>(bytesRemainingInVariablePart);
689 int numberOfBytesConsumed = 0;
690 while (numberOfBytesConsumed < bytesRemainingInVariablePart) {
691 checkMaxMessageLengthRemaining(--maxAllowedRemainingBytes);
692 int reasonCode = buffer.readUnsignedByte();
693 numberOfBytesConsumed++;
694 grantedQos.add(reasonCode);
695 }
696 validateNoBytesRemain(numberOfBytesConsumed);
697 this.maxAllowedRemainingBytes = maxAllowedRemainingBytes;
698 return new MqttSubAckPayload(grantedQos);
699 }
700
701 private MqttUnsubAckPayload decodeUnsubAckPayload(
702 ByteBuf buffer, int maxAllowedRemainingBytes) {
703 int bytesRemainingInVariablePart = this.bytesRemainingInVariablePart;
704 final List<Short> reasonCodes = new ArrayList<Short>(bytesRemainingInVariablePart);
705 int numberOfBytesConsumed = 0;
706 while (numberOfBytesConsumed < bytesRemainingInVariablePart) {
707 checkMaxMessageLengthRemaining(--maxAllowedRemainingBytes);
708 short reasonCode = buffer.readUnsignedByte();
709 numberOfBytesConsumed++;
710 reasonCodes.add(reasonCode);
711 }
712 validateNoBytesRemain(numberOfBytesConsumed);
713 this.maxAllowedRemainingBytes = maxAllowedRemainingBytes;
714 return new MqttUnsubAckPayload(reasonCodes);
715 }
716
717 private MqttUnsubscribePayload decodeUnsubscribePayload(
718 ByteBuf buffer, int maxAllowedRemainingBytes) {
719 final List<String> unsubscribeTopics = new ArrayList<String>();
720 int numberOfBytesConsumed = 0;
721 while (numberOfBytesConsumed < bytesRemainingInVariablePart) {
722 final Result<String> decodedTopicName = decodeString(buffer, maxAllowedRemainingBytes);
723 numberOfBytesConsumed += decodedTopicName.numberOfBytesConsumed;
724 maxAllowedRemainingBytes -= decodedTopicName.numberOfBytesConsumed;
725 unsubscribeTopics.add(decodedTopicName.value);
726 }
727 validateNoBytesRemain(numberOfBytesConsumed);
728 this.maxAllowedRemainingBytes = maxAllowedRemainingBytes;
729 return new MqttUnsubscribePayload(unsubscribeTopics);
730 }
731
732 private ByteBuf decodePublishPayload(ByteBuf buffer, int maxAllowedRemainingBytes) {
733 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes - bytesRemainingInVariablePart);
734 ByteBuf payload = buffer.readRetainedSlice(bytesRemainingInVariablePart);
735 this.maxAllowedRemainingBytes = maxAllowedRemainingBytes - bytesRemainingInVariablePart;
736 return payload;
737 }
738
739 private void validateNoBytesRemain(int numberOfBytesConsumed) {
740 bytesRemainingInVariablePart -= numberOfBytesConsumed;
741 if (bytesRemainingInVariablePart != 0) {
742 throw new DecoderException(
743 "non-zero remaining payload bytes: " +
744 bytesRemainingInVariablePart + " (" + mqttFixedHeader.messageType() + ')');
745 }
746 }
747
748 private Result<String> decodeString(ByteBuf buffer, int maxAllowedRemainingBytes) {
749 return decodeString(buffer, 0, Integer.MAX_VALUE, maxAllowedRemainingBytes);
750 }
751
752 private Result<String> decodeString(ByteBuf buffer, int minBytes, int maxBytes, int maxAllowedRemainingBytes) {
753 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes -= 2);
754 int size = decodeMsbLsb(buffer);
755 int numberOfBytesConsumed = 2;
756 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes - size);
757 if (size < minBytes || size > maxBytes) {
758 buffer.skipBytes(size);
759 numberOfBytesConsumed += size;
760 return new Result<String>(null, numberOfBytesConsumed);
761 }
762 final String s;
763 if (strictUtf8Validation) {
764 s = readStrictUtf8(buffer, size);
765 } else {
766 s = buffer.toString(buffer.readerIndex(), size, CharsetUtil.UTF_8);
767 buffer.skipBytes(size);
768 }
769 numberOfBytesConsumed += size;
770 return new Result<String>(s, numberOfBytesConsumed);
771 }
772
773
774
775
776
777
778 private String readStrictUtf8(ByteBuf buffer, int length) {
779 if (length == 0) {
780 return "";
781 }
782 final int readerIndex = buffer.readerIndex();
783 final ByteBuffer nioBuf;
784 if (buffer.nioBufferCount() == 1) {
785 nioBuf = buffer.nioBuffer(readerIndex, length);
786 } else {
787
788
789 byte[] tmp = new byte[length];
790 buffer.getBytes(readerIndex, tmp);
791 nioBuf = ByteBuffer.wrap(tmp);
792 }
793 if (utf8Decoder == null) {
794 utf8Decoder = CharsetUtil.UTF_8.newDecoder()
795 .onMalformedInput(CodingErrorAction.REPORT)
796 .onUnmappableCharacter(CodingErrorAction.REPORT);
797 }
798 utf8Decoder.reset();
799 final String s;
800 try {
801 s = utf8Decoder.decode(nioBuf).toString();
802 } catch (CharacterCodingException e) {
803 buffer.skipBytes(length);
804 throw new DecoderException("invalid UTF-8 string in MQTT packet", e);
805 }
806 buffer.skipBytes(length);
807
808
809 if (s.indexOf('\u0000') >= 0) {
810 throw new DecoderException("MQTT UTF-8 Encoded String must not contain U+0000");
811 }
812 return s;
813 }
814
815
816
817
818
819 private byte[] decodeByteArray(ByteBuf buffer, int maxAllowedRemainingBytes) {
820 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes -= 2);
821 int size = decodeMsbLsb(buffer);
822 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes - size);
823 byte[] bytes = new byte[size];
824 buffer.readBytes(bytes);
825 return bytes;
826 }
827
828
829 private static long packInts(int a, int b) {
830 return (((long) a) << 32) | (b & 0xFFFFFFFFL);
831 }
832
833 private static int unpackA(long ints) {
834 return (int) (ints >> 32);
835 }
836
837 private static int unpackB(long ints) {
838 return (int) ints;
839 }
840
841
842
843
844 private static int decodeMsbLsb(ByteBuf buffer) {
845 int min = 0;
846 int max = 65535;
847 short msbSize = buffer.readUnsignedByte();
848 short lsbSize = buffer.readUnsignedByte();
849 int result = msbSize << 8 | lsbSize;
850 if (result < min || result > max) {
851 result = -1;
852 }
853 return result;
854 }
855
856
857
858
859
860
861
862
863 private long decodeVariableByteInteger(ByteBuf buffer, int maxAllowedRemainingBytes) {
864 int remainingLength = 0;
865 int multiplier = 1;
866
867 for (int i = 0; i < 4; i++) {
868 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes - (i + 1));
869 short digit = buffer.readUnsignedByte();
870 remainingLength += (digit & 127) * multiplier;
871
872 if ((digit & 128) == 0) {
873 return packInts(remainingLength, i + 1);
874 }
875
876 multiplier *= 128;
877 }
878
879 throw new DecoderException("MQTT protocol limits Remaining Length to 4 bytes");
880 }
881
882 private static final class Result<T> {
883
884 private final T value;
885 private final int numberOfBytesConsumed;
886
887 Result(T value, int numberOfBytesConsumed) {
888 this.value = value;
889 this.numberOfBytesConsumed = numberOfBytesConsumed;
890 }
891 }
892
893 private Result<MqttProperties> decodeProperties(ByteBuf buffer, int maxAllowedRemainingBytes) {
894 final long propertiesLength = decodeVariableByteInteger(buffer, maxAllowedRemainingBytes);
895 int totalPropertiesLength = unpackA(propertiesLength);
896
897
898
899 final int propertiesLengthNumBytes = unpackB(propertiesLength);
900 int numberOfBytesConsumed = unpackB(propertiesLength);
901 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes - numberOfBytesConsumed);
902 if (totalPropertiesLength > 0) {
903 checkMaxMessageLengthRemaining(
904 maxAllowedRemainingBytes - numberOfBytesConsumed - totalPropertiesLength);
905
906
907
908
909
910
911 buffer.getByte(buffer.readerIndex() + totalPropertiesLength - 1);
912 }
913
914 MqttProperties decodedProperties = new MqttProperties();
915 while (numberOfBytesConsumed - propertiesLengthNumBytes < totalPropertiesLength) {
916 long propertyId = decodeVariableByteInteger(buffer, maxAllowedRemainingBytes - numberOfBytesConsumed);
917 final int propertyIdValue = unpackA(propertyId);
918 numberOfBytesConsumed += unpackB(propertyId);
919 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes - numberOfBytesConsumed);
920 MqttProperties.MqttPropertyType propertyType = MqttProperties.MqttPropertyType.valueOf(propertyIdValue);
921 switch (propertyType) {
922 case PAYLOAD_FORMAT_INDICATOR:
923 case REQUEST_PROBLEM_INFORMATION:
924 case REQUEST_RESPONSE_INFORMATION:
925 case MAXIMUM_QOS:
926 case RETAIN_AVAILABLE:
927 case WILDCARD_SUBSCRIPTION_AVAILABLE:
928 case SUBSCRIPTION_IDENTIFIER_AVAILABLE:
929 case SHARED_SUBSCRIPTION_AVAILABLE:
930 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes - numberOfBytesConsumed - 1);
931 final int b1 = buffer.readUnsignedByte();
932 numberOfBytesConsumed++;
933 decodedProperties.add(new IntegerProperty(propertyIdValue, b1));
934 break;
935 case SERVER_KEEP_ALIVE:
936 case RECEIVE_MAXIMUM:
937 case TOPIC_ALIAS_MAXIMUM:
938 case TOPIC_ALIAS:
939 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes - numberOfBytesConsumed - 2);
940 final int int2BytesResult = decodeMsbLsb(buffer);
941 numberOfBytesConsumed += 2;
942 decodedProperties.add(new IntegerProperty(propertyIdValue, int2BytesResult));
943 break;
944 case PUBLICATION_EXPIRY_INTERVAL:
945 case SESSION_EXPIRY_INTERVAL:
946 case WILL_DELAY_INTERVAL:
947 case MAXIMUM_PACKET_SIZE:
948 checkMaxMessageLengthRemaining(maxAllowedRemainingBytes - numberOfBytesConsumed - 4);
949 final int maxPacketSize = buffer.readInt();
950 numberOfBytesConsumed += 4;
951 decodedProperties.add(new IntegerProperty(propertyIdValue, maxPacketSize));
952 break;
953 case SUBSCRIPTION_IDENTIFIER:
954 long vbIntegerResult = decodeVariableByteInteger(buffer,
955 maxAllowedRemainingBytes - numberOfBytesConsumed);
956 numberOfBytesConsumed += unpackB(vbIntegerResult);
957 decodedProperties.add(new IntegerProperty(propertyIdValue, unpackA(vbIntegerResult)));
958 break;
959 case CONTENT_TYPE:
960 case RESPONSE_TOPIC:
961 case ASSIGNED_CLIENT_IDENTIFIER:
962 case AUTHENTICATION_METHOD:
963 case RESPONSE_INFORMATION:
964 case SERVER_REFERENCE:
965 case REASON_STRING:
966 final Result<String> stringResult = decodeString(buffer,
967 maxAllowedRemainingBytes - numberOfBytesConsumed);
968 numberOfBytesConsumed += stringResult.numberOfBytesConsumed;
969 decodedProperties.add(new MqttProperties.StringProperty(propertyIdValue, stringResult.value));
970 break;
971 case USER_PROPERTY:
972 final Result<String> keyResult = decodeString(buffer,
973 maxAllowedRemainingBytes - numberOfBytesConsumed);
974 final Result<String> valueResult = decodeString(buffer,
975 maxAllowedRemainingBytes - numberOfBytesConsumed - keyResult.numberOfBytesConsumed);
976 numberOfBytesConsumed += keyResult.numberOfBytesConsumed;
977 numberOfBytesConsumed += valueResult.numberOfBytesConsumed;
978 decodedProperties.add(new MqttProperties.UserProperty(keyResult.value, valueResult.value));
979 break;
980 case CORRELATION_DATA:
981 case AUTHENTICATION_DATA:
982 final byte[] binaryDataResult = decodeByteArray(buffer,
983 maxAllowedRemainingBytes - numberOfBytesConsumed);
984 numberOfBytesConsumed += binaryDataResult.length + 2;
985 decodedProperties.add(new MqttProperties.BinaryProperty(propertyIdValue, binaryDataResult));
986 break;
987 default:
988
989 throw new DecoderException("Unknown property type: " + propertyType);
990 }
991 }
992
993 return new Result<MqttProperties>(decodedProperties, numberOfBytesConsumed);
994 }
995 }