emqx: auth、acl集成

dao/src/main/java/cc/iotkit/dao/DeviceAclRepository.java

@@ -1,18 +0,0 @@
-package cc.iotkit.dao;
-
-import cc.iotkit.model.device.DeviceAclInfo;
-import org.springframework.data.mongodb.repository.MongoRepository;
-import org.springframework.stereotype.Repository;
-
-import java.util.List;
-
-@Repository
-public interface DeviceAclRepository extends MongoRepository<DeviceAclInfo, String> {
-
-    DeviceAclInfo findByProductKeyAndDeviceName(String productKey, String deviceName);
-
-    DeviceAclInfo findByClientId(String clientId);
-
-    List<DeviceAclInfo> findByDeviceName(String deviceName);
-
-}

dao/src/main/java/cc/iotkit/dao/DeviceAuthRepository.java

@@ -1,18 +0,0 @@
-package cc.iotkit.dao;
-
-import cc.iotkit.model.device.DeviceAuthInfo;
-import org.springframework.data.mongodb.repository.MongoRepository;
-import org.springframework.stereotype.Repository;
-
-import java.util.List;
-
-@Repository
-public interface DeviceAuthRepository extends MongoRepository<DeviceAuthInfo, String> {
-
-    DeviceAuthInfo findByProductKeyAndDeviceName(String productKey, String deviceName);
-
-    DeviceAuthInfo findByClientId(String clientId);
-
-    List<DeviceAuthInfo> findByDeviceName(String deviceName);
-
-}

manager/src/main/resources/application.yml

@@ -10,19 +10,19 @@ spring:
 
   data:
     mongodb:
-      uri: mongodb://填写mongodb地址/admin
+      uri: mongodb://iotkit:123456@192.168.100.118:27017/?authSource=iotkit
       database: iotkit
 
   elasticsearch:
     rest:
-      uris: http://elasticsearch 连接地址
+      uris: http://192.168.100.118:9200
       connection-timeout: 10s
 
   redis:
-    host: redis地址
+    host: 192.168.100.50
     port: 6379
     database: 0
-    password:
+    password: 5tgbNHY^
 
   cache:
     cache-names: foo,bar
@@ -33,10 +33,10 @@ spring:
       matching-strategy: ant_path_matcher
 
 aliyun:
-  bucketId: 填写阿里云bucketId
+  bucketId: iotkit-images
   endpoint: oss-cn-shenzhen.aliyuncs.com
-  accessKeyId: 填写阿里云accessKeyId
-  accessKeySecret: 填写阿里云accessKeySecret
+  accessKeyId: LTAI4FmGGouSnScYrGe7rQLA
+  accessKeySecret: 9S0xjffwmGW2B3HsE4Efw1TqUy8YXC
 
 sa-token:
   # token名称 (同时也是cookie名称)
@@ -53,14 +53,24 @@ sa-token:
   token-style: uuid
   # 是否输出操作日志
   is-log: false
-  
+
+security:
+  oauth2:
+    resource:
+      filter-order: 3
+
+oauth2:
+  auth-server-url: http://127.0.0.1:8086
+
 pulsar:
-  broker: pulsar://pulsar broker地址:6650
-  service: http://pulsar 服务地址:8080
+  broker: pulsar://192.168.100.66:6650
+  service: http://192.168.100.66:8080
 
 app:
   systemRole: iot_system_user
-
+  aligenie:
+    push:
+      device:
 mqtt:
-  url: tcp://填写mqtt连接地址
+  url: tcp:/192.168.100.118:1883
 

model/src/main/java/cc/iotkit/model/device/DeviceAclInfo.java

@@ -1,38 +0,0 @@
-package cc.iotkit.model.device;
-
-import cc.iotkit.model.Owned;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-import org.springframework.data.annotation.Id;
-import org.springframework.data.mongodb.core.mapping.Document;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-@Data
-@Document
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-public class DeviceAclInfo {
-
-    @Id
-    private String id;
-
-    private String deviceId;
-
-    private String productKey;
-
-    private String deviceName;
-
-    private String topic;
-
-    private String access;
-
-    private String parentId;
-
-}

model/src/main/java/cc/iotkit/model/device/DeviceAuthInfo.java

@@ -1,40 +0,0 @@
-package cc.iotkit.model.device;
-
-import cc.iotkit.model.Owned;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-import org.springframework.data.annotation.Id;
-import org.springframework.data.mongodb.core.mapping.Document;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-@Data
-@Document
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-public class DeviceAuthInfo {
-
-    @Id
-    private String id;
-
-    private String deviceId;
-
-    private String productKey;
-
-    private String deviceName;
-
-    private String clientId;
-
-    private String userName;
-
-    private String password;
-
-    private String salt;
-
-}

protocol-gateway/component-server/src/main/java/cc/iotkit/comps/DeviceMessageHandler.java

@@ -48,6 +48,8 @@ public class DeviceMessageHandler implements IMessageHandler {
         this.component = component;
         this.converter = converter;
         this.deviceBehaviourService = deviceBehaviourService;
+
+        engine.put("component", component);
         scriptObj = engine.eval(String.format("new (function () {\n%s})()", script));
     }
 
