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