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 ef3ec0acd fix(dlq): enforce broker topology guard on the selected DLQ
msgId lookup (#3112)
ef3ec0acd is described below
commit ef3ec0acdf7ca3bce175953e1f2bbe25c5001c4d
Author: 烤化の初雪 <[email protected]>
AuthorDate: Fri Sep 4 16:11:13 2026 +0800
fix(dlq): enforce broker topology guard on the selected DLQ msgId lookup
(#3112)
Signed-off-by: unbridled-41
<[email protected]>
Co-authored-by: unbridled-41
<[email protected]>
---
.../provider/apache/BrokerTopologyGuards.java | 110 +++++++++++++++++++++
.../provider/apache/RocketMQDLQProvider.java | 3 +
.../provider/apache/RocketMQMessageProvider.java | 79 +--------------
.../provider/apache/RocketMQDLQProviderTest.java | 96 ++++++++++++++++++
4 files changed, 212 insertions(+), 76 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/BrokerTopologyGuards.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/BrokerTopologyGuards.java
new file mode 100644
index 000000000..3ca0fca6f
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/BrokerTopologyGuards.java
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+package org.apache.rocketmq.studio.provider.apache;
+
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.common.message.MessageDecoder;
+import org.apache.rocketmq.common.message.MessageId;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+import org.springframework.util.StringUtils;
+
+/**
+ * Shared guards for offset-style message ids, which embed a broker address
that
+ * {@code MQAdminImpl#viewMessage} connects to directly. Ids whose embedded
address is
+ * outside the selected instance topology must be rejected before that call.
+ */
+@Slf4j
+public final class BrokerTopologyGuards {
+
+ private BrokerTopologyGuards() {}
+
+ /**
+ * Returns true when the msgId either does not decode as an offset id (in
which case
+ * MQAdminImpl takes the unique-key lookup through the topic route and
needs no guard)
+ * or its embedded broker address belongs to the selected instance
topology. When the
+ * topology itself cannot be verified, reject instead of handing an
unverified address
+ * to remoting.
+ */
+ public static boolean isWithinKnownBrokerTopology(MQAdminExt admin, String
msgId) {
+ MessageId messageId;
+ try {
+ messageId = MessageDecoder.decodeMessageId(msgId);
+ } catch (Exception e) {
+ return true;
+ }
+ try {
+ return validatedBrokerAddr(admin, msgId, messageId) != null;
+ } catch (Exception e) {
+ log.warn("Could not verify broker topology for msgId={}: {}",
msgId, e.getMessage());
+ return false;
+ }
+ }
+
+ /** Returns the embedded broker address when it is a known endpoint,
otherwise null. */
+ public static String validatedBrokerAddr(MQAdminExt admin, String msgId,
MessageId messageId) throws Exception {
+ String brokerAddr = decodedBrokerAddr(messageId);
+ if (!StringUtils.hasText(brokerAddr)) {
+ return null;
+ }
+ if (knownBrokerEndpoints(admin).contains(brokerAddr)) {
+ return brokerAddr;
+ }
+ log.warn("Rejecting decoded broker address {} for msgId={} because it
is not a known broker endpoint"
+ + " for the selected instance", brokerAddr, msgId);
+ return null;
+ }
+
+ static String decodedBrokerAddr(MessageId messageId) {
+ SocketAddress address = messageId.getAddress();
+ if (!(address instanceof InetSocketAddress)) {
+ return null;
+ }
+ InetSocketAddress inet = (InetSocketAddress) address;
+ if (inet.getAddress() == null) {
+ return null;
+ }
+ return inet.getAddress().getHostAddress() + ":" + inet.getPort();
+ }
+
+ static Set<String> knownBrokerEndpoints(MQAdminExt admin) throws Exception
{
+ ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
+ if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null
+ || clusterInfo.getBrokerAddrTable().isEmpty()) {
+ return Collections.emptySet();
+ }
+ Set<String> endpoints = new HashSet<>();
+ for (BrokerData brokerData :
clusterInfo.getBrokerAddrTable().values()) {
+ if (brokerData == null || brokerData.getBrokerAddrs() == null
+ || brokerData.getBrokerAddrs().isEmpty()) {
+ continue;
+ }
+ for (String brokerAddr : brokerData.getBrokerAddrs().values()) {
+ if (StringUtils.hasText(brokerAddr)) {
+ endpoints.add(brokerAddr.trim());
+ }
+ }
+ }
+ return endpoints;
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
index acb5922ed..c0d10c60f 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
@@ -266,6 +266,9 @@ public class RocketMQDLQProvider implements DLQProvider {
List<MessageExt> resolved = new ArrayList<>(selected.size());
for (String msgId : selected) {
try {
+ if
(!BrokerTopologyGuards.isWithinKnownBrokerTopology(admin, msgId)) {
+ continue;
+ }
MessageExt deadLetter = admin.viewMessage(dlqTopic, msgId);
if (deadLetter != null) {
resolved.add(deadLetter);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
index 789aa2bc0..1c112bfa6 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
@@ -27,8 +27,6 @@ import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageId;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
-import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
-import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.protocol.route.QueueData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
@@ -51,8 +49,6 @@ import org.springframework.util.StringUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
-import java.net.InetSocketAddress;
-import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.nio.charset.CharacterCodingException;
@@ -61,7 +57,6 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Base64;
-import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -142,7 +137,7 @@ public class RocketMQMessageProvider implements
MessageProvider {
MessageExt messageExt = null;
if (StringUtils.hasText(topic)) {
try {
- if (isWithinKnownBrokerTopology(adminExt, msgId)) {
+ if (BrokerTopologyGuards.isWithinKnownBrokerTopology(adminExt,
msgId)) {
messageExt = adminExt.viewMessage(topic, msgId);
}
} catch (Exception e) {
@@ -165,7 +160,7 @@ public class RocketMQMessageProvider implements
MessageProvider {
private MessageExt viewMessageByOffsetId(DefaultMQAdminExt adminExt,
String topic, String msgId) {
try {
MessageId messageId = MessageDecoder.decodeMessageId(msgId);
- String brokerAddr = validatedBrokerAddr(adminExt, msgId,
messageId);
+ String brokerAddr =
BrokerTopologyGuards.validatedBrokerAddr(adminExt, msgId, messageId);
if (!StringUtils.hasText(brokerAddr)) {
return null;
}
@@ -179,74 +174,6 @@ public class RocketMQMessageProvider implements
MessageProvider {
}
}
- private String validatedBrokerAddr(DefaultMQAdminExt adminExt, String
msgId, MessageId messageId) throws Exception {
- String brokerAddr = decodedBrokerAddr(messageId);
- if (!StringUtils.hasText(brokerAddr)) {
- return null;
- }
- if (knownBrokerEndpoints(adminExt).contains(brokerAddr)) {
- return brokerAddr;
- }
- log.warn("Rejecting decoded broker address {} for msgId={} because it
is not a known broker endpoint"
- + " for the selected instance", brokerAddr, msgId);
- return null;
- }
-
- /**
- * Offset-style message ids embed a broker address that {@code
MQAdminImpl#viewMessage} connects
- * to directly, so ids whose embedded address is outside the selected
instance topology must be
- * rejected before that call. Ids that do not decode as offset ids take
MQAdminImpl's unique-key
- * lookup, which resolves brokers from the topic route and needs no guard.
When the topology
- * itself cannot be verified, reject too — mirroring the fallback path's
behavior — instead of
- * handing an unverified address to remoting.
- */
- private boolean isWithinKnownBrokerTopology(DefaultMQAdminExt adminExt,
String msgId) {
- MessageId messageId;
- try {
- messageId = MessageDecoder.decodeMessageId(msgId);
- } catch (Exception e) {
- return true;
- }
- try {
- return validatedBrokerAddr(adminExt, msgId, messageId) != null;
- } catch (Exception e) {
- log.warn("Could not verify broker topology for msgId={}: {}",
msgId, e.getMessage());
- return false;
- }
- }
-
- private String decodedBrokerAddr(MessageId messageId) {
- SocketAddress address = messageId.getAddress();
- if (!(address instanceof InetSocketAddress)) {
- return null;
- }
- InetSocketAddress inet = (InetSocketAddress) address;
- if (inet.getAddress() == null) {
- return null;
- }
- return inet.getAddress().getHostAddress() + ":" + inet.getPort();
- }
-
- private Set<String> knownBrokerEndpoints(DefaultMQAdminExt adminExt)
throws Exception {
- ClusterInfo clusterInfo = adminExt.examineBrokerClusterInfo();
- if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null
- || clusterInfo.getBrokerAddrTable().isEmpty()) {
- return Collections.emptySet();
- }
- Set<String> endpoints = new HashSet<>();
- for (BrokerData brokerData :
clusterInfo.getBrokerAddrTable().values()) {
- if (brokerData == null || brokerData.getBrokerAddrs() == null ||
brokerData.getBrokerAddrs().isEmpty()) {
- continue;
- }
- for (String brokerAddr : brokerData.getBrokerAddrs().values()) {
- if (StringUtils.hasText(brokerAddr)) {
- endpoints.add(brokerAddr.trim());
- }
- }
- }
- return endpoints;
- }
-
private List<MessageRecordVO> queryByKey(DefaultMQAdminExt adminExt,
String topic, String key,
String tag, long begin, long end)
{
try {
@@ -592,7 +519,7 @@ public class RocketMQMessageProvider implements
MessageProvider {
if (StringUtils.hasText(topic)) {
try {
MessageExt messageExt = null;
- if (isWithinKnownBrokerTopology(adminExt, msgId)) {
+ if (BrokerTopologyGuards.isWithinKnownBrokerTopology(adminExt,
msgId)) {
messageExt = adminExt.viewMessage(topic, msgId);
}
if (messageExt != null) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
index a842d271f..b44c4d691 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
@@ -25,10 +25,13 @@ import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.cluster.broker.MqClientPool;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
@@ -37,6 +40,7 @@ import
org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.dlq.DLQExportResultVO;
import org.apache.rocketmq.studio.instance.dlq.DLQGroupVO;
import org.apache.rocketmq.studio.instance.dlq.DLQMessageVO;
+import org.apache.rocketmq.studio.instance.dlq.DLQResendResultVO;
import org.apache.rocketmq.studio.ops.audit.AuditService;
import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.junit.jupiter.api.BeforeEach;
@@ -47,9 +51,12 @@ import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
+import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
@@ -310,6 +317,73 @@ class RocketMQDLQProviderTest {
verify(pullConsumer, never()).pull(any(MessageQueue.class),
anyString(), anyLong(), anyInt());
}
+ @Test
+ void resendSelectedMessagesRejectsForgedMsgIdOutsideKnownTopology() throws
Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ String forgedMsgId = MessageDecoder.createMessageId(new
InetSocketAddress("10.2.3.4", 10911), 12345L);
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithBrokerAddresses("172.30.10.100:10911"));
+ stubExistingTarget("target-topic");
+
+ DLQResendResultVO result = provider.resendMessages(
+ "instance-a", "group-a", List.of(forgedMsgId), "target-topic");
+
+ assertThat(result.getMatched()).isZero();
+ assertThat(result.getResent()).isZero();
+ verify(adminExt, never()).viewMessage(anyString(), anyString());
+ verify(runtimeAdminClientResolver,
never()).executeProducer(anyString(), any());
+ }
+
+ @Test
+ void resendSelectedMessagesResolvesInTopologyMsgIdNormally() throws
Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ String msgId = MessageDecoder.createMessageId(new
InetSocketAddress("172.30.10.100", 10911), 12345L);
+ MessageExt deadLetter = new MessageExt();
+ deadLetter.setMsgId(msgId);
+ deadLetter.setTopic(dlqTopic);
+ deadLetter.setBody(new byte[] {1, 2, 3});
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithBrokerAddresses("172.30.10.100:10911"));
+ when(adminExt.viewMessage(dlqTopic, msgId)).thenReturn(deadLetter);
+ stubExistingTarget("target-topic");
+ SendResult sendResult = new SendResult();
+ sendResult.setSendStatus(SendStatus.SEND_OK);
+ when(dlqProducer.send(any(Message.class))).thenReturn(sendResult);
+
+ DLQResendResultVO result = provider.resendMessages(
+ "instance-a", "group-a", List.of(msgId), "target-topic");
+
+ assertThat(result.getMatched()).isEqualTo(1);
+ assertThat(result.getResent()).isEqualTo(1);
+ assertThat(result.getOutcome()).isEqualTo("SUCCESS");
+ verify(adminExt).viewMessage(dlqTopic, msgId);
+ }
+
+ @Test
+ void resendSelectedMessagesPassesNonOffsetIdsThroughToViewMessage() throws
Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ when(adminExt.viewMessage(dlqTopic, "uniq-key-1"))
+ .thenThrow(new IllegalStateException("unique key lookup
handled by MQAdminImpl"));
+
+ DLQResendResultVO result = provider.resendMessages(
+ "instance-a", "group-a", List.of("uniq-key-1"), null);
+
+ assertThat(result.getMatched()).isZero();
+ verify(adminExt).viewMessage(dlqTopic, "uniq-key-1");
+ verify(adminExt, never()).examineBrokerClusterInfo();
+ }
+
+ @Test
+ void resendSelectedMessagesFailsClosedWhenTopologyCannotBeVerified()
throws Exception {
+ String msgId = MessageDecoder.createMessageId(new
InetSocketAddress("172.30.10.100", 10911), 12345L);
+ when(adminExt.examineBrokerClusterInfo()).thenThrow(new
IllegalStateException("nameserver unreachable"));
+
+ DLQResendResultVO result = provider.resendMessages(
+ "instance-a", "group-a", List.of(msgId), null);
+
+ assertThat(result.getMatched()).isZero();
+ verify(adminExt, never()).viewMessage(anyString(), anyString());
+ verify(runtimeAdminClientResolver,
never()).executeProducer(anyString(), any());
+ }
+
@Test
void
resendMessagesShouldNormalizeGroupNameBeforeBuildingDlqTopicAndAuditing()
throws Exception {
String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
@@ -704,4 +778,26 @@ class RocketMQDLQProviderTest {
assertThat(exported.getMessages()).isEmpty();
assertThat(exported.getLimit()).isEqualTo(5000);
}
+
+ private void stubExistingTarget(String topic) throws Exception {
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of(topic));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
+ }
+
+ private static ClusterInfo clusterInfoWithBrokerAddresses(String...
brokerAddresses) {
+ ClusterInfo clusterInfo = new ClusterInfo();
+ Map<String, BrokerData> brokerAddrTable = new HashMap<>();
+ for (int index = 0; index < brokerAddresses.length; index++) {
+ BrokerData brokerData = new BrokerData();
+ brokerData.setBrokerName("broker-" + index);
+ brokerData.setCluster("cluster-a");
+ HashMap<Long, String> brokerAddrs = new HashMap<>();
+ brokerAddrs.put(0L, brokerAddresses[index]);
+ brokerData.setBrokerAddrs(brokerAddrs);
+ brokerAddrTable.put(brokerData.getBrokerName(), brokerData);
+ }
+ clusterInfo.setBrokerAddrTable(brokerAddrTable);
+ return clusterInfo;
+ }
}