@@ -64,6 +66,7 @@ public class DeviceMessageHandler implements IMessageHandler {
             if (!(data instanceof Map)) {
                 throw new BizException("script result data is incorrect");
             }
+
             Map<String, Object> dataMap = (Map) data;
             //获取动作数据
             Action action = getAction(result.get("action"));

protocol-gateway/component-server/src/main/java/cc/iotkit/comps/service/DeviceBehaviourService.java

@@ -199,6 +199,7 @@ public class DeviceBehaviourService {
                                   boolean online) {
         DeviceInfo device = deviceRepository.findByProductKeyAndDeviceName(productKey, deviceName);
         if (device == null) {
+            log.warn(String.format("productKey: %s,device: %s,online: %s",productKey,device,online));
             throw new BizException("device does not exist");
         }
         deviceStateChange(device, online);

protocol-gateway/emqx-component/src/main/java/cc/iotkit/comp/emqx/AuthVerticle.java

@@ -1,14 +1,9 @@
 package cc.iotkit.comp.emqx;
 
-import cc.iotkit.common.Constants;
-import cc.iotkit.common.utils.JsonUtil;
 import cc.iotkit.comp.IMessageHandler;
-import cc.iotkit.comp.utils.SpringUtils;
-import cc.iotkit.dao.DeviceRepository;
 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.http.HttpMethod;
 import io.vertx.core.http.HttpServer;
-import io.vertx.core.json.JsonObject;
 import io.vertx.ext.web.Router;
 import io.vertx.ext.web.handler.BodyHandler;
 import lombok.extern.slf4j.Slf4j;
@@ -44,17 +39,11 @@ public class AuthVerticle extends AbstractVerticle {
 
         //第二步 配置Router解析url
         backendRouter.route(HttpMethod.POST, "/mqtt/auth").handler(rc -> {
-            JsonObject json = rc.getBodyAsJson();
-
-            String clientid = json.getString("clientid", "");
-            String username = json.getString("username", "");
-            String password = json.getString("password", "");
-
-            log.info(String.format("clientid: %s, username: %s, password: %s", clientid, username, password));
-
+            String json = rc.getBodyAsString();
             try {
-
-                //executor.onReceive(new HashMap<>(), "auth", json.toString());
+                Map<String, Object> head = new HashMap<>();
+                head.put("topic", "/mqtt/auth");
+                executor.onReceive(head, "auth", json);
                 rc.response().setStatusCode(200)
                         .end();
             } catch (Throwable e) {
@@ -64,30 +53,11 @@ public class AuthVerticle extends AbstractVerticle {
             }
         });
         backendRouter.route(HttpMethod.POST, "/mqtt/acl").handler(rc -> {
-            JsonObject json = rc.getBodyAsJson();
+            String json = rc.getBodyAsString();
             try {
-                String clientid = json.getString("clientid", "");
-                String topic = json.getString("topic", "");
-                String access = json.getString("access", "").equals("1") ? "subscribe" : "publish"; //1 - subscribe, 2 - publish
-
-                log.info(String.format("clientid: %s, username: %s, password: %s", clientid, topic, access));
-
-
                 Map<String, Object> head = new HashMap<>();
-                head.put("topic", topic);
-
-                /**
-                 * 1、匹配clientId, 匹配topic (topic白名单)
-                 */
-                if (topic.matches(Constants.MQTT.DEVICE_SUBSCRIBE_TOPIC)) {
-                    DeviceRepository deviceRepository = SpringUtils.getBean(DeviceRepository.class);
-
-                    String dd = JsonUtil.toJsonString(deviceRepository.findAll().get(0));
-                    log.info(dd);
-
-                    executor.onReceive(head, access, json.toString());
-                }
-
+                head.put("topic", "/mqtt/acl");
+                executor.onReceive(head, "acl", json);
 
                 rc.response().setStatusCode(200)
                         .end();

protocol-gateway/emqx-component/src/main/java/cc/iotkit/comp/emqx/EmqxDeviceComponent.java

@@ -7,6 +7,7 @@ import cc.iotkit.comp.AbstractDeviceComponent;
 import cc.iotkit.comp.CompConfig;
 import cc.iotkit.comp.IMessageHandler;
 import cc.iotkit.comp.model.DeviceState;
+import cc.iotkit.comp.utils.SpringUtils;
 import cc.iotkit.converter.DeviceMessage;
 import cc.iotkit.converter.ThingService;
 import cc.iotkit.model.device.message.ThingModelMessage;
@@ -19,15 +20,14 @@ import io.vertx.mqtt.MqttClient;
 import io.vertx.mqtt.MqttClientOptions;
 import lombok.*;
 import org.apache.commons.beanutils.BeanUtils;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.lang.reflect.InvocationTargetException;
 import java.nio.charset.Charset;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.CountDownLatch;
 
 
@@ -42,6 +42,10 @@ public class EmqxDeviceComponent extends AbstractDeviceComponent {
     private EmqxConfig mqttConfig;
     MqttClient client;
 
+    //组件mqtt clientId，默认通过mqtt auth验证。
+    private Set<String> compMqttClientIdList = new HashSet<>();
+
+
     private final Map<String, Device> deviceChildToParent = new HashMap<>();
 
     private TransparentConverter transparentConverter = new TransparentConverter();
@@ -56,6 +60,8 @@ public class EmqxDeviceComponent extends AbstractDeviceComponent {
     @Override
     public void start() {
         try {
+            compMqttClientIdList.add(mqttConfig.getClientId());
+
             authVerticle.setExecutor(getHandler());
             countDownLatch = new CountDownLatch(1);
             Future<String> future = vertx.deployVerticle(authVerticle);
@@ -78,19 +84,18 @@ public class EmqxDeviceComponent extends AbstractDeviceComponent {
                     .setCleanSession(true)
                     .setKeepAliveInterval(60);
 
+
             if (mqttConfig.isSsl()) {
                 options.setSsl(true)
                         .setTrustAll(true);
             }
             client = MqttClient.create(vertx, options);
 
-
             // handler will be called when we have a message in topic we subscribe for
             /*client.publishHandler(p -> {
                 log.info("Client received message on [{}] payload [{}] with QoS [{}]", p.topicName(), p.payload().toString(Charset.defaultCharset()), p.qosLevel());
             });*/
 
-
             List<String> topics = mqttConfig.getSubscribeTopics();
             Map<String, Integer> subscribes = new HashMap<>();
 
@@ -115,77 +120,14 @@ public class EmqxDeviceComponent extends AbstractDeviceComponent {
 
                 try {
                     IMessageHandler messageHandler = getHandler();
-
                     if (messageHandler != null) {
                         Map<String, Object> head = new HashMap<>();
                         head.put("topic", topic);
-                        if (topic.equals("/sys/client/connected")) {
-                            JsonNode payloadJson = JsonUtil.parse(payload);
-                            String clientId = payloadJson.get("clientid").textValue();
-                            log.warn("client connection connected,clientId:{}", clientId);
-                            head.put("clientId", clientId);
-                            messageHandler.onReceive(head, "connect", payload);
-                            return;
-                        }
-
-                        //连接断开
-                        if (topic.equals("/sys/client/disconnected")) {
-                            JsonNode payloadJson = JsonUtil.parse(payload);
-                            String clientId = payloadJson.get("clientid").textValue();
-                            log.warn("client connection closed,clientId:{}", clientId);
-                            head.put("clientId", clientId);
-                            messageHandler.onReceive(head, "disconnect", payload);
-                            return;
-                        }
-
-                        /**
-                        ** 子设备在线离线状态（topic: ^/sys/.+/.+/c/#$）**： 改为从 从 acl 访问控制 获取离线在线状态。
-
-
-                        if (topic.equals("/sys/session/subscribed")) {
-                            JsonNode payloadJson = JsonUtil.parse(payload);
-                            String _topic = payloadJson.get("topic").textValue();
-
-                            //在线
-                            if (_topic.matches(Constants.MQTT.DEVICE_SUBSCRIBE_TOPIC)) {
-                                //head.put("topic", _topic);
-                                String clientId = payloadJson.get("clientid").textValue();
-                                log.warn("session subscribe, topic:{}", _topic);
-                                head.put("clientId", clientId);
-                                messageHandler.onReceive(head, "subscribe", payload);
-                            }
-
-                            return;
-                        }
-
-
-                        if (topic.equals("/sys/session/unsubscribed")) {
-                            JsonNode payloadJson = JsonUtil.parse(payload);
-                            String _topic = payloadJson.get("topic").textValue();
-
-                            //离线
-                            if (_topic.matches(Constants.MQTT.DEVICE_SUBSCRIBE_TOPIC)) {
-                                //head.put("topic", _topic);
-                                String clientId = payloadJson.get("clientid").textValue();
-                                log.warn("session unsubscribe, topic:{}", _topic);
-                                head.put("clientId", clientId);
-                                messageHandler.onReceive(head, "unsubscribe", payload);
-                            }
-
-                            return;
-                        }*/
-
-                        String[] parts = topic.split("/");
-                        if (parts.length < 5) {
-                            log.error("message topic is illegal.");
-                            return;
-                        }
-
                         messageHandler.onReceive(head, "", payload);
 
                     }
                 } catch (Exception e) {
-                    log.error("message topic is illegal.", e);
+                    log.error("message is illegal.", e);
                 }
             });
 
@@ -207,10 +149,6 @@ public class EmqxDeviceComponent extends AbstractDeviceComponent {
                 log.error("client fail: ", event.getCause());
             });
 
-            /** client.pingResponseHandler(s -> {
-             log.info("We have just received PINGRESP packet");
-             });*/
-
         } catch (Throwable e) {
             throw new BizException("start emqx auth component error", e);
         }
@@ -311,6 +249,11 @@ public class EmqxDeviceComponent extends AbstractDeviceComponent {
         return transparentConverter.encode(service, device);
     }
 
+    public Object getCompMqttClientIdList(){
+        String[] result =  compMqttClientIdList.toArray(new String[0]);
+        return JsonUtil.toJsonString(result);
+    }
+
     @Data
     public static class Message {
         private String topic;