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 ed43c71027..2fe4543806 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 @@ -396,7 +396,6 @@ public class DefaultTransportApiService implements TransportApiService { // 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(), @@ -405,37 +404,34 @@ public class DefaultTransportApiService implements TransportApiService { 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; + log.error("[{}] Failed checking relation for device {}", + gateway.getId(), + existingDevice.getId(), + e); + throw new RuntimeException( + "Failed checking relation for device " + existingDevice.getId(), + e + ); } - // If the device is found but not related to this gateway, it's a security breach + // If the device is found but not related to this gateway 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."); + log.debug("[{}] Device [{}] exists but is not related to gateway. " + + "Skipping rename and allowing creation of Sparkplug device [{}].", + gateway.getId(), + existingDevice.getId(), + requestMsg.getDeviceName()); + + return null; } // 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()) { + if (StringUtils.isEmpty(existingDevice.getLabel())) { existingDevice.setLabel(deviceId); } 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 2adc7b08a5..c3c72e916c 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,6 @@ 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; @@ -58,6 +57,7 @@ import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import static java.util.concurrent.Executors.newFixedThreadPool; 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; @@ -108,7 +108,8 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte protected static final String metricBirthName_Int32 = "Device Metric int32"; protected Set sparkplugAttributesMetricNames; - public void beforeSparkplugTest(boolean isCreateDevices) throws Exception { + public void beforeSparkplugTest() throws Exception { + MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() .gatewayName(edgeNodeDeviceName) .isSparkplug(true) @@ -116,24 +117,26 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte .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 seedLegacyAndFullPathDevices() throws Exception { + // 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 { @@ -208,7 +211,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte .setTimestamp(ts) .setSeq(getSeqNum()); String deviceIdName = deviceId + "_" + i; - String deviceName = groupId + ":" + edgeNode + ":" + deviceIdName; + String deviceName = groupId + DEVICE_NAME_SPLIT_SEPARATOR + edgeNode + DEVICE_NAME_SPLIT_SEPARATOR + deviceIdName; payloadBirthDevice.addMetrics(metric); if (client.isConnected()) { 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, @@ -304,7 +307,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte protected void renameCollisionWhenTargetNameAlreadyExists_Test() throws Exception { long ts = calendar.getTimeInMillis(); String shortName = deviceId + "_1"; // Created in beforeTest - String fullPathName = groupId + ":" + edgeNode + ":" + shortName; + String fullPathName = groupId + DEVICE_NAME_SPLIT_SEPARATOR + edgeNode + DEVICE_NAME_SPLIT_SEPARATOR + shortName; // Manually create a device that already has the "new" full-path name to trigger a collision createDevice(fullPathName, deviceProfile.getName(), false); @@ -318,7 +321,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte // 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, + client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + "/DBIRTH/" + edgeNode + TOPIC_SPLIT_SEPARATOR + shortName, payload.build().toByteArray(), 0, false); await("Checking stability after collision") @@ -352,10 +355,10 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte .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, + client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + "/DBIRTH/" + edgeNode + TOPIC_SPLIT_SEPARATOR + strangerName, payload.build().toByteArray(), 0, false); - String expectedFullPath = groupId + ":" + edgeNode + ":" + strangerName; + String expectedFullPath = groupId + DEVICE_NAME_SPLIT_SEPARATOR + edgeNode + DEVICE_NAME_SPLIT_SEPARATOR + 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") @@ -394,10 +397,10 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte .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, + client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + "/DBIRTH/" + edgeNode + TOPIC_SPLIT_SEPARATOR + strangerName, payload.build().toByteArray(), 0, false); - String expectedFullPath = groupId + ":" + edgeNode + ":" + strangerName; + String expectedFullPath = groupId + DEVICE_NAME_SPLIT_SEPARATOR + edgeNode + DEVICE_NAME_SPLIT_SEPARATOR + strangerName; await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> doGet("/api/tenant/devices?deviceName=" + expectedFullPath, Device.class, status().isNotFound()) ); @@ -438,7 +441,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte String concurrentDeviceName = "concurrent_device"; clientWithCorrectNodeAccessTokenWithNDEATH(); - java.util.concurrent.ExecutorService executor = java.util.concurrent.Executors.newFixedThreadPool(threadCount); + java.util.concurrent.ExecutorService executor = newFixedThreadPool(threadCount); long ts = calendar.getTimeInMillis(); for (int i = 0; i < threadCount; i++) { @@ -446,7 +449,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte try { SparkplugBProto.Payload.Builder payload = SparkplugBProto.Payload.newBuilder() .setTimestamp(ts).setSeq(0); - client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/DBIRTH/" + edgeNode + "/" + concurrentDeviceName, + client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + "/DBIRTH/" + edgeNode + TOPIC_SPLIT_SEPARATOR + concurrentDeviceName, payload.build().toByteArray(), 0, false); } catch (Exception e) { log.error("Concurrent publish failed", e); @@ -454,9 +457,9 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte }); } - String expectedName = groupId + ":" + edgeNode + ":" + concurrentDeviceName; + String expectedName = groupId + DEVICE_NAME_SPLIT_SEPARATOR + edgeNode + DEVICE_NAME_SPLIT_SEPARATOR + concurrentDeviceName; await("Wait for concurrent registration result") - .atMost(40, TimeUnit.SECONDS) // Restored to 40s as requested + .atMost(40, TimeUnit.SECONDS) .until(() -> doGet("/api/tenant/devices?deviceName=" + expectedName, Device.class) != null); executor.shutdown(); @@ -506,7 +509,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte .setTimestamp(ts) .setSeq(getSeqNum()); String deviceIdName = deviceId + "_1"; - String deviceName = groupId + ":" + edgeNode + ":" + deviceIdName; + String deviceName = groupId + DEVICE_NAME_SPLIT_SEPARATOR + edgeNode + DEVICE_NAME_SPLIT_SEPARATOR + deviceIdName; payloadBirthDevice.addMetrics(metric); if (client.isConnected()) { 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 8cd713ad7d..37abade6e7 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(false); + beforeSparkplugTest(); } @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 b080cfbff0..3826b59bdc 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(false); + beforeSparkplugTest(); } @After 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 index 92583e655c..08fc4ea9a5 100644 --- 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 @@ -34,7 +34,8 @@ public class MqttV5ClientSparkplugBConnectionDevicesCreatingBeforeTest extends A */ @Before public void beforeTest() throws Exception { - beforeSparkplugTest(true); + beforeSparkplugTest(); + seedLegacyAndFullPathDevices(); } @After 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 0d5e73e2f0..01038c21f1 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(false); + beforeSparkplugTest(); } @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 08ade0274d..e90f6e94da 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(false); + beforeSparkplugTest(); } @After 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 5d9f453415..bafe2d81d2 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(false); + beforeSparkplugTest(); } @After 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 94c24f5a1d..99f10eb176 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 @@ -261,7 +261,7 @@ public abstract class AbstractGatewaySessionHandler { ack(msg, MqttReasonCodes.PubAck.SUCCESS); log.trace("[{}][{}][{}] onDeviceConnectOk: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName); @@ -277,6 +277,10 @@ public abstract class AbstractGatewaySessionHandler onDeviceConnect(String deviceName, String deviceType) { + return onDeviceConnect(deviceName, deviceType, false); + } + ListenableFuture onDeviceConnect(String deviceName, String deviceType, boolean isSparkplug) { T result = devices.get(deviceName); if (result == null) { @@ -902,7 +906,7 @@ public abstract class AbstractGatewaySessionHandler onSuccess, Consumer onFailure) { - ListenableFuture deviceCtxFuture = onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE, false); + ListenableFuture deviceCtxFuture = onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE); process(deviceCtxFuture, onSuccess, onFailure); } 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 6cb566d445..1a088344c8 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 @@ -115,26 +115,23 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler 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); - } + 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()); - } catch (IllegalArgumentException | DuplicateKeyException e) { - throw new RuntimeException(e); - } + } + return ctx; + }, MoreExecutors.directExecutor()); } Set attributesMetricNames = ((MqttDeviceProfileTransportConfiguration) deviceSessionCtx .getDeviceProfile().getProfileData().getTransportConfiguration()).getSparkplugAttributesMetricNames();