This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 56b1bb638 fix: harden cluster, ACL, client, and polling behavior
(#2362)
56b1bb638 is described below
commit 56b1bb638c189a5aff76f5441e6e5b4c024382cd
Author: yyqdbngt <[email protected]>
AuthorDate: Fri Aug 21 17:10:16 2026 +0800
fix: harden cluster, ACL, client, and polling behavior (#2362)
* fix(cluster): isolate repository topology snapshots
* fix(provider): reject duplicate vendor registrations
* fix(web): serialize visible page polls
* feat(acl): support IPv6 whitelist ranges
* fix(mock): scope client connections by instance
* fix: mask credentials on Unicode boundaries
---
.../cluster/broker/ClusterRepositoryImpl.java | 55 ++++++++-
.../studio/common/util/CredentialUtils.java | 34 +++++-
.../studio/instance/acl/IpRangeMatcher.java | 78 +++++++-----
.../studio/provider/InstanceProviderRegistry.java | 16 ++-
.../cluster/broker/ClusterRepositoryImplTest.java | 25 ++++
.../studio/common/util/CredentialUtilsTest.java | 64 ++++++++++
.../studio/instance/acl/IpRangeMatcherTest.java | 30 ++++-
.../provider/InstanceProviderRegistryTest.java | 35 ++++++
web/src/hooks/useVisiblePolling.test.tsx | 131 +++++++++++++++++++++
web/src/hooks/useVisiblePolling.ts | 13 +-
web/src/mock/clients.ts | 9 ++
web/src/services/connectionsService.test.ts | 34 +++++-
web/src/services/connectionsService.ts | 11 +-
13 files changed, 480 insertions(+), 55 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterRepositoryImpl.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterRepositoryImpl.java
index 58d631106..a07d0c5ed 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterRepositoryImpl.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterRepositoryImpl.java
@@ -70,7 +70,7 @@ public class ClusterRepositoryImpl implements
ClusterRepository {
// observe a partially updated snapshot (new config, old updatedAt).
store.computeIfPresent(clusterId, (id, cluster) -> {
ClusterVO updated = defensiveCopy(cluster);
- updated.setConfig(config);
+ updated.setConfig(copyConfig(config));
updated.setGmtModified(LocalDateTime.now());
log.info("Config updated for cluster: {}", clusterId);
return updated;
@@ -89,11 +89,15 @@ public class ClusterRepositoryImpl implements
ClusterRepository {
.endpoint(cluster.getEndpoint())
.status(cluster.getStatus())
.version(cluster.getVersion())
- // new ArrayList forces a fresh list even when the source is
already immutable
- // (List.copyOf would return the source reference for
immutable inputs).
- .brokers(cluster.getBrokers() == null ? null : new
ArrayList<>(cluster.getBrokers()))
- .proxies(cluster.getProxies() == null ? null : new
ArrayList<>(cluster.getProxies()))
- .nameServers(cluster.getNameServers() == null ? null : new
ArrayList<>(cluster.getNameServers()))
+ .brokers(cluster.getBrokers() == null ? null : new
ArrayList<>(cluster.getBrokers().stream()
+ .map(this::copyBroker)
+ .toList()))
+ .proxies(cluster.getProxies() == null ? null : new
ArrayList<>(cluster.getProxies().stream()
+ .map(this::copyProxy)
+ .toList()))
+ .nameServers(cluster.getNameServers() == null ? null : new
ArrayList<>(cluster.getNameServers().stream()
+ .map(this::copyNameServer)
+ .toList()))
.config(copyConfig(cluster.getConfig()))
.topicCount(cluster.getTopicCount())
.groupCount(cluster.getGroupCount())
@@ -105,6 +109,45 @@ public class ClusterRepositoryImpl implements
ClusterRepository {
return copy;
}
+ private BrokerVO copyBroker(BrokerVO broker) {
+ if (broker == null) {
+ return null;
+ }
+ return BrokerVO.builder()
+ .name(broker.getName())
+ .addr(broker.getAddr())
+ .version(broker.getVersion())
+ .status(broker.getStatus())
+ .diskUsage(broker.getDiskUsage())
+ .tpsIn(broker.getTpsIn())
+ .tpsOut(broker.getTpsOut())
+ .runtimeStatsAvailable(broker.isRuntimeStatsAvailable())
+ .build();
+ }
+
+ private ProxyVO copyProxy(ProxyVO proxy) {
+ if (proxy == null) {
+ return null;
+ }
+ return ProxyVO.builder()
+ .addr(proxy.getAddr())
+ .status(proxy.getStatus())
+ .connections(proxy.getConnections())
+ .grpcPort(proxy.getGrpcPort())
+ .remotingPort(proxy.getRemotingPort())
+ .build();
+ }
+
+ private NameServerVO copyNameServer(NameServerVO nameServer) {
+ if (nameServer == null) {
+ return null;
+ }
+ return NameServerVO.builder()
+ .addr(nameServer.getAddr())
+ .status(nameServer.getStatus())
+ .build();
+ }
+
private ClusterConfigVO copyConfig(ClusterConfigVO config) {
if (config == null) {
return null;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/common/util/CredentialUtils.java
b/server/src/main/java/org/apache/rocketmq/studio/common/util/CredentialUtils.java
index 718a5757b..b8c2ed372 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/common/util/CredentialUtils.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/common/util/CredentialUtils.java
@@ -61,18 +61,42 @@ public final class CredentialUtils {
}
/**
- * Keeps the first and last few characters visible, hides everything else;
short values are
- * fully masked.
+ * Keeps the first and last few Unicode code points visible and hides
everything else. Short
+ * values and malformed UTF-16 input are fully masked. Using code-point
boundaries prevents a
+ * supplementary character from being split into an isolated surrogate in
API responses.
*/
public static String mask(String value) {
if (value == null || value.isEmpty()) {
return value;
}
- if (value.length() < MIN_PARTIALLY_MASKED_CREDENTIAL_CHARS) {
+ if (hasUnpairedSurrogate(value)) {
return CREDENTIAL_MASK;
}
- return value.substring(0, VISIBLE_CREDENTIAL_CHARS)
+ int codePointCount = value.codePointCount(0, value.length());
+ if (codePointCount < MIN_PARTIALLY_MASKED_CREDENTIAL_CHARS) {
+ return CREDENTIAL_MASK;
+ }
+ int visiblePrefixEnd = value.offsetByCodePoints(0,
VISIBLE_CREDENTIAL_CHARS);
+ int visibleSuffixStart = value.offsetByCodePoints(
+ 0, codePointCount - VISIBLE_CREDENTIAL_CHARS);
+ return value.substring(0, visiblePrefixEnd)
+ CREDENTIAL_MASK
- + value.substring(value.length() - VISIBLE_CREDENTIAL_CHARS);
+ + value.substring(visibleSuffixStart);
+ }
+
+ private static boolean hasUnpairedSurrogate(String value) {
+ for (int index = 0; index < value.length(); index++) {
+ char current = value.charAt(index);
+ if (Character.isHighSurrogate(current)) {
+ if (index + 1 >= value.length()
+ || !Character.isLowSurrogate(value.charAt(index + 1)))
{
+ return true;
+ }
+ index++;
+ } else if (Character.isLowSurrogate(current)) {
+ return true;
+ }
+ }
+ return false;
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/IpRangeMatcher.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/IpRangeMatcher.java
index e9d9d74a8..1514eb630 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/IpRangeMatcher.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/IpRangeMatcher.java
@@ -17,6 +17,10 @@
package org.apache.rocketmq.studio.instance.acl;
+import java.net.Inet6Address;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Arrays;
import java.util.regex.Pattern;
/**
@@ -35,13 +39,11 @@ public final class IpRangeMatcher {
private IpRangeMatcher() {
}
- /**
- * Matches a strict dotted-quad IPv4 literal (each octet 0-255). Avoids
{@link InetAddress#getByName}
- * on purpose: that method performs DNS resolution, which would make ACL
whitelist validation
- * network-dependent and non-deterministic.
- */
+ /** Matches a strict dotted-quad IPv4 literal (each octet 0-255). */
private static final Pattern IPV4_LITERAL =
Pattern.compile("^(25[0-5]|2[0-4]\\d|1?\\d?\\d)(\\.(25[0-5]|2[0-4]\\d|1?\\d?\\d)){3}$");
+ private static final Pattern IPV6_LITERAL_CHARACTERS =
+ Pattern.compile("^[0-9a-fA-F:.]+$");
private static boolean isIpv4Literal(String value) {
return value != null && IPV4_LITERAL.matcher(value).matches();
@@ -52,9 +54,9 @@ public final class IpRangeMatcher {
*
* <ul>
* <li>{@code 0.0.0.0}, {@code 0.0.0.0/0} or {@code ::/0} match any
non-blank ip.</li>
- * <li>An entry without a {@code /} is matched by exact equality.</li>
- * <li>An entry of the form {@code x.x.x.x/n} is matched against the
IPv4 subnet.</li>
- * <li>Any unparseable input (or IPv6 CIDR other than {@code ::/0})
returns {@code false}.</li>
+ * <li>An entry without a {@code /} matches the equivalent IPv4 or IPv6
literal.</li>
+ * <li>IPv4 and IPv6 CIDR entries are matched using their respective
prefix widths.</li>
+ * <li>Any unparseable input or address-family mismatch returns {@code
false}.</li>
* </ul>
*
* @param ip the address being checked (IPv4 or IPv6)
@@ -65,6 +67,7 @@ public final class IpRangeMatcher {
if (ip == null || ip.isBlank() || cidrOrIp == null ||
cidrOrIp.isBlank()) {
return false;
}
+ String target = ip.trim();
String entry = cidrOrIp.trim();
if (WILDCARD_V4.equals(entry) || WILDCARD_V4_CIDR.equals(entry) ||
WILDCARD_V6_CIDR.equals(entry)) {
@@ -73,7 +76,9 @@ public final class IpRangeMatcher {
int slash = entry.indexOf('/');
if (slash < 0) {
- return ip.trim().equals(entry);
+ byte[] targetBytes = parseAddressLiteral(target);
+ byte[] entryBytes = parseAddressLiteral(entry);
+ return targetBytes != null && Arrays.equals(targetBytes,
entryBytes);
}
String baseIp = entry.substring(0, slash);
@@ -84,15 +89,12 @@ public final class IpRangeMatcher {
} catch (NumberFormatException e) {
return false;
}
- if (prefix < 0 || prefix > 32) {
- return false;
- }
- if (!isIpv4Literal(ip.trim()) || !isIpv4Literal(baseIp)) {
+ byte[] targetBytes = parseAddressLiteral(target);
+ byte[] baseBytes = parseAddressLiteral(baseIp);
+ if (targetBytes == null || baseBytes == null || targetBytes.length !=
baseBytes.length
+ || prefix < 0 || prefix > baseBytes.length * Byte.SIZE) {
return false;
}
-
- byte[] targetBytes = ipToBytes(ip.trim());
- byte[] baseBytes = ipToBytes(baseIp);
int bits = prefix;
for (int i = 0; i < targetBytes.length && bits > 0; i++) {
int mask = (bits >= 8) ? 0xFF : (0xFF << (8 - bits)) & 0xFF;
@@ -105,23 +107,37 @@ public final class IpRangeMatcher {
}
/**
- * Converts a validated dotted-quad IPv4 literal to its 4-byte
representation.
- * Callers must ensure the input passes {@link #isIpv4Literal(String)}
first.
+ * Parses an IPv4 or IPv6 literal without resolving hostnames. IPv6 input
is restricted to
+ * address-literal characters before using {@link
InetAddress#getByName(String)}, so this path
+ * cannot issue a DNS query. Scoped addresses are intentionally rejected
because interface
+ * names are host-specific and cannot form portable ACL entries.
*/
- private static byte[] ipToBytes(String ip) {
- String[] parts = ip.split("\\.");
- byte[] bytes = new byte[4];
- for (int i = 0; i < 4; i++) {
- bytes[i] = (byte) Integer.parseInt(parts[i]);
+ private static byte[] parseAddressLiteral(String value) {
+ if (isIpv4Literal(value)) {
+ String[] parts = value.split("\\.");
+ byte[] bytes = new byte[4];
+ for (int i = 0; i < bytes.length; i++) {
+ bytes[i] = (byte) Integer.parseInt(parts[i]);
+ }
+ return bytes;
+ }
+ if (value == null || !value.contains(":")
+ || !IPV6_LITERAL_CHARACTERS.matcher(value).matches()) {
+ return null;
+ }
+ try {
+ InetAddress address = InetAddress.getByName(value);
+ return address instanceof Inet6Address ? address.getAddress() :
null;
+ } catch (UnknownHostException exception) {
+ return null;
}
- return bytes;
}
/**
* Returns {@code true} when {@code cidrOrIp} is a well-formed whitelist
entry: a wildcard
- * ({@code 0.0.0.0}, {@code 0.0.0.0/0}, {@code ::/0}), a bare IPv4
address, or an IPv4 CIDR with a
- * prefix between 0 and 32. Used to validate ACL 2.0 {@code whiteSet}
entries before they are
- * applied.
+ * ({@code 0.0.0.0}, {@code 0.0.0.0/0}, {@code ::/0}), a bare IPv4/IPv6
address, or a CIDR with a
+ * prefix valid for that address family. Used to validate ACL 2.0 {@code
whiteSet} entries before
+ * they are applied.
*/
public static boolean isValidRange(String cidrOrIp) {
if (cidrOrIp == null || cidrOrIp.isBlank()) {
@@ -133,7 +149,7 @@ public final class IpRangeMatcher {
}
int slash = entry.indexOf('/');
if (slash < 0) {
- return isIpv4Literal(entry);
+ return parseAddressLiteral(entry) != null;
}
String baseIp = entry.substring(0, slash);
String prefixStr = entry.substring(slash + 1);
@@ -143,9 +159,7 @@ public final class IpRangeMatcher {
} catch (NumberFormatException e) {
return false;
}
- if (prefix < 0 || prefix > 32) {
- return false;
- }
- return isIpv4Literal(baseIp);
+ byte[] baseBytes = parseAddressLiteral(baseIp);
+ return baseBytes != null && prefix >= 0 && prefix <= baseBytes.length
* Byte.SIZE;
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProviderRegistry.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProviderRegistry.java
index ebfce9eee..ceb77a1bb 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProviderRegistry.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProviderRegistry.java
@@ -37,11 +37,23 @@ public class InstanceProviderRegistry {
public InstanceProviderRegistry(List<InstanceProvider> providerList,
List<CloudCatalogProvider> catalogList,
InstanceRepository instanceRepository) {
- providerList.forEach(provider -> providers.put(provider.vendor(),
provider));
- catalogList.forEach(catalog -> catalogs.put(catalog.vendor(),
catalog));
+ providerList.forEach(provider -> registerProvider(provider.vendor(),
provider));
+ catalogList.forEach(catalog -> registerCatalog(catalog.vendor(),
catalog));
this.instanceRepository = instanceRepository;
}
+ private void registerProvider(InstanceVendor vendor, InstanceProvider
provider) {
+ if (providers.putIfAbsent(vendor, provider) != null) {
+ throw new IllegalStateException("Duplicate instance provider
registered for vendor " + vendor);
+ }
+ }
+
+ private void registerCatalog(InstanceVendor vendor, CloudCatalogProvider
catalog) {
+ if (catalogs.putIfAbsent(vendor, catalog) != null) {
+ throw new IllegalStateException("Duplicate cloud catalog provider
registered for vendor " + vendor);
+ }
+ }
+
public InstanceProvider forVendor(InstanceVendor vendor) {
InstanceProvider provider = providers.get(vendor);
if (provider == null) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterRepositoryImplTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterRepositoryImplTest.java
index a0cb555bc..721b9a1c5 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterRepositoryImplTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterRepositoryImplTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.studio.cluster.broker;
+import org.apache.rocketmq.studio.cluster.config.ClusterConfigVO;
import org.junit.jupiter.api.Test;
import java.util.List;
@@ -48,13 +49,37 @@ class ClusterRepositoryImplTest {
ClusterVO first = repository.findById("cluster-001").orElseThrow();
first.setName("mutated");
first.getConfig().setFileReservedTime(1);
+ first.getBrokers().get(0).setName("mutated-broker");
+ first.getProxies().get(0).setAddr("mutated-proxy");
+ first.getNameServers().get(0).setAddr("mutated-nameserver");
+ first.getTpsHistory().set(0, 999);
ClusterVO second = repository.findById("cluster-001").orElseThrow();
// Mutating the returned copy must not affect the cached cluster.
assertThat(second.getName()).isEqualTo("rmq-cluster-prod");
assertThat(first.getBrokers()).isNotSameAs(second.getBrokers());
+
assertThat(first.getBrokers().get(0)).isNotSameAs(second.getBrokers().get(0));
+ assertThat(second.getBrokers().get(0).getName()).isEqualTo("broker-a");
+
assertThat(second.getProxies().get(0).getAddr()).isEqualTo("10.0.0.10:8081");
+
assertThat(second.getNameServers().get(0).getAddr()).isEqualTo("10.0.0.20:9876");
+ assertThat(first.getTpsHistory()).isNotSameAs(second.getTpsHistory());
+ assertThat(second.getTpsHistory().get(0)).isEqualTo(1200);
assertThat(second.getConfig()).isNotSameAs(first.getConfig());
assertThat(second.getConfig().getFileReservedTime()).isEqualTo(72);
}
+
+ @Test
+ void updateConfigShouldNotRetainCallerOwnedObject() {
+ ClusterRepositoryImpl repository = new ClusterRepositoryImpl(true);
+ ClusterConfigVO config = ClusterConfigVO.builder()
+ .fileReservedTime(24)
+ .build();
+
+ repository.updateConfig("cluster-001", config);
+ config.setFileReservedTime(1);
+
+
assertThat(repository.findById("cluster-001").orElseThrow().getConfig().getFileReservedTime())
+ .isEqualTo(24);
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/common/util/CredentialUtilsTest.java
b/server/src/test/java/org/apache/rocketmq/studio/common/util/CredentialUtilsTest.java
index 698b6c56a..8fc57c9f8 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/common/util/CredentialUtilsTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/common/util/CredentialUtilsTest.java
@@ -30,4 +30,68 @@ class CredentialUtilsTest {
assertThat(CredentialUtils.decodeBase64(legacyValue)).isEqualTo(legacyValue);
}
+
+ @Test
+ void maskShouldPreserveAsciiBehavior() {
+ assertThat(CredentialUtils.mask("abcdefghijklmnopq"))
+ .isEqualTo("abcd****nopq");
+ assertThat(CredentialUtils.mask("abcdefghijklmnop"))
+ .isEqualTo("****");
+ }
+
+ @Test
+ void maskShouldUseSupplementaryCodePointBoundaries() {
+ String value = "abc🚀" + "123456789" + "🛰xyz";
+
+ String masked = CredentialUtils.mask(value);
+
+ assertThat(masked).isEqualTo("abc🚀****🛰xyz");
+ assertThat(hasIsolatedSurrogate(masked)).isFalse();
+ }
+
+ @Test
+ void maskShouldKeepFourCodePointsAtEachEnd() {
+ String value = "🚀abc" + "123456789" + "xyz🛰";
+
+ assertThat(CredentialUtils.mask(value)).isEqualTo("🚀abc****xyz🛰");
+ }
+
+ @Test
+ void maskShouldMeasureShortValuesInCodePoints() {
+ assertThat(CredentialUtils.mask("🚀".repeat(16))).isEqualTo("****");
+ assertThat(CredentialUtils.mask("🚀".repeat(17)))
+ .isEqualTo("🚀".repeat(4) + "****" + "🚀".repeat(4));
+ }
+
+ @Test
+ void maskShouldPreserveNullAndEmptyValues() {
+ assertThat(CredentialUtils.mask(null)).isNull();
+ assertThat(CredentialUtils.mask("")).isEmpty();
+ }
+
+ @Test
+ void maskShouldFullyHideMalformedUtf16WithoutLeakingSurrogates() {
+ String malformed = "abcd" + '\uD83D' + "1234567890123456";
+
+ String masked = CredentialUtils.mask(malformed);
+
+ assertThat(masked).isEqualTo("****");
+ assertThat(hasIsolatedSurrogate(masked)).isFalse();
+ }
+
+ private static boolean hasIsolatedSurrogate(String value) {
+ for (int index = 0; index < value.length(); index++) {
+ char current = value.charAt(index);
+ if (Character.isHighSurrogate(current)) {
+ if (index + 1 >= value.length()
+ || !Character.isLowSurrogate(value.charAt(index + 1)))
{
+ return true;
+ }
+ index++;
+ } else if (Character.isLowSurrogate(current)) {
+ return true;
+ }
+ }
+ return false;
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/IpRangeMatcherTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/IpRangeMatcherTest.java
index 7c34f9a0b..32123b2e7 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/IpRangeMatcherTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/IpRangeMatcherTest.java
@@ -76,8 +76,19 @@ class IpRangeMatcherTest {
}
@Test
- void ipv6CidrOtherThanWildcardIsNotMatched() {
- assertThat(IpRangeMatcher.isInRange("2001:db8::1",
"2001:db8::/32")).isFalse();
+ void ipv6CidrMatchesEquivalentAddressesInSubnet() {
+ assertThat(IpRangeMatcher.isInRange("2001:db8::1",
"2001:db8::/32")).isTrue();
+ assertThat(IpRangeMatcher.isInRange("2001:db8:10::abcd",
"2001:db8:10::/64")).isTrue();
+ assertThat(IpRangeMatcher.isInRange("2001:db9::1",
"2001:db8::/32")).isFalse();
+ assertThat(IpRangeMatcher.isInRange("2001:db8:11::1",
"2001:db8:10::/64")).isFalse();
+ }
+
+ @Test
+ void ipv6ExactAndSingleHostCidrUseCanonicalAddressBytes() {
+ assertThat(IpRangeMatcher.isInRange(
+ "2001:0db8:0000:0000:0000:0000:0000:0001",
"2001:db8::1")).isTrue();
+ assertThat(IpRangeMatcher.isInRange("2001:db8::1",
"2001:db8::1/128")).isTrue();
+ assertThat(IpRangeMatcher.isInRange("2001:db8::2",
"2001:db8::1/128")).isFalse();
}
@Test
@@ -86,6 +97,15 @@ class IpRangeMatcherTest {
assertThat(IpRangeMatcher.isInRange("192.168.1.10",
"2001:db8::/64")).isFalse();
}
+ @Test
+ void malformedIpv6AndInvalidPrefixesReturnFalse() {
+ assertThat(IpRangeMatcher.isInRange("2001:db8::1",
"2001:db8::/129")).isFalse();
+ assertThat(IpRangeMatcher.isInRange("2001:db8::1",
"2001:db8::/-1")).isFalse();
+ assertThat(IpRangeMatcher.isInRange("2001:db8::1",
"2001:db8::/abc")).isFalse();
+ assertThat(IpRangeMatcher.isInRange("2001:db8::1",
"2001:db8::1::2/64")).isFalse();
+ assertThat(IpRangeMatcher.isInRange("fe80::1%eth0",
"fe80::/10")).isFalse();
+ }
+
@Test
void leadingAndTrailingWhitespaceIsTrimmed() {
assertThat(IpRangeMatcher.isInRange(" 192.168.1.42 ", " 192.168.1.0/24
")).isTrue();
@@ -104,6 +124,9 @@ class IpRangeMatcherTest {
assertThat(IpRangeMatcher.isValidRange("192.168.1.10")).isTrue();
assertThat(IpRangeMatcher.isValidRange("192.168.1.0/24")).isTrue();
assertThat(IpRangeMatcher.isValidRange("10.0.0.1/32")).isTrue();
+ assertThat(IpRangeMatcher.isValidRange("2001:db8::1")).isTrue();
+ assertThat(IpRangeMatcher.isValidRange("2001:db8::/32")).isTrue();
+ assertThat(IpRangeMatcher.isValidRange("2001:db8::1/128")).isTrue();
}
@Test
@@ -115,5 +138,8 @@ class IpRangeMatcherTest {
assertThat(IpRangeMatcher.isValidRange("192.168.1.0/")).isFalse();
assertThat(IpRangeMatcher.isValidRange("10.0.0.1/abc")).isFalse();
assertThat(IpRangeMatcher.isValidRange("256.1.1.1")).isFalse();
+ assertThat(IpRangeMatcher.isValidRange("2001:db8::/129")).isFalse();
+ assertThat(IpRangeMatcher.isValidRange("2001:db8::1::2")).isFalse();
+ assertThat(IpRangeMatcher.isValidRange("fe80::1%eth0")).isFalse();
}
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/InstanceProviderRegistryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/InstanceProviderRegistryTest.java
index 84433c9bd..20f4b377e 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/InstanceProviderRegistryTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/InstanceProviderRegistryTest.java
@@ -101,9 +101,44 @@ class InstanceProviderRegistryTest {
.satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(501));
}
+ @Test
+ void constructorShouldRejectDuplicateProvidersForVendorTest() {
+ InstanceProvider duplicate = stubProvider(InstanceVendor.APACHE);
+
+ assertThatThrownBy(() -> new InstanceProviderRegistry(
+ List.of(apacheProvider, duplicate), List.of(),
instanceRepository))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("Duplicate instance provider registered for vendor
APACHE");
+ }
+
+ @Test
+ void constructorShouldRejectDuplicateCatalogsForVendorTest() {
+ CloudCatalogProvider first = stubCatalog(InstanceVendor.ALIYUN);
+ CloudCatalogProvider duplicate = stubCatalog(InstanceVendor.ALIYUN);
+
+ assertThatThrownBy(() -> new InstanceProviderRegistry(
+ List.of(apacheProvider), List.of(first, duplicate),
instanceRepository))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("Duplicate cloud catalog provider registered for
vendor ALIYUN");
+ }
+
+ @Test
+ void catalogForShouldReturnRegisteredCatalogTest() {
+ CloudCatalogProvider catalog = stubCatalog(InstanceVendor.ALIYUN);
+ registry = new InstanceProviderRegistry(List.of(apacheProvider),
List.of(catalog), instanceRepository);
+
+
assertThat(registry.catalogFor(InstanceVendor.ALIYUN)).isSameAs(catalog);
+ }
+
private InstanceProvider stubProvider(InstanceVendor vendor) {
InstanceProvider provider = mock(InstanceProvider.class);
when(provider.vendor()).thenReturn(vendor);
return provider;
}
+
+ private CloudCatalogProvider stubCatalog(InstanceVendor vendor) {
+ CloudCatalogProvider catalog = mock(CloudCatalogProvider.class);
+ when(catalog.vendor()).thenReturn(vendor);
+ return catalog;
+ }
}
diff --git a/web/src/hooks/useVisiblePolling.test.tsx
b/web/src/hooks/useVisiblePolling.test.tsx
new file mode 100644
index 000000000..6682fe69b
--- /dev/null
+++ b/web/src/hooks/useVisiblePolling.test.tsx
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { act, renderHook } from '@testing-library/react';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { useVisiblePolling } from './useVisiblePolling';
+
+function deferred() {
+ let resolve!: () => void;
+ let reject!: (reason?: unknown) => void;
+ const promise = new Promise<void>((resolvePromise, rejectPromise) => {
+ resolve = resolvePromise;
+ reject = rejectPromise;
+ });
+ return { promise, resolve, reject };
+}
+
+describe('useVisiblePolling', () => {
+ let visibilityState = 'visible';
+
+ beforeEach(() => {
+ vi.useFakeTimers();
+ vi.spyOn(document, 'visibilityState', 'get').mockImplementation(
+ () => visibilityState as DocumentVisibilityState,
+ );
+ });
+
+ afterEach(() => {
+ vi.useRealTimers();
+ vi.restoreAllMocks();
+ });
+
+ it('does not overlap interval and visibility-triggered polls', async () => {
+ const first = deferred();
+ const poll =
vi.fn().mockReturnValueOnce(first.promise).mockResolvedValue(undefined);
+
+ renderHook(() => useVisiblePolling(true, 1_000, poll));
+
+ await act(async () => {
+ vi.advanceTimersByTime(1_000);
+ await Promise.resolve();
+ });
+ expect(poll).toHaveBeenCalledTimes(1);
+
+ await act(async () => {
+ vi.advanceTimersByTime(2_000);
+ document.dispatchEvent(new Event('visibilitychange'));
+ await Promise.resolve();
+ });
+ expect(poll).toHaveBeenCalledTimes(1);
+
+ await act(async () => {
+ first.resolve();
+ await first.promise;
+ await Promise.resolve();
+ });
+
+ await act(async () => {
+ vi.advanceTimersByTime(1_000);
+ await Promise.resolve();
+ });
+ expect(poll).toHaveBeenCalledTimes(2);
+ });
+
+ it('releases the in-flight guard after a rejected poll', async () => {
+ const first = deferred();
+ const poll =
vi.fn().mockReturnValueOnce(first.promise).mockResolvedValue(undefined);
+
+ renderHook(() => useVisiblePolling(true, 1_000, poll));
+ await act(async () => {
+ vi.advanceTimersByTime(1_000);
+ await Promise.resolve();
+ first.reject(new Error('temporary failure'));
+ await Promise.resolve();
+ await Promise.resolve();
+ });
+
+ await act(async () => {
+ vi.advanceTimersByTime(1_000);
+ await Promise.resolve();
+ });
+ expect(poll).toHaveBeenCalledTimes(2);
+ });
+
+ it('stops scheduling polls after unmount', async () => {
+ const poll = vi.fn().mockResolvedValue(undefined);
+ const { unmount } = renderHook(() => useVisiblePolling(true, 1_000, poll));
+
+ unmount();
+ await act(async () => {
+ vi.advanceTimersByTime(2_000);
+ document.dispatchEvent(new Event('visibilitychange'));
+ await Promise.resolve();
+ });
+
+ expect(poll).not.toHaveBeenCalled();
+ });
+
+ it('waits for a hidden page to become visible', async () => {
+ visibilityState = 'hidden';
+ const poll = vi.fn().mockResolvedValue(undefined);
+
+ renderHook(() => useVisiblePolling(true, 1_000, poll));
+ await act(async () => {
+ vi.advanceTimersByTime(2_000);
+ await Promise.resolve();
+ });
+ expect(poll).not.toHaveBeenCalled();
+
+ visibilityState = 'visible';
+ await act(async () => {
+ document.dispatchEvent(new Event('visibilitychange'));
+ await Promise.resolve();
+ });
+ expect(poll).toHaveBeenCalledTimes(1);
+ });
+});
diff --git a/web/src/hooks/useVisiblePolling.ts
b/web/src/hooks/useVisiblePolling.ts
index 0d586e052..bcdc2cd2f 100644
--- a/web/src/hooks/useVisiblePolling.ts
+++ b/web/src/hooks/useVisiblePolling.ts
@@ -25,10 +25,17 @@ export function useVisiblePolling(
useEffect(() => {
if (!enabled) return;
+ let pollInFlight = false;
const pollWhenVisible = () => {
- if (document.visibilityState === 'visible') {
- void poll();
- }
+ if (document.visibilityState !== 'visible' || pollInFlight) return;
+
+ pollInFlight = true;
+ void Promise.resolve()
+ .then(poll)
+ .catch(() => undefined)
+ .finally(() => {
+ pollInFlight = false;
+ });
};
const intervalId = window.setInterval(pollWhenVisible, intervalMs);
document.addEventListener('visibilitychange', pollWhenVisible);
diff --git a/web/src/mock/clients.ts b/web/src/mock/clients.ts
index 6ece01ac8..029f79d20 100644
--- a/web/src/mock/clients.ts
+++ b/web/src/mock/clients.ts
@@ -31,6 +31,15 @@ export interface ClientConnection {
/* ─── Mock Data ─── */
+/**
+ * Associates mock client inventories with the NameServer endpoints that own
them. The real
+ * endpoint is NameServer-scoped, so mock mode must not expose every cluster
for every selection.
+ */
+export const mockClientClusterByNamesrvAddr: Readonly<Record<string, string>>
= {
+ '10.101.2.1:9876': 'ns-prod',
+ '10.102.5.1:9876': 'ns-pre',
+};
+
export const mockClients: ClientConnection[] = [
{
clientId: '[email protected]:49152',
diff --git a/web/src/services/connectionsService.test.ts
b/web/src/services/connectionsService.test.ts
index 393c767e3..625eb02d7 100644
--- a/web/src/services/connectionsService.test.ts
+++ b/web/src/services/connectionsService.test.ts
@@ -25,9 +25,31 @@ vi.mock('../config', () => ({
import { listConnections } from './connectionsService';
describe('connectionsService mock connections', () => {
- it('returns defensive copies after applying filters', async () => {
+ it('isolates client inventories by the required NameServer address', async
() => {
+ const production = await listConnections({ namesrvAddr: '10.101.2.1:9876'
});
+ const preproduction = await listConnections({ namesrvAddr:
'10.102.5.1:9876' });
+
+ expect(production).not.toHaveLength(0);
+ expect(preproduction).not.toHaveLength(0);
+ expect(production.every((connection) => connection.clusterName ===
'ns-prod')).toBe(true);
+ expect(preproduction.every((connection) => connection.clusterName ===
'ns-pre')).toBe(true);
+ expect(new Set(production.map((connection) =>
connection.clientId))).not.toEqual(
+ new Set(preproduction.map((connection) => connection.clientId)),
+ );
+ });
+
+ it('returns an empty inventory for an unknown NameServer address', async ()
=> {
+ await expect(listConnections({ namesrvAddr: '10.103.9.1:9876'
})).resolves.toEqual([]);
+ });
+
+ it('requires the same NameServer parameter as the real endpoint', async ()
=> {
+ await expect(listConnections()).rejects.toThrow('namesrvAddr is required');
+ await expect(listConnections({ namesrvAddr: ' '
})).rejects.toThrow('namesrvAddr is required');
+ });
+
+ it('returns defensive copies after applying NameServer, cluster, and type
filters', async () => {
const connections = await listConnections({
- namesrvAddr: 'namesrv-1:9876',
+ namesrvAddr: '10.101.2.1:9876',
clusterId: 'ns-prod',
type: 'Consumer',
});
@@ -38,7 +60,7 @@ describe('connectionsService mock connections', () => {
connections[0].address = '127.0.0.1:8081';
const fresh = await listConnections({
- namesrvAddr: 'namesrv-1:9876',
+ namesrvAddr: '10.101.2.1:9876',
clusterId: 'ns-prod',
type: 'Consumer',
});
@@ -49,4 +71,10 @@ describe('connectionsService mock connections', () => {
expect(fresh.every((connection) => connection.clusterName ===
'ns-prod')).toBe(true);
expect(fresh.every((connection) => connection.type ===
'Consumer')).toBe(true);
});
+
+ it('does not leak another NameServer when a conflicting cluster filter is
supplied', async () => {
+ await expect(
+ listConnections({ namesrvAddr: '10.101.2.1:9876', clusterId: 'ns-pre' }),
+ ).resolves.toEqual([]);
+ });
});
diff --git a/web/src/services/connectionsService.ts
b/web/src/services/connectionsService.ts
index d4adc7284..44eba5a96 100644
--- a/web/src/services/connectionsService.ts
+++ b/web/src/services/connectionsService.ts
@@ -1,7 +1,7 @@
import { isMockMode } from './dataMode';
import * as connApi from '../api/connections';
import type { ClientConnection, ClientConnectionQuery } from
'../api/connections';
-import { mockClients } from '../mock/clients';
+import { mockClientClusterByNamesrvAddr, mockClients } from '../mock/clients';
function copyConnection(connection: ClientConnection): ClientConnection {
return { ...connection };
@@ -9,7 +9,14 @@ function copyConnection(connection: ClientConnection):
ClientConnection {
export async function listConnections(params?: ClientConnectionQuery):
Promise<ClientConnection[]> {
if (isMockMode()) {
- let result = [...mockClients];
+ const namesrvAddr = params?.namesrvAddr?.trim();
+ if (!namesrvAddr) {
+ throw new Error('namesrvAddr is required');
+ }
+ const instanceCluster = mockClientClusterByNamesrvAddr[namesrvAddr];
+ if (!instanceCluster) return [];
+
+ let result = mockClients.filter((connection) => connection.clusterName ===
instanceCluster);
if (params?.clusterId)
result = result.filter((connection) => connection.clusterName ===
params.clusterId);
if (params?.type) result = result.filter((c) => c.type === params.type);
