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 0ffdc4a1a fix(provider): validate before Tencent ACL and DLQ writes
(#6173)
0ffdc4a1a is described below
commit 0ffdc4a1afcdc7548ad4d607674422157494d6af
Author: Xiao Yang <[email protected]>
AuthorDate: Mon Oct 12 16:15:59 2026 +0800
fix(provider): validate before Tencent ACL and DLQ writes (#6173)
Two provider paths that issued a remote write before validating the input
that decides what the write does.
- #6173 `TencentAclService.createRule` derives `permRead` / `permWrite`
from `hasPermissionAction(rule, "SUB")` and `(rule, "PUB")` at `:248-249`,
whose underlying `hasAction` (`:292-295`) is an `anyMatch(equalsIgnoreCase)`
that silently ignores an unknown action, and then unconditionally calls
`client.ModifyRole(request)` at `:255-256` — so `["PBU"]`, a typo for `PUB`,
switched both permissions off on the cloud role and still returned 200, and
`["PUB","PBU"]` quietly dropped the SUB [...]
- #6251 both `resendMessages` overloads run `validateResendTargetTopic`
only when the caller passed a `targetTopic` (`RocketMQDLQProvider.java:202-203`
for the time-range entry, `:274-275` for the selected-messages one), but when
it is absent `resolveTargetTopic` (`:766-782`) infers the destination from
`PROPERTY_DLQ_ORIGIN_TOPIC` or `PROPERTY_RETRY_TOPIC` and `resendOne` hands
that name to a producer routing by topic across the whole endpoint — so a
sibling cluster sharing the NameSe [...]
Tests: `TencentAclServiceTest` adds a `@ParameterizedTest` over seven
invalid action lists — null, empty, `["PBU"]`, `["PUB","PBU"]`, `[""]`, `[" SUB
"]`, `["SUB", null]` — each asserting the 400 together with
`verifyNoInteractions(clientFactory, client)` so no cloud call can slip
through, plus a case feeding `["pub","sUb","pub"]` through `updateRule` to pin
that case-insensitive matching and de-duplication still work;
`RocketMQDLQProviderTest` adds two `@ParameterizedTest` methods ov [...]
Folded: #6251 was closed as folded into this commit (same author).
---
docs/api-spec.md | 6 ++
.../provider/apache/RocketMQDLQProvider.java | 9 ++
.../studio/provider/tencent/TencentAclService.java | 10 ++
.../provider/apache/RocketMQDLQProviderTest.java | 115 +++++++++++++++++++++
.../provider/tencent/TencentAclServiceTest.java | 44 ++++++++
5 files changed, 184 insertions(+)
diff --git a/docs/api-spec.md b/docs/api-spec.md
index e5b0c7fd2..b7a8842e8 100644
--- a/docs/api-spec.md
+++ b/docs/api-spec.md
@@ -1754,6 +1754,12 @@ POST /api/dlq/resend
Topic，未指定 `targetTopic` 时即原 Topic）、`reason`（`string`，归一化后的简短失败原因，最长 256 字符）。
§9.4「重发选中的死信消息」返回同一结构，两条重投路径共用该 VO。
+> **跨集群拒绝（`409`）**：显式传入 `targetTopic` 时，若它解析出的集群不是死信 Topic（`%DLQ%{groupName}`）
+> 所属集群的子集，返回 `409 targetTopic resolves to a different cluster than
%DLQ%{groupName}; resend rejected: {targetTopic}`
+> （`RocketMQDLQProvider.requireSameClusterAsDlqTopic`）。**省略 `targetTopic`
时同样校验**：服务端先按每条死信
+> 推断目的地（即其原 Topic），对推断出的每一个目的地跑同一次校验，任一跨集群即整批 `409` 拒绝，
+> 不会出现「重投了一半」。§9.4 的按 `msgIds` 重投走同一套校验。
+
### 9.3 分页获取死信消息明细
```
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 2e8159c78..efc9bf91b 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
@@ -582,6 +582,15 @@ public class RocketMQDLQProvider implements DLQProvider {
if (deadLetters.isEmpty()) {
return batch;
}
+ if (!StringUtils.hasText(targetTopic)) {
+ Set<String> destinations = deadLetters.stream()
+ .map(deadLetter -> resolveTargetTopic(deadLetter,
targetTopic))
+ .filter(StringUtils::hasText)
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+ for (String destination : destinations) {
+ requireSameClusterAsDlqTopic(instanceId,
MixAll.DLQ_GROUP_TOPIC_PREFIX + groupName, destination);
+ }
+ }
try {
runtimeAdminClientResolver.executeProducer(instanceId, producer ->
{
for (MessageExt deadLetter : deadLetters) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentAclService.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentAclService.java
index ba1e048ff..cf2f5b896 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentAclService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentAclService.java
@@ -310,6 +310,16 @@ public class TencentAclService {
throw new BusinessException(400,
"Tencent Cloud roles only support ALLOW ACL rules");
}
+ if (rule.getActions() == null || rule.getActions().isEmpty()) {
+ throw new BusinessException(400, "Tencent Cloud ACL rules require
at least one action");
+ }
+ for (String action : rule.getActions()) {
+ if (!"PUB".equalsIgnoreCase(action) &&
!"SUB".equalsIgnoreCase(action)
+ && !"ALL".equalsIgnoreCase(action)) {
+ throw new BusinessException(400,
+ "Tencent Cloud ACL rules only support PUB, SUB or ALL
actions");
+ }
+ }
}
private static boolean isOptionalValue(String actual, String expected) {
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 27ba07a9f..4374fe9b9 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
@@ -54,6 +54,8 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
@@ -352,6 +354,119 @@ class RocketMQDLQProviderTest {
verify(runtimeAdminClientResolver,
never()).executeProducer(anyString(), any());
}
+ @ParameterizedTest
+ @ValueSource(strings = {MessageConst.PROPERTY_DLQ_ORIGIN_TOPIC,
MessageConst.PROPERTY_RETRY_TOPIC})
+ void
rangeResendShouldRejectInferredForeignTargetsBeforeSendingAnyMessageTest(String
originProperty)
+ throws Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ MessageExt local = deadLetterWithOrigin("local-msg", dlqTopic,
originProperty, "local-topic");
+ MessageExt foreign = deadLetterWithOrigin("foreign-msg", dlqTopic,
originProperty, "foreign-topic");
+ MessageQueue queue = new MessageQueue(dlqTopic, "broker-a", 0);
+
when(pullConsumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(Set.of(queue));
+ when(pullConsumer.searchOffset(eq(queue), anyLong())).thenReturn(0L,
2L);
+ when(pullConsumer.pull(eq(queue), eq("*"), anyLong(), anyInt()))
+ .thenReturn(new PullResult(PullStatus.FOUND, 2L, 0L, 2L,
List.of(local, foreign)));
+
when(adminExt.examineTopicRouteInfo(dlqTopic)).thenReturn(routeInCluster("cluster-a"));
+
when(adminExt.examineTopicRouteInfo("local-topic")).thenReturn(routeInCluster("cluster-a"));
+
when(adminExt.examineTopicRouteInfo("foreign-topic")).thenReturn(routeInCluster("cluster-b"));
+
+ assertThatThrownBy(() -> provider.resendMessages("instance-a",
"group-a", 100L, 200L, null))
+ .isInstanceOfSatisfying(BusinessException.class, exception -> {
+ assertThat(exception.getCode()).isEqualTo(409);
+
assertThat(exception.getMessage()).contains("foreign-topic", "different
cluster");
+ });
+ verify(runtimeAdminClientResolver,
never()).executeProducer(anyString(), any());
+ verifyNoInteractions(dlqProducer);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {MessageConst.PROPERTY_DLQ_ORIGIN_TOPIC,
MessageConst.PROPERTY_RETRY_TOPIC})
+ void
selectedResendShouldRejectInferredForeignTargetsBeforeSendingAnyMessageTest(String
originProperty)
+ throws Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ when(adminExt.viewMessage(dlqTopic, "local-msg"))
+ .thenReturn(deadLetterWithOrigin("local-msg", dlqTopic,
originProperty, "local-topic"));
+ when(adminExt.viewMessage(dlqTopic, "foreign-msg"))
+ .thenReturn(deadLetterWithOrigin("foreign-msg", dlqTopic,
originProperty, "foreign-topic"));
+
when(adminExt.examineTopicRouteInfo(dlqTopic)).thenReturn(routeInCluster("cluster-a"));
+
when(adminExt.examineTopicRouteInfo("local-topic")).thenReturn(routeInCluster("cluster-a"));
+
when(adminExt.examineTopicRouteInfo("foreign-topic")).thenReturn(routeInCluster("cluster-b"));
+
+ assertThatThrownBy(() -> provider.resendMessages(
+ "instance-a", "group-a", List.of("local-msg", "foreign-msg"),
null))
+ .isInstanceOfSatisfying(BusinessException.class, exception -> {
+ assertThat(exception.getCode()).isEqualTo(409);
+
assertThat(exception.getMessage()).contains("foreign-topic", "different
cluster");
+ });
+ verify(runtimeAdminClientResolver,
never()).executeProducer(anyString(), any());
+ verifyNoInteractions(dlqProducer);
+ }
+
+ @Test
+ void
originalResendShouldAllowSameClusterTargetsAndPreserveMissingOriginFailuresTest()
throws Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ when(adminExt.viewMessage(dlqTopic,
"origin-msg")).thenReturn(deadLetterWithOrigin(
+ "origin-msg", dlqTopic,
MessageConst.PROPERTY_DLQ_ORIGIN_TOPIC, "orders"));
+ when(adminExt.viewMessage(dlqTopic,
"retry-msg")).thenReturn(deadLetterWithOrigin(
+ "retry-msg", dlqTopic, MessageConst.PROPERTY_RETRY_TOPIC,
"orders"));
+ when(adminExt.viewMessage(dlqTopic,
"missing-origin-msg")).thenReturn(deadLetterWithOrigin(
+ "missing-origin-msg", dlqTopic,
MessageConst.PROPERTY_DLQ_ORIGIN_TOPIC, " "));
+
lenient().when(adminExt.examineTopicRouteInfo(dlqTopic)).thenReturn(routeInCluster("cluster-a"));
+
lenient().when(adminExt.examineTopicRouteInfo("orders")).thenReturn(routeInCluster("cluster-a"));
+ 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("origin-msg", "retry-msg", "missing-origin-msg"),
null);
+
+ assertThat(result.getMatched()).isEqualTo(3);
+ assertThat(result.getResent()).isEqualTo(2);
+ assertThat(result.getFailed()).isEqualTo(1);
+ assertThat(result.getOutcome()).isEqualTo("PARTIAL");
+ assertThat(result.getFailures()).singleElement().satisfies(failure -> {
+ assertThat(failure.getMsgId()).isEqualTo("missing-origin-msg");
+ assertThat(failure.getReason()).isEqualTo("No target topic could
be resolved");
+ });
+ ArgumentCaptor<Message> sent = ArgumentCaptor.forClass(Message.class);
+ verify(dlqProducer, times(2)).send(sent.capture());
+
assertThat(sent.getAllValues()).extracting(Message::getTopic).containsExactly("orders",
"orders");
+ verify(adminExt).examineTopicRouteInfo("orders");
+ }
+
+ @Test
+ void
originalResendShouldLeaveUnresolvableRoutesToTheExistingSendFailureTest()
throws Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ when(adminExt.viewMessage(dlqTopic,
"origin-msg")).thenReturn(deadLetterWithOrigin(
+ "origin-msg", dlqTopic,
MessageConst.PROPERTY_DLQ_ORIGIN_TOPIC, "orders"));
+
lenient().when(adminExt.examineTopicRouteInfo(dlqTopic)).thenReturn(routeInCluster("cluster-a"));
+ lenient().when(adminExt.examineTopicRouteInfo("orders"))
+ .thenThrow(new MQClientException("No topic route info for
orders", null));
+ when(dlqProducer.send(any(Message.class)))
+ .thenThrow(new MQClientException("No topic route info for
orders", null));
+
+ DLQResendResultVO result = provider.resendMessages(
+ "instance-a", "group-a", List.of("origin-msg"), null);
+
+ assertThat(result.getResent()).isZero();
+ assertThat(result.getFailed()).isEqualTo(1);
+ assertThat(result.getOutcome()).isEqualTo("FAILED");
+ assertThat(result.getFailures()).singleElement().satisfies(failure ->
+ assertThat(failure.getReason()).contains("No topic route info
for orders"));
+ verify(dlqProducer).send(any(Message.class));
+ }
+
+ private static MessageExt deadLetterWithOrigin(String msgId, String
dlqTopic,
+ String originProperty,
String originTopic) {
+ MessageExt message = new MessageExt();
+ message.setMsgId(msgId);
+ message.setTopic(dlqTopic);
+ message.setStoreTimestamp(150L);
+ message.setBody(new byte[] {1, 2, 3});
+ MessageAccessor.putProperty(message, originProperty, originTopic);
+ return message;
+ }
+
private static TopicRouteData routeInCluster(String cluster) {
BrokerData broker = new BrokerData();
broker.setCluster(cluster);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentAclServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentAclServiceTest.java
index dae359cd2..152242b4e 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentAclServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentAclServiceTest.java
@@ -29,13 +29,17 @@ import org.apache.rocketmq.studio.instance.acl.AclUserVO;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
+import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.stream.IntStream;
+import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -46,6 +50,7 @@ import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -272,6 +277,45 @@ class TencentAclServiceTest {
assertThat(created.getActions()).containsExactly("PUB", "SUB");
}
+ @ParameterizedTest
+ @MethodSource("invalidRuleActions")
+ void
ruleMutationsShouldRejectInvalidActionsBeforeChangingRolePermissionsTest(List<String>
actions) {
+ AclRuleVO rule = AclRuleVO.builder()
+ .principal("reader-role")
+ .resource("*")
+ .actions(actions)
+ .build();
+
+ assertThatThrownBy(() -> service.createRule(INSTANCE_ID, rule))
+ .isInstanceOf(BusinessException.class)
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(400));
+ assertThatThrownBy(() -> service.updateRule(INSTANCE_ID, rule))
+ .isInstanceOf(BusinessException.class)
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(400));
+
+ verifyNoInteractions(clientFactory, client);
+ }
+
+ static Stream<List<String>> invalidRuleActions() {
+ return Stream.of(null, List.of(), List.of("PBU"), List.of("PUB",
"PBU"),
+ List.of(""), List.of(" SUB "), Arrays.asList("SUB", null));
+ }
+
+ @Test
+ void updateRuleShouldPreserveCaseInsensitivePermissionActionsTest() throws
Exception {
+ AclRuleVO updated = service.updateRule(INSTANCE_ID, AclRuleVO.builder()
+ .principal("reader-role")
+ .resource("*")
+ .actions(List.of("pub", "sUb", "pub"))
+ .build());
+
+ ArgumentCaptor<ModifyRoleRequest> requestCaptor =
ArgumentCaptor.forClass(ModifyRoleRequest.class);
+ verify(client).ModifyRole(requestCaptor.capture());
+ assertThat(requestCaptor.getValue().getPermRead()).isTrue();
+ assertThat(requestCaptor.getValue().getPermWrite()).isTrue();
+ assertThat(updated.getActions()).containsExactly("PUB", "SUB");
+ }
+
@Test
void createRuleShouldRejectResourceScopedTencentRulesTest() {
AclRuleVO rule = AclRuleVO.builder()
