添加android客户端签名验证功能

broker/src/main/java/io/netty/handler/codec/mqtt/MqttConnectPayload.java

@@ -32,6 +32,7 @@ public final class MqttConnectPayload {
     private final byte[] willMessage;
     private final String userName;
     private final byte[] password;
+    private final byte[] signature;
 
     /**
      * @deprecated use {@link MqttConnectPayload#MqttConnectPayload(String,
@@ -50,7 +51,7 @@ public final class MqttConnectPayload {
           willTopic,
           willMessage.getBytes(CharsetUtil.UTF_8),
           userName,
-          password.getBytes(CharsetUtil.UTF_8));
+          password.getBytes(CharsetUtil.UTF_8), null);
     }
 
     public MqttConnectPayload(
@@ -64,7 +65,7 @@ public final class MqttConnectPayload {
                 willTopic,
                 willMessage,
                 userName,
-                password);
+                password, null);
     }
 
     public MqttConnectPayload(
@@ -73,13 +74,15 @@ public final class MqttConnectPayload {
             String willTopic,
             byte[] willMessage,
             String userName,
-            byte[] password) {
+            byte[] password,
+            byte[] signature) {
         this.clientIdentifier = clientIdentifier;
         this.willProperties = MqttProperties.withEmptyDefaults(willProperties);
         this.willTopic = willTopic;
         this.willMessage = willMessage;
         this.userName = userName;
         this.password = password;
+        this.signature = signature;
     }
 
     public String clientIdentifier() {
@@ -122,6 +125,10 @@ public final class MqttConnectPayload {
         return password;
     }
 
+    public byte[] signatureInBytes() {
+        return signature;
+    }
+
     @Override
     public String toString() {
         return new StringBuilder(StringUtil.simpleClassName(this))

broker/src/main/java/io/netty/handler/codec/mqtt/MqttConnectReturnCode.java

@@ -29,6 +29,7 @@ public enum MqttConnectReturnCode {
     CONNECTION_REFUSED_NOT_AUTHORIZED((byte) 0x05),
     CONNECTION_REFUSED_UNEXPECT_NODE((byte) 0X06),
     CONNECTION_REFUSED_SESSION_NOT_EXIST((byte) 0x07),
+    CONNECTION_REFUSED_SIGNATURE_FAILURE((byte) 0x08),
     //MQTT 5 codes
     CONNECTION_REFUSED_UNSPECIFIED_ERROR((byte) 0x80),
     CONNECTION_REFUSED_MALFORMED_PACKET((byte) 0x81),

broker/src/main/java/io/netty/handler/codec/mqtt/MqttDecoder.java

@@ -146,7 +146,7 @@ public final class MqttDecoder extends ReplayingDecoder<MqttDecoder.DecoderState
      * bytes for the remaining length.
      *
      * @see
-     * https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html#_Toc442180841
+     * <a href="https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html#_Toc442180841">...</a>
      *
      * @param buffer the buffer to decode from
      * @return the fixed header
@@ -534,7 +534,7 @@ public final class MqttDecoder extends ReplayingDecoder<MqttDecoder.DecoderState
 
         final MqttProperties willProperties;
         if (mqttConnectVariableHeader.isWillFlag()) {
-            if (mqttVersion == MqttVersion.MQTT_5) {
+            if (mqttVersion.protocolLevel() <= MqttVersion.MQTT_5.protocolLevel()) {
                 final Result<MqttProperties> propertiesResult = decodeProperties(buffer);
                 willProperties = propertiesResult.value;
                 numberOfBytesConsumed += propertiesResult.numberOfBytesConsumed;
@@ -550,6 +550,7 @@ public final class MqttDecoder extends ReplayingDecoder<MqttDecoder.DecoderState
         }
         Result<String> decodedUserName = null;
         byte[] decodedPassword = null;
+        byte[] decodedSignature = null;
         if (mqttConnectVariableHeader.hasUserName()) {
             decodedUserName = decodeString(buffer);
             numberOfBytesConsumed += decodedUserName.numberOfBytesConsumed;
@@ -558,6 +559,10 @@ public final class MqttDecoder extends ReplayingDecoder<MqttDecoder.DecoderState
             decodedPassword = decodeByteArray(buffer);
             numberOfBytesConsumed += decodedPassword.length + 2;
         }
+        if (mqttConnectVariableHeader.isWillRetain()) {
+            decodedSignature = decodeByteArray(buffer);
+            numberOfBytesConsumed += decodedSignature.length + 2;
+        }
 
         final MqttConnectPayload mqttConnectPayload =
                 new MqttConnectPayload(
@@ -566,7 +571,9 @@ public final class MqttDecoder extends ReplayingDecoder<MqttDecoder.DecoderState
                         decodedWillTopic != null ? decodedWillTopic.value : null,
                         decodedWillMessage,
                         decodedUserName != null ? decodedUserName.value : null,
-                        decodedPassword);
+                        decodedPassword,
+                        decodedSignature
+                    );
         return new Result<MqttConnectPayload>(mqttConnectPayload, numberOfBytesConsumed);
     }
 

broker/src/main/java/io/netty/handler/codec/mqtt/MqttMessageBuilders.java

@@ -208,7 +208,7 @@ public final class MqttMessageBuilders {
                             keepAliveSecs,
                             properties);
             MqttConnectPayload mqttConnectPayload =
-                    new MqttConnectPayload(clientId, willProperties, willTopic, willMessage, username, password);
+                    new MqttConnectPayload(clientId, willProperties, willTopic, willMessage, username, password, null);
             return new MqttConnectMessage(mqttFixedHeader, mqttConnectVariableHeader, mqttConnectPayload);
         }
     }