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