add mqtt连接 添加账密

fixed 标签id错误

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/auth/MqttDefaultAuth.java

@@ -1,12 +1,17 @@
 package org.jetlinks.community.network.mqtt.auth;
 
+import cn.hutool.core.util.StrUtil;
+import org.apache.commons.codec.digest.DigestUtils;
+import org.jetlinks.core.Value;
 import org.jetlinks.core.device.AuthenticationRequest;
 import org.jetlinks.core.device.AuthenticationResponse;
 import org.jetlinks.core.device.DeviceOperator;
+import org.jetlinks.core.device.MqttAuthenticationRequest;
 import org.jetlinks.supports.official.JetLinksAuthenticator;
 import reactor.core.publisher.Mono;
 
 import javax.annotation.Nonnull;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author lifang
@@ -18,6 +23,46 @@ import javax.annotation.Nonnull;
 public class MqttDefaultAuth extends JetLinksAuthenticator {
     @Override
     public Mono<AuthenticationResponse> authenticate(@Nonnull AuthenticationRequest request, @Nonnull DeviceOperator deviceOperation) {
-        return Mono.just(AuthenticationResponse.success(deviceOperation.getDeviceId()));
+        if (request instanceof MqttAuthenticationRequest) {
+            MqttAuthenticationRequest mqtt = ((MqttAuthenticationRequest) request);
+            // secureId|timestamp
+            String username = mqtt.getUsername();
+            // md5(secureId|timestamp|secureKey)
+            String password = mqtt.getPassword();
+            String requestSecureId;
+            try {
+//                String[] arr = username.split("[|]");
+//                if (arr.length <= 1) {
+//                    return Mono.just(AuthenticationResponse.error(401, "用户名格式错误"));
+//                }
+//                requestSecureId = arr[0];
+//                long time = Long.parseLong(arr[1]);
+                //和设备时间差大于5分钟则认为无效
+//                if (Math.abs(System.currentTimeMillis() - time) > TimeUnit.MINUTES.toMillis(5)) {
+//                    return Mono.just(AuthenticationResponse.error(401, "设备时间不同步"));
+//                }
+                return deviceOperation
+                    .getConfigs("secureId", "secureKey")
+                    .map(conf -> {
+                        String secureId = conf.getValue("secureId").map(Value::asString).orElse(null);
+
+                        String secureKey = conf
+                            .getValue("secureKey")
+                            .map(Value::asString)
+                            .orElse(null);
+                        //签名
+                        String digest = DigestUtils.md5Hex(username + "|" + secureKey);
+                        if ((StrUtil.isEmpty(secureId)||username.equals(secureId))
+                            && (StrUtil.isEmpty(secureKey)||password.equals(secureKey))) {
+                            return AuthenticationResponse.success(deviceOperation.getDeviceId());
+                        } else {
+                            return AuthenticationResponse.error(401, "密钥错误");
+                        }
+                    });
+            } catch (NumberFormatException e) {
+                return Mono.just(AuthenticationResponse.error(401, "用户名格式错误"));
+            }
+        }
+        return Mono.just(AuthenticationResponse.error(400, "不支持的授权类型:" + request));
     }
 }

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttServerDeviceGateway.java

@@ -124,8 +124,8 @@ class MqttServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGate
             .publishOn(Schedulers.parallel())
             .flatMap(this::handleConnection)
             .flatMap(tuple3 -> handleAuthResponse(tuple3.getT1(), tuple3.getT2(), tuple3.getT3()))
-            .flatMap(this::handleSubscriptionTopic)
-            .flatMap(this::handleUnSubscriptionTopic)
+            .doOnNext(this::handleSubscriptionTopic)
+            .doOnNext(this::handleUnSubscriptionTopic)
             .flatMap(tp -> handleAcceptedMqttConnection(tp.getT1(), tp.getT2(), tp.getT3()), Integer.MAX_VALUE)
             .onErrorContinue((err, obj) -> log.error("处理MQTT连接失败", err))
             .subscriberContext(ReactiveLogger.start("network", mqttServer.getId()))
@@ -216,8 +216,8 @@ class MqttServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGate
             }));
     }
     //处理已经建立连接的MQTT连接的主题订阅
-    private Mono<Tuple3<MqttConnection, DeviceOperator, MqttConnectionSession>> handleSubscriptionTopic(Tuple3<MqttConnection, DeviceOperator, MqttConnectionSession> tuple3) {
-        return tuple3.getT1()
+    private void handleSubscriptionTopic(Tuple3<MqttConnection, DeviceOperator, MqttConnectionSession> tuple3) {
+         tuple3.getT1()
             .handleSubscribe(true)
             .doOnNext(topic->{
                 MqttSubscribeMessage message = topic.getMessage();
@@ -227,13 +227,12 @@ class MqttServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGate
             .flatMap(ignore->
                 eventBus.publish(String.format("/dashboard/device/%s/changed/topics",
                     tuple3.getT2().getDeviceId()),new TimeSyncMessage()))
-            .collectList()
-            .map(ignore->tuple3);
+            .subscribe();
     }
 
     //取消MQTT连接的主题订阅
-    private Mono<Tuple3<MqttConnection, DeviceOperator, MqttConnectionSession>> handleUnSubscriptionTopic(Tuple3<MqttConnection, DeviceOperator, MqttConnectionSession> tuple3) {
-        return tuple3.getT1()
+    private void handleUnSubscriptionTopic(Tuple3<MqttConnection, DeviceOperator, MqttConnectionSession> tuple3) {
+         tuple3.getT1()
             .handleUnSubscribe(true)
             .doOnNext(topic->{
                 MqttUnsubscribeMessage message = topic.getMessage();
@@ -242,8 +241,7 @@ class MqttServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGate
             .flatMap(ignore->
                 eventBus.publish(String.format("/dashboard/device/%s/changed/topics",
                     tuple3.getT2().getDeviceId()),new TimeSyncMessage()))
-            .collectList()
-            .map(ignore->tuple3);
+            .subscribe();
     }
 
     //处理已经建立连接的MQTT连接

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/response/DeviceDetail.java

@@ -182,11 +182,11 @@ public class DeviceDetail {
         this.tags = new ArrayList<>(map.values());
         this.tags.sort(Comparator.comparing(DeviceTagEntity::getCreateTime));
 
-        if (StringUtils.hasText(id)) {
-            for (DeviceTagEntity tag : getTags()) {
-                tag.setId(DeviceTagEntity.createTagId(id, tag.getKey()));
-            }
-        }
+//        if (StringUtils.hasText(id)) {
+//            for (DeviceTagEntity tag : getTags()) {
+//                tag.setId(DeviceTagEntity.createTagId(id, tag.getKey()));
+//            }
+//        }
         return this;
     }