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 cc97b1c83 fix(acl): read remote rules only from the configured cluster
(#6502)
cc97b1c83 is described below
commit cc97b1c834f6c75526211beb4efdd82de91ad724
Author: Apulupie <[email protected]>
AuthorDate: Thu Oct 15 18:13:04 2026 +0800
fix(acl): read remote rules only from the configured cluster (#6502)
#6196 narrowed discoverClusters, refreshClusterDetail and discoverBrokers
to the cluster
configured for the instance, but ApacheAclReadService.listRules still
walked every entry of
the shared brokerAddrTable. On an instance bound to cluster A whose
NameServer also serves
cluster B, the remote-rule read returned cluster B's brokers in
policiesByBroker /
failuresByBroker, so the selected instance surfaced another cluster's ACL.
The broker set is now resolved once through the same configuredClusterName
lookup the
cluster provider uses, and a configured cluster whose membership is missing
from the
topology fails closed with 503 instead of falling back to the whole table -
the same
condition and status discoverBrokers raises. An instance that names no
cluster keeps
reading every broker of the endpoint.
---
.../studio/instance/acl/ApacheAclReadService.java | 32 ++++++++++-
.../instance/acl/ApacheAclReadServiceTest.java | 67 +++++++++++++++++++++-
2 files changed, 96 insertions(+), 3 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/ApacheAclReadService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/ApacheAclReadService.java
index 78735a875..a4b729333 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/ApacheAclReadService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/ApacheAclReadService.java
@@ -22,12 +22,16 @@ import org.apache.rocketmq.remoting.protocol.body.AclInfo;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
+import java.util.TreeSet;
@Service
@RequiredArgsConstructor
@@ -35,6 +39,8 @@ public class ApacheAclReadService {
private final RuntimeAdminClientResolver runtimeAdminClientResolver;
public RemoteAclReadResult listRules(String instanceId, String subject,
String resource) {
+ String configuredCluster = StringUtils.hasText(instanceId)
+ ? runtimeAdminClientResolver.configuredClusterName(instanceId)
: null;
return runtimeAdminClientResolver.execute(instanceId, admin -> {
Map<String, List<RemoteAclPolicyVO>> policies = new
LinkedHashMap<>();
Map<String, String> failures = new LinkedHashMap<>();
@@ -42,8 +48,8 @@ public class ApacheAclReadService {
if (clusterInfo == null || clusterInfo.getBrokerAddrTable() ==
null) {
return result(policies, failures);
}
- for (BrokerData broker :
clusterInfo.getBrokerAddrTable().values()) {
- String address = masterAddress(broker);
+ for (String brokerName : scopedBrokerNames(clusterInfo,
configuredCluster)) {
+ String address =
masterAddress(clusterInfo.getBrokerAddrTable().get(brokerName));
if (address == null || policies.containsKey(address) ||
failures.containsKey(address)) {
continue;
}
@@ -57,6 +63,28 @@ public class ApacheAclReadService {
});
}
+ /**
+ * The broker set this read is allowed to touch. An instance bound to one
cluster on a NameServer
+ * shared by several must not surface another cluster's ACL, so the
topology is narrowed the same
+ * way {@code RocketMQClusterProvider.discoverBrokers} narrows it; an
instance that names no
+ * cluster keeps every broker of the endpoint. A configured cluster whose
membership is missing
+ * from the topology cannot be resolved, so the read fails closed instead
of reading the whole
+ * cluster table (mirrors the 503 that broker discovery raises for the
same condition).
+ */
+ private Set<String> scopedBrokerNames(ClusterInfo clusterInfo, String
configuredCluster) {
+ Set<String> names = new
TreeSet<>(clusterInfo.getBrokerAddrTable().keySet());
+ if (configuredCluster == null) {
+ return names;
+ }
+ Map<String, Set<String>> clusters = clusterInfo.getClusterAddrTable();
+ if (clusters == null || clusters.containsKey(configuredCluster)
+ && clusters.get(configuredCluster) == null) {
+ throw new BusinessException(503, "Broker cluster membership is
unavailable");
+ }
+ names.retainAll(clusters.getOrDefault(configuredCluster, Set.of()));
+ return names;
+ }
+
private List<RemoteAclPolicyVO> mapPolicies(List<AclInfo> brokerPolicies) {
if (brokerPolicies == null) {
return List.of();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/ApacheAclReadServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/ApacheAclReadServiceTest.java
index 51c79466b..ff7464ad4 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/ApacheAclReadServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/ApacheAclReadServiceTest.java
@@ -28,11 +28,15 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class ApacheAclReadServiceTest {
@@ -109,6 +113,62 @@ class ApacheAclReadServiceTest {
});
}
+ @Test
+ void readsRemoteRulesOnlyFromTheConfiguredClusterTest() throws Exception {
+ RuntimeAdminClientResolver resolver =
mock(RuntimeAdminClientResolver.class);
+ MQAdminExt admin = mock(MQAdminExt.class);
+ when(admin.examineBrokerClusterInfo()).thenReturn(sharedTopology());
+ when(admin.listAcl("10.0.0.11:10911", null,
null)).thenReturn(List.of());
+
when(resolver.configuredClusterName("selected")).thenReturn("DefaultCluster");
+ executeWith(resolver, admin, "selected");
+
+ RemoteAclReadResult result = new
ApacheAclReadService(resolver).listRules("selected", null, null);
+
+
assertThat(result.getPoliciesByBroker()).containsOnlyKeys("10.0.0.11:10911");
+ verify(admin, never()).listAcl(eq("10.0.0.12:10911"), any(), any());
+ }
+
+ @Test
+ void readsRemoteRulesFromEveryBrokerWhenTheInstanceNamesNoClusterTest()
throws Exception {
+ RuntimeAdminClientResolver resolver =
mock(RuntimeAdminClientResolver.class);
+ MQAdminExt admin = mock(MQAdminExt.class);
+ when(admin.examineBrokerClusterInfo()).thenReturn(sharedTopology());
+ when(admin.listAcl(any(), any(), any())).thenReturn(List.of());
+ when(resolver.configuredClusterName("selected")).thenReturn(null);
+ executeWith(resolver, admin, "selected");
+
+ RemoteAclReadResult result = new
ApacheAclReadService(resolver).listRules("selected", null, null);
+
+
assertThat(result.getPoliciesByBroker()).containsOnlyKeys("10.0.0.11:10911",
"10.0.0.12:10911");
+ }
+
+ @Test
+ void reportsUnavailableClusterMembershipInsteadOfEveryBrokerTest() throws
Exception {
+ RuntimeAdminClientResolver resolver =
mock(RuntimeAdminClientResolver.class);
+ MQAdminExt admin = mock(MQAdminExt.class);
+ ClusterInfo withoutClusterTable = sharedTopology();
+ withoutClusterTable.setClusterAddrTable(null);
+ when(admin.examineBrokerClusterInfo()).thenReturn(withoutClusterTable);
+
when(resolver.configuredClusterName("selected")).thenReturn("DefaultCluster");
+ executeWith(resolver, admin, "selected");
+
+ assertThatThrownBy(() -> new
ApacheAclReadService(resolver).listRules("selected", null, null))
+
.isInstanceOf(org.apache.rocketmq.studio.common.exception.BusinessException.class)
+ .hasMessageContaining("Broker cluster membership is
unavailable");
+ }
+
+ private ClusterInfo sharedTopology() {
+ BrokerData first = new BrokerData("DefaultCluster", "broker-a",
+ new HashMap<>(Map.of(0L, "10.0.0.11:10911")));
+ BrokerData second = new BrokerData("OtherCluster", "broker-b",
+ new HashMap<>(Map.of(0L, "10.0.0.12:10911")));
+ ClusterInfo clusterInfo = new ClusterInfo();
+ clusterInfo.setClusterAddrTable(new HashMap<>(Map.of(
+ "DefaultCluster", Set.of("broker-a"), "OtherCluster",
Set.of("broker-b"))));
+ clusterInfo.setBrokerAddrTable(new HashMap<>(Map.of("broker-a", first,
"broker-b", second)));
+ return clusterInfo;
+ }
+
private ClusterInfo clusterInfo(String address) {
BrokerData broker = new BrokerData();
broker.setBrokerAddrs(new HashMap<>(Map.of(0L, address)));
@@ -118,7 +178,12 @@ class ApacheAclReadServiceTest {
}
private void executeWith(RuntimeAdminClientResolver resolver, MQAdminExt
admin) throws Exception {
- when(resolver.execute(eq("instance-1"), any())).thenAnswer(invocation
-> {
+ executeWith(resolver, admin, "instance-1");
+ }
+
+ private void executeWith(RuntimeAdminClientResolver resolver, MQAdminExt
admin, String instanceId)
+ throws Exception {
+ when(resolver.execute(eq(instanceId), any())).thenAnswer(invocation ->
{
MqAdminExtFactory.AdminAction<?> action =
invocation.getArgument(1);
return action.apply(admin);
});
