diff --git a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java index 48bd497c6e..ed43c71027 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java +++ b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java @@ -22,6 +22,7 @@ import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; import com.google.protobuf.ByteString; +import org.thingsboard.server.common.data.relation.RelationTypeGroup; import org.thingsboard.server.exception.EntitiesLimitExceededException; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; @@ -114,6 +115,7 @@ import java.util.stream.Collectors; import static org.thingsboard.server.service.transport.BasicCredentialsValidationResult.PASSWORD_MISMATCH; import static org.thingsboard.server.service.transport.BasicCredentialsValidationResult.VALID; +import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.DEVICE_NAME_SPLIT_SEPARATOR; /** * Created by ashvayka on 05.10.18. @@ -330,84 +332,203 @@ public class DefaultTransportApiService implements TransportApiService { } private TransportApiResponseMsg handle(GetOrCreateDeviceFromGatewayRequestMsg requestMsg) { - DeviceId gatewayId = new DeviceId(new UUID(requestMsg.getGatewayIdMSB(), requestMsg.getGatewayIdLSB())); + DeviceId gatewayId = toDeviceId(requestMsg); Device gateway = deviceService.findDeviceById(TenantId.SYS_TENANT_ID, gatewayId); - Lock deviceCreationLock = deviceCreationLocks.computeIfAbsent(requestMsg.getDeviceName(), id -> new ReentrantLock()); - deviceCreationLock.lock(); + String deviceName = requestMsg.getDeviceName(); + Lock lock = deviceCreationLocks.computeIfAbsent(deviceName, k -> new ReentrantLock()); + lock.lock(); try { - Device device = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), requestMsg.getDeviceName()); - if (device == null) { - TenantId tenantId = gateway.getTenantId(); - device = new Device(); - device.setTenantId(tenantId); - device.setName(requestMsg.getDeviceName()); - device.setType(requestMsg.getDeviceType()); - device.setCustomerId(gateway.getCustomerId()); - DeviceProfile deviceProfile = deviceProfileCache.findOrCreateDeviceProfile(gateway.getTenantId(), requestMsg.getDeviceType()); - - device.setDeviceProfileId(deviceProfile.getId()); - ObjectNode additionalInfo = JacksonUtil.newObjectNode(); - additionalInfo.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayId.toString()); - device.setAdditionalInfo(additionalInfo); - device = deviceService.saveDevice(device); - - relationService.saveRelation(tenantId, new EntityRelation(gateway.getId(), device.getId(), "Created")); - - TbMsgMetaData metaData = new TbMsgMetaData(); - CustomerId customerId = gateway.getCustomerId(); - if (customerId != null && !customerId.isNullUid()) { - metaData.putValue("customerId", customerId.toString()); - } - metaData.putValue("gatewayId", gatewayId.toString()); - - DeviceId deviceId = device.getId(); - JsonNode entityNode = JacksonUtil.valueToTree(device); - TbMsg tbMsg = TbMsg.newMsg() - .type(TbMsgType.ENTITY_CREATED) - .originator(deviceId) - .customerId(customerId) - .copyMetaData(metaData) - .dataType(TbMsgDataType.JSON) - .data(JacksonUtil.toString(entityNode)) - .build(); - tbClusterService.pushMsgToRuleEngine(tenantId, deviceId, tbMsg, null); - } else { - JsonNode deviceAdditionalInfo = device.getAdditionalInfo(); - if (deviceAdditionalInfo == null) { - deviceAdditionalInfo = JacksonUtil.newObjectNode(); - } - if (deviceAdditionalInfo.isObject() && - (!deviceAdditionalInfo.has(DataConstants.LAST_CONNECTED_GATEWAY) - || !gatewayId.toString().equals(deviceAdditionalInfo.get(DataConstants.LAST_CONNECTED_GATEWAY).asText()))) { - ObjectNode newDeviceAdditionalInfo = (ObjectNode) deviceAdditionalInfo; - newDeviceAdditionalInfo.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayId.toString()); - deviceService.saveDevice(device); - } - } - GetOrCreateDeviceFromGatewayResponseMsg.Builder builder = GetOrCreateDeviceFromGatewayResponseMsg.newBuilder() - .setDeviceInfo(ProtoUtils.toDeviceInfoProto(device)); - DeviceProfile deviceProfile = deviceProfileCache.get(device.getTenantId(), device.getDeviceProfileId()); - if (deviceProfile != null) { - builder.setDeviceProfile(ProtoUtils.toProto(deviceProfile)); - } else { - log.warn("[{}] Failed to find device profile [{}] for device. ", device.getId(), device.getDeviceProfileId()); - } - return TransportApiResponseMsg.newBuilder() - .setGetOrCreateDeviceResponseMsg(builder.build()) - .build(); + Device device = findOrCreateDevice(requestMsg, gateway, gatewayId); + updateLastConnectedGateway(device, gatewayId); + return buildResponse(device); } catch (JsonProcessingException e) { - log.warn("[{}] Failed to lookup device by gateway id and name: [{}]", gatewayId, requestMsg.getDeviceName(), e); + log.warn("[{}] Failed to process device [{}]", gatewayId, deviceName, e); throw new RuntimeException(e); } catch (EntitiesLimitExceededException e) { - log.warn("[{}][{}] API limit exception: [{}]", e.getTenantId(), gatewayId, e.getMessage()); - return TransportApiResponseMsg.newBuilder() - .setGetOrCreateDeviceResponseMsg( - GetOrCreateDeviceFromGatewayResponseMsg.newBuilder() - .setError(TransportProtos.TransportApiRequestErrorCode.ENTITY_LIMIT)) - .build(); + return buildLimitErrorResponse(e, gatewayId); } finally { - deviceCreationLock.unlock(); + lock.unlock(); + } + } + + private DeviceId toDeviceId(GetOrCreateDeviceFromGatewayRequestMsg requestMsg) { + return new DeviceId(new UUID( + requestMsg.getGatewayIdMSB(), + requestMsg.getGatewayIdLSB() + )); + } + + private Device findOrCreateDevice(GetOrCreateDeviceFromGatewayRequestMsg requestMsg, + Device gateway, + DeviceId gatewayId) throws JsonProcessingException { + TenantId tenantId = gateway.getTenantId(); + String deviceName = requestMsg.getDeviceName(); + Device device = deviceService.findDeviceByTenantIdAndName(tenantId, deviceName); + if (device != null) { + return device; + } + String[] topicPath = requestMsg.getDeviceName().split(DEVICE_NAME_SPLIT_SEPARATOR); + device = tryRenameSparkplugDevice(requestMsg, gateway, topicPath); + if (device != null) { + return device; } + device = createNewDevice(requestMsg, gateway, gatewayId, topicPath); + pushCreatedEvent(device, gateway); + return device; + } + + private Device tryRenameSparkplugDevice(GetOrCreateDeviceFromGatewayRequestMsg requestMsg, Device gateway, String[] topicPath) { + if (!requestMsg.getIsSparkplug()) { + return null; + } + + if (topicPath.length != 3) { + return null; + } + + String deviceId = topicPath[2]; + Device existingDevice = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), deviceId); + + if (existingDevice == null) { + return null; + } + + // Security check: verify that the device was created by this gateway + boolean isRelated = false; + try { + // Security check: verify that the device was originally created by this gateway + isRelated = relationService.checkRelation( + gateway.getTenantId(), + gateway.getId(), + existingDevice.getId(), + "Created", + RelationTypeGroup.COMMON + ); + } catch (Exception e) { + // Log the error from the relation service but return null to allow potential recovery + log.error("[{}] Error checking relation for device {}", gateway.getId(), existingDevice.getId(), e); + return null; + } + + // If the device is found but not related to this gateway, it's a security breach + if (!isRelated) { + log.error("[{}] Security breach attempt! Gateway tried to rename device [{}] without 'Created' relation.", + gateway.getId(), existingDevice.getId()); + // Throwing exception to halt the entire connection process + throw new RuntimeException("Security breach attempt! Unauthorized device rename."); + } + + // Logic for renaming the device if it's related and no naming conflicts exist + boolean changed = false; + String newName = requestMsg.getDeviceName(); + + if (!newName.equals(existingDevice.getName())) { + // Check if the new name is already taken by another device + Device conflictDevice = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), newName); + + if (conflictDevice != null) { + log.warn("[{}] Cannot rename device [{}] to [{}]: name already exists!", + gateway.getId(), existingDevice.getId(), newName); + return existingDevice; + } + + existingDevice.setName(newName); + + // Update label only if it's empty to avoid overwriting user changes + if (existingDevice.getLabel() == null || existingDevice.getLabel().isEmpty()) { + existingDevice.setLabel(deviceId); + } + + changed = true; + } + + return changed ? deviceService.saveDevice(existingDevice) : existingDevice; + } + + private Device createNewDevice(GetOrCreateDeviceFromGatewayRequestMsg requestMsg, + Device gateway, + DeviceId gatewayId, String[] topicPath) { + TenantId tenantId = gateway.getTenantId(); + Device device = new Device(); + device.setTenantId(tenantId); + device.setName(requestMsg.getDeviceName()); + if (requestMsg.getIsSparkplug()) { + if (topicPath.length == 3) device.setLabel(topicPath[2]); + } + device.setType(requestMsg.getDeviceType()); + device.setCustomerId(gateway.getCustomerId()); + DeviceProfile profile = + deviceProfileCache.findOrCreateDeviceProfile(tenantId, requestMsg.getDeviceType()); + device.setDeviceProfileId(profile.getId()); + ObjectNode additionalInfo = JacksonUtil.newObjectNode(); + additionalInfo.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayId.toString()); + device.setAdditionalInfo(additionalInfo); + device = deviceService.saveDevice(device); + relationService.saveRelation( + tenantId, + new EntityRelation(gateway.getId(), device.getId(), "Created") + ); + return device; + } + + private void updateLastConnectedGateway(Device device, DeviceId gatewayId) { + String gatewayIdStr = gatewayId.toString(); + JsonNode info = device.getAdditionalInfo(); + ObjectNode objectNode = (info instanceof ObjectNode) + ? (ObjectNode) info + : JacksonUtil.newObjectNode(); + if (!objectNode.has(DataConstants.LAST_CONNECTED_GATEWAY) + || !gatewayIdStr.equals(objectNode.get(DataConstants.LAST_CONNECTED_GATEWAY).asText())) { + objectNode.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayIdStr); + device.setAdditionalInfo(objectNode); + deviceService.saveDevice(device); + } + } + + private void pushCreatedEvent(Device device, Device gateway) { + TenantId tenantId = gateway.getTenantId(); + CustomerId customerId = gateway.getCustomerId(); + TbMsgMetaData metaData = new TbMsgMetaData(); + metaData.putValue("gatewayId", gateway.getId().toString()); + if (customerId != null && !customerId.isNullUid()) { + metaData.putValue("customerId", customerId.toString()); + } + JsonNode entityNode = JacksonUtil.valueToTree(device); + TbMsg msg = TbMsg.newMsg() + .type(TbMsgType.ENTITY_CREATED) + .originator(device.getId()) + .customerId(customerId) + .copyMetaData(metaData) + .dataType(TbMsgDataType.JSON) + .data(JacksonUtil.toString(entityNode)) + .build(); + tbClusterService.pushMsgToRuleEngine(tenantId, device.getId(), msg, null); + } + + private TransportApiResponseMsg buildResponse(Device device) throws JsonProcessingException { + GetOrCreateDeviceFromGatewayResponseMsg.Builder builder = + GetOrCreateDeviceFromGatewayResponseMsg.newBuilder() + .setDeviceInfo(ProtoUtils.toDeviceInfoProto(device)); + DeviceProfile profile = + deviceProfileCache.get(device.getTenantId(), device.getDeviceProfileId()); + if (profile != null) { + builder.setDeviceProfile(ProtoUtils.toProto(profile)); + } + return TransportApiResponseMsg.newBuilder() + .setGetOrCreateDeviceResponseMsg(builder.build()) + .build(); + } + + private TransportApiResponseMsg buildLimitErrorResponse(EntitiesLimitExceededException e, + DeviceId gatewayId) { + log.warn("[{}][{}] API limit exception: [{}]", + e.getTenantId(), gatewayId, e.getMessage()); + return TransportApiResponseMsg.newBuilder() + .setGetOrCreateDeviceResponseMsg( + GetOrCreateDeviceFromGatewayResponseMsg.newBuilder() + .setError(TransportProtos.TransportApiRequestErrorCode.ENTITY_LIMIT) + ) + .build(); } private TransportApiResponseMsg handle(ProvisionDeviceRequestMsg requestMsg) { diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java index 962d85ec7c..2ed3e0d161 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java @@ -88,7 +88,7 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg assertNotNull(accessToken); } if (config.getGatewayName() != null) { - savedGateway = createDevice(config.getGatewayName(), deviceProfile.getName(), true); + savedGateway = createDevice(config.getGatewayName(), deviceProfile.getName(), !config.isSparkplug); DeviceCredentials gatewayCredentials = doGet("/api/device/" + savedGateway.getId().getId().toString() + "/credentials", DeviceCredentials.class); assertNotNull(gatewayCredentials); diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java index a9bdbb55db..2adc7b08a5 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java @@ -31,7 +31,9 @@ import org.junit.Assert; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.TransportPayloadType; +import org.thingsboard.server.common.data.asset.AssetInfo; import org.thingsboard.server.common.data.exception.ThingsboardException; +import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.BooleanDataEntry; import org.thingsboard.server.common.data.kv.DoubleDataEntry; @@ -39,6 +41,7 @@ import org.thingsboard.server.common.data.kv.JsonDataEntry; import org.thingsboard.server.common.data.kv.LongDataEntry; import org.thingsboard.server.common.data.kv.StringDataEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.gen.transport.mqtt.SparkplugBProto; import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; @@ -57,6 +60,7 @@ import java.util.concurrent.atomic.AtomicReference; import static org.awaitility.Awaitility.await; import static org.eclipse.paho.mqttv5.common.packet.MqttWireMessage.MESSAGE_TYPE_CONNACK; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; import static org.thingsboard.common.util.JacksonUtil.newArrayNode; import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.Bytes; import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.Int16; @@ -67,8 +71,13 @@ import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataTyp import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.UInt32; import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.UInt64; import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.UInt8; +import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState.ONLINE; +import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.STATE; +import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.messageName; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.createMetric; +import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.DEVICE_NAME_SPLIT_SEPARATOR; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_ROOT_SPB_V_1_0; +import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_SEPARATOR; /** * Created by nickAS21 on 12.01.23 @@ -82,6 +91,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte protected ThreadLocalRandom random = ThreadLocalRandom.current(); protected static final String groupId = "SparkplugBGroupId"; + protected static final String edgeNodeDeviceName = "Test Connect Sparkplug client node"; protected static final String edgeNode = "SparkpluBNode"; protected static final String keysBdSeq = "bdSeq"; protected static final String alias = "Failed Telemetry/Attribute proto sparkplug payload. SparkplugMessageType "; @@ -98,14 +108,32 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte protected static final String metricBirthName_Int32 = "Device Metric int32"; protected Set sparkplugAttributesMetricNames; - public void beforeSparkplugTest() throws Exception { + public void beforeSparkplugTest(boolean isCreateDevices) throws Exception { MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() - .gatewayName("Test Connect Sparkplug client node") + .gatewayName(edgeNodeDeviceName) .isSparkplug(true) .sparkplugAttributesMetricNames(sparkplugAttributesMetricNames) .transportPayloadType(TransportPayloadType.PROTOBUF) .build(); processBeforeTest(configProperties); + if (isCreateDevices) { + // 1. Create the first device with a short name (legacy style) + String deviceName1 = deviceId + "_1"; + Device device1 = createDevice(deviceName1, deviceProfile.getName(), false); + + // 2. Establish 'Created' relation so the transport identifies this gateway as the owner + String relationType = "Created"; + EntityRelation relation1 = createFromRelation(savedGateway, device1, relationType); + doPost("/api/relation", relation1).andExpect(status().isOk()); + + // 3. Create the second device with a full-path name + String deviceName2 = groupId + DEVICE_NAME_SPLIT_SEPARATOR + edgeNode + DEVICE_NAME_SPLIT_SEPARATOR + deviceId + "_2"; + Device device2 = createDevice(deviceName2, deviceProfile.getName(), false); + + // 4. Establish 'Created' relation for the second device as well + EntityRelation relation2 = createFromRelation(savedGateway, device2, relationType); + doPost("/api/relation", relation2).andExpect(status().isOk()); + } } public void clientWithCorrectNodeAccessTokenWithNDEATH() throws Exception { @@ -146,7 +174,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte options.setSessionExpiryInterval(0L); options.setUserName(gatewayAccessToken); String nameSpace = nameSpaceBad.length == 0 ? TOPIC_ROOT_SPB_V_1_0 : nameSpaceBad[0]; - String topic = nameSpace + "/" + groupId + "/" + SparkplugMessageType.NDEATH.name() + "/" + edgeNode; + String topic = nameSpace + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.NDEATH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode; // The NDEATH message MUST set the MQTT Will QoS to 1 and Retained flag to false MqttMessage msg = new MqttMessage(); msg.setId(0); @@ -159,38 +187,35 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte protected List connectClientWithCorrectAccessTokenWithNDEATHCreatedDevices(int cntDevices, long ts) throws Exception { List devices = new ArrayList<>(); clientWithCorrectNodeAccessTokenWithNDEATH(); - String keyInt = "Node Metric int32"; + MetricDataType metricDataType = Int32; + String key = "Node Metric int32"; int valueDeviceInt32 = 1024; - SparkplugBProto.Payload.Metric metricInt = createMetric(valueDeviceInt32, ts, keyInt, Int32, -1L); - String keyStringEmpty = "Node Metric String Empty"; - String valueDeviceStringEmpty = ""; - SparkplugBProto.Payload.Metric metricStringEmpty = createMetric(valueDeviceStringEmpty, ts, keyStringEmpty, MetricDataType.String, -1L); + SparkplugBProto.Payload.Metric metric = createMetric(valueDeviceInt32, ts, key, metricDataType, -1L); SparkplugBProto.Payload.Builder payloadBirthNode = SparkplugBProto.Payload.newBuilder() .setTimestamp(ts) .setSeq(getBdSeqNum()); - payloadBirthNode.addMetrics(metricInt); - payloadBirthNode.addMetrics(metricStringEmpty); + payloadBirthNode.addMetrics(metric); payloadBirthNode.setTimestamp(ts); if (client.isConnected()) { - client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode, + client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.NBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode, payloadBirthNode.build().toByteArray(), 0, false); } valueDeviceInt32 = 4024; - metricInt = createMetric(valueDeviceInt32, ts, metricBirthName_Int32, metricBirthDataType_Int32, -1L); + metric = createMetric(valueDeviceInt32, ts, metricBirthName_Int32, metricBirthDataType_Int32, -1L); for (int i = 0; i < cntDevices; i++) { SparkplugBProto.Payload.Builder payloadBirthDevice = SparkplugBProto.Payload.newBuilder() .setTimestamp(ts) .setSeq(getSeqNum()); - String deviceName = deviceId + "_" + i; - - payloadBirthDevice.addMetrics(metricInt); + String deviceIdName = deviceId + "_" + i; + String deviceName = groupId + ":" + edgeNode + ":" + deviceIdName; + payloadBirthDevice.addMetrics(metric); if (client.isConnected()) { - client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.DBIRTH.name() + "/" + edgeNode + "/" + deviceName, + client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.DBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode + TOPIC_SPLIT_SEPARATOR + deviceIdName, payloadBirthDevice.build().toByteArray(), 0, false); AtomicReference device = new AtomicReference<>(); - await(alias + "find device [" + deviceName + "] after created") - .atMost(200, TimeUnit.SECONDS) + await(alias + "find device [" + deviceIdName + "] after created") + .atMost(40, TimeUnit.SECONDS) .ignoreExceptions() .until(() -> { device.set(doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class)); @@ -198,13 +223,265 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte }); devices.add(device.get()); } - } Assert.assertEquals(cntDevices, devices.size()); return devices; } + protected void connectClientWithCorrectAccessTokenWithNDEATHDevicesCreatingBefore_Test(int cntDevices) throws Exception { + long ts = calendar.getTimeInMillis(); + List devices = new ArrayList<>(); + clientWithCorrectNodeAccessTokenWithNDEATH(); + MetricDataType metricDataType = Int32; + String key = "Node Metric int32"; + int valueDeviceInt32 = 1024; + SparkplugBProto.Payload.Metric metric = createMetric(valueDeviceInt32, ts, key, metricDataType, -1L); + SparkplugBProto.Payload.Builder payloadBirthNode = SparkplugBProto.Payload.newBuilder() + .setTimestamp(ts) + .setSeq(getBdSeqNum()); + payloadBirthNode.addMetrics(metric); + payloadBirthNode.setTimestamp(ts); + if (client.isConnected()) { + client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.NBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode, + payloadBirthNode.build().toByteArray(), 0, false); + } + + valueDeviceInt32 = 4024; + metric = createMetric(valueDeviceInt32, ts, metricBirthName_Int32, metricBirthDataType_Int32, -1L); + // as old device name -> deviceId + String deviceIdNameLabel1 = deviceId + "_1"; + + if (client.isConnected()) { + SparkplugBProto.Payload.Builder payloadBirthDevice1 = SparkplugBProto.Payload.newBuilder() + .setTimestamp(ts) + .setSeq(getSeqNum()); + payloadBirthDevice1.addMetrics(metric); + client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.DBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode + TOPIC_SPLIT_SEPARATOR + deviceIdNameLabel1, + payloadBirthDevice1.build().toByteArray(), 0, false); + + } + String deviceName1 = groupId + DEVICE_NAME_SPLIT_SEPARATOR + edgeNode + DEVICE_NAME_SPLIT_SEPARATOR + deviceIdNameLabel1; + AtomicReference device1 = new AtomicReference<>(); + await(alias + "find device [" + deviceName1 + "] before connecting") + .atMost(40, TimeUnit.SECONDS) + .until(() -> { + device1.set(doGet("/api/tenant/devices?deviceName=" + deviceName1, Device.class)); + return device1.get() != null; + }); + devices.add(device1.get()); + + // as new device name -> groupId + ":" + edgeNode + ":" + deviceId; + String deviceIdName2 = deviceId + "_2"; + if (client.isConnected()) { + SparkplugBProto.Payload.Builder payloadBirthDevice2 = SparkplugBProto.Payload.newBuilder() + .setTimestamp(ts) + .setSeq(getSeqNum()); + payloadBirthDevice2.addMetrics(metric); + client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.DBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode + TOPIC_SPLIT_SEPARATOR + deviceIdName2, + payloadBirthDevice2.build().toByteArray(), 0, false); + } + String deviceName2 = groupId + DEVICE_NAME_SPLIT_SEPARATOR + edgeNode + DEVICE_NAME_SPLIT_SEPARATOR + deviceIdName2; + AtomicReference device2 = new AtomicReference<>(); + await(alias + "find device [" + deviceName2 + "] before connecting") + .atMost(40, TimeUnit.SECONDS) + .until(() -> { + device2.set(doGet("/api/tenant/devices?deviceName=" + deviceName2, Device.class)); + return device2.get() != null; + }); + devices.add(device2.get()); + Assert.assertEquals(cntDevices, devices.size()); + state_ONLINE_ALL (devices, calendar.getTimeInMillis()); + // Without full topic: as it was in the old version. When deviceId is updated to full theme, Label is also updated to old deviceId + Assert.assertEquals(deviceIdNameLabel1, device1.get().getLabel()); + // // With a full topic: if new. When creating a device by a client to a full topic, if the Label was not filled in - we do not touch it. + Assert.assertNull(device2.get().getLabel()); + } + + /** + * Coverage: Rename when a device with the target full-path name already exists (collision). + */ + protected void renameCollisionWhenTargetNameAlreadyExists_Test() throws Exception { + long ts = calendar.getTimeInMillis(); + String shortName = deviceId + "_1"; // Created in beforeTest + String fullPathName = groupId + ":" + edgeNode + ":" + shortName; + + // Manually create a device that already has the "new" full-path name to trigger a collision + createDevice(fullPathName, deviceProfile.getName(), false); + + clientWithCorrectNodeAccessTokenWithNDEATH(); + + SparkplugBProto.Payload.Builder payload = SparkplugBProto.Payload.newBuilder() + .setTimestamp(ts) + .setSeq(getSeqNum()); + payload.addMetrics(createMetric(123, ts, metricBirthName_Int32, metricBirthDataType_Int32, -1L)); + + // Gateway sends DBIRTH for the short name. + // Transport will try to rename it but should find a conflict and handle it gracefully. + client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/DBIRTH/" + edgeNode + "/" + shortName, + payload.build().toByteArray(), 0, false); + + await("Checking stability after collision") + .atMost(40, TimeUnit.SECONDS) + .until(() -> { + Device oldDevice = doGet("/api/tenant/devices?deviceName=" + shortName, Device.class); + Device conflictDevice = doGet("/api/tenant/devices?deviceName=" + fullPathName, Device.class); + // Both devices must still exist, proving no exception crashed the process + return oldDevice != null && conflictDevice != null; + }); + } + + /** + * Coverage: The privilege concern — attempt to rename a device not owned by the gateway. + * This test verifies that the original device's ID remains unchanged, meaning it was not hijacked. + */ + protected void unauthorizedRenameAttemptBad_Test() throws Exception { + long ts = calendar.getTimeInMillis(); + String strangerName = "unauthorized_device_rename"; + + // 1. Create a "stranger" device via API (it has no 'Created' relation to the gateway) + Device stranger = new Device(); + stranger.setName(strangerName); + stranger.setType("default"); + doPost("/api/device", stranger); + final DeviceId originalStrangerId = stranger.getId(); + + clientWithCorrectNodeAccessTokenWithNDEATH(); + + SparkplugBProto.Payload.Builder payload = SparkplugBProto.Payload.newBuilder() + .setTimestamp(ts).setSeq(getSeqNum()); + + // 2. Unauthorized gateway attempts to rename this device via Sparkplug topic path + client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/DBIRTH/" + edgeNode + "/" + strangerName, + payload.build().toByteArray(), 0, false); + + String expectedFullPath = groupId + ":" + edgeNode + ":" + strangerName; + + // 3. Verify security: the original device must still be linked to its short name with the same ID + await("Verify original device was not hijacked") + .atMost(40, TimeUnit.SECONDS) + .pollDelay(2, TimeUnit.SECONDS) + .untilAsserted(() -> { + // Check if the original device still exists with its original ID + Device currentStranger = doGet("/api/tenant/devices?deviceName=" + strangerName, Device.class); + Assert.assertNotNull("Original device disappeared!", currentStranger); + Assert.assertEquals("Security breach: Original device ID changed!", originalStrangerId, currentStranger.getId()); + + // Even if the gateway created a NEW device with a full path, it must have a different ID + Device newDevice = doGet("/api/tenant/devices?deviceName=" + expectedFullPath, Device.class); + if (newDevice != null) { + Assert.assertNotEquals("Stranger device was successfully hijacked (IDs match)!", originalStrangerId, newDevice.getId()); + } + }); + } + + /** + * Coverage: The privilege concern — attempt to rename a device not owned by the gateway. + */ + protected void unauthorizedRenameAttempt_Test() throws Exception { + long ts = calendar.getTimeInMillis(); + String strangerName = "unauthorized_device_rename"; + + // Create a device without a "Created" relation to the gateway + Device stranger = new Device(); + stranger.setName(strangerName); + stranger.setType("default"); + doPost("/api/device", stranger); + + clientWithCorrectNodeAccessTokenWithNDEATH(); + + SparkplugBProto.Payload.Builder payload = SparkplugBProto.Payload.newBuilder() + .setTimestamp(ts).setSeq(getSeqNum()); + + // Unauthorized gateway attempts to rename the device via Sparkplug topic + client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/DBIRTH/" + edgeNode + "/" + strangerName, + payload.build().toByteArray(), 0, false); + + String expectedFullPath = groupId + ":" + edgeNode + ":" + strangerName; + await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> + doGet("/api/tenant/devices?deviceName=" + expectedFullPath, Device.class, status().isNotFound()) + ); + } + + protected void state_ONLINE_ALL (List devices, long ts) { + TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new StringDataEntry(messageName(STATE), ONLINE.name())); + await(alias + messageName(STATE) + ", device: " + savedGateway.getName()) + .atMost(40, TimeUnit.SECONDS) + .until(() -> { + var foundEntry = tsService.findAllLatest(tenantId, savedGateway.getId()).get().stream() + .filter(tsKv -> tsKv.getKey().equals(tsKvEntry.getKey())) + .filter(tsKv -> tsKv.getValue().equals(tsKvEntry.getValue())) + .filter(tsKv -> tsKv.getTs() == tsKvEntry.getTs()) + .findFirst(); + return foundEntry.isPresent(); + }); + + for (Device device : devices) { + await(alias + messageName(STATE) + ", device: " + device.getName()) + .atMost(40, TimeUnit.SECONDS) + .until(() -> { + var foundEntry = tsService.findAllLatest(tenantId, device.getId()).get().stream() + .filter(tsKv -> tsKv.getKey().equals(tsKvEntry.getKey())) + .filter(tsKv -> tsKv.getValue().equals(tsKvEntry.getValue())) + .filter(tsKv -> tsKv.getTs() == tsKvEntry.getTs()) + .findFirst(); + return foundEntry.isPresent(); + }); + } + } + + /** + * Coverage: Concurrent first-message registration with the lock mechanism. + */ + protected void concurrentFirstMessageRegistration_Test() throws Exception { + int threadCount = 5; + String concurrentDeviceName = "concurrent_device"; + clientWithCorrectNodeAccessTokenWithNDEATH(); + + java.util.concurrent.ExecutorService executor = java.util.concurrent.Executors.newFixedThreadPool(threadCount); + long ts = calendar.getTimeInMillis(); + + for (int i = 0; i < threadCount; i++) { + executor.submit(() -> { + try { + SparkplugBProto.Payload.Builder payload = SparkplugBProto.Payload.newBuilder() + .setTimestamp(ts).setSeq(0); + client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/DBIRTH/" + edgeNode + "/" + concurrentDeviceName, + payload.build().toByteArray(), 0, false); + } catch (Exception e) { + log.error("Concurrent publish failed", e); + } + }); + } + + String expectedName = groupId + ":" + edgeNode + ":" + concurrentDeviceName; + await("Wait for concurrent registration result") + .atMost(40, TimeUnit.SECONDS) // Restored to 40s as requested + .until(() -> doGet("/api/tenant/devices?deviceName=" + expectedName, Device.class) != null); + + executor.shutdown(); + } + + /** + * Coverage: Sparkplug-message handling when msgId <= 0 (#7). + * Verifies that the transport does not close the session for Sparkplug clients using msgId 0. + */ + protected void sparkplugSessionStaysAliveWithZeroMsgId_Test() throws Exception { + // clientMqttV5ConnectWithNDEATH internally sets msgId = 0 for the Will message. + // This validates that the connection is accepted despite msgId being 0. + IMqttToken connectionResult = clientMqttV5ConnectWithNDEATH(calendar.getTimeInMillis(), 0, -1L); + Assert.assertTrue("Sparkplug connection should be successful with msgId=0", client.isConnected()); + + // Publish NBIRTH message which usually goes through the aggregate callback. + // This verifies that msgId=0 in the callback does not trigger closeDeviceSession. + connectionWithNBirth(Int32, "test_metric_msgId_0", 555); + + // Awaitility to ensure the session remains open after processing. + await("Verify Sparkplug session remains open after receiving msgId=0") + .atMost(40, TimeUnit.SECONDS) + .until(() -> client.isConnected()); + } + protected List connectClientWithCorrectAccessTokenWithNDEATHWithAliasCreatedDevices(long ts) throws Exception { List devices = new ArrayList<>(); Long alias = 0L; @@ -219,7 +496,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte payloadBirthNode.addMetrics(metric); payloadBirthNode.setTimestamp(ts); if (client.isConnected()) { - client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode, + client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.NBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode, payloadBirthNode.build().toByteArray(), 0, false); } @@ -228,15 +505,16 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte SparkplugBProto.Payload.Builder payloadBirthDevice = SparkplugBProto.Payload.newBuilder() .setTimestamp(ts) .setSeq(getSeqNum()); - String deviceName = deviceId + "_" + 1; + String deviceIdName = deviceId + "_1"; + String deviceName = groupId + ":" + edgeNode + ":" + deviceIdName; payloadBirthDevice.addMetrics(metric); if (client.isConnected()) { - client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.DBIRTH.name() + "/" + edgeNode + "/" + deviceName, + client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.DBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode + TOPIC_SPLIT_SEPARATOR + deviceIdName, payloadBirthDevice.build().toByteArray(), 0, false); AtomicReference device = new AtomicReference<>(); await(alias + "find device [" + deviceName + "] after created") - .atMost(200, TimeUnit.SECONDS) + .atMost(40, TimeUnit.SECONDS) .ignoreExceptions() .until(() -> { device.set(doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class)); @@ -282,7 +560,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte listKeys.add(metricKey); if (client.isConnected()) { - client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode, + client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.NBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode, payloadBirthNode.build().toByteArray(), 0, false); } return listKeys; @@ -525,4 +803,9 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte } } + private EntityRelation createFromRelation(Device mainDevice, Device device, String relationType) { + return new EntityRelation(mainDevice.getId(), device.getId(), relationType); + } + + } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/attributes/MqttV5ClientSparkplugBAttributesInProfileTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/attributes/MqttV5ClientSparkplugBAttributesInProfileTest.java index 37abade6e7..8cd713ad7d 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/attributes/MqttV5ClientSparkplugBAttributesInProfileTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/attributes/MqttV5ClientSparkplugBAttributesInProfileTest.java @@ -33,7 +33,7 @@ public class MqttV5ClientSparkplugBAttributesInProfileTest extends AbstractMqttV public void beforeTest() throws Exception { sparkplugAttributesMetricNames = new HashSet<>(); sparkplugAttributesMetricNames.add(metricBirthName_Int32); - beforeSparkplugTest(); + beforeSparkplugTest(false); } @After diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/attributes/MqttV5ClientSparkplugBAttributesTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/attributes/MqttV5ClientSparkplugBAttributesTest.java index 3826b59bdc..b080cfbff0 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/attributes/MqttV5ClientSparkplugBAttributesTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/attributes/MqttV5ClientSparkplugBAttributesTest.java @@ -29,7 +29,7 @@ public class MqttV5ClientSparkplugBAttributesTest extends AbstractMqttV5ClientSp @Before public void beforeTest() throws Exception { - beforeSparkplugTest(); + beforeSparkplugTest(false); } @After diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/AbstractMqttV5ClientSparkplugConnectionTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/AbstractMqttV5ClientSparkplugConnectionTest.java index 8f9943b393..1402e5c4c3 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/AbstractMqttV5ClientSparkplugConnectionTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/AbstractMqttV5ClientSparkplugConnectionTest.java @@ -37,10 +37,10 @@ import java.util.concurrent.atomic.AtomicReference; import static org.awaitility.Awaitility.await; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState.OFFLINE; -import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState.ONLINE; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.STATE; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.messageName; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_ROOT_SPB_V_1_0; +import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_SEPARATOR; /** * Created by nickAS21 on 12.01.23 @@ -95,31 +95,7 @@ public abstract class AbstractMqttV5ClientSparkplugConnectionTest extends Abstra protected void processConnectClientWithCorrectAccessTokenWithNDEATH_State_ONLINE_ALL(int cntDevices) throws Exception { long ts = calendar.getTimeInMillis(); List devices = connectClientWithCorrectAccessTokenWithNDEATHCreatedDevices(cntDevices, ts); - - TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new StringDataEntry(messageName(STATE), ONLINE.name())); - await(alias + messageName(STATE) + ", device: " + savedGateway.getName()) - .atMost(40, TimeUnit.SECONDS) - .until(() -> { - var foundEntry = tsService.findAllLatest(tenantId, savedGateway.getId()).get().stream() - .filter(tsKv -> tsKv.getKey().equals(tsKvEntry.getKey())) - .filter(tsKv -> tsKv.getValue().equals(tsKvEntry.getValue())) - .filter(tsKv -> tsKv.getTs() == tsKvEntry.getTs()) - .findFirst(); - return foundEntry.isPresent(); - }); - - for (Device device : devices) { - await(alias + messageName(STATE) + ", device: " + device.getName()) - .atMost(40, TimeUnit.SECONDS) - .until(() -> { - var foundEntry = tsService.findAllLatest(tenantId, device.getId()).get().stream() - .filter(tsKv -> tsKv.getKey().equals(tsKvEntry.getKey())) - .filter(tsKv -> tsKv.getValue().equals(tsKvEntry.getValue())) - .filter(tsKv -> tsKv.getTs() == tsKvEntry.getTs()) - .findFirst(); - return foundEntry.isPresent(); - }); - } + state_ONLINE_ALL (devices, ts); } protected void processConnectClientWithCorrectAccessTokenWithNDEATH_State_ONLINE_All_Then_OneDeviceOFFLINE(int cntDevices, int indexDeviceDisconnect) throws Exception { @@ -135,7 +111,7 @@ public abstract class AbstractMqttV5ClientSparkplugConnectionTest extends Abstra if (client.isConnected()) { List devicesList = new ArrayList<>(devices); Device device = devicesList.get(indexDeviceDisconnect); - client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.DDEATH.name() + "/" + edgeNode + "/" + device.getName(), + client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.DDEATH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode + TOPIC_SPLIT_SEPARATOR + device.getName(), payloadDeathDevice.build().toByteArray(), 0, false); await(alias + messageName(STATE) + ", device: " + device.getName()) .atMost(40, TimeUnit.SECONDS) diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/MqttV5ClientSparkplugBConnectionDevicesCreatingBeforeTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/MqttV5ClientSparkplugBConnectionDevicesCreatingBeforeTest.java new file mode 100644 index 0000000000..92583e655c --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/MqttV5ClientSparkplugBConnectionDevicesCreatingBeforeTest.java @@ -0,0 +1,71 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.transport.mqtt.sparkplug.connection; + +import org.eclipse.paho.mqttv5.common.MqttException; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.thingsboard.server.dao.service.DaoSqlTest; + +/** + * Created by nickAS21 on 16.02.26 + */ +@DaoSqlTest +public class MqttV5ClientSparkplugBConnectionDevicesCreatingBeforeTest extends AbstractMqttV5ClientSparkplugConnectionTest { + + /** + * String deviceName_1 = deviceId + "_1"; Only name device. Without a complete topic: how it was in the old version. + * String deviceName_2 = groupId + DEVICE_NAME_SPLIT_REGEXP + edgeNode + DEVICE_NAME_SPLIT_REGEXP + deviceId + "_2"; With complete topic: how it was in the new version. + * @throws Exception + */ + @Before + public void beforeTest() throws Exception { + beforeSparkplugTest(true); + } + + @After + public void afterTest() throws MqttException { + if (client.isConnected()) { + client.disconnect(); + } + } + + @Test + public void testClientWithCorrectAccessTokenWithNDEATHTwoDevicesCreatingBeforeFirstNameDeviceIdSecondNameFull() throws Exception { + connectClientWithCorrectAccessTokenWithNDEATHDevicesCreatingBefore_Test(2); + } + + @Test + public void testRenameWhenDeviceFullPathAlreadyExists_Collision() throws Exception { + renameCollisionWhenTargetNameAlreadyExists_Test(); + } + + @Test + public void testUnauthorizedRenameAttempt() throws Exception { + unauthorizedRenameAttempt_Test(); + } + + @Test + public void testConcurrentFirstMessageRegistration() throws Exception { + concurrentFirstMessageRegistration_Test(); + } + + @Test + public void testSparkplugSessionStaysAliveWithZeroMsgId() throws Exception { + sparkplugSessionStaysAliveWithZeroMsgId_Test(); + } +} diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/MqttV5ClientSparkplugBConnectionTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/MqttV5ClientSparkplugBConnectionTest.java index 01038c21f1..0d5e73e2f0 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/MqttV5ClientSparkplugBConnectionTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/MqttV5ClientSparkplugBConnectionTest.java @@ -29,7 +29,7 @@ public class MqttV5ClientSparkplugBConnectionTest extends AbstractMqttV5ClientSp @Before public void beforeTest() throws Exception { - beforeSparkplugTest(); + beforeSparkplugTest(false); } @After diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/MqttV5RpcSparkplugTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/MqttV5RpcSparkplugTest.java index e36b4e6260..08ade0274d 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/MqttV5RpcSparkplugTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/MqttV5RpcSparkplugTest.java @@ -28,7 +28,7 @@ public class MqttV5RpcSparkplugTest extends AbstractMqttV5RpcSparkplugTest { @Before public void beforeTest() throws Exception { - beforeSparkplugTest(); + beforeSparkplugTest(false); } @After @@ -47,6 +47,7 @@ public class MqttV5RpcSparkplugTest extends AbstractMqttV5RpcSparkplugTest { public void testClientDeviceWithCorrectAccessTokenPublish_TwoWayRpc_Success() throws Exception { processClientDeviceWithCorrectAccessTokenPublish_TwoWayRpc_Success(); } + @Test public void testClientDeviceWithCorrectAccessTokenPublishWithAlias_TwoWayRpc_Success() throws Exception { processClientDeviceWithCorrectAccessTokenPublishWithAlias_TwoWayRpc_Success(); diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/AbstractMqttV5ClientSparkplugTelemetryTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/AbstractMqttV5ClientSparkplugTelemetryTest.java index 319b61b7f3..8f2d0659af 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/AbstractMqttV5ClientSparkplugTelemetryTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/AbstractMqttV5ClientSparkplugTelemetryTest.java @@ -30,6 +30,7 @@ import java.util.concurrent.atomic.AtomicReference; import static org.awaitility.Awaitility.await; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_ROOT_SPB_V_1_0; +import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_SEPARATOR; /** * Created by nickAS21 on 12.01.23 @@ -67,7 +68,7 @@ public abstract class AbstractMqttV5ClientSparkplugTelemetryTest extends Abstrac createdAddMetricValuePrimitiveTsKv(listTsKvEntry, listKeys, ndataPayload, ts); if (client.isConnected()) { - client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + messageTypeName + "/" + edgeNode, + client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + messageTypeName + TOPIC_SPLIT_SEPARATOR + edgeNode, ndataPayload.build().toByteArray(), 0, false); } @@ -96,7 +97,7 @@ public abstract class AbstractMqttV5ClientSparkplugTelemetryTest extends Abstrac createdAddMetricValueArraysPrimitiveTsKv(listTsKvEntry, listKeys, ndataPayload, ts); if (client.isConnected()) { - client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + messageTypeName + "/" + edgeNode, + client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + messageTypeName + TOPIC_SPLIT_SEPARATOR + edgeNode, ndataPayload.build().toByteArray(), 0, false); } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/MqttV5ClientSparkplugBTelemetryTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/MqttV5ClientSparkplugBTelemetryTest.java index bafe2d81d2..5d9f453415 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/MqttV5ClientSparkplugBTelemetryTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/MqttV5ClientSparkplugBTelemetryTest.java @@ -29,7 +29,7 @@ public class MqttV5ClientSparkplugBTelemetryTest extends AbstractMqttV5ClientSpa @Before public void beforeTest() throws Exception { - beforeSparkplugTest(); + beforeSparkplugTest(false); } @After diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 68f7052c69..8cbc2ec7ac 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -470,6 +470,7 @@ message GetOrCreateDeviceFromGatewayRequestMsg { int64 gatewayIdLSB = 2; string deviceName = 3; string deviceType = 4; + bool isSparkplug = 5; } message GetOrCreateDeviceFromGatewayResponseMsg { diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java index 8ea6480142..94c24f5a1d 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java @@ -68,6 +68,7 @@ import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor; import org.thingsboard.server.transport.mqtt.adaptors.ProtoMqttAdaptor; import org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService; import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState; +import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopic; import java.util.ArrayList; import java.util.Collections; @@ -260,7 +261,7 @@ public abstract class AbstractGatewaySessionHandler { ack(msg, MqttReasonCodes.PubAck.SUCCESS); log.trace("[{}][{}][{}] onDeviceConnectOk: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName); @@ -276,7 +277,7 @@ public abstract class AbstractGatewaySessionHandler onDeviceConnect(String deviceName, String deviceType) { + ListenableFuture onDeviceConnect(String deviceName, String deviceType, boolean isSparkplug) { T result = devices.get(deviceName); if (result == null) { Lock deviceCreationLock = deviceCreationLockMap.computeIfAbsent(deviceName, s -> new ReentrantLock()); @@ -284,7 +285,7 @@ public abstract class AbstractGatewaySessionHandler getDeviceCreationFuture(String deviceName, String deviceType) { + ListenableFuture onDeviceConnectSparkplug(SparkplugTopic topic, String deviceType) { + String fullPath = topic.getNodeDeviceNameAllPath(); + // Primary lookup: try to find the device by its full-path name (standard for new devices) + T result = devices.get(fullPath); + + if (result == null) { + // Secondary lookup (Legacy Fallback): check for the short name if full path is not found. + // This supports devices migrated from older versions. + String shortName = topic.getNodeDeviceName(); + result = devices.get(shortName); + } + + if (result != null) { + return Futures.immediateFuture(result); + } else { + // If not found in cache at all, proceed with connection/creation using full path + return onDeviceConnect(fullPath, deviceType, true); + } + } + + private ListenableFuture getDeviceCreationFuture(String deviceName, String deviceType, boolean isSparkplug) { final SettableFuture futureToSet = SettableFuture.create(); ListenableFuture future = deviceFutures.putIfAbsent(deviceName, futureToSet); if (future != null) { @@ -309,6 +330,7 @@ public abstract class AbstractGatewaySessionHandler() { @Override @@ -844,8 +866,16 @@ public abstract class AbstractGatewaySessionHandler onSuccess, Consumer onFailure) { - ListenableFuture deviceCtxFuture = onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE); + ListenableFuture deviceCtxFuture = onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE, false); process(deviceCtxFuture, onSuccess, onFailure); } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugDeviceSessionContext.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugDeviceSessionContext.java index 2bd0d7702f..9a619ccd4b 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugDeviceSessionContext.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugDeviceSessionContext.java @@ -128,5 +128,5 @@ public class SparkplugDeviceSessionContext extends AbstractGatewayDeviceSessionC rpcRequest.getMethodName() + ". " + e.getMessage()); } } - + } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java index d7cea8c61b..6cb566d445 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java @@ -21,7 +21,6 @@ import com.google.common.util.concurrent.MoreExecutors; import com.google.gson.JsonSyntaxException; import io.netty.handler.codec.mqtt.MqttMessage; import io.netty.handler.codec.mqtt.MqttPublishMessage; -import io.netty.handler.codec.mqtt.MqttReasonCodes; import io.netty.handler.codec.mqtt.MqttTopicSubscription; import lombok.Getter; import lombok.extern.slf4j.Slf4j; @@ -61,8 +60,8 @@ import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetr import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.createMetric; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.fromSparkplugBMetricToKeyValueProto; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.validatedValueByTypeMetric; -import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_REGEXP; -import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_STATE_REGEXP; +import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_SEPARATOR; +import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_STATE_SEPARATOR; @Slf4j @SpecVersion(spec = "sparkplug", version = "3.0.0") @@ -106,9 +105,7 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler contextListenableFuture; if (topic.isNode()) { if (topic.isType(NBIRTH)) { @@ -118,19 +115,26 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler deviceCtx = onDeviceConnectProto(topic); - contextListenableFuture = Futures.transform(deviceCtx, ctx -> { - if (topic.isType(DBIRTH)) { - sendSparkplugStateOnTelemetry(ctx.getSessionInfo(), deviceName, ONLINE, - sparkplugBProto.getTimestamp()); - try { - ctx.setDeviceBirthMetrics(sparkplugBProto.getMetricsList()); - } catch (IllegalArgumentException | DuplicateKeyException e) { + try { + deviceName = checkDeviceName(topic.getNodeDeviceNameAllPath()); + ListenableFuture deviceCtx = this.onDeviceConnectProto(topic); + String finalDeviceName = deviceName; + contextListenableFuture = Futures.transform(deviceCtx, ctx -> { + if (topic.isType(DBIRTH)) { + sendSparkplugStateOnTelemetry(ctx.getSessionInfo(), finalDeviceName, ONLINE, + sparkplugBProto.getTimestamp()); + try { + ctx.setDeviceBirthMetrics(sparkplugBProto.getMetricsList()); + } catch (IllegalArgumentException | DuplicateKeyException e) { + log.error("[{}] Failed to set birth metrics", finalDeviceName, e); throw new RuntimeException(e); + } } - } - return ctx; - }, MoreExecutors.directExecutor()); + return ctx; + }, MoreExecutors.directExecutor()); + } catch (IllegalArgumentException | DuplicateKeyException e) { + throw new RuntimeException(e); + } } Set attributesMetricNames = ((MqttDeviceProfileTransportConfiguration) deviceSessionCtx .getDeviceProfile().getProfileData().getTransportConfiguration()).getSparkplugAttributesMetricNames(); @@ -200,7 +204,7 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler= 4 && splitTopic.length <= 5 && splitTopic[0].equals(this.sparkplugTopicNode.getNamespace()) && splitTopic[1].equals(this.sparkplugTopicNode.getGroupId()) && diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopic.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopic.java index 2988c903f7..9b0cb62ca6 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopic.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopic.java @@ -21,8 +21,9 @@ import org.thingsboard.server.common.data.exception.ThingsboardErrorCode; import org.thingsboard.server.common.data.exception.ThingsboardException; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.parseMessageType; +import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.DEVICE_NAME_SPLIT_SEPARATOR; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_ROOT_SPB_V_1_0; -import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_REGEXP; +import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_SEPARATOR; /** * Created by nickAS21 on 12.12.22 @@ -196,7 +197,7 @@ public class SparkplugTopic { try { if (isValidIdElementToUTF8(topicString)) { SparkplugMessageType messageType; - String[] splitTopic = topicString.split(TOPIC_SPLIT_REGEXP); + String[] splitTopic = topicString.split(TOPIC_SPLIT_SEPARATOR); if (TOPIC_ROOT_SPB_V_1_0.equals(splitTopic[0])) { if (splitTopic.length == 3) { messageType = parseMessageType(splitTopic[1]); @@ -329,6 +330,17 @@ public class SparkplugTopic { return isNode() ? edgeNodeId : deviceId; } + public String getNodeDeviceNameAllPath() { + StringBuilder sb = new StringBuilder(); + if (hostApplicationId == null) { + sb.append(getGroupId()).append(DEVICE_NAME_SPLIT_SEPARATOR).append(getEdgeNodeId()); + if (getDeviceId() != null) { + sb.append(DEVICE_NAME_SPLIT_SEPARATOR).append(getDeviceId()); + } + } + return sb.toString(); + } + public static boolean isValidIdElementToUTF8(String deviceIdElement) { if (deviceIdElement == null) { return false; diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopicService.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopicService.java index 74ea6858fe..377b759957 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopicService.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopicService.java @@ -34,8 +34,9 @@ public class SparkplugTopicService { private static final Map SPLIT_TOPIC_CACHE = new HashMap<>(); public static final String TOPIC_ROOT_SPB_V_1_0 = "spBv1.0"; public static final String TOPIC_ROOT_CERT_SP = "$sparkplug/certificates/"; - public static final String TOPIC_SPLIT_REGEXP = "/"; - public static final String TOPIC_STATE_REGEXP = TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_REGEXP + STATE.name() + TOPIC_SPLIT_REGEXP; + public static final String TOPIC_SPLIT_SEPARATOR = "/"; + public static final String DEVICE_NAME_SPLIT_SEPARATOR = ":"; + public static final String TOPIC_STATE_SEPARATOR = TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + STATE.name() + TOPIC_SPLIT_SEPARATOR; public static SparkplugTopic getSplitTopic(String topic) throws ThingsboardException { SparkplugTopic sparkplugTopic = SPLIT_TOPIC_CACHE.get(topic);