Merge branch 'develop' of gitee.com:pnoker/iot-dc3 into develop

 Conflicts:

	dc3-center/dc3-auth/src/main/resources/application.yml

	dc3-center/dc3-data/src/main/resources/application.yml

	dc3-common/dc3-base/pom.xml

dc3-center/dc3-auth/src/main/resources/application.yml

@@ -12,8 +12,6 @@ spring:
       - mysql
       - redis
       - feign
-      #用于本地启动
-      #- dev
   main:
     allow-bean-definition-overriding: true
   mvc:

dc3-center/dc3-data/src/main/resources/application.yml

@@ -20,8 +20,6 @@ spring:
       - mongo
       - redis
       - feign
-      #用于本地启动
-      #- dev
   main:
     allow-bean-definition-overriding: true
   mvc:

dc3-common/dc3-base/pom.xml

@@ -32,6 +32,7 @@
         <jwt.version>0.9.1</jwt.version>
         <hutool.version>5.6.3</hutool.version>
         <fastjson.version>1.2.76</fastjson.version>
+        <bouncycastle.version>1.46</bouncycastle.version>
     </properties>
 
     <dependencies>
@@ -96,10 +97,10 @@
             <version>${fastjson.version}</version>
         </dependency>
 
-        <!-- Bouncy Castle -->
         <dependency>
             <groupId>org.bouncycastle</groupId>
-            <artifactId>bcpkix-jdk15on</artifactId>
+            <artifactId>bcprov-jdk16</artifactId>
+            <version>${bouncycastle.version}</version>
         </dependency>
 
     </dependencies>

dc3-common/dc3-base/src/main/java/com/dc3/common/utils/KeyUtil.java

@@ -19,13 +19,25 @@ import io.jsonwebtoken.Claims;
 import io.jsonwebtoken.JwtBuilder;
 import io.jsonwebtoken.Jwts;
 import io.jsonwebtoken.SignatureAlgorithm;
+import lombok.extern.slf4j.Slf4j;
+import org.bouncycastle.jce.provider.BouncyCastleProvider;
+import org.bouncycastle.openssl.PEMReader;
 
 import javax.crypto.Cipher;
 import javax.crypto.KeyGenerator;
 import javax.crypto.SecretKey;
 import javax.crypto.spec.SecretKeySpec;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManagerFactory;
+import java.io.ByteArrayInputStream;
+import java.io.InputStreamReader;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.security.*;
+import java.security.cert.X509Certificate;
 import java.security.interfaces.RSAPrivateKey;
 import java.security.interfaces.RSAPublicKey;
 import java.security.spec.KeySpec;
@@ -39,6 +51,7 @@ import java.util.Date;
  *
  * @author pnoker
  */
+@Slf4j
 public class KeyUtil {
 
     /**
@@ -182,4 +195,49 @@ public class KeyUtil {
                 .getBody();
     }
 
+    public static SSLSocketFactory getSocketFactory(final String caCrtFile, final String crtFile, final String keyFile, final String password) {
+        try {
+            Security.addProvider(new BouncyCastleProvider());
+
+            // load CA certificate
+            PEMReader reader = new PEMReader(new InputStreamReader(new ByteArrayInputStream(Files.readAllBytes(Paths.get(caCrtFile)))));
+            X509Certificate caCert = (X509Certificate) reader.readObject();
+            reader.close();
+
+            // load client certificate
+            reader = new PEMReader(new InputStreamReader(new ByteArrayInputStream(Files.readAllBytes(Paths.get(crtFile)))));
+            X509Certificate cert = (X509Certificate) reader.readObject();
+            reader.close();
+
+            // load client private key
+            reader = new PEMReader(new InputStreamReader(new ByteArrayInputStream(Files.readAllBytes(Paths.get(keyFile)))), password::toCharArray);
+            KeyPair key = (KeyPair) reader.readObject();
+            reader.close();
+
+            // CA certificate is used to authenticate server
+            KeyStore caKs = KeyStore.getInstance(KeyStore.getDefaultType());
+            caKs.load(null, null);
+            caKs.setCertificateEntry("cacertfile", caCert);
+            TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+            tmf.init(caKs);
+
+            // client key and certificates are sent to server so it can authenticate us
+            KeyStore ks = KeyStore.getInstance(KeyStore.getDefaultType());
+            ks.load(null, null);
+            ks.setCertificateEntry("certfile", cert);
+            ks.setKeyEntry("keyfile", key.getPrivate(), password.toCharArray(), new java.security.cert.Certificate[]{cert});
+            KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+            kmf.init(ks, password.toCharArray());
+
+            // finally, create SSL socket factory
+            SSLContext context = SSLContext.getInstance("TLSv1.2");
+            context.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
+
+            return context.getSocketFactory();
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
+        }
+        return null;
+    }
+
 }

dc3-driver/dc3-driver-mqtt/src/main/java/com/dc3/driver/bean/MqttProperty.java

@@ -33,25 +33,49 @@ import java.util.List;
 @Validated({Insert.class, Update.class})
 @ConfigurationProperties(prefix = "driver.mqtt")
 public class MqttProperty {
+    @NotBlank(message = "url can't be empty")
+    private String url;
+
+    @NotBlank(message = "auth type can't be empty")
+    private AuthTypeEnum authType;
+
     @NotBlank(message = "username can't be empty")
     private String username;
 
     @NotBlank(message = "password can't be empty")
     private String password;
 
-    @NotBlank(message = "url can't be empty")
-    private String url;
+    private String caCrt = "/home/mi/Code/mi/emqx-docker/certs/ca.crt";
+    private String clientKeyPass = "";
+    private String clientKey = "/home/mi/Code/mi/emqx-docker/certs/server.key";
+    private String clientCrt = "/home/mi/Code/mi/emqx-docker/certs/server.crt";
 
     @Size(min = 1, message = "at least one qos")
     private List<Integer> qos;
 
+    @Size(min = 1, message = "at least one topic")
+    private List<String> topics;
+
     @NotNull(message = "keep alive interval can't be empty")
     private Integer keepAlive;
 
     @NotNull(message = "completion timeout can't be empty")
     private Integer completionTimeout;
 
-    @Size(min = 1, message = "at least one topic")
-    private List<String> topics;
+    public enum AuthTypeEnum {
+
+        NONE("none"),
+        CLIENT_ID("clientId"),
+        USERNAME("username"),
+        X509("x509");
+
+        @Getter
+        @Setter
+        private String authType;
+
+        AuthTypeEnum(String authType) {
+            this.authType = authType;
+        }
+    }
 
 }

dc3-driver/dc3-driver-mqtt/src/main/java/com/dc3/driver/config/MqttConfig.java

@@ -13,6 +13,8 @@
 
 package com.dc3.driver.config;
 
+import cn.hutool.core.util.StrUtil;
+import com.dc3.common.utils.KeyUtil;
 import com.dc3.driver.bean.MqttProperty;
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
@@ -55,16 +57,6 @@ public class MqttConfig {
     @Resource
     private MqttProperty mqttProperty;
 
-    @Bean
-    public MqttConnectOptions getMqttConnectOptions() {
-        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
-        mqttConnectOptions.setUserName(mqttProperty.getUsername());
-        mqttConnectOptions.setPassword(mqttProperty.getPassword().toCharArray());
-        mqttConnectOptions.setServerURIs(new String[]{mqttProperty.getUrl()});
-        mqttConnectOptions.setKeepAliveInterval(mqttProperty.getKeepAlive());
-        return mqttConnectOptions;
-    }
-
     @Bean
     public MqttPahoClientFactory mqttClientFactory() {
         DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
@@ -72,6 +64,36 @@ public class MqttConfig {
         return factory;
     }
 
+    @Bean
+    public MqttConnectOptions getMqttConnectOptions() {
+        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
+
+
+        // username & password
+        if (mqttProperty.getAuthType().equals(MqttProperty.AuthTypeEnum.USERNAME)) {
+            mqttConnectOptions.setUserName(mqttProperty.getUsername());
+            mqttConnectOptions.setPassword(mqttProperty.getPassword().toCharArray());
+        }
+
+        // tls x509
+        if (mqttProperty.getAuthType().equals(MqttProperty.AuthTypeEnum.X509)) {
+            mqttConnectOptions.setSocketFactory(KeyUtil.getSocketFactory(
+                    mqttProperty.getCaCrt(),
+                    mqttProperty.getClientCrt(),
+                    mqttProperty.getClientKey(),
+                    StrUtil.isBlank(mqttProperty.getClientKeyPass()) ? "" : mqttProperty.getClientKeyPass()
+            ));
+            if (!StrUtil.isBlank(mqttProperty.getUsername()) && !StrUtil.isBlank(mqttProperty.getPassword())) {
+                mqttConnectOptions.setUserName(mqttProperty.getUsername());
+                mqttConnectOptions.setPassword(mqttProperty.getPassword().toCharArray());
+            }
+        }
+
+        mqttConnectOptions.setServerURIs(new String[]{mqttProperty.getUrl()});
+        mqttConnectOptions.setKeepAliveInterval(mqttProperty.getKeepAlive());
+        return mqttConnectOptions;
+    }
+
     @Bean
     public MessageChannel defaultMqttInputChannel() {
         return new DirectChannel();
@@ -104,6 +126,11 @@ public class MqttConfig {
         return new DirectChannel();
     }
 
+    @Bean
+    public MessageChannel mqttOutputChannel() {
+        return new DirectChannel();
+    }
+
     @Bean
     public MessageProducer inbound() {
         MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
@@ -119,12 +146,7 @@ public class MqttConfig {
     }
 
     @Bean
-    public MessageChannel mqttOutChannel() {
-        return new DirectChannel();
-    }
-
-    @Bean
-    @ServiceActivator(inputChannel = "mqttOutChannel")
+    @ServiceActivator(inputChannel = "mqttOutputChannel")
     public MessageHandler outbound() {
         MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId, mqttClientFactory());
         messageHandler.setAsync(true);

dc3-driver/dc3-driver-mqtt/src/main/java/com/dc3/driver/service/mqtt/MqttSendHandler.java

@@ -20,7 +20,7 @@ import org.springframework.messaging.handler.annotation.Header;
 /**
  * @author pnoker
  */
-@MessagingGateway(defaultRequestChannel = "mqttOutChannel")
+@MessagingGateway(defaultRequestChannel = "mqttOutputChannel")
 public interface MqttSendHandler {
     /**
      * 使用 Default Topic & Default Qos 发送数据

dc3-driver/dc3-driver-mqtt/src/main/resources/META-INF/spring-configuration-metadata.json

@@ -0,0 +1,9 @@
+{
+  "properties": [
+    {
+      "name": "driver.mqtt.auth-type",
+      "type": "com.dc3.driver.bean.MqttProperty.AuthTypeEnum",
+      "description": "mqtt auth type."
+    }
+  ]
+}

dc3-driver/dc3-driver-mqtt/src/main/resources/application.yml

@@ -35,11 +35,14 @@ driver:
         data-type: static/url
         data: jsonString
   mqtt:
+    url: tcp://dc3-emqx:8883
+    auth-type: X509
     username: dc3
     password: dc3
-    url: tcp://dc3-rabbitmq:1883
-    #也可以连接到其他的mqtt服务器，例如：emqx，你可以在docker-compose.yml中找到，并启动它
-    #url: tcp://dc3-emqx:1884
+    ca-crt: ./certs/ca.crt
+    client-key-pass: emqx-mqtt
+    client-key: ./certs/client.key
+    client-crt: ./certs/client.crt
     qos:
       - 0
     topics:

pom.xml

@@ -348,10 +348,17 @@
                 <plugin>
                     <groupId>org.springframework.boot</groupId>
                     <artifactId>spring-boot-maven-plugin</artifactId>
-                    <version>${spring-boot.version}</version>
+                    <version>2.3.8.RELEASE</version>
                     <configuration>
                         <finalName>${project.build.finalName}</finalName>
                     </configuration>
+                    <executions>
+                        <execution>
+                            <goals>
+                                <goal>repackage</goal>
+                            </goals>
+                        </execution>
+                    </executions>
                 </plugin>
                 <!-- 过滤文件 -->
                 <plugin>