From 418dd84f77d0354fc61f28866f23cdb23619516b Mon Sep 17 00:00:00 2001 From: Wouter Born Date: Thu, 5 Nov 2026 16:27:01 +0100 Subject: [PATCH 1/3] Prevent cross-gateway asset data leakage - Introduce V2 asset ID mapping using all 22 characters of the gateway ID to reduce mapping collisions - Preserve legacy V1 mapping for gateways registered before the upgrade to version 1.34.0 - Track first activation timestamps of normalized application versions to maintain a stable mapping cutoff across upgrades, restarts and rollbacks - Enforce gateway asset ownership for creates, updates, deletions and attribute events using persisted asset ancestry - Add database locking to prevent concurrent changes from bypassing ownership checks - Reject conflicting asset IDs and invalid synchronization inventories to prevent unintended asset modifications or deletions - Abort gateway synchronization when asset operations fail instead of continuing with inconsistent state - Add unit and integration tests covering ID mapping, version tracking, ownership isolation and synchronization failures Closes #2427 --- .../manager/asset/AssetProcessingService.java | 9 +- .../manager/asset/AssetStorageService.java | 72 +++++- .../manager/gateway/GatewayConnector.java | 143 ++++++++---- .../manager/gateway/GatewayService.java | 42 +++- .../persistence/ApplicationVersion.java | 45 ++++ .../system/ApplicationVersionService.java | 109 +++++++++ .../org.openremote.model.ContainerService | 1 + .../V20261105_1530__Application_version.sql | 6 + .../manager/gateway/GatewayMappingTest.groovy | 193 ++++++++++++++++ .../persistence/ApplicationVersionTest.groovy | 52 +++++ .../AbstractGatewayCompatibilityTest.groovy | 6 +- .../test/gateway/GatewayOwnershipTest.groovy | 215 ++++++++++++++++++ .../test/gateway/GatewayTest.groovy | 5 +- .../ApplicationVersionServiceTest.groovy | 199 ++++++++++++++++ 14 files changed, 1038 insertions(+), 59 deletions(-) create mode 100644 manager/src/main/java/org/openremote/manager/persistence/ApplicationVersion.java create mode 100644 manager/src/main/java/org/openremote/manager/system/ApplicationVersionService.java create mode 100644 manager/src/main/resources/org/openremote/manager/setup/database/V20261105_1530__Application_version.sql create mode 100644 manager/src/test/groovy/org/openremote/manager/gateway/GatewayMappingTest.groovy create mode 100644 manager/src/test/groovy/org/openremote/manager/persistence/ApplicationVersionTest.groovy create mode 100644 test/src/test/groovy/org/openremote/test/gateway/GatewayOwnershipTest.groovy create mode 100644 test/src/test/groovy/org/openremote/test/system/ApplicationVersionServiceTest.groovy diff --git a/manager/src/main/java/org/openremote/manager/asset/AssetProcessingService.java b/manager/src/main/java/org/openremote/manager/asset/AssetProcessingService.java index 089ca47276..568dc0dca9 100644 --- a/manager/src/main/java/org/openremote/manager/asset/AssetProcessingService.java +++ b/manager/src/main/java/org/openremote/manager/asset/AssetProcessingService.java @@ -24,6 +24,7 @@ import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.Tags; import io.micrometer.core.instrument.Timer; +import jakarta.persistence.LockModeType; import jakarta.validation.ConstraintViolation; import java.time.Instant; import java.util.*; @@ -421,10 +422,14 @@ protected boolean processAttributeEvent(AttributeEvent event) throws AssetProces Event result = persistenceService.doReturningTransaction( em -> { + // Hold the database row lock through ownership checks and the attribute write + // so a concurrent move or deletion cannot invalidate gateway authorization. // TODO: Retrieve optimised DTO rather than whole asset - Asset asset = assetStorageService.find(em, event.getId(), true); + Asset asset = + em.find(Asset.class, event.getId(), LockModeType.PESSIMISTIC_WRITE); - if (asset == null) { + // Treat pending deletion as missing so queued events cannot modify the asset. + if (asset == null || asset.isDeletePending()) { throw new AssetProcessingException(ASSET_NOT_FOUND, event.getId()); } diff --git a/manager/src/main/java/org/openremote/manager/asset/AssetStorageService.java b/manager/src/main/java/org/openremote/manager/asset/AssetStorageService.java index 8da18d4b57..042baa2585 100644 --- a/manager/src/main/java/org/openremote/manager/asset/AssetStorageService.java +++ b/manager/src/main/java/org/openremote/manager/asset/AssetStorageService.java @@ -32,6 +32,7 @@ import static org.openremote.model.value.MetaItemType.ACCESS_RESTRICTED_READ; import jakarta.persistence.EntityManager; +import jakarta.persistence.LockModeType; import jakarta.persistence.NoResultException; import jakarta.persistence.Query; import jakarta.persistence.TypedQuery; @@ -785,7 +786,20 @@ public > T merge( T existingAsset = TextUtil.isNullOrEmpty(asset.getId()) ? null - : (T) em.find(Asset.class, asset.getId()); + : (T) em.find(Asset.class, asset.getId(), LockModeType.PESSIMISTIC_WRITE); + + if (requestingGatewayAsset != null) { + String gatewayId = requestingGatewayAsset.getId(); + if (existingAsset != null) { + requireGatewayOwnership(em, existingAsset, gatewayId, false); + } + Asset parent = + em.find(Asset.class, asset.getParentId(), LockModeType.PESSIMISTIC_READ); + requireGatewayOwnership(em, parent, gatewayId, true); + if (!parent.getRealm().equals(asset.getRealm())) { + throw new IllegalStateException("Gateway asset realm mismatch"); + } + } if (existingAsset != null) { @@ -999,6 +1013,11 @@ public > T merge( existingAsset.setParentId(asset.getParentId()); existingAsset.setAccessPublicRead(asset.isAccessPublicRead()); updatedAsset = em.merge(existingAsset); + } else if (existingAsset == null) { + // INSERT must fail on a concurrent ID collision, never merge the new owner's + // row. + em.persist(asset); + updatedAsset = asset; } else { updatedAsset = em.merge(asset); } @@ -1149,7 +1168,18 @@ public DeleteResult deleteWithResult(List assetIds, boolean skipGatewayC * * @return DeleteResult indicating whether the deletion was accepted or rejected */ + public boolean deleteGatewayAssets(List assetIds, String gatewayId) { + Objects.requireNonNull(gatewayId); + List ids = assetIds.stream().distinct().sorted().toList(); + return markAssetsForDeletionAndQueue(ids, ids, gatewayId).accepted(); + } + protected DeleteResult markAssetsForDeletionAndQueue(List ids, List assetIds) { + return markAssetsForDeletionAndQueue(ids, assetIds, null); + } + + protected DeleteResult markAssetsForDeletionAndQueue( + List ids, List assetIds, String gatewayId) { if (ids.isEmpty()) { return DeleteResult.accepted(CompletableFuture.completedFuture(true)); } @@ -1183,6 +1213,15 @@ protected DeleteResult markAssetsForDeletionAndQueue(List ids, List asset : assets) { + em.lock(asset, LockModeType.PESSIMISTIC_WRITE); + em.refresh(asset); + if (gatewayId != null) { + requireGatewayOwnership(em, asset, gatewayId, false); + } + } assets.forEach(asset -> asset.setDeletePending(true)); }); @@ -1963,6 +2002,37 @@ public void storeUserAssetLinks(List userAssetLinks) { }); } + /** + * Validate persisted ancestry, not connector caches or an untrusted proposed parent. Ancestor + * read locks prevent a concurrent move/deletion until the caller's transaction ends. + */ + public void requireGatewayOwnership( + EntityManager em, Asset asset, String gatewayId, boolean allowGateway) { + if (asset == null || (!allowGateway && gatewayId.equals(asset.getId()))) { + throw new IllegalStateException("Gateway asset ownership conflict: " + gatewayId); + } + Set visited = new HashSet<>(); + String owner = null; + Asset current = asset; + while (current != null) { + if (!visited.add(current.getId()) || current.isDeletePending()) { + throw new IllegalStateException("Invalid gateway asset ancestry"); + } + if (current instanceof GatewayAsset) { + owner = current.getId(); + } + String parentId = current.getParentId(); + current = + parentId == null ? null : em.find(Asset.class, parentId, LockModeType.PESSIMISTIC_READ); + if (parentId != null && current == null) { + throw new IllegalStateException("Missing gateway asset ancestor"); + } + } + if (!gatewayId.equals(owner)) { + throw new IllegalStateException("Gateway asset ownership conflict: " + asset.getId()); + } + } + public R withAssetLock(String assetId, Supplier action) { try { assetLocks.lock(assetId); diff --git a/manager/src/main/java/org/openremote/manager/gateway/GatewayConnector.java b/manager/src/main/java/org/openremote/manager/gateway/GatewayConnector.java index 9bd9cd5b38..ac8d0e3d8f 100644 --- a/manager/src/main/java/org/openremote/manager/gateway/GatewayConnector.java +++ b/manager/src/main/java/org/openremote/manager/gateway/GatewayConnector.java @@ -20,6 +20,7 @@ import static org.openremote.model.syslog.SyslogCategory.GATEWAY; +import java.time.Instant; import java.util.*; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicReference; @@ -63,6 +64,7 @@ public class GatewayConnector { ASSET_ID_MAPPERS = new ConcurrentHashMap<>(); protected final String realm; protected final String gatewayId; + protected final int assetIdMappingVersion; protected final AssetStorageService assetStorageService; protected final ExecutorService executorService; protected final ScheduledExecutorService scheduledExecutorService; @@ -108,7 +110,8 @@ protected GatewayConnector( ScheduledExecutorService scheduledExecutorService, GatewayService gatewayService, TimerService timerService, - GatewayAsset gateway) { + GatewayAsset gateway, + Instant mappingV2ActivatedOn) { this.assetStorageService = assetStorageService; this.assetProcessingService = assetProcessingService; @@ -119,6 +122,8 @@ protected GatewayConnector( this.disabled = gateway.getDisabled().orElse(false); this.realm = gateway.getRealm(); this.gatewayId = gateway.getId(); + this.assetIdMappingVersion = + getAssetIdMappingVersion(gateway.getCreatedOn(), mappingV2ActivatedOn); this.gatewayAsset = gateway; // Setup static inbound event handling @@ -490,12 +495,12 @@ protected SharedEvent authoriseGatewayEvent(SharedEvent event) { if (event instanceof AssetInfo assetInfo) { // Map the gateway Asset ID to the central instance - String assetId = mapAssetId(gatewayId, assetInfo.getId(), false); + String assetId = mapAssetId(assetInfo.getId(), false); // Only use the gateway ID as the parent asset ID if the event is an asset event String parentAssetId = assetInfo.getParentId() == null ? event instanceof AssetEvent ? gatewayId : null - : mapAssetId(gatewayId, assetInfo.getParentId(), false); + : mapAssetId(assetInfo.getParentId(), false); // Events for the gateway asset are not allowed over the connector if (Objects.equals(assetId, gatewayId)) { @@ -639,33 +644,50 @@ protected synchronized void onSyncAssetsResponse(AssetsEvent e) { boolean isInitialResponse = ASSET_READ_EVENT_NAME_INITIAL.equalsIgnoreCase(messageId); if (isInitialResponse) { + if (e.getAssets() == null) { + throw new IllegalStateException("Missing gateway sync inventory"); + } // Put assets in hierarchical order Map gatewayAssetIdParentIdMap = - e.getAssets() == null - ? Collections.emptyMap() - : e.getAssets().stream() - .collect( - HashMap::new, (m, v) -> m.put(v.getId(), v.getParentId()), HashMap::putAll); + e.getAssets().stream() + .collect(HashMap::new, (m, v) -> m.put(v.getId(), v.getParentId()), HashMap::putAll); + + if (gatewayAssetIdParentIdMap.size() != e.getAssets().size()) { + throw new IllegalStateException("Duplicate gateway asset IDs in sync inventory"); + } + gatewayAssetIdParentIdMap.forEach( + (id, parent) -> { + validateAssetId(id); + if (parent != null && !gatewayAssetIdParentIdMap.containsKey(parent)) { + throw new IllegalStateException("Missing parent in gateway sync inventory"); + } + }); ToIntFunction> assetLevelExtractor = asset -> { int level = 0; String parentId = asset.getParentId(); + Set visited = new HashSet<>(); + visited.add(asset.getId()); while (parentId != null) { + if (!visited.add(parentId)) { + throw new IllegalStateException("Cycle in gateway sync inventory"); + } level++; parentId = gatewayAssetIdParentIdMap.get(parentId); } return level; }; + // Validate every chain, including single-asset inventories where sorting needs no + // comparisons. + e.getAssets().forEach(assetLevelExtractor::applyAsInt); syncAssetIds = - e.getAssets() == null - ? Collections.emptyList() - : e.getAssets().stream() - .sorted(Comparator.comparingInt(assetLevelExtractor)) - .map(Asset::getId) - .collect(Collectors.toList()); + e.getAssets().stream() + .sorted(Comparator.comparingInt(assetLevelExtractor)) + .map(Asset::getId) + .collect(Collectors.toList()); if (syncAssetIds.isEmpty()) { deleteObsoleteLocalAssets(); @@ -708,7 +730,8 @@ protected synchronized void onSyncAssetsResponse(AssetsEvent e) { return remove; }); - if (returnedAssets.size() != requestedAssetIds.size() + if (returnedAssets.stream().map(Asset::getId).distinct().count() != returnedAssets.size() + || returnedAssets.size() != requestedAssetIds.size() || !returnedAssets.stream() .allMatch(asset -> requestedAssetIds.contains(asset.getId()))) { LOG.warning( @@ -761,17 +784,7 @@ protected synchronized void onSyncAssetsResponse(AssetsEvent e) { syncAssetIds.remove(assetEvent.getId()); } else if (assetEvent.getCause() == AssetEvent.Cause.CREATE) { syncAssetIds.add(assetEvent.getId()); - try { - saveAssetLocally(assetEvent.getAsset()); - } catch (Exception ex) { - LOG.log( - Level.WARNING, - "Failed to add new gateway asset (Asset=" - + assetEvent.getAsset() - + "): " - + getGatewayIdString(), - ex); - } + saveAssetLocally(assetEvent.getAsset()); } else { refreshAssets.add(assetEvent.getId()); } @@ -816,15 +829,13 @@ protected void deleteObsoleteLocalAssets() { List obsoleteLocalAssetIds = localAssets.stream() .map(Asset::getId) - .filter(id -> !syncAssetIds.contains(mapAssetId(gatewayId, id, true))) + .filter(id -> !syncAssetIds.contains(mapAssetId(id, true))) .toList(); if (!obsoleteLocalAssetIds.isEmpty()) { boolean deleted = deleteAssetsLocally(obsoleteLocalAssetIds); if (!deleted) { - LOG.warning( - "Failed to delete obsolete local assets; assets are not correctly synced: " - + getGatewayIdString()); + throw new IllegalStateException("Failed to delete obsolete gateway assets: " + gatewayId); } } } @@ -842,28 +853,27 @@ protected synchronized void onAssetEvent(AssetEvent e) { switch (e.getCause()) { case CREATE, READ, UPDATE -> saveAssetLocally(e.getAsset()); - case DELETE -> - deleteAssetsLocally(Collections.singletonList(mapAssetId(gatewayId, e.getId(), false))); + case DELETE -> { + if (!deleteAssetsLocally(Collections.singletonList(mapAssetId(e.getId(), false)))) { + throw new IllegalStateException("Gateway asset deletion rejected"); + } + } } } protected void onAttributeEvent(AttributeEvent e) { // Just push the event through the processing chain - publishAttributeEvent( + assetProcessingService.sendAttributeEvent( new AttributeEvent( - mapAssetId(gatewayId, e.getId(), false), - e.getName(), - e.getValue().orElse(null), - e.getTimestamp())); + mapAssetId(e.getId(), false), e.getName(), e.getValue().orElse(null), e.getTimestamp()), + EVENT_SOURCE + ":" + gatewayId); } protected > T saveAssetLocally(T asset) { String assetId = asset.getId(); - asset.setId(mapAssetId(gatewayId, assetId, false)); + asset.setId(mapAssetId(assetId, false)); asset.setParentId( - asset.getParentId() != null - ? mapAssetId(gatewayId, asset.getParentId(), false) - : gatewayId); + asset.getParentId() != null ? mapAssetId(asset.getParentId(), false) : gatewayId); asset.setRealm(realm); LOG.finest( "Creating/updating asset: Asset ID=" @@ -881,7 +891,7 @@ protected boolean deleteAssetsLocally(List assetIds) { + Arrays.toString(assetIds.toArray()) + ": " + getGatewayIdString()); - return assetStorageService.delete(assetIds, true); + return assetStorageService.deleteGatewayAssets(assetIds, gatewayId); } public GatewayAsset getGatewayAsset() { @@ -897,12 +907,37 @@ protected String getGatewayIdString() { return "GatewayID=" + gatewayId; } - /** - * An easily reversible mathematical way of ensuring gateway asset IDs are unique by incrementing - * the first two characters by adding the first two characters of the gateway ID for inbound IDs - * and the reverse for outbound. - */ - public static String mapAssetId(String gatewayId, String assetId, boolean outbound) { + /** Gateways created before this installation's V2 upgrade retain V1 for all their assets. */ + public static int getAssetIdMappingVersion(Instant createdOn, Instant mappingV2ActivatedOn) { + Objects.requireNonNull(createdOn, "Gateway creation timestamp is required"); + Objects.requireNonNull(mappingV2ActivatedOn, "Gateway mapping cutoff is required"); + return createdOn.isBefore(mappingV2ActivatedOn) ? 1 : 2; + } + + /** Use the mapping selected from the persisted creation timestamp when connecting. */ + public String mapAssetId(String assetId, boolean outbound) { + return mapAssetId(gatewayId, assetId, outbound, assetIdMappingVersion); + } + + public static String mapAssetId(String gatewayId, String assetId, boolean outbound, int version) { + validateAssetId(gatewayId); + validateAssetId(assetId); + return switch (version) { + case 1 -> mapAssetIdV1(gatewayId, assetId, outbound); + case 2 -> mapAssetIdV2(gatewayId, assetId, outbound); + default -> + throw new IllegalArgumentException("Unsupported gateway mapping version: " + version); + }; + } + + private static void validateAssetId(String id) { + if (id == null || !id.matches(org.openremote.model.Constants.ASSET_ID_REGEXP)) { + throw new IllegalArgumentException("Invalid gateway asset ID: " + id); + } + } + + /** Legacy V1 mapping. Keep its arithmetic unchanged for existing registrations. */ + private static String mapAssetIdV1(String gatewayId, String assetId, boolean outbound) { Pair, Function> gatewayIdMappers = ASSET_ID_MAPPERS.computeIfAbsent( gatewayId, @@ -935,4 +970,16 @@ public static String mapAssetId(String gatewayId, String assetId, boolean outbou return outbound ? gatewayIdMappers.value.apply(assetId) : gatewayIdMappers.key.apply(assetId); } + + /** Position-by-position Base62 index addition; subtraction reverses the mapping. */ + private static String mapAssetIdV2(String gatewayId, String assetId, boolean outbound) { + StringBuilder mapped = new StringBuilder(22); + for (int i = 0; i < 22; i++) { + int gatewayIndex = ALPHA_NUMERIC_CHARACTERS.indexOf((int) gatewayId.charAt(i)); + int assetIndex = ALPHA_NUMERIC_CHARACTERS.indexOf((int) assetId.charAt(i)); + int index = Math.floorMod(assetIndex + (outbound ? -gatewayIndex : gatewayIndex), 62); + mapped.append((char) ALPHA_NUMERIC_CHARACTERS.get(index).intValue()); + } + return mapped.toString(); + } } diff --git a/manager/src/main/java/org/openremote/manager/gateway/GatewayService.java b/manager/src/main/java/org/openremote/manager/gateway/GatewayService.java index 421cff9e25..bf1014f931 100644 --- a/manager/src/main/java/org/openremote/manager/gateway/GatewayService.java +++ b/manager/src/main/java/org/openremote/manager/gateway/GatewayService.java @@ -21,7 +21,6 @@ import static org.apache.camel.builder.PredicateBuilder.or; import static org.openremote.container.persistence.PersistenceService.PERSISTENCE_TOPIC; import static org.openremote.container.persistence.PersistenceService.isPersistenceEventForEntityType; -import static org.openremote.manager.gateway.GatewayConnector.mapAssetId; import static org.openremote.model.Constants.*; import static org.openremote.model.syslog.SyslogCategory.GATEWAY; import static org.openremote.model.util.MapAccess.getInteger; @@ -29,6 +28,7 @@ import jakarta.persistence.EntityManager; import java.time.Duration; +import java.time.Instant; import java.util.*; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicBoolean; @@ -51,6 +51,7 @@ import org.openremote.manager.rules.RulesetStorageService; import org.openremote.manager.security.ManagerIdentityService; import org.openremote.manager.security.ManagerKeycloakIdentityProvider; +import org.openremote.manager.system.ApplicationVersionService; import org.openremote.manager.web.ManagerWebService; import org.openremote.model.Container; import org.openremote.model.ContainerService; @@ -85,6 +86,7 @@ public class GatewayService extends RouteBuilder implements ContainerService { // Need a high priority so that event authorizer can get the events before other authorizers public static final int PRIORITY = HIGH_PRIORITY + 100; + public static final String ASSET_ID_MAPPING_V2_VERSION = "1.34.0"; public static final String GATEWAY_CLIENT_ID_PREFIX = "gateway-"; public static final String OR_GATEWAY_TUNNEL_SSH_KEY_FILE = "OR_GATEWAY_TUNNEL_SSH_KEY_FILE"; private static final Logger LOG = @@ -106,6 +108,7 @@ public class GatewayService extends RouteBuilder implements ContainerService { protected ExecutorService executorService; protected ScheduledExecutorService scheduledExecutorService; protected TimerService timerService; + protected ApplicationVersionService applicationVersionService; protected String tunnelSSHHostname; protected String tunnelHostname; protected int tunnelSSHPort; @@ -188,6 +191,7 @@ public int getPriority() { @Override public void init(Container container) throws Exception { assetStorageService = container.getService(AssetStorageService.class); + applicationVersionService = container.getService(ApplicationVersionService.class); assetProcessingService = container.getService(AssetProcessingService.class); identityService = container.getService(ManagerIdentityService.class); clientEventService = container.getService(ClientEventService.class); @@ -242,6 +246,13 @@ public boolean intercept(EntityManager em, AttributeEvent event) getInteger(container.getConfig(), OR_GATEWAY_TUNNEL_AUTO_CLOSE_MINUTES, 0); } + protected Instant getGatewayMappingV2ActivatedOn() { + return applicationVersionService + .getFirstActivationAtOrAfter(ASSET_ID_MAPPING_V2_VERSION) + .orElseThrow( + () -> new IllegalStateException("No activation recorded for gateway mapping V2")); + } + @Override public void start(Container container) throws Exception { @@ -285,7 +296,8 @@ public void start(Container container) throws Exception { scheduledExecutorService, this, timerService, - gateway); + gateway, + getGatewayMappingV2ActivatedOn()); gatewayConnectorMap.put(gateway.getId().toLowerCase(Locale.ROOT), connector); // Get IDs of all assets under this gateway @@ -404,7 +416,18 @@ protected void onGatewayMessageIntercept(Exchange exchange) { public boolean onAttributeEventIntercepted(EntityManager em, AttributeEvent event) throws AssetProcessingException { - // If the event came from a gateway then we don't want to process it here + // The authenticated connector's identity survives queuing until this transaction. + if (event.getSource() != null + && event.getSource().startsWith(GatewayConnector.EVENT_SOURCE + ":")) { + String gatewayId = event.getSource().substring(GatewayConnector.EVENT_SOURCE.length() + 1); + Asset asset = + em.find(Asset.class, event.getId(), jakarta.persistence.LockModeType.PESSIMISTIC_WRITE); + assetStorageService.requireGatewayOwnership(em, asset, gatewayId, false); + event.getMeta().remove(MetaItemType.AGENT_LINK); + return false; + } + + // Internal status updates for the registration itself. if (getClass().getSimpleName().equals(event.getSource())) { // Clear out agentlink meta so agent interceptor doesn't try intercepting the event event.getMeta().remove(MetaItemType.AGENT_LINK); @@ -526,13 +549,19 @@ public boolean onAttributeEventIntercepted(EntityManager em, AttributeEvent even + ", gatewayId=" + gatewayId + ")"); + Asset persisted = + em.find(Asset.class, event.getId(), jakarta.persistence.LockModeType.PESSIMISTIC_READ); + assetStorageService.requireGatewayOwnership(em, persisted, gatewayId, false); connector.sendMessageToGateway( new AttributeEvent( - mapAssetId(gatewayId, event.getId(), true), + connector.mapAssetId(event.getId(), true), event.getName(), event.getValue().orElse(null), event.getTimestamp()) - .setParentId(mapAssetId(gatewayId, event.getParentId(), true)) + .setParentId( + gatewayId.equals(persisted.getParentId()) + ? null + : connector.mapAssetId(persisted.getParentId(), true)) .setRealm(event.getRealm())); // Consume this event as it is for a gateway descendant and we've sent it to that gateway for @@ -863,7 +892,8 @@ protected void processGatewayChange( scheduledExecutorService, this, timerService, - gateway); + gateway, + getGatewayMappingV2ActivatedOn()); gatewayConnectorMap.put(gateway.getId().toLowerCase(Locale.ROOT), connector); } case UPDATE -> { diff --git a/manager/src/main/java/org/openremote/manager/persistence/ApplicationVersion.java b/manager/src/main/java/org/openremote/manager/persistence/ApplicationVersion.java new file mode 100644 index 0000000000..8e3d89e056 --- /dev/null +++ b/manager/src/main/java/org/openremote/manager/persistence/ApplicationVersion.java @@ -0,0 +1,45 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.manager.persistence; + +import java.util.regex.Pattern; + +/** Normalized application versions used as activation-history keys. */ +public final class ApplicationVersion { + /** + * Accepts {@code major.minor.patch[-qualifier][+build.metadata]} as a complete version string. + * Each numeric version component is zero or a positive decimal integer without leading zeros. + * Optional qualifier and build suffixes contain nonempty, dot-separated identifiers made of ASCII + * letters, digits, or hyphens (for example, {@code SNAPSHOT}, {@code M1}, or {@code beta.2}). + * Numeric suffix identifiers may have leading zeros; this is not strict SemVer validation. Both + * suffixes are discarded during normalization, leaving only {@code major.minor.patch}. + */ + private static final Pattern VERSION = + Pattern.compile( + "(0|[1-9][0-9]*)\\.(0|[1-9][0-9]*)\\.(0|[1-9][0-9]*)(?:-[0-9A-Za-z-]+(?:\\.[0-9A-Za-z-]+)*)?(?:\\+[0-9A-Za-z-]+(?:\\.[0-9A-Za-z-]+)*)?"); + + private ApplicationVersion() {} + + public static String normalize(String version) { + if (version == null || !VERSION.matcher(version).matches()) { + throw new IllegalArgumentException("Invalid OpenRemote application version: " + version); + } + return version.split("[-+]", 2)[0]; + } +} diff --git a/manager/src/main/java/org/openremote/manager/system/ApplicationVersionService.java b/manager/src/main/java/org/openremote/manager/system/ApplicationVersionService.java new file mode 100644 index 0000000000..5da7d9b939 --- /dev/null +++ b/manager/src/main/java/org/openremote/manager/system/ApplicationVersionService.java @@ -0,0 +1,109 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.manager.system; + +import java.time.Instant; +import java.util.List; +import java.util.Optional; +import org.openremote.container.persistence.PersistenceService; +import org.openremote.manager.persistence.ApplicationVersion; +import org.openremote.model.Container; +import org.openremote.model.ContainerService; + +/** Records the first activation of each normalized application version for other services. */ +public class ApplicationVersionService implements ContainerService { + + protected PersistenceService persistenceService; + + @Override + public int getPriority() { + return PersistenceService.PRIORITY + 10; + } + + @Override + public void init(Container container) { + persistenceService = container.getService(PersistenceService.class); + } + + @Override + public void start(Container container) { + // Persistence has completed migrations and started before version history is initialized. + recordApplicationVersion(VersionInfo.getManagerVersion()); + } + + @Override + public void stop(Container container) {} + + /** + * Returns the first recorded activation of a normalized version, ignoring any build suffixes. The + * timestamp is stored in {@code updated_on} and is never refreshed on reactivation. + */ + public Optional getFirstActivation(String applicationVersion) { + String version = ApplicationVersion.normalize(applicationVersion); + return persistenceService.doReturningTransaction( + em -> { + List timestamps = + em.createNativeQuery( + "SELECT updated_on FROM application_version WHERE version = :version", + Instant.class) + .setParameter("version", version) + .getResultList(); + return timestamps.isEmpty() + ? Optional.empty() + : Optional.of((Instant) timestamps.getFirst()); + }); + } + + /** + * Returns when this installation first ran the given version or a later version. This also + * supports installations that skipped the requested release. Versions are compared numerically; + * qualifiers and build metadata are ignored. Empty means no qualifying activation was recorded. + */ + public Optional getFirstActivationAtOrAfter(String applicationVersion) { + String version = ApplicationVersion.normalize(applicationVersion); + return persistenceService.doReturningTransaction( + em -> + Optional.ofNullable( + (Instant) + em.createNativeQuery( + """ + SELECT MIN(updated_on) FROM application_version + WHERE CAST(string_to_array(version, '.') AS NUMERIC[]) >= + CAST(string_to_array(:version, '.') AS NUMERIC[]) + """, + Instant.class) + .setParameter("version", version) + .getSingleResult())); + } + + protected void recordApplicationVersion(String applicationVersion) { + String version = ApplicationVersion.normalize(applicationVersion); + // Atomic insert preserves the first timestamp across concurrent startups. + persistenceService.doTransaction( + em -> + em.createNativeQuery( + """ + INSERT INTO application_version (version, updated_on) + VALUES (:version, clock_timestamp()) + ON CONFLICT (version) DO NOTHING + """) + .setParameter("version", version) + .executeUpdate()); + } +} diff --git a/manager/src/main/resources/META-INF/services/org.openremote.model.ContainerService b/manager/src/main/resources/META-INF/services/org.openremote.model.ContainerService index 47f7117a59..e8434dd514 100644 --- a/manager/src/main/resources/META-INF/services/org.openremote.model.ContainerService +++ b/manager/src/main/resources/META-INF/services/org.openremote.model.ContainerService @@ -1,6 +1,7 @@ org.openremote.manager.rules.geofence.ORConsoleGeofenceAssetAdapter org.openremote.container.timer.TimerService org.openremote.manager.persistence.ManagerPersistenceService +org.openremote.manager.system.ApplicationVersionService org.openremote.manager.security.ManagerIdentityService org.openremote.manager.setup.SetupService org.openremote.manager.event.ClientEventService diff --git a/manager/src/main/resources/org/openremote/manager/setup/database/V20261105_1530__Application_version.sql b/manager/src/main/resources/org/openremote/manager/setup/database/V20261105_1530__Application_version.sql new file mode 100644 index 0000000000..d68a3e4bb5 --- /dev/null +++ b/manager/src/main/resources/org/openremote/manager/setup/database/V20261105_1530__Application_version.sql @@ -0,0 +1,6 @@ +-- Only first activations from this migration onwards are recorded. +CREATE TABLE application_version ( + version TEXT PRIMARY KEY, + -- First activation timestamp; subsequent startups never change it. + updated_on TIMESTAMP WITH TIME ZONE NOT NULL +); diff --git a/manager/src/test/groovy/org/openremote/manager/gateway/GatewayMappingTest.groovy b/manager/src/test/groovy/org/openremote/manager/gateway/GatewayMappingTest.groovy new file mode 100644 index 0000000000..9e844d9a5a --- /dev/null +++ b/manager/src/test/groovy/org/openremote/manager/gateway/GatewayMappingTest.groovy @@ -0,0 +1,193 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.manager.gateway + +import org.openremote.container.timer.TimerService +import org.openremote.manager.asset.AssetProcessingService +import org.openremote.manager.asset.AssetStorageService +import org.openremote.model.asset.Asset +import org.openremote.model.asset.AssetEvent +import org.openremote.model.asset.AssetsEvent +import org.openremote.model.asset.impl.GatewayAsset +import org.openremote.model.asset.impl.ThingAsset +import org.openremote.model.util.UniqueIdentifierGenerator +import spock.lang.Specification + +import java.time.Instant +import java.util.concurrent.ExecutorService +import java.util.concurrent.ScheduledExecutorService +import java.util.concurrent.ScheduledFuture + +class GatewayMappingTest extends Specification { + def setupSpec() { + org.openremote.model.util.ValueUtil.initialise(null) + } + + static final String GATEWAY = "4H1sjYjbYMnvQRsJnkmcrs" + static final String OTHER = "4HQRYPyyW46j5pWoaKz0B3" + static final String EDGE = "0123456789ABCDEFGHIJKL" + + def "V#version is reversible across Base62 IDs"() { + expect: + 200.times { + def id = UniqueIdentifierGenerator.generateId() + def mapped = GatewayConnector.mapAssetId(GATEWAY, id, false, version) + assert mapped ==~ /[a-zA-Z0-9]{22}/ + assert mapped == GatewayConnector.mapAssetId(GATEWAY, id, false, version) + assert GatewayConnector.mapAssetId(GATEWAY, mapped, true, version) == id + } + true + + where: + version << [1, 2] + } + + def "legacy V1 output remains unchanged and V2 separates same-prefix gateways"() { + expect: + GatewayConnector.mapAssetId(GATEWAY, EDGE, false, 1) == "Qb23456789ABCDEFGHIJKL" + GatewayConnector.mapAssetId(GATEWAY, EDGE, false, 1) == GatewayConnector.mapAssetId(OTHER, EDGE, false, 1) + GatewayConnector.mapAssetId(GATEWAY, EDGE, false, 2) != GatewayConnector.mapAssetId(OTHER, EDGE, false, 2) + } + + def "mapping rejects malformed input"() { + when: + GatewayConnector.mapAssetId(gateway, edge, false, version) + + then: + thrown(IllegalArgumentException) + + where: + gateway | edge | version + null | EDGE | 1 + GATEWAY | null | 2 + "short" | EDGE | 2 + GATEWAY | "!" * 22 | 1 + GATEWAY | EDGE | 3 + } + + def "gateway mapping uses the fixed upgrade timestamp (#createdOn)"() { + expect: + GatewayConnector.getAssetIdMappingVersion(Instant.parse(createdOn), Instant.parse("2026-11-05T12:00:00Z")) == version + + where: + createdOn | version + "2026-11-05T11:59:59.999999Z" | 1 + "2026-11-05T12:00:00Z" | 2 + "2026-11-05T12:00:00.000001Z" | 2 + } + + def "missing mapping timestamps are rejected"() { + when: + GatewayConnector.getAssetIdMappingVersion(createdOn, cutoff) + + then: + thrown(NullPointerException) + + where: + createdOn | cutoff + null | Instant.EPOCH + Instant.EPOCH | null + } + + def "mapping is frozen for a connector and reconstructed from creation time"() { + given: + def gateway = new GatewayAsset("New").setId(GATEWAY).setRealm("test").setCreatedOn(Instant.parse("2026-11-06T00:00:00Z")) + def first = connector(gateway, Stub(AssetStorageService)) + def reconnected = connector(gateway, Stub(AssetStorageService)) + gateway.setCreatedOn(Instant.EPOCH) + + expect: + first.mapAssetId(EDGE, false) == GatewayConnector.mapAssetId(GATEWAY, EDGE, false, 2) + reconnected.mapAssetId(EDGE, false) == first.mapAssetId(EDGE, false) + } + + def "invalid inventory never triggers cleanup"() { + given: + def storage = Mock(AssetStorageService) + def gateway = new GatewayAsset("Test").setId(GATEWAY).setRealm("test") + def connector = connector(gateway, storage) + connector.@sessionId.set("session") + connector.initialSyncInProgress = true + connector.syncProcessorFuture = Stub(ScheduledFuture) + connector.expectedSyncResponseName = "INITIAL" + def response = new AssetsEvent(inventory) + response.setMessageID("INITIAL") + + when: + connector.onGatewayEvent(response) + + then: + !connector.connected + 0 * storage._ + + where: + inventory << [ + null, + [new ThingAsset("Invalid").setId("bad")], + [new ThingAsset("Duplicate").setId(EDGE), new ThingAsset("Duplicate").setId(EDGE)], + [new ThingAsset("Cycle").setId(EDGE).setParentId(EDGE)], + [new ThingAsset("Missing parent").setId(EDGE).setParentId("a" * 22)] + ] + } + + def "sync failure stops before cleanup including cached creates"() { + given: + def storage = Mock(AssetStorageService) + def gateway = new GatewayAsset("Test").setId(GATEWAY).setRealm("test") + def connector = connector(gateway, storage) + connector.@sessionId.set("session") + connector.initialSyncInProgress = true + connector.syncProcessorFuture = Stub(ScheduledFuture) + connector.expectedSyncResponseName = "BATCH0" + connector.syncAssetIds = [EDGE] + connector.syncIndex = 0 + connector.cachedAttributeEvents = [] + connector.cachedAssetEvents = cachedFailure ? [new AssetEvent(AssetEvent.Cause.CREATE, + new ThingAsset("Cached").setId("a" * 22))] : [] + def response = new AssetsEvent([new ThingAsset("Edge").setId(EDGE)]) + response.setMessageID("BATCH0") + + when: + connector.onGatewayEvent(response) + + then: + (cachedFailure ? 2 : 1) * storage.merge(_ as Asset, true, gateway, null) >> { + if (!cachedFailure || it[0].name == "Cached") { + throw new IllegalStateException("ownership conflict") + } + it[0] + } + 0 * storage.findAll(_) + 0 * storage.deleteGatewayAssets(_, _) + !connector.connected + + where: + cachedFailure << [false, true] + } + + private GatewayConnector connector(GatewayAsset gateway, AssetStorageService storage) { + if (gateway.createdOn == null) { + gateway.setCreatedOn(Instant.EPOCH) + } + new GatewayConnector(storage, Stub(AssetProcessingService), Stub(ExecutorService), + Stub(ScheduledExecutorService), Stub(GatewayService), Stub(TimerService) { + getNow() >> Instant.now() + }, gateway, Instant.parse("2026-11-05T12:00:00Z")) + } +} diff --git a/manager/src/test/groovy/org/openremote/manager/persistence/ApplicationVersionTest.groovy b/manager/src/test/groovy/org/openremote/manager/persistence/ApplicationVersionTest.groovy new file mode 100644 index 0000000000..7358184c26 --- /dev/null +++ b/manager/src/test/groovy/org/openremote/manager/persistence/ApplicationVersionTest.groovy @@ -0,0 +1,52 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.manager.persistence + +import spock.lang.Specification + +class ApplicationVersionTest extends Specification { + def "normalize #version"() { + expect: + ApplicationVersion.normalize(version) == "1.2.0" + + where: + version << [ + "1.2.0", + "1.2.0-SNAPSHOT", + "1.2.0-M1", + "1.2.0-RC1", + "1.2.0-beta.2", + "1.2.0-alpha", + "1.2.0-LOCAL", + "1.2.0+build.42", + "1.2.0-RC1+build.42" + ] + } + + def "reject unexpected version #version"() { + when: + ApplicationVersion.normalize(version) + + then: + thrown(IllegalArgumentException) + + where: + version << [null, "", "1.2", "v1.2.0", "01.2.0", "1.2.0-", "1.2.0+", "1.2.0/foo", "1.2.0-beta..2", "1.2.0 "] + } +} diff --git a/test/src/test/groovy/org/openremote/test/gateway/AbstractGatewayCompatibilityTest.groovy b/test/src/test/groovy/org/openremote/test/gateway/AbstractGatewayCompatibilityTest.groovy index d59f7542d3..e7352e6c5b 100644 --- a/test/src/test/groovy/org/openremote/test/gateway/AbstractGatewayCompatibilityTest.groovy +++ b/test/src/test/groovy/org/openremote/test/gateway/AbstractGatewayCompatibilityTest.groovy @@ -23,6 +23,7 @@ import org.apache.http.client.utils.URIBuilder import org.openremote.agent.protocol.io.AbstractNettyIOClient import org.openremote.agent.protocol.websocket.WebsocketIOClient import org.openremote.manager.asset.AssetStorageService +import org.openremote.manager.gateway.GatewayConnector import org.openremote.manager.gateway.GatewayService import org.openremote.manager.security.ManagerIdentityService import org.openremote.manager.security.ManagerKeycloakIdentityProvider @@ -44,10 +45,13 @@ import org.openremote.setup.integration.ManagerTestSetup import spock.lang.Specification import spock.util.concurrent.PollingConditions -import static org.openremote.manager.gateway.GatewayConnector.mapAssetId abstract class AbstractGatewayCompatibilityTest extends Specification implements ManagerContainerTrait { + private static String mapAssetId(String gatewayId, String assetId, boolean outbound) { + GatewayConnector.mapAssetId(gatewayId, assetId, outbound, 2) + } + // Frozen AssetsEvent JSON captured from BETA (c706a7fb78542967c68eb97dcf6ef06673475ff7, // be467d6^), 1.0.0 (be467d6dfaa3875c5d48b65146cecb5bc5b8e30b), and 1.1.0 // (c7c1c4c962c6954282d1ec38a1361fb3a99177ad). All three serialize these assets identically. diff --git a/test/src/test/groovy/org/openremote/test/gateway/GatewayOwnershipTest.groovy b/test/src/test/groovy/org/openremote/test/gateway/GatewayOwnershipTest.groovy new file mode 100644 index 0000000000..baa96aa7c6 --- /dev/null +++ b/test/src/test/groovy/org/openremote/test/gateway/GatewayOwnershipTest.groovy @@ -0,0 +1,215 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.test.gateway + +import org.openremote.manager.asset.AssetProcessingException +import org.openremote.manager.asset.AssetStorageService +import org.openremote.manager.asset.AssetProcessingService +import org.openremote.manager.gateway.GatewayConnector +import org.openremote.manager.gateway.GatewayService +import org.openremote.container.persistence.PersistenceService +import org.openremote.manager.system.ApplicationVersionService +import org.openremote.manager.setup.SetupService +import org.openremote.manager.system.VersionInfo +import org.openremote.model.asset.impl.GatewayAsset +import org.openremote.model.asset.impl.ThingAsset +import org.openremote.model.attribute.Attribute +import org.openremote.model.attribute.AttributeEvent +import org.openremote.model.value.ValueType +import org.openremote.setup.integration.ManagerTestSetup +import org.openremote.test.ManagerContainerTrait +import spock.lang.Specification +import spock.util.concurrent.PollingConditions + +class GatewayOwnershipTest extends Specification implements ManagerContainerTrait { + def setupSpec() { + startContainer(defaultConfig(), defaultServices()) + } + + def "new registrations use V2 across updates, qualified restarts and rollback"() { + given: + def storage = container.getService(AssetStorageService) + def versions = container.getService(ApplicationVersionService) + def gateways = container.getService(GatewayService) + def realm = container.getService(SetupService).getTaskOfType(ManagerTestSetup).realmBuildingName + def gateway = storage.merge(new GatewayAsset("V2 registration").setRealm(realm)) + def cutoff = versions.getFirstActivationAtOrAfter("1.34.0").orElseThrow() + def createdOn = gateway.createdOn + def mapId = { + -> gateways.gatewayConnectorMap[gateway.id.toLowerCase(Locale.ROOT)].mapAssetId("b" * 22, false) + } + new PollingConditions(timeout: 15).eventually { + assert gateways.gatewayConnectorMap[gateway.id.toLowerCase(Locale.ROOT)] != null + } + + expect: + GatewayConnector.getAssetIdMappingVersion(createdOn, cutoff) == 2 + mapId() == GatewayConnector.mapAssetId(gateway.id, "b" * 22, false, 2) + + when: "ordinary updates try to change the creation timestamp" + gateway = storage.find(gateway.id, true) as GatewayAsset + gateway = storage.merge(gateway.setCreatedOn(java.time.Instant.EPOCH).setName("Updated")) + gateway = storage.find(gateway.id, true) as GatewayAsset + + then: + gateway.createdOn == createdOn + + when: "qualified builds reactivate and the application rolls back and upgrades" + [ + VersionInfo.managerVersion, + VersionInfo.managerVersion, + "0.9.0-SNAPSHOT", + VersionInfo.managerVersion + ].each { version -> + versions.recordApplicationVersion(version) + assert versions.getFirstActivationAtOrAfter("1.34.0").orElseThrow() == cutoff + } + gateways.stop(container) + gateways.start(container) + + then: + mapId() == GatewayConnector.mapAssetId(gateway.id, "b" * 22, false, 2) + !storage.find(gateway.id, true).attributes.has("assetIdMappingVersion") + } + + def "gateways created before the upgrade retain V1 across updates and service restarts"() { + given: + def storage = container.getService(AssetStorageService) + def persistence = container.getService(PersistenceService) + def versions = container.getService(ApplicationVersionService) + def realm = container.getService(SetupService).getTaskOfType(ManagerTestSetup).realmBuildingName + def gateway = storage.merge(new GatewayAsset("Legacy fixture").setRealm(realm)) + new PollingConditions(timeout: 15).eventually { + assert container.getService(GatewayService).gatewayConnectorMap[gateway.id.toLowerCase(Locale.ROOT)] != null + } + // Simulate a registration that already existed before the migration. + persistence.doTransaction { em -> + em.createNativeQuery("UPDATE asset SET created_on = :createdOn WHERE id = :id") + .setParameter("createdOn", versions.getFirstActivationAtOrAfter("1.34.0").orElseThrow().minusSeconds(1)) + .setParameter("id", gateway.id).executeUpdate() + } + gateway = storage.find(gateway.id, true) as GatewayAsset + + when: + gateway = storage.merge(gateway.setName("Updated legacy gateway")) + def version = GatewayConnector.getAssetIdMappingVersion(gateway.createdOn, versions.getFirstActivationAtOrAfter("1.34.0").orElseThrow()) + def centralId = GatewayConnector.mapAssetId(gateway.id, "b" * 22, false, version) + def child = storage.merge(new ThingAsset("Legacy child").setId(centralId).setParentId(gateway.id).setRealm(realm), true, gateway, null) + def gateways = container.getService(GatewayService) + gateways.stop(container) + gateways.start(container) + + then: + version == 1 + child.id == GatewayConnector.mapAssetId(gateway.id, "b" * 22, false, 1) + gateways.gatewayConnectorMap[gateway.id.toLowerCase(Locale.ROOT)].mapAssetId("b" * 22, false) == child.id + !gateway.attributes.has("assetIdMappingVersion") + } + + def "gateway cannot overwrite, move, delete or write another owner's asset (V#otherVersion collision)"() { + given: + def storage = container.getService(AssetStorageService) + def persistence = container.getService(PersistenceService) + def versions = container.getService(ApplicationVersionService) + def realm = container.getService(SetupService).getTaskOfType(ManagerTestSetup).realmBuildingName + def owner = storage.merge(new GatewayAsset("Owner").setRealm(realm)) + def attacker = storage.merge(new GatewayAsset("Attacker").setRealm(realm)) + def gatewayService = container.getService(GatewayService) + new PollingConditions(timeout: 15).eventually { + assert gatewayService.gatewayConnectorMap[owner.id.toLowerCase(Locale.ROOT)] != null + assert gatewayService.gatewayConnectorMap[attacker.id.toLowerCase(Locale.ROOT)] != null + } + if (otherVersion == 1) { + persistence.doTransaction { em -> + em.createNativeQuery("UPDATE asset SET created_on = :createdOn WHERE id = :id") + .setParameter("createdOn", versions.getFirstActivationAtOrAfter("1.34.0").orElseThrow().minusSeconds(1)) + .setParameter("id", owner.id).executeUpdate() + } + owner = storage.find(owner.id, true) as GatewayAsset + } + def targetId = GatewayConnector.mapAssetId(owner.id, "b" * 22, false, otherVersion) + def target = storage.merge(new ThingAsset("Owned").setId(targetId).setParentId(owner.id).setRealm(realm) + .addAttributes(new Attribute<>("test", ValueType.TEXT, "original")), true, owner, null) + def maliciousEdgeId = GatewayConnector.mapAssetId(attacker.id, target.id, true, 2) + assert GatewayConnector.mapAssetId(attacker.id, maliciousEdgeId, false, 2) == target.id + + when: "incoming create/update claims an occupied mapped ID" + storage.merge(new ThingAsset("Attack").setId(target.id).setParentId(attacker.id).setRealm(realm), true, attacker, null) + + then: + thrown(IllegalStateException) + storage.find(target.id, true).parentId == owner.id + storage.find(target.id, true).name == "Owned" + + when: "new asset names a parent outside its subtree" + storage.merge(new ThingAsset("Attack parent").setId("z" * 22).setParentId(target.id).setRealm(realm), true, attacker, null) + + then: + thrown(IllegalStateException) + storage.find("z" * 22, true) == null + + when: "cleanup includes an owned asset and a conflicting asset" + def ownChild = storage.merge(new ThingAsset("Own child").setId(org.openremote.model.util.UniqueIdentifierGenerator.generateId()).setParentId(attacker.id).setRealm(realm), true, attacker, null) + def deleted = storage.deleteGatewayAssets([ownChild.id, target.id], attacker.id) + + then: "the whole deletion batch is rejected" + !deleted + !storage.find(target.id, true).deletePending + !storage.find(ownChild.id, true).deletePending + + when: "a queued incoming attribute write retains the authenticated gateway identity" + container.getService(AssetProcessingService).processAttributeEvent( + new AttributeEvent(target.id, "test", "attack").setSource(GatewayConnector.EVENT_SOURCE + ":" + attacker.id)) + + then: + thrown(AssetProcessingException) + storage.find(target.id, true).getAttribute("test").get().value.get() == "original" + + when: "persisted ownership authorizes the actual owner" + container.getService(AssetProcessingService).processAttributeEvent( + new AttributeEvent(target.id, "test", "authorized", target.getAttribute("test").get().timestamp.get() + 1L).setSource(GatewayConnector.EVENT_SOURCE + ":" + owner.id)) + + then: + storage.find(target.id, true).getAttribute("test").get().value.get() == "authorized" + + when: "an incorrect routing cache tries to send the target to another gateway" + def wrongConnector = gatewayService.gatewayConnectorMap[attacker.id.toLowerCase(Locale.ROOT)] + def sent = [] + wrongConnector.@sessionId.set("test-session") + wrongConnector.gatewayMessageConsumer = { sent.add(it) } + gatewayService.assetIdGatewayIdMap.put(target.id, attacker.id) + container.getService(AssetProcessingService).processAttributeEvent( + new AttributeEvent(target.id, "test", "outbound")) + + then: + thrown(AssetProcessingException) + sent.empty + + cleanup: + if (wrongConnector != null) { + wrongConnector.@sessionId.set(null) + } + if (gatewayService != null && target != null && owner != null) { + gatewayService.assetIdGatewayIdMap.put(target.id, owner.id) + } + + where: + otherVersion << [1, 2] + } +} diff --git a/test/src/test/groovy/org/openremote/test/gateway/GatewayTest.groovy b/test/src/test/groovy/org/openremote/test/gateway/GatewayTest.groovy index b007b01be0..ae3baf0806 100644 --- a/test/src/test/groovy/org/openremote/test/gateway/GatewayTest.groovy +++ b/test/src/test/groovy/org/openremote/test/gateway/GatewayTest.groovy @@ -74,7 +74,6 @@ import java.util.function.Consumer import java.util.stream.Collectors import java.util.stream.IntStream -import static org.openremote.manager.gateway.GatewayConnector.mapAssetId import static org.openremote.manager.gateway.GatewayService.getGatewayClientId import static org.openremote.manager.security.ManagerIdentityProvider.OR_ADMIN_PASSWORD import static org.openremote.manager.security.ManagerIdentityProvider.OR_ADMIN_PASSWORD_DEFAULT @@ -86,6 +85,10 @@ import static org.openremote.model.value.ValueType.* class GatewayTest extends Specification implements ManagerContainerTrait { + private static String mapAssetId(String gatewayId, String assetId, boolean outbound) { + GatewayConnector.mapAssetId(gatewayId, assetId, outbound, 2) + } + def setupSpec() { // Mock the GatewayClientService so we can use a different port for authentication on the GatewayClientConnector def services = defaultServices().toList() diff --git a/test/src/test/groovy/org/openremote/test/system/ApplicationVersionServiceTest.groovy b/test/src/test/groovy/org/openremote/test/system/ApplicationVersionServiceTest.groovy new file mode 100644 index 0000000000..437c31f187 --- /dev/null +++ b/test/src/test/groovy/org/openremote/test/system/ApplicationVersionServiceTest.groovy @@ -0,0 +1,199 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.test.system + +import org.openremote.container.persistence.PersistenceService +import org.openremote.manager.persistence.ApplicationVersion +import org.openremote.manager.system.ApplicationVersionService +import org.openremote.manager.system.VersionInfo +import org.openremote.test.ManagerContainerTrait +import spock.lang.Specification + +import java.util.concurrent.Callable +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit + +class ApplicationVersionServiceTest extends Specification implements ManagerContainerTrait { + def setupSpec() { + startContainer(defaultConfig(), defaultServices()) + } + + def "activation history records each normalized version once across upgrades, restarts and rollbacks"() { + given: + def persistence = container.getService(PersistenceService) + def versions = container.getService(ApplicationVersionService) + def savedRows = snapshotHistory(persistence) + def rows = { + -> persistence.doReturningTransaction { em -> + em.createNativeQuery("SELECT version, updated_on FROM application_version").resultList.collectEntries { + [(it[0]): it[1]] + } + } + } + + expect: "startup has recorded the running application after migrations" + rows().containsKey(ApplicationVersion.normalize(VersionInfo.managerVersion)) + + when: + persistence.doTransaction { + it.createNativeQuery("DELETE FROM application_version").executeUpdate() + } + versions.recordApplicationVersion("1.0.0") + def first = rows() + def firstActivation = versions.getFirstActivation("1.0.0").orElseThrow() + versions.recordApplicationVersion("1.0.0") + + then: + rows() == first + first.keySet() == ["1.0.0"] as Set + versions.getFirstActivation("1.0.0").orElseThrow() == firstActivation + + when: + versions.recordApplicationVersion("1.1.0") + def upgraded = rows() + versions.recordApplicationVersion("1.1.0") + + then: + rows() == upgraded + upgraded["1.1.0"] > first["1.0.0"] + + when: + versions.recordApplicationVersion("1.0.0") + + then: "rollback preserves the original activation" + rows() == upgraded + rows().size() == 2 + versions.getFirstActivation("1.0.0").orElseThrow() == firstActivation + + when: + ["1.2.0-SNAPSHOT", "1.2.0-SNAPSHOT", "1.2.0-M1", "1.2.0-RC1", "1.2.0-beta.2", "1.2.0+build.1"].each { version -> + def before = rows()["1.2.0"] + versions.recordApplicationVersion(version) + assert before == null || rows()["1.2.0"] == before + } + def qualified = rows() + versions.recordApplicationVersion("1.2.0") + + then: "the release retains the first activation of its normalized version" + rows() == qualified + + when: + versions.recordApplicationVersion("1.3.0") + + then: + rows()["1.3.0"] > qualified["1.2.0"] + + cleanup: + restoreHistory(persistence, savedRows) + } + + def "release queries handle skipped versions, numeric ordering, qualifiers and rollbacks"() { + given: + def persistence = container.getService(PersistenceService) + def versions = container.getService(ApplicationVersionService) + def savedRows = snapshotHistory(persistence) + persistence.doTransaction { + it.createNativeQuery("DELETE FROM application_version").executeUpdate() + } + + expect: + versions.getFirstActivation("1.34.0").empty + versions.getFirstActivationAtOrAfter("1.34.0").empty + + when: "an earlier release does not satisfy a numerically higher release query" + versions.recordApplicationVersion("1.9.0") + + then: + versions.getFirstActivationAtOrAfter("1.34.0").empty + + when: "this installation upgrades directly to a later development release" + versions.recordApplicationVersion("1.40.0-SNAPSHOT") + def cutoff = versions.getFirstActivation("1.40.0").orElseThrow() + versions.recordApplicationVersion("1.40.0-SNAPSHOT") + versions.recordApplicationVersion("1.34.0-M1") + versions.recordApplicationVersion("1.34.0") + versions.recordApplicationVersion("1.9.0") + + then: "the earliest qualifying activation remains fixed across restarts and rollbacks" + versions.getFirstActivationAtOrAfter("1.34.0-RC1+build.2").orElseThrow() == cutoff + versions.getFirstActivation("1.40.0-beta.2").orElseThrow() == cutoff + versions.getFirstActivation("1.34.0").orElseThrow() > cutoff + versions.getFirstActivationAtOrAfter("2.0.0").empty + versions.getFirstActivation("1.9.0").present + + when: "another service lifecycle restart records the running version" + versions.stop(container) + versions.start(container) + + then: + versions.getFirstActivationAtOrAfter("1.34.0").orElseThrow() == cutoff + + cleanup: + restoreHistory(persistence, savedRows) + } + + def "concurrent activations preserve one timestamp for a normalized version"() { + given: + def persistence = container.getService(PersistenceService) + def versions = container.getService(ApplicationVersionService) + def savedRows = snapshotHistory(persistence) + persistence.doTransaction { + it.createNativeQuery("DELETE FROM application_version").executeUpdate() + } + def executor = Executors.newFixedThreadPool(4) + + when: + def startups = (["2.0.0-SNAPSHOT", "2.0.0-RC1", "2.0.0"] * 4).collect { version -> + executor.submit({ + -> + versions.recordApplicationVersion(version) + versions.getFirstActivation("2.0.0").orElseThrow() + } as Callable) + } + def activations = startups.collect { it.get(10, TimeUnit.SECONDS) } + + then: + activations.toSet().size() == 1 + snapshotHistory(persistence).size() == 1 + + cleanup: + executor?.shutdownNow() + executor?.awaitTermination(10, TimeUnit.SECONDS) + restoreHistory(persistence, savedRows) + } + + private List snapshotHistory(PersistenceService persistence) { + persistence.doReturningTransaction { em -> + em.createNativeQuery("SELECT version, updated_on FROM application_version").resultList + } + } + + private void restoreHistory(PersistenceService persistence, List rows) { + if (persistence == null || rows == null) { + return + } + persistence.doTransaction { em -> + em.createNativeQuery("DELETE FROM application_version").executeUpdate() + rows.each { row -> + em.createNativeQuery("INSERT INTO application_version (version, updated_on) VALUES (:version, :activation)") + .setParameter("version", row[0]).setParameter("activation", row[1]).executeUpdate() + } + } + } +} From 926ed42d9cff6d206e1b4229cf55d2fd01ac6e2f Mon Sep 17 00:00:00 2001 From: Wouter Born Date: Fri, 6 Nov 2026 12:13:22 +0100 Subject: [PATCH 2/3] Reuse VersionUtil --- .../persistence/ApplicationVersion.java | 45 ----- .../system/ApplicationVersionService.java | 8 +- .../VersionInfoTest.groovy} | 34 +--- .../openremote/model/util/VersionUtil.java | 16 ++ .../model/util/VersionUtilTest.groovy | 190 ++++++++++++++++++ .../ApplicationVersionServiceTest.groovy | 4 +- .../test/system/VersionUtilTest.groovy | 133 ------------ 7 files changed, 218 insertions(+), 212 deletions(-) delete mode 100644 manager/src/main/java/org/openremote/manager/persistence/ApplicationVersion.java rename manager/src/test/groovy/org/openremote/manager/{persistence/ApplicationVersionTest.groovy => system/VersionInfoTest.groovy} (54%) create mode 100644 model/src/test/groovy/org/openremote/model/util/VersionUtilTest.groovy delete mode 100644 test/src/test/groovy/org/openremote/test/system/VersionUtilTest.groovy diff --git a/manager/src/main/java/org/openremote/manager/persistence/ApplicationVersion.java b/manager/src/main/java/org/openremote/manager/persistence/ApplicationVersion.java deleted file mode 100644 index 8e3d89e056..0000000000 --- a/manager/src/main/java/org/openremote/manager/persistence/ApplicationVersion.java +++ /dev/null @@ -1,45 +0,0 @@ -/* - * Copyright 2026, OpenRemote Inc. - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the GNU Affero General Public License as - * published by the Free Software Foundation, either version 3 of the - * License, or (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Affero General Public License for more details. - * - * You should have received a copy of the GNU Affero General Public License - * along with this program. If not, see . - * - * SPDX-License-Identifier: AGPL-3.0-or-later - */ -package org.openremote.manager.persistence; - -import java.util.regex.Pattern; - -/** Normalized application versions used as activation-history keys. */ -public final class ApplicationVersion { - /** - * Accepts {@code major.minor.patch[-qualifier][+build.metadata]} as a complete version string. - * Each numeric version component is zero or a positive decimal integer without leading zeros. - * Optional qualifier and build suffixes contain nonempty, dot-separated identifiers made of ASCII - * letters, digits, or hyphens (for example, {@code SNAPSHOT}, {@code M1}, or {@code beta.2}). - * Numeric suffix identifiers may have leading zeros; this is not strict SemVer validation. Both - * suffixes are discarded during normalization, leaving only {@code major.minor.patch}. - */ - private static final Pattern VERSION = - Pattern.compile( - "(0|[1-9][0-9]*)\\.(0|[1-9][0-9]*)\\.(0|[1-9][0-9]*)(?:-[0-9A-Za-z-]+(?:\\.[0-9A-Za-z-]+)*)?(?:\\+[0-9A-Za-z-]+(?:\\.[0-9A-Za-z-]+)*)?"); - - private ApplicationVersion() {} - - public static String normalize(String version) { - if (version == null || !VERSION.matcher(version).matches()) { - throw new IllegalArgumentException("Invalid OpenRemote application version: " + version); - } - return version.split("[-+]", 2)[0]; - } -} diff --git a/manager/src/main/java/org/openremote/manager/system/ApplicationVersionService.java b/manager/src/main/java/org/openremote/manager/system/ApplicationVersionService.java index 5da7d9b939..2e063f2601 100644 --- a/manager/src/main/java/org/openremote/manager/system/ApplicationVersionService.java +++ b/manager/src/main/java/org/openremote/manager/system/ApplicationVersionService.java @@ -22,9 +22,9 @@ import java.util.List; import java.util.Optional; import org.openremote.container.persistence.PersistenceService; -import org.openremote.manager.persistence.ApplicationVersion; import org.openremote.model.Container; import org.openremote.model.ContainerService; +import org.openremote.model.util.VersionUtil; /** Records the first activation of each normalized application version for other services. */ public class ApplicationVersionService implements ContainerService { @@ -55,7 +55,7 @@ public void stop(Container container) {} * timestamp is stored in {@code updated_on} and is never refreshed on reactivation. */ public Optional getFirstActivation(String applicationVersion) { - String version = ApplicationVersion.normalize(applicationVersion); + String version = VersionUtil.normalizeVersion(applicationVersion); return persistenceService.doReturningTransaction( em -> { List timestamps = @@ -76,7 +76,7 @@ public Optional getFirstActivation(String applicationVersion) { * qualifiers and build metadata are ignored. Empty means no qualifying activation was recorded. */ public Optional getFirstActivationAtOrAfter(String applicationVersion) { - String version = ApplicationVersion.normalize(applicationVersion); + String version = VersionUtil.normalizeVersion(applicationVersion); return persistenceService.doReturningTransaction( em -> Optional.ofNullable( @@ -93,7 +93,7 @@ WHERE CAST(string_to_array(version, '.') AS NUMERIC[]) >= } protected void recordApplicationVersion(String applicationVersion) { - String version = ApplicationVersion.normalize(applicationVersion); + String version = VersionUtil.normalizeVersion(applicationVersion); // Atomic insert preserves the first timestamp across concurrent startups. persistenceService.doTransaction( em -> diff --git a/manager/src/test/groovy/org/openremote/manager/persistence/ApplicationVersionTest.groovy b/manager/src/test/groovy/org/openremote/manager/system/VersionInfoTest.groovy similarity index 54% rename from manager/src/test/groovy/org/openremote/manager/persistence/ApplicationVersionTest.groovy rename to manager/src/test/groovy/org/openremote/manager/system/VersionInfoTest.groovy index 7358184c26..576aafee41 100644 --- a/manager/src/test/groovy/org/openremote/manager/persistence/ApplicationVersionTest.groovy +++ b/manager/src/test/groovy/org/openremote/manager/system/VersionInfoTest.groovy @@ -16,37 +16,15 @@ * * SPDX-License-Identifier: AGPL-3.0-or-later */ -package org.openremote.manager.persistence +package org.openremote.manager.system import spock.lang.Specification -class ApplicationVersionTest extends Specification { - def "normalize #version"() { +class VersionInfoTest extends Specification { + def "loadVersion should load manager version from resources"() { expect: - ApplicationVersion.normalize(version) == "1.2.0" - - where: - version << [ - "1.2.0", - "1.2.0-SNAPSHOT", - "1.2.0-M1", - "1.2.0-RC1", - "1.2.0-beta.2", - "1.2.0-alpha", - "1.2.0-LOCAL", - "1.2.0+build.42", - "1.2.0-RC1+build.42" - ] - } - - def "reject unexpected version #version"() { - when: - ApplicationVersion.normalize(version) - - then: - thrown(IllegalArgumentException) - - where: - version << [null, "", "1.2", "v1.2.0", "01.2.0", "1.2.0-", "1.2.0+", "1.2.0/foo", "1.2.0-beta..2", "1.2.0 "] + !VersionInfo.loadVersion().isBlank() + VersionInfo.loadVersion() == VersionInfo.VERSION + VersionInfo.loadVersion() == VersionInfo.getManagerVersion() } } diff --git a/model/src/main/java/org/openremote/model/util/VersionUtil.java b/model/src/main/java/org/openremote/model/util/VersionUtil.java index 0e3d670adf..98f0b0fc8e 100644 --- a/model/src/main/java/org/openremote/model/util/VersionUtil.java +++ b/model/src/main/java/org/openremote/model/util/VersionUtil.java @@ -78,4 +78,20 @@ protected static int parsePart(String part) { return 0; } } + + /** + * Normalizes a three-component version using the same numeric parsing as {@link + * #compareVersions}. Qualifiers and build metadata are ignored, leading zeros are removed, and + * nonnumeric or overflowing components become zero. + */ + public static String normalizeVersion(String version) { + if (version == null) { + throw new IllegalArgumentException("Invalid OpenRemote application version: " + version); + } + String[] parts = version.split("[-+]", 2)[0].split("\\.", -1); + if (parts.length != 3) { + throw new IllegalArgumentException("Expected major.minor.patch version: " + version); + } + return parsePart(parts[0]) + "." + parsePart(parts[1]) + "." + parsePart(parts[2]); + } } diff --git a/model/src/test/groovy/org/openremote/model/util/VersionUtilTest.groovy b/model/src/test/groovy/org/openremote/model/util/VersionUtilTest.groovy new file mode 100644 index 0000000000..132438c76e --- /dev/null +++ b/model/src/test/groovy/org/openremote/model/util/VersionUtilTest.groovy @@ -0,0 +1,190 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.model.util + +import spock.lang.Specification + +class VersionUtilTest extends Specification { + + def "compareVersions('#v1', '#v2') should return #expected"() { + expect: + Integer.signum(VersionUtil.compareVersions(v1, v2)) == Integer.signum(expected) + + where: + v1 | v2 || expected + "1.0.0" | "1.0.0" || 0 + "1.0.0" | "1.0.1" || -1 + "1.0.1" | "1.0.0" || 1 + "1.0" | "1.0.0" || 0 + "1.0.0" | "1.0" || 0 + "2.0.0" | "1.9.9" || 1 + "1.9.9" | "2.0.0" || -1 + "1.2.3" | "1.2.3.4" || -1 + "1.2.3.4" | "1.2.3" || 1 + "10.0.0" | "2.0.0" || 1 + "2.0.0" | "10.0.0" || -1 + "1.0" | "1.0.0.0" || 0 + "01.002.0003" | "1.2.3" || 0 + "1.0.0+build" | "1.0.0" || 0 + "1.0.2147483647" | "1.0.2" || 1 + "1.0.2147483648" | "1.0.0" || 0 + "" | "0.0.0" || 0 + "0.0.0" | "" || 0 + "" | "1.0.0" || -1 + "1.0.0" | "" || 1 + } + + def "compareVersions with qualifiers: '#v1' vs '#v2'"() { + expect: + Integer.signum(VersionUtil.compareVersions(v1, v2)) == Integer.signum(expected) + + where: + v1 | v2 || expected + "1.0.0" | "1.0.0-SNAPSHOT" || 0 // numeric parts are equal + "1.1.0-SNAPSHOT" | "1.0.0" || 1 // 1.1 > 1.0 + "2.0-beta" | "1.9" || 1 // 2 > 1 + "1.0-alpha" | "1.0-beta" || 0 // both parse to 1.0 + } + + def "compareVersions should handle empty segments"() { + expect: + VersionUtil.compareVersions("1..0", "1.0.0") == 0 + VersionUtil.compareVersions("1.0.", "1.0.0") == 0 + VersionUtil.compareVersions("1.0.0", "1..0") == 0 + VersionUtil.compareVersions("1.0.0", "1.0.") == 0 + VersionUtil.compareVersions("1..2", "1.0.1") > 0 + VersionUtil.compareVersions("1.0.1", "1..2") < 0 + } + + def "compareVersions should throw NullPointerException for null inputs"() { + when: + VersionUtil.compareVersions(null, "1.0.0") + + then: + thrown(NullPointerException) + + when: + VersionUtil.compareVersions("1.0.0", null) + + then: + thrown(NullPointerException) + } + + def "isVersionGreater should work correctly"() { + expect: + VersionUtil.isVersionGreater("2.0.0", "1.0.0") + !VersionUtil.isVersionGreater("1.0.0", "2.0.0") + !VersionUtil.isVersionGreater("1.0.0", "1.0.0") + } + + def "isVersionGreaterOrEqual should work correctly"() { + expect: + VersionUtil.isVersionGreaterOrEqual("2.0.0", "1.0.0") + VersionUtil.isVersionGreaterOrEqual("1.0.0", "1.0.0") + !VersionUtil.isVersionGreaterOrEqual("1.0.0", "2.0.0") + } + + def "isVersionEqual should work correctly"() { + expect: + VersionUtil.isVersionEqual("1.0.0", "1.0.0") + VersionUtil.isVersionEqual("1.0", "1.0.0") + !VersionUtil.isVersionEqual("1.0.0", "1.0.1") + !VersionUtil.isVersionEqual("1.0.1", "1.0.0") + } + + def "isVersionLess should work correctly"() { + expect: + VersionUtil.isVersionLess("1.0.0", "2.0.0") + !VersionUtil.isVersionLess("2.0.0", "1.0.0") + !VersionUtil.isVersionLess("1.0.0", "1.0.0") + } + + def "isVersionLessOrEqual should work correctly"() { + expect: + VersionUtil.isVersionLessOrEqual("1.0.0", "2.0.0") + VersionUtil.isVersionLessOrEqual("1.0.0", "1.0.0") + !VersionUtil.isVersionLessOrEqual("2.0.0", "1.0.0") + } + + def "parsePart('#part') returns #expected"() { + expect: + VersionUtil.parsePart(part) == expected + + where: + part | expected + "123" | 123 + "1-SNAPSHOT" | 1 + "42-alpha" | 42 + "3+build.42" | 3 + "0" | 0 + "0012" | 12 + "" | 0 + "beta" | 0 + "-1" | 0 + "2147483647" | Integer.MAX_VALUE + "2147483647-SNAPSHOT" | Integer.MAX_VALUE + "2147483648" | 0 + "999999999999999999999" | 0 + } + + def "normalize #version"() { + expect: + VersionUtil.normalizeVersion(version) == "1.2.0" + + where: + version << [ + "1.2.0", + "1.2.0-SNAPSHOT", + "1.2.0-M1", + "1.2.0-RC1", + "1.2.0-beta.2", + "1.2.0-alpha", + "1.2.0-LOCAL", + "1.2.0+build.42", + "1.2.0-RC1+build.42" + ] + } + + def "normalize numeric components in #version to #normalized"() { + expect: + VersionUtil.normalizeVersion(version) == normalized + + where: + version | normalized + "0.0.0" | "0.0.0" + "12.34.56" | "12.34.56" + "01.02.003" | "1.2.3" + "01.02.003-beta.2" | "1.2.3" + "1..3" | "1.0.3" + "1.2." | "1.2.0" + "1.beta.3" | "1.0.3" + "1.2147483648.3" | "1.0.3" + } + + def "reject unexpected version #version"() { + when: + VersionUtil.normalizeVersion(version) + + then: + thrown(IllegalArgumentException) + + where: + version << [null, "", "1.2", "1.2.3.4"] + } +} diff --git a/test/src/test/groovy/org/openremote/test/system/ApplicationVersionServiceTest.groovy b/test/src/test/groovy/org/openremote/test/system/ApplicationVersionServiceTest.groovy index 437c31f187..525064068e 100644 --- a/test/src/test/groovy/org/openremote/test/system/ApplicationVersionServiceTest.groovy +++ b/test/src/test/groovy/org/openremote/test/system/ApplicationVersionServiceTest.groovy @@ -19,9 +19,9 @@ package org.openremote.test.system import org.openremote.container.persistence.PersistenceService -import org.openremote.manager.persistence.ApplicationVersion import org.openremote.manager.system.ApplicationVersionService import org.openremote.manager.system.VersionInfo +import org.openremote.model.util.VersionUtil import org.openremote.test.ManagerContainerTrait import spock.lang.Specification @@ -48,7 +48,7 @@ class ApplicationVersionServiceTest extends Specification implements ManagerCont } expect: "startup has recorded the running application after migrations" - rows().containsKey(ApplicationVersion.normalize(VersionInfo.managerVersion)) + rows().containsKey(VersionUtil.normalizeVersion(VersionInfo.managerVersion)) when: persistence.doTransaction { diff --git a/test/src/test/groovy/org/openremote/test/system/VersionUtilTest.groovy b/test/src/test/groovy/org/openremote/test/system/VersionUtilTest.groovy deleted file mode 100644 index e4526ef3dd..0000000000 --- a/test/src/test/groovy/org/openremote/test/system/VersionUtilTest.groovy +++ /dev/null @@ -1,133 +0,0 @@ -/* - * Copyright 2026, OpenRemote Inc. - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the GNU Affero General Public License as - * published by the Free Software Foundation, either version 3 of the - * License, or (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Affero General Public License for more details. - * - * You should have received a copy of the GNU Affero General Public License - * along with this program. If not, see . - * - * SPDX-License-Identifier: AGPL-3.0-or-later - */ -package org.openremote.test.system - -import org.openremote.manager.system.VersionInfo -import org.openremote.model.util.VersionUtil -import spock.lang.Specification -import spock.lang.Unroll - -class VersionUtilTest extends Specification { - - @Unroll - def "compareVersions('#v1', '#v2') should return #expected"() { - expect: - Integer.signum(VersionUtil.compareVersions(v1, v2)) == Integer.signum(expected) - - where: - v1 | v2 || expected - "1.0.0" | "1.0.0" || 0 - "1.0.0" | "1.0.1" || -1 - "1.0.1" | "1.0.0" || 1 - "1.0" | "1.0.0" || 0 - "1.0.0" | "1.0" || 0 - "2.0.0" | "1.9.9" || 1 - "1.9.9" | "2.0.0" || -1 - "1.2.3" | "1.2.3.4" || -1 - "1.2.3.4" | "1.2.3" || 1 - "10.0.0" | "2.0.0" || 1 - "2.0.0" | "10.0.0" || -1 - "1.0" | "1.0.0.0" || 0 - } - - @Unroll - def "compareVersions with qualifiers: '#v1' vs '#v2'"() { - expect: - Integer.signum(VersionUtil.compareVersions(v1, v2)) == Integer.signum(expected) - - where: - v1 | v2 || expected - "1.0.0" | "1.0.0-SNAPSHOT"|| 0 // numeric parts are equal - "1.1.0-SNAPSHOT"| "1.0.0" || 1 // 1.1 > 1.0 - "2.0-beta" | "1.9" || 1 // 2 > 1 - "1.0-alpha" | "1.0-beta" || 0 // both parse to 1.0 - } - - def "compareVersions should handle empty segments"() { - expect: - VersionUtil.compareVersions("1..0", "1.0.0") == 0 - VersionUtil.compareVersions("1.0.", "1.0.0") == 0 - } - - def "compareVersions should throw NullPointerException for null inputs"() { - when: - VersionUtil.compareVersions(null, "1.0.0") - - then: - thrown(NullPointerException) - - when: - VersionUtil.compareVersions("1.0.0", null) - - then: - thrown(NullPointerException) - } - - def "isVersionGreater should work correctly"() { - expect: - VersionUtil.isVersionGreater("2.0.0", "1.0.0") == true - VersionUtil.isVersionGreater("1.0.0", "2.0.0") == false - VersionUtil.isVersionGreater("1.0.0", "1.0.0") == false - } - - def "isVersionGreaterOrEqual should work correctly"() { - expect: - VersionUtil.isVersionGreaterOrEqual("2.0.0", "1.0.0") == true - VersionUtil.isVersionGreaterOrEqual("1.0.0", "1.0.0") == true - VersionUtil.isVersionGreaterOrEqual("1.0.0", "2.0.0") == false - } - - def "isVersionEqual should work correctly"() { - expect: - VersionUtil.isVersionEqual("1.0.0", "1.0.0") == true - VersionUtil.isVersionEqual("1.0", "1.0.0") == true - VersionUtil.isVersionEqual("1.0.0", "1.0.1") == false - } - - def "isVersionLess should work correctly"() { - expect: - VersionUtil.isVersionLess("1.0.0", "2.0.0") == true - VersionUtil.isVersionLess("2.0.0", "1.0.0") == false - VersionUtil.isVersionLess("1.0.0", "1.0.0") == false - } - - def "isVersionLessOrEqual should work correctly"() { - expect: - VersionUtil.isVersionLessOrEqual("1.0.0", "2.0.0") == true - VersionUtil.isVersionLessOrEqual("1.0.0", "1.0.0") == true - VersionUtil.isVersionLessOrEqual("2.0.0", "1.0.0") == false - } - - def "parsePart should handle various formats"() { - expect: - VersionUtil.parsePart("123") == 123 - VersionUtil.parsePart("1-SNAPSHOT") == 1 - VersionUtil.parsePart("42-alpha") == 42 - VersionUtil.parsePart("0") == 0 - VersionUtil.parsePart("") == 0 - VersionUtil.parsePart("beta") == 0 - } - - def "loadVersion should load manager version from resources"() { - expect: - !VersionInfo.loadVersion().isBlank() - VersionInfo.loadVersion() == VersionInfo.VERSION - VersionInfo.loadVersion() == VersionInfo.getManagerVersion() - } -} From a00b311553243f995b43fe433d7c0ee31e1445b7 Mon Sep 17 00:00:00 2001 From: Wouter Born Date: Fri, 6 Nov 2026 15:43:41 +0100 Subject: [PATCH 3/3] Preserve empty gateway inventories and safe asset inserts - Serialize empty asset inventories so gateways can complete valid empty syncs - Use inserts for gateway assets to prevent ID collisions from overwriting existing assets - Add regression tests for inventory serialization and independent asset creation --- .../manager/asset/AssetStorageService.java | 2 +- .../openremote/model/asset/AssetsEvent.java | 4 ++ .../model/asset/AssetsEventTest.groovy | 37 +++++++++++++++++++ .../test/assets/AssetIntegrityTest.groovy | 22 +++++++++++ 4 files changed, 64 insertions(+), 1 deletion(-) create mode 100644 model/src/test/groovy/org/openremote/model/asset/AssetsEventTest.groovy diff --git a/manager/src/main/java/org/openremote/manager/asset/AssetStorageService.java b/manager/src/main/java/org/openremote/manager/asset/AssetStorageService.java index 042baa2585..f1f12dd448 100644 --- a/manager/src/main/java/org/openremote/manager/asset/AssetStorageService.java +++ b/manager/src/main/java/org/openremote/manager/asset/AssetStorageService.java @@ -1013,7 +1013,7 @@ public > T merge( existingAsset.setParentId(asset.getParentId()); existingAsset.setAccessPublicRead(asset.isAccessPublicRead()); updatedAsset = em.merge(existingAsset); - } else if (existingAsset == null) { + } else if (existingAsset == null && requestingGatewayAsset != null) { // INSERT must fail on a concurrent ID collision, never merge the new owner's // row. em.persist(asset); diff --git a/model/src/main/java/org/openremote/model/asset/AssetsEvent.java b/model/src/main/java/org/openremote/model/asset/AssetsEvent.java index 62e6d75b65..19570c708a 100644 --- a/model/src/main/java/org/openremote/model/asset/AssetsEvent.java +++ b/model/src/main/java/org/openremote/model/asset/AssetsEvent.java @@ -19,6 +19,7 @@ package org.openremote.model.asset; import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonProperty; import java.util.List; import java.util.stream.Collectors; @@ -27,6 +28,9 @@ /** This event is used in response to a {@link ReadAssetsEvent}. */ public class AssetsEvent extends SharedEvent { + // Empty inventories must survive serialization so gateways can distinguish them from missing + // data. + @JsonInclude(JsonInclude.Include.NON_ABSENT) protected List> assets; @JsonCreator diff --git a/model/src/test/groovy/org/openremote/model/asset/AssetsEventTest.groovy b/model/src/test/groovy/org/openremote/model/asset/AssetsEventTest.groovy new file mode 100644 index 0000000000..25c0407969 --- /dev/null +++ b/model/src/test/groovy/org/openremote/model/asset/AssetsEventTest.groovy @@ -0,0 +1,37 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.model.asset + +import org.openremote.model.event.shared.SharedEvent +import org.openremote.model.util.ValueUtil +import spock.lang.Specification + +class AssetsEventTest extends Specification { + def "empty inventory remains distinct from missing inventory on the wire"() { + when: + def json = ValueUtil.asJSON(new AssetsEvent(inventory)).orElseThrow() + def restored = ValueUtil.JSON.readValue(json, SharedEvent) as AssetsEvent + + then: + restored.assets == inventory + + where: + inventory << [[], null] + } +} diff --git a/test/src/test/groovy/org/openremote/test/assets/AssetIntegrityTest.groovy b/test/src/test/groovy/org/openremote/test/assets/AssetIntegrityTest.groovy index 87e03736d5..a2153f76d0 100644 --- a/test/src/test/groovy/org/openremote/test/assets/AssetIntegrityTest.groovy +++ b/test/src/test/groovy/org/openremote/test/assets/AssetIntegrityTest.groovy @@ -52,6 +52,28 @@ import static org.openremote.model.Constants.* class AssetIntegrityTest extends Specification implements ManagerContainerTrait { + def "creating assets from a reused template returns independent persisted instances"() { + given: + startContainer(defaultConfig(), defaultServices()) + def storage = container.getService(AssetStorageService) + def template = new ThingAsset("First").setRealm(MASTER_REALM) + + when: + def first = storage.merge(template) + def firstId = first.id + template.setId(null).setName("Second") + def second = storage.merge(template) + + then: + !first.is(template) + !second.is(template) + first.id == firstId + first.id != second.id + first.name == "First" + storage.find(first.id, true).name == "First" + storage.find(second.id, true).name == "Second" + } + def "Test asset changes as superuser"() { given: "the server container is started" def container = startContainer(defaultConfig(), defaultServices())