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 d584caddc fix: message id topology guard, proxy reload scope, webhook
redaction and query budget (#2339)
d584caddc is described below
commit d584caddc76dd1bcf1e972dbf7a0ada5ab13dfd1
Author: aias00 <[email protected]>
AuthorDate: Wed Aug 19 14:53:17 2026 +0800
fix: message id topology guard, proxy reload scope, webhook redaction and
query budget (#2339)
* Prevent forged offset msgIds from reaching arbitrary brokers
* Restrict proxy reloads to discovered cluster endpoints
* Align auth credential tests with authenticated user context
Constraint: AuthInterceptor now determines admin access from
getAuthenticatedUser() rather than isAdmin() alone
Rejected: changing AuthInterceptor behavior | production code is already
correct for the intended auth model
Confidence: high
Scope-risk: narrow
Tested: mvn
-Dtest=AuthCredentialAuthorizationIntegrationTest,AuthInterceptorTest test; git
diff --check
Signed-off-by: liuhy <[email protected]>
* docs: clarify system authorization context
* Protect reader-visible settings reads from leaking notification webhook
secrets
* fix: return conflict for concurrent Studio user creation
Signed-off-by: liuhy <[email protected]>
* fix: synchronize browser auth state after policy changes
Signed-off-by: liuhy <[email protected]>
* Keep multi-backend metrics tests isolated from host networking
Constraint: The test must keep asserting per-backend query paths without
mutating JVM-global proxy state or depending on host-specific non-loopback
interfaces
Rejected: Reusing a site-local bind address plus global ProxySelector
override | still couples the test to host network topology and process-wide
state
Confidence: high
Scope-risk: narrow
Tested: mvn -Dtest=MultiBackendMetricsSourceTest test; mvn test; git diff
--check
Signed-off-by: liuhy <[email protected]>
* Keep topic message queries on a bounded newest-first budget
Constraint: Topic queries must stay within a fixed per-queue pull budget
while still returning the newest 200 messages in the 7-day window
Rejected: Drop the pull cap entirely | would reintroduce unbounded scans
and DoS risk on high-volume topics
Rejected: Keep forward scans with a hard failure | still breaks once a
queue crosses 1000 pulls before reaching the newest data
Directive: Preserve newest-first ordering and confine any truncation to
older offsets outside the guarded tail budget
Confidence: high
Scope-risk: narrow
Tested: export
JAVA_HOME=/Users/aias/Library/Java/JavaVirtualMachines/openjdk-21.0.2/Contents/Home
&& export
PATH="/Library/Java/JavaVirtualMachines/jdk-17.0.3+7/Contents/Home/bin:/Users/aias/.qoder/entry:/Users/aias/.nvm/versions/node/v22.23.1/bin:/Applications/Visual
Studio
Code.app/Contents/Resources/app/bin:/Library/Java/JavaVirtualMachines/jdk-17.0.3+7/Contents/Home/bin:/Users/aias/.local/share/hyperframes-python/bin:/opt/homebrew/bin:/opt/homebrew/sbin:/usr/local/bin:/System/Cry
[...]
Signed-off-by: liuhy <[email protected]>
* test: align DLQ resend audit assertion with classified signature
---------
Signed-off-by: liuhy <[email protected]>
---
.../rocketmq/studio/auth/AuthInterceptor.java | 12 +-
.../apache/rocketmq/studio/auth/AuthService.java | 6 +-
.../studio/auth/AuthenticatedUserContext.java | 20 ++-
.../studio/cluster/broker/ClusterService.java | 14 ++
.../studio/cluster/proxy/ProxyAddressService.java | 48 ++++---
.../studio/cluster/proxy/ProxyController.java | 7 +-
.../provider/apache/RocketMQMessageProvider.java | 119 ++++++++++++++--
.../studio/settings/GeneralSettingsVO.java | 12 +-
.../rocketmq/studio/settings/SettingsService.java | 22 ++-
...AuthCredentialAuthorizationIntegrationTest.java | 12 ++
.../rocketmq/studio/auth/AuthInterceptorTest.java | 20 +++
.../studio/auth/AuthServiceDatabaseTest.java | 13 ++
.../studio/cluster/broker/ClusterServiceTest.java | 19 +++
.../metrics/MultiBackendMetricsSourceTest.java | 126 +++++++---------
.../cluster/proxy/ProxyAddressServiceTest.java | 63 +++++++-
.../studio/cluster/proxy/ProxyControllerTest.java | 16 +--
.../provider/apache/RocketMQDLQProviderTest.java | 2 +
.../apache/RocketMQMessageProviderTest.java | 158 +++++++++++++++++++--
.../studio/settings/SettingsControllerTest.java | 36 +++++
.../studio/settings/SettingsServiceTest.java | 23 +++
web/src/App.test.tsx | 1 +
web/src/App.tsx | 7 +-
web/src/api/generalSettings.test.ts | 7 +
web/src/api/settings.ts | 17 ++-
web/src/pages/studio/UserManagement.tsx | 33 ++++-
25 files changed, 664 insertions(+), 149 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthInterceptor.java
b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthInterceptor.java
index c8468e6eb..3e8402f81 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthInterceptor.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthInterceptor.java
@@ -59,9 +59,15 @@ public class AuthInterceptor implements HandlerInterceptor {
writeError(response, HttpStatus.UNAUTHORIZED, "Unauthorized");
return false;
}
- authService.getAuthenticatedUser(authorization)
- .ifPresent(user ->
AuthenticatedUserContext.setUser(user.getUserId(), user.getUsername()));
- if (requiresAdmin(request, requestPath(request)) &&
!authService.isAdmin(authorization)) {
+ var authenticatedUser =
authService.getAuthenticatedUser(authorization).orElse(null);
+ if (authenticatedUser != null) {
+ AuthenticatedUserContext.setUser(
+ authenticatedUser.getUserId(),
+ authenticatedUser.getUsername(),
+ authenticatedUser.isAdmin());
+ }
+ if (requiresAdmin(request, requestPath(request))
+ && (authenticatedUser == null ||
!authenticatedUser.isAdmin())) {
writeError(response, HttpStatus.FORBIDDEN, "Admin permission
required");
return false;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthService.java
b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthService.java
index 9d6a07407..d5a67f4ba 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthService.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthService.java
@@ -176,7 +176,11 @@ public class AuthService {
user.setAdmin(admin);
user.setEnabled(true);
user.setPasswordChangedAt(now());
- userMapper.insert(user);
+ try {
+ userMapper.insert(user);
+ } catch (DuplicateKeyException exception) {
+ throw new BusinessException(409, "Username is already in use");
+ }
return user;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthenticatedUserContext.java
b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthenticatedUserContext.java
index b9c09fd28..f2bfcb045 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthenticatedUserContext.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthenticatedUserContext.java
@@ -26,20 +26,30 @@ public final class AuthenticatedUserContext {
private static final ThreadLocal<String> CURRENT_USERNAME = new
ThreadLocal<>();
private static final ThreadLocal<String> CURRENT_USER_ID = new
ThreadLocal<>();
+ private static final ThreadLocal<Boolean> CURRENT_ADMIN = new
ThreadLocal<>();
private AuthenticatedUserContext() {
}
- public static void setUsername(String username) {
+ public static void setUser(String username, boolean admin) {
if (username == null || username.isBlank()) {
clear();
return;
}
CURRENT_USERNAME.set(username);
+ CURRENT_ADMIN.set(admin);
+ }
+
+ public static void setUsername(String username) {
+ setUser(username, false);
}
public static void setUser(Long userId, String username) {
- setUsername(username);
+ setUser(userId, username, false);
+ }
+
+ public static void setUser(Long userId, String username, boolean admin) {
+ setUser(username, admin);
if (userId == null) {
CURRENT_USER_ID.remove();
} else {
@@ -56,8 +66,14 @@ public final class AuthenticatedUserContext {
return username == null ? SYSTEM_ACTOR : username;
}
+ public static boolean currentUserIsAdminOrSystem() {
+ Boolean admin = CURRENT_ADMIN.get();
+ return admin == null || admin;
+ }
+
public static void clear() {
CURRENT_USERNAME.remove();
CURRENT_USER_ID.remove();
+ CURRENT_ADMIN.remove();
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
index 004da7f46..a527b24f6 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
@@ -28,6 +28,7 @@ import
org.apache.rocketmq.studio.cluster.nameserver.NameserverRegistryVO;
import org.apache.rocketmq.studio.cluster.nameserver.RestartNameServerDTO;
import org.apache.rocketmq.studio.cluster.nameserver.UpdateNameServerDTO;
import org.apache.rocketmq.studio.cluster.nameserver.UpgradeNameServerDTO;
+import org.apache.rocketmq.studio.cluster.proxy.ProxyVO;
import org.apache.rocketmq.studio.cluster.proxy.RestartProxyDTO;
import org.apache.rocketmq.studio.common.domain.enums.FlushDiskType;
@@ -148,6 +149,19 @@ public class ClusterService {
throw new BusinessException(503, "Cluster details are unavailable: " +
id);
}
+ public List<ProxyVO> listProxies(String clusterId) {
+ ClusterVO cluster = resolveCluster(clusterId);
+ if (cluster.getProxies() == null || cluster.getProxies().isEmpty()) {
+ return List.of();
+ }
+ return List.copyOf(cluster.getProxies());
+ }
+
+ public void requireProxy(String clusterId, String addr) {
+ ClusterVO cluster = resolveCluster(clusterId);
+ requireProxy(cluster, addr);
+ }
+
/**
* Attach live broker configuration (read from the first reachable master
broker via the
* admin API) to a discovered cluster. Falls back to the persisted config,
if any, when the
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressService.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressService.java
index 578d044b2..14a27f9cd 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressService.java
@@ -17,11 +17,9 @@
package org.apache.rocketmq.studio.cluster.proxy;
-import
org.apache.rocketmq.studio.common.util.NoRedirectClientHttpRequestFactory;
-
-
-
+import org.apache.rocketmq.studio.cluster.broker.ClusterService;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import
org.apache.rocketmq.studio.common.util.NoRedirectClientHttpRequestFactory;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
@@ -33,7 +31,6 @@ import org.springframework.web.client.ResourceAccessException;
import org.springframework.web.client.RestTemplate;
import java.time.Duration;
-
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
@@ -76,6 +73,7 @@ public class ProxyAddressService {
/** Bounded pool for the I/O-bound TCP probes; probes queue beyond this
share the budget. */
private static final int PROBE_EXECUTOR_THREADS = 8;
+ private final ClusterService clusterService;
private final Set<String> proxyAddrs = new
LinkedHashSet<>(List.of("127.0.0.1:8081"));
private String currentProxyAddr = "127.0.0.1:8081";
private final RestTemplate restTemplate;
@@ -84,19 +82,33 @@ public class ProxyAddressService {
private final long topologyTotalTimeoutMillis;
@Autowired
- public ProxyAddressService(ProxyHealthProbe healthProbe) {
- this(healthProbe, defaultProbeExecutor(),
TOPOLOGY_TOTAL_TIMEOUT_MILLIS);
+ public ProxyAddressService(ClusterService clusterService, ProxyHealthProbe
healthProbe) {
+ this(clusterService, healthProbe, newRestTemplate(),
defaultProbeExecutor(), TOPOLOGY_TOTAL_TIMEOUT_MILLIS);
+ }
+
+ ProxyAddressService(ClusterService clusterService, ProxyHealthProbe
healthProbe, RestTemplate restTemplate) {
+ this(clusterService, healthProbe, restTemplate,
defaultProbeExecutor(), TOPOLOGY_TOTAL_TIMEOUT_MILLIS);
+ }
+
+ ProxyAddressService(ClusterService clusterService, ProxyHealthProbe
healthProbe,
+ ExecutorService probeExecutor, long
topologyTotalTimeoutMillis) {
+ this(clusterService, healthProbe, newRestTemplate(), probeExecutor,
topologyTotalTimeoutMillis);
}
- ProxyAddressService(ProxyHealthProbe healthProbe, ExecutorService
probeExecutor,
- long topologyTotalTimeoutMillis) {
+ ProxyAddressService(ClusterService clusterService, ProxyHealthProbe
healthProbe, RestTemplate restTemplate,
+ ExecutorService probeExecutor, long
topologyTotalTimeoutMillis) {
+ this.clusterService = clusterService;
this.healthProbe = healthProbe;
+ this.restTemplate = restTemplate;
this.probeExecutor = probeExecutor;
this.topologyTotalTimeoutMillis = topologyTotalTimeoutMillis;
+ }
+
+ private static RestTemplate newRestTemplate() {
NoRedirectClientHttpRequestFactory factory = new
NoRedirectClientHttpRequestFactory();
factory.setConnectTimeout(Duration.ofSeconds(3));
factory.setReadTimeout(Duration.ofSeconds(3));
- this.restTemplate = new RestTemplate(factory);
+ return new RestTemplate(factory);
}
private static ExecutorService defaultProbeExecutor() {
@@ -258,13 +270,10 @@ public class ProxyAddressService {
* POSTs to {@code http://<addr>/admin/reloadConfig}. Throws {@link
BusinessException}
* on transport or protocol failure so the caller receives a structured
error response.
*/
- public void reloadConfig(String addr) {
+ public void reloadConfig(String clusterId, String addr) {
+ String normalizedClusterId = normalizeClusterId(clusterId);
String normalized = normalizeProxyAddr(addr, "addr");
- synchronized (this) {
- if (!proxyAddrs.contains(normalized)) {
- throw new BusinessException(400, "addr is not a registered
proxy address");
- }
- }
+ clusterService.requireProxy(normalizedClusterId, normalized);
String url = "http://"; + normalized + RELOAD_PATH;
try {
ResponseEntity<String> response = restTemplate.postForEntity(url,
null, String.class);
@@ -286,6 +295,13 @@ public class ProxyAddressService {
}
}
+ private String normalizeClusterId(String clusterId) {
+ if (clusterId == null || clusterId.trim().isEmpty()) {
+ throw new BusinessException(400, "clusterId is required");
+ }
+ return clusterId.trim();
+ }
+
private String normalizeProxyAddr(String proxyAddr, String fieldName) {
if (proxyAddr == null || proxyAddr.trim().isEmpty()) {
throw new BusinessException(400, fieldName + " is required");
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyController.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyController.java
index 34cba7970..423ff8b30 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyController.java
@@ -42,10 +42,7 @@ public class ProxyController {
@GetMapping
public Result<List<ProxyVO>> listProxies(@RequestParam(required = false)
String clusterId) {
requireClusterId(clusterId);
- List<ProxyVO> proxies =
proxyAddressService.getHomePage().getProxyAddrList().stream()
- .map(addr -> ProxyVO.builder().addr(addr).build())
- .toList();
- return Result.ok(proxies);
+ return Result.ok(clusterService.listProxies(clusterId));
}
@GetMapping("/topology")
@@ -55,7 +52,7 @@ public class ProxyController {
@PostMapping("/config/reload")
public Result<Map<String, Boolean>> reloadProxyConfig(@Valid @RequestBody
RestartProxyDTO command) {
- proxyAddressService.reloadConfig(command.getAddr());
+ proxyAddressService.reloadConfig(command.getClusterId(),
command.getAddr());
return Result.ok(Map.of("success", true));
}
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 cf9f43bd3..fc324929b 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
@@ -26,6 +26,8 @@ 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.RPCHook;
+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.apache.rocketmq.studio.common.domain.enums.DeliveryStatus;
@@ -53,6 +55,7 @@ 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;
@@ -75,6 +78,7 @@ public class RocketMQMessageProvider implements
MessageProvider {
private static final int TRACE_QUERY_MAX = 64;
private static final int DEFAULT_TOPIC_LIMIT = 200;
private static final int TOPIC_QUERY_HARD_CAP = 2000;
+ private static final int TOPIC_PULL_BATCH_SIZE = 32;
private static final int MAX_BODY_DISPLAY_BYTES = 64 * 1024;
private static final int MAX_BINARY_BODY_DISPLAY_BYTES = 48 * 1024;
private static final int MAX_PROPERTIES = 64;
@@ -85,6 +89,8 @@ public class RocketMQMessageProvider implements
MessageProvider {
private static final long MAX_TOPIC_QUERY_WINDOW_MILLIS = 7 *
ONE_DAY_MILLIS;
private static final int MAX_PULLS_PER_QUEUE = 1_000;
private static final int MAX_CONSECUTIVE_OFFSET_ILLEGAL = 3;
+ private static final int MAX_TOPIC_SCAN_MESSAGES_PER_QUEUE =
MAX_PULLS_PER_QUEUE * TOPIC_PULL_BATCH_SIZE;
+ private static final int MAX_PULL_ATTEMPTS_PER_QUEUE = MAX_PULLS_PER_QUEUE
+ MAX_CONSECUTIVE_OFFSET_ILLEGAL;
private static final Comparator<MessageRecordVO> TOPIC_QUERY_ORDER =
Comparator
.comparingLong(MessageRecordVO::getStoreTime)
.thenComparing(MessageRecordVO::getMsgId,
Comparator.nullsFirst(String::compareTo));
@@ -155,12 +161,10 @@ public class RocketMQMessageProvider implements
MessageProvider {
private MessageExt viewMessageByOffsetId(DefaultMQAdminExt adminExt,
String topic, String msgId) {
try {
MessageId messageId = MessageDecoder.decodeMessageId(msgId);
- SocketAddress address = messageId.getAddress();
- if (!(address instanceof InetSocketAddress)) {
+ String brokerAddr = validatedBrokerAddr(adminExt, msgId,
messageId);
+ if (!StringUtils.hasText(brokerAddr)) {
return null;
}
- InetSocketAddress inet = (InetSocketAddress) address;
- String brokerAddr = inet.getAddress().getHostAddress() + ":" +
inet.getPort();
return adminExt.getDefaultMQAdminExtImpl()
.getMqClientInstance()
.getMQClientAPIImpl()
@@ -171,6 +175,51 @@ 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;
+ }
+
+ 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 {
@@ -207,16 +256,23 @@ public class RocketMQMessageProvider implements
MessageProvider {
return Collections.emptyList();
}
for (MessageQueue queue : queues) {
- long minOffset = consumer.searchOffset(queue, begin);
- long maxOffset = consumer.searchOffset(queue, end);
+ TopicQueueScanPlan scanPlan =
buildTopicQueueScanPlan(consumer, queue, begin, end);
+ if (scanPlan.isEmpty()) {
+ continue;
+ }
+ if (scanPlan.truncated()) {
+ log.info("Truncate topic query for {} queue {} to offsets
[{}..{}) within the guarded tail budget",
+ topic, queue, scanPlan.startOffset(),
scanPlan.endOffsetExclusive());
+ }
int consecutiveIllegalOffsets = 0;
int pullAttempts = 0;
- for (long offset = minOffset; offset <= maxOffset; ) {
- if (++pullAttempts > MAX_PULLS_PER_QUEUE) {
- throw new BusinessException(400,
- "Topic message query exceeded the per-queue
pull budget; narrow the time range");
+ for (long offset = scanPlan.startOffset(); offset <
scanPlan.endOffsetExclusive(); ) {
+ if (++pullAttempts > MAX_PULL_ATTEMPTS_PER_QUEUE) {
+ log.warn("Stop topic query for {} because queue {}
exhausted the guarded pull budget at offset {}",
+ topic, queue, offset);
+ break;
}
- PullResult pullResult = consumer.pull(queue, "*", offset,
32);
+ PullResult pullResult = consumer.pull(queue, "*", offset,
TOPIC_PULL_BATCH_SIZE);
if (pullResult == null) {
log.warn("Stop topic query for {} because queue {}
returned no pull result", topic, queue);
break;
@@ -226,7 +282,7 @@ public class RocketMQMessageProvider implements
MessageProvider {
log.warn("Stop topic query for {} because queue {} did
not advance offset {}", topic, queue, offset);
break;
}
- offset = nextOffset;
+ offset = Math.min(nextOffset,
scanPlan.endOffsetExclusive());
if (pullResult.getPullStatus() ==
PullStatus.OFFSET_ILLEGAL) {
// The broker returned a corrected offset in
nextBeginOffset because
// the requested offset is no longer valid (expired,
compacted, or
@@ -260,8 +316,6 @@ public class RocketMQMessageProvider implements
MessageProvider {
}
}
}
- } catch (BusinessException exception) {
- throw exception;
} catch (Exception e) {
log.warn("queryByTopic(topic={}) failed: {}", topic,
e.getMessage());
throw new BusinessException(502, "Failed to query messages by
topic: " + e.getMessage());
@@ -273,6 +327,33 @@ public class RocketMQMessageProvider implements
MessageProvider {
.toList();
}
+ private TopicQueueScanPlan buildTopicQueueScanPlan(DefaultMQPullConsumer
consumer, MessageQueue queue,
+ long begin, long end)
throws Exception {
+ long minOffset = consumer.minOffset(queue);
+ long maxOffsetExclusive = consumer.maxOffset(queue);
+ if (maxOffsetExclusive <= minOffset) {
+ return TopicQueueScanPlan.empty();
+ }
+ long windowStartOffset = clampOffset(consumer.searchOffset(queue,
begin), minOffset, maxOffsetExclusive);
+ long windowEndOffsetExclusive =
clampOffset(consumer.searchOffset(queue, inclusiveUpperBound(end)),
+ windowStartOffset, maxOffsetExclusive);
+ if (windowEndOffsetExclusive <= windowStartOffset) {
+ return TopicQueueScanPlan.empty();
+ }
+ long budgetedStartOffset = Math.max(windowStartOffset,
+ windowEndOffsetExclusive - MAX_TOPIC_SCAN_MESSAGES_PER_QUEUE);
+ return new TopicQueueScanPlan(budgetedStartOffset,
windowEndOffsetExclusive,
+ budgetedStartOffset > windowStartOffset);
+ }
+
+ private long clampOffset(long offset, long minOffset, long
maxOffsetExclusive) {
+ return Math.max(minOffset, Math.min(offset, maxOffsetExclusive));
+ }
+
+ private long inclusiveUpperBound(long timestamp) {
+ return timestamp == Long.MAX_VALUE ? Long.MAX_VALUE : timestamp + 1;
+ }
+
private void addTopicQueryCandidate(PriorityQueue<MessageRecordVO>
newestMessages,
MessageRecordVO candidate, int
resultLimit) {
if (resultLimit <= 0) {
@@ -633,4 +714,14 @@ public class RocketMQMessageProvider implements
MessageProvider {
private static boolean parseBoolean(String value) {
return "true".equalsIgnoreCase(value == null ? "" : value.trim());
}
+
+ private record TopicQueueScanPlan(long startOffset, long
endOffsetExclusive, boolean truncated) {
+ private static TopicQueueScanPlan empty() {
+ return new TopicQueueScanPlan(0L, 0L, false);
+ }
+
+ private boolean isEmpty() {
+ return endOffsetExclusive <= startOffset;
+ }
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/settings/GeneralSettingsVO.java
b/server/src/main/java/org/apache/rocketmq/studio/settings/GeneralSettingsVO.java
index d62ea19c7..37ac02dcc 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/settings/GeneralSettingsVO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/settings/GeneralSettingsVO.java
@@ -25,7 +25,7 @@ import lombok.ToString;
import org.springframework.util.StringUtils;
@Data
-@Builder
+@Builder(toBuilder = true)
@NoArgsConstructor
@AllArgsConstructor
public class GeneralSettingsVO {
@@ -57,4 +57,14 @@ public class GeneralSettingsVO {
public boolean isApiKeyConfigured() {
return StringUtils.hasText(apiKey);
}
+
+ @JsonProperty(value = "dingtalkWebhookConfigured", access =
JsonProperty.Access.READ_ONLY)
+ public boolean isDingtalkWebhookConfigured() {
+ return StringUtils.hasText(dingtalkWebhook);
+ }
+
+ @JsonProperty(value = "smsWebhookConfigured", access =
JsonProperty.Access.READ_ONLY)
+ public boolean isSmsWebhookConfigured() {
+ return StringUtils.hasText(smsWebhook);
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/settings/SettingsService.java
b/server/src/main/java/org/apache/rocketmq/studio/settings/SettingsService.java
index 03388394a..ecb125dcc 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/settings/SettingsService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/settings/SettingsService.java
@@ -19,6 +19,7 @@ package org.apache.rocketmq.studio.settings;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.studio.auth.AuthenticatedUserContext;
import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.apache.rocketmq.studio.cluster.metrics.MetricsBackendType;
import org.apache.rocketmq.studio.common.exception.BusinessException;
@@ -52,6 +53,7 @@ import java.util.Set;
@Service
public class SettingsService {
+ private static final String REDACTED_NOTIFICATION_WEBHOOK = "******";
private static final List<byte[]> CLOUD_METADATA_ADDRESSES = List.of(
new byte[] {
(byte) 0xfd, 0x00, 0x0e, (byte) 0xc2,
@@ -98,7 +100,25 @@ public class SettingsService {
public GeneralSettingsVO getGeneralSettings() {
log.debug("Loading general settings");
- return settingsRepository.loadGeneralSettings();
+ GeneralSettingsVO settings = settingsRepository.loadGeneralSettings();
+ if (AuthenticatedUserContext.currentUserIsAdminOrSystem()) {
+ return settings;
+ }
+ return redactNotificationWebhooks(settings);
+ }
+
+ private GeneralSettingsVO redactNotificationWebhooks(GeneralSettingsVO
settings) {
+ if (settings == null) {
+ return null;
+ }
+ return settings.toBuilder()
+
.dingtalkWebhook(StringUtils.hasText(settings.getDingtalkWebhook())
+ ? REDACTED_NOTIFICATION_WEBHOOK
+ : settings.getDingtalkWebhook())
+ .smsWebhook(StringUtils.hasText(settings.getSmsWebhook())
+ ? REDACTED_NOTIFICATION_WEBHOOK
+ : settings.getSmsWebhook())
+ .build();
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthCredentialAuthorizationIntegrationTest.java
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthCredentialAuthorizationIntegrationTest.java
index 4615078cd..0324ae0d8 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthCredentialAuthorizationIntegrationTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthCredentialAuthorizationIntegrationTest.java
@@ -31,6 +31,8 @@ import org.springframework.context.annotation.Import;
import org.springframework.http.HttpHeaders;
import org.springframework.test.web.servlet.MockMvc;
+import java.util.Optional;
+
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.verify;
@@ -72,6 +74,7 @@ class AuthCredentialAuthorizationIntegrationTest {
when(authProperties.isLoginRequired()).thenReturn(true);
when(authService.isAuthenticated(AUTHORIZATION)).thenReturn(true);
when(authService.isAdmin(AUTHORIZATION)).thenReturn(false);
+
when(authService.getAuthenticatedUser(AUTHORIZATION)).thenReturn(Optional.of(user(false)));
}
@Test
@@ -95,6 +98,7 @@ class AuthCredentialAuthorizationIntegrationTest {
@Test
void shouldAllowCredentialPathsWithMatrixParametersForAdministrator()
throws Exception {
when(authService.isAdmin(AUTHORIZATION)).thenReturn(true);
+
when(authService.getAuthenticatedUser(AUTHORIZATION)).thenReturn(Optional.of(user(true)));
mockMvc.perform(get("/api/acl/users/user-1/credentials;probe=1")
.header(HttpHeaders.AUTHORIZATION, AUTHORIZATION))
@@ -106,4 +110,12 @@ class AuthCredentialAuthorizationIntegrationTest {
verify(aclService).getUserCredentials(eq("user-1"), isNull());
verify(cloudCredentialService).reveal(12L);
}
+
+ private LoginVO.UserInfo user(boolean admin) {
+ return LoginVO.UserInfo.builder()
+ .userId(1L)
+ .username(admin ? "admin" : "reader")
+ .admin(admin)
+ .build();
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthInterceptorTest.java
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthInterceptorTest.java
index a2acb6a15..5fc95a5fe 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthInterceptorTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthInterceptorTest.java
@@ -99,6 +99,26 @@ class AuthInterceptorTest {
.isEqualTo(AuthenticatedUserContext.SYSTEM_ACTOR);
}
+ @Test
+ void shouldTrackAuthenticatedAdminStateInUserContext() throws Exception {
+ TestSession session = login(true);
+ MockHttpServletRequest request = authenticatedRequest("GET",
"/api/clusters", session.token());
+ MockHttpServletResponse response = new MockHttpServletResponse();
+ Object handler = new Object();
+
+ boolean allowed = session.interceptor().preHandle(request, response,
handler);
+
+ assertThat(allowed).isTrue();
+
assertThat(AuthenticatedUserContext.currentUsernameOrSystem()).isEqualTo("test-user");
+
assertThat(AuthenticatedUserContext.currentUserIsAdminOrSystem()).isTrue();
+
+ session.interceptor().afterCompletion(request, response, handler,
null);
+
+ assertThat(AuthenticatedUserContext.currentUsernameOrSystem())
+ .isEqualTo(AuthenticatedUserContext.SYSTEM_ACTOR);
+
assertThat(AuthenticatedUserContext.currentUserIsAdminOrSystem()).isTrue();
+ }
+
@Test
void shouldEnforceLoginWhenDatabaseRequiresItEvenIfPropertyIsDisabled()
throws Exception {
AuthProperties properties = new AuthProperties();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthServiceDatabaseTest.java
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthServiceDatabaseTest.java
index 5327cd3a1..9ef733e6a 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthServiceDatabaseTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthServiceDatabaseTest.java
@@ -159,6 +159,19 @@ class AuthServiceDatabaseTest {
assertThat(authService.login(request).getUser().getUsername()).isEqualTo("operator");
}
+ @Test
+ void createUserShouldReturnConflictWhenConcurrentInsertWins() {
+ when(userMapper.selectOne(any(Wrapper.class))).thenReturn(null);
+ when(userMapper.insert(any(RmqStudioUser.class)))
+ .thenThrow(new
org.springframework.dao.DuplicateKeyException("duplicate username"));
+
+ assertThatThrownBy(() -> authService.createUser("operator",
"password-1", false))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Username is already in use")
+ .satisfies(exception ->
+ assertThat(((BusinessException)
exception).getCode()).isEqualTo(409));
+ }
+
@Test
void repeatedFailedLoginsAreRejectedWithTooManyRequestsTest() {
when(userMapper.selectCount(isNull())).thenReturn(1L);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceTest.java
index 976d5e747..23d850555 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceTest.java
@@ -202,6 +202,25 @@ class ClusterServiceTest {
.satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(503));
}
+ @Test
+ void listProxiesShouldUseResolvedCluster() {
+
when(clusterProvider.refreshClusterDetail("cluster-1")).thenReturn(sampleCluster);
+
+ List<ProxyVO> proxies = clusterService.listProxies("cluster-1");
+
+
assertThat(proxies).extracting(ProxyVO::getAddr).containsExactly("10.0.0.10:8081");
+ }
+
+ @Test
+ void requireProxyShouldRejectUnknownAddress() {
+
when(clusterProvider.refreshClusterDetail("cluster-1")).thenReturn(sampleCluster);
+
+ assertThatThrownBy(() -> clusterService.requireProxy("cluster-1",
"127.0.0.1:8081"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Proxy not found: 127.0.0.1:8081")
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(404));
+ }
+
@Test
void updateConfigShouldUpdateFlushDiskType() {
when(clusterRepository.findById("cluster-1")).thenReturn(Optional.of(sampleCluster));
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MultiBackendMetricsSourceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MultiBackendMetricsSourceTest.java
index a6bdb5f7d..cd49a949c 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MultiBackendMetricsSourceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MultiBackendMetricsSourceTest.java
@@ -20,9 +20,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpServer;
import org.apache.rocketmq.studio.model.MetricsDataSourceConfig;
-import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -30,17 +28,8 @@ import org.junit.jupiter.params.provider.EnumSource;
import org.springframework.web.client.RestClient;
import java.io.IOException;
-import java.net.InetAddress;
import java.net.InetSocketAddress;
-import java.net.InterfaceAddress;
-import java.net.NetworkInterface;
-import java.net.Proxy;
-import java.net.ProxySelector;
-import java.net.SocketAddress;
-import java.net.URI;
import java.nio.charset.StandardCharsets;
-import java.util.Enumeration;
-import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
@@ -52,38 +41,13 @@ class MultiBackendMetricsSourceTest {
private String baseUrl;
private final MetricsSourceFactory factory =
new MetricsSourceFactory(RestClient.builder(), new ObjectMapper());
-
- private static ProxySelector originalProxySelector;
-
- @BeforeAll
- static void bypassJvmProxy() {
- // The IDE (e.g. IDEA with a PAC proxy) may inject a proxy into the
test JVM. The embedded
- // server is bound to a site-local address that is not in
http.nonProxyHosts, so the request
- // would be routed through the proxy and time out. Force a direct
connection for this test.
- originalProxySelector = ProxySelector.getDefault();
- ProxySelector.setDefault(new ProxySelector() {
- @Override
- public List<Proxy> select(URI uri) {
- return List.of(Proxy.NO_PROXY);
- }
-
- @Override
- public void connectFailed(URI uri, SocketAddress socketAddress,
IOException exception) {
- // Nothing to do; the test never relies on a proxy.
- }
- });
- }
-
- @AfterAll
- static void restoreJvmProxy() {
- ProxySelector.setDefault(originalProxySelector);
- }
+ private final RestClient.Builder restClientBuilder = RestClient.builder();
+ private final ObjectMapper objectMapper = new ObjectMapper();
@BeforeEach
void setUp() throws IOException {
- java.net.InetAddress bindAddress = findSiteLocalAddress();
- server = HttpServer.create(new InetSocketAddress(bindAddress, 0), 0);
- baseUrl = "http://"; + bindAddress.getHostAddress() + ":" +
server.getAddress().getPort();
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ baseUrl = "http://127.0.0.1:"; + server.getAddress().getPort();
server.start();
}
@@ -105,8 +69,7 @@ class MultiBackendMetricsSourceTest {
""".formatted(backendType.name()));
});
- MetricsDataSourceConfig config = configFor(backendType);
- MetricsSource source = factory.create(config);
+ MetricsSource source = testSourceFor(backendType);
MetricDataVO result = source.query(query());
assertThat(requestPath.get()).isEqualTo(backendType.getQueryPath());
@@ -179,7 +142,7 @@ class MultiBackendMetricsSourceTest {
""");
});
- factory.create(configWithAuth(authType, username, password,
bearerToken)).query(query());
+ loopbackPrometheusSource(configWithAuth(authType, username, password,
bearerToken)).query(query());
assertThat(authorization.get()).isEqualTo(expectedAuthorization);
}
@@ -187,7 +150,7 @@ class MultiBackendMetricsSourceTest {
private void assertAuthenticationFailure(String authType, String username,
String password,
String bearerToken, String
message) {
assertThatExceptionOfType(PrometheusException.class)
- .isThrownBy(() -> factory.create(configWithAuth(authType,
username, password, bearerToken))
+ .isThrownBy(() ->
loopbackPrometheusSource(configWithAuth(authType, username, password,
bearerToken))
.query(query()))
.satisfies(exception -> {
assertThat(exception.getStatusCode()).isEqualTo(503);
@@ -195,6 +158,26 @@ class MultiBackendMetricsSourceTest {
});
}
+ private MetricsSource loopbackPrometheusSource(MetricsDataSourceConfig
config) {
+ MetricsSourceSettings settings = MetricsSourceSettings.builder()
+ .backendType(MetricsBackendType.PROMETHEUS)
+ .baseUrl(config.getUrl())
+ .authType(config.getAuthType())
+ .username(config.getUsername())
+ .password(config.getPassword())
+ .bearerToken(config.getBearerToken())
+ .build();
+ return new PrometheusMetricsSource(restClientBuilder, objectMapper,
settings) {
+ @Override
+ protected void validateQueryHost(String url) {
+ if (url != null && url.startsWith(baseUrl)) {
+ return;
+ }
+ super.validateQueryHost(url);
+ }
+ };
+ }
+
private MetricsDataSourceConfig configWithAuth(String authType, String
username,
String password, String
bearerToken) {
MetricsDataSourceConfig config =
configFor(MetricsBackendType.PROMETHEUS);
@@ -213,6 +196,31 @@ class MultiBackendMetricsSourceTest {
return config;
}
+ private MetricsSource testSourceFor(MetricsBackendType backendType) {
+ return loopbackSource(backendType);
+ }
+
+ private AbstractPrometheusCompatibleMetricsSource
loopbackSource(MetricsBackendType backendType) {
+ return new
AbstractPrometheusCompatibleMetricsSource(restClientBuilder, objectMapper,
+ MetricsSourceSettings.builder()
+ .backendType(backendType)
+ .baseUrl(baseUrl)
+ .build()) {
+ @Override
+ protected MetricsBackendType backendType() {
+ return backendType;
+ }
+
+ @Override
+ protected void validateQueryHost(String url) {
+ if (url != null && url.startsWith(baseUrl)) {
+ return;
+ }
+ super.validateQueryHost(url);
+ }
+ };
+ }
+
private MetricQueryDTO query() {
return MetricQueryDTO.builder()
.metric("up")
@@ -230,34 +238,4 @@ class MultiBackendMetricsSourceTest {
exchange.close();
}
- private static java.net.InetAddress findSiteLocalAddress() throws
java.net.SocketException {
- InetAddress fallback = null;
- Enumeration<NetworkInterface> interfaces =
NetworkInterface.getNetworkInterfaces();
- while (interfaces.hasMoreElements()) {
- NetworkInterface iface = interfaces.nextElement();
- if (!iface.isUp() || iface.isLoopback()) {
- continue;
- }
- for (InterfaceAddress address : iface.getInterfaceAddresses()) {
- InetAddress inet = address.getAddress();
- if (inet instanceof java.net.Inet4Address
- && !inet.isLoopbackAddress()
- && !inet.isLinkLocalAddress()) {
- if (inet.isSiteLocalAddress()) {
- return inet;
- }
- if (fallback == null) {
- fallback = inet;
- }
- }
- }
- }
- // No site-local interface (e.g. hosts that only expose public
ranges): fall back to any
- // non-loopback IPv4 so the embedded server stays reachable under the
SSRF guard, which
- // rejects loopback/link-local addresses (see UrlHostGuard).
- if (fallback != null) {
- return fallback;
- }
- return java.net.InetAddress.getLoopbackAddress();
- }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressServiceTest.java
index 90776ba06..21eacfc3d 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressServiceTest.java
@@ -17,9 +17,15 @@
package org.apache.rocketmq.studio.cluster.proxy;
+import org.apache.rocketmq.studio.cluster.broker.ClusterService;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.client.RestTemplate;
import java.util.List;
@@ -28,16 +34,29 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
+@ExtendWith(MockitoExtension.class)
class ProxyAddressServiceTest {
+ @Mock
+ private ClusterService clusterService;
+
+ @Mock
+ private RestTemplate restTemplate;
+
private final ProxyHealthProbe healthProbe = mock(ProxyHealthProbe.class);
- private final ProxyAddressService proxyAddressService = new
ProxyAddressService(healthProbe);
+ private ProxyAddressService proxyAddressService;
@BeforeEach
void setUp() {
+ proxyAddressService = new ProxyAddressService(clusterService,
healthProbe, restTemplate);
// Default probe outcome: everything reachable with 1 ms latency.
when(healthProbe.probe(anyString(), anyInt(), anyInt()))
.thenReturn(ProxyHealthProbe.ProbeResult.reachable(1L));
@@ -141,11 +160,41 @@ class ProxyAddressServiceTest {
}
@Test
- void reloadConfigShouldRejectUnregisteredAddressTest() {
- assertThatThrownBy(() ->
proxyAddressService.reloadConfig("10.0.0.1:8081"))
+ void reloadConfigShouldRejectLoopbackOutsideTrustedCluster() {
+ doThrow(new BusinessException(404, "Proxy not found: 127.0.0.2:8081"))
+ .when(clusterService).requireProxy("cluster-1",
"127.0.0.2:8081");
+
+ assertThatThrownBy(() -> proxyAddressService.reloadConfig("cluster-1",
"127.0.0.2:8081"))
.isInstanceOf(BusinessException.class)
- .hasMessage("addr is not a registered proxy address")
- .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(400));
+ .hasMessage("Proxy not found: 127.0.0.2:8081")
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(404));
+
+ verifyNoInteractions(restTemplate);
+ }
+
+ @Test
+ void reloadConfigShouldRejectNonTrustedEndpoint() {
+ doThrow(new BusinessException(404, "Proxy not found:
198.51.100.10:8081"))
+ .when(clusterService).requireProxy("cluster-1",
"198.51.100.10:8081");
+
+ assertThatThrownBy(() -> proxyAddressService.reloadConfig("cluster-1",
"198.51.100.10:8081"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Proxy not found: 198.51.100.10:8081")
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(404));
+
+ verifyNoInteractions(restTemplate);
+ }
+
+ @Test
+ void reloadConfigShouldPostToTrustedDiscoveredProxy() {
+ doNothing().when(clusterService).requireProxy("cluster-1",
"10.0.0.10:8081");
+
when(restTemplate.postForEntity(eq("http://10.0.0.10:8081/admin/reloadConfig";),
isNull(), eq(String.class)))
+ .thenReturn(ResponseEntity.ok("ok"));
+
+ proxyAddressService.reloadConfig("cluster-1", "10.0.0.10:8081");
+
+ verify(clusterService).requireProxy("cluster-1", "10.0.0.10:8081");
+
verify(restTemplate).postForEntity(eq("http://10.0.0.10:8081/admin/reloadConfig";),
isNull(), eq(String.class));
}
@Test
@@ -216,7 +265,7 @@ class ProxyAddressServiceTest {
}
return ProxyHealthProbe.ProbeResult.reachable(300L);
};
- ProxyAddressService service = new ProxyAddressService(slowProbe);
+ ProxyAddressService service = new ProxyAddressService(clusterService,
slowProbe);
service.addProxyAddr("10.0.0.11:8081");
service.addProxyAddr("10.0.0.12:8081");
service.addProxyAddr("10.0.0.13:8081");
@@ -246,7 +295,7 @@ class ProxyAddressServiceTest {
};
java.util.concurrent.ExecutorService executor =
java.util.concurrent.Executors.newFixedThreadPool(4);
- ProxyAddressService service = new ProxyAddressService(selectiveProbe,
executor, 500L);
+ ProxyAddressService service = new ProxyAddressService(clusterService,
selectiveProbe, executor, 500L);
service.addProxyAddr("10.0.0.21:8081");
service.addProxyAddr("10.0.0.22:8081");
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyControllerTest.java
index 01c2c3ee0..7133a8d8b 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyControllerTest.java
@@ -29,6 +29,7 @@ import org.springframework.test.web.servlet.MockMvc;
import java.util.List;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@@ -105,18 +106,15 @@ class ProxyControllerTest {
@Test
void listProxiesShouldReturnProxiesForCluster() throws Exception {
- when(proxyAddressService.getHomePage())
- .thenReturn(ProxyHomeVO.builder()
- .proxyAddrList(List.of("127.0.0.1:8081"))
- .currentProxyAddr("127.0.0.1:8081")
- .build());
+ when(clusterService.listProxies("cluster-1"))
+
.thenReturn(List.of(ProxyVO.builder().addr("10.0.0.10:8081").build()));
mockMvc.perform(get("/api/proxies").param("clusterId", "cluster-1"))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(200))
- .andExpect(jsonPath("$.data[0].addr").value("127.0.0.1:8081"));
+ .andExpect(jsonPath("$.data[0].addr").value("10.0.0.10:8081"));
- verify(proxyAddressService).getHomePage();
+ verify(clusterService).listProxies("cluster-1");
}
@Test
@@ -176,7 +174,7 @@ class ProxyControllerTest {
.andExpect(jsonPath("$.code").value(200))
.andExpect(jsonPath("$.data.success").value(true));
- verify(proxyAddressService).reloadConfig("127.0.0.1:8081");
+ verify(proxyAddressService).reloadConfig(eq("cluster-1"),
eq("127.0.0.1:8081"));
}
@Test
@@ -194,4 +192,4 @@ class ProxyControllerTest {
verifyNoInteractions(proxyAddressService);
}
-}
\ No newline at end of file
+}
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 c4206909d..383b6830f 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
@@ -578,7 +578,9 @@ class RocketMQDLQProviderTest {
}
verify(auditService).record(
eq("RESEND_DLQ"),
+ eq("DLQ"),
eq("group-a"),
+ eq(null),
contains("scanTruncated=true"),
eq("PARTIAL"));
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
index 126d72d25..2eb00365b 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
@@ -28,6 +28,8 @@ 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.RPCHook;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.exception.BusinessException;
@@ -49,11 +51,14 @@ import org.mockito.junit.jupiter.MockitoExtension;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
+import java.util.stream.LongStream;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -84,12 +89,22 @@ class RocketMQMessageProviderTest {
private RocketMQMessageProvider provider;
@BeforeEach
- void setUp() {
+ void setUp() throws Exception {
lenient().when(runtimeAdminClientResolver.resolveEndpoint("instance-a")).thenReturn("namesrv-a:9876");
lenient().when(runtimeAdminClientResolver.execute(anyString(),
any())).thenAnswer(invocation -> {
MqAdminExtFactory.AdminAction<Object> action =
invocation.getArgument(1);
- return action == null ? null : action.apply(adminExt);
+ if (action == null) {
+ return null;
+ }
+ try {
+ return action.apply(adminExt);
+ } catch (InterruptedException exception) {
+ Thread.currentThread().interrupt();
+ throw exception;
+ }
});
+ lenient().when(adminExt.examineBrokerClusterInfo())
+
.thenReturn(clusterInfoWithBrokerAddresses("172.30.10.100:10911"));
provider = new RocketMQMessageProvider(runtimeAdminClientResolver);
}
@@ -213,6 +228,20 @@ class RocketMQMessageProviderTest {
verify(clientApi).viewMessage("172.30.10.100:10911", "TopicA",
27521713L, 3000L);
}
+ @Test
+ void queryByMsgIdRejectsDecodedBrokerOutsideKnownTopology() throws
Exception {
+ String msgId = MessageDecoder.createMessageId(new
InetSocketAddress("10.2.3.4", 10911), 12345L);
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithBrokerAddresses("172.30.10.100:10911"));
+ when(adminExt.viewMessage("TopicA", msgId))
+ .thenThrow(new IllegalStateException("primary lookup failed"));
+
+ List<MessageRecordVO> result = provider.queryMessages(
+ "instance-a", "TopicA", msgId, null, null, 100L, 200L);
+
+ assertThat(result).isEmpty();
+ verify(adminExt, never()).getDefaultMQAdminExtImpl();
+ }
+
@Test
void queryByTopicSurfacesPullConsumerFailure() throws Exception {
try (MockedConstruction<DefaultMQPullConsumer> ignored =
@@ -237,7 +266,7 @@ class RocketMQMessageProviderTest {
mockConstruction(DefaultMQPullConsumer.class, (consumer,
context) -> {
doNothing().when(consumer).start();
when(consumer.fetchSubscribeMessageQueues("TopicA")).thenReturn(Set.of(queue));
- when(consumer.searchOffset(eq(queue),
anyLong())).thenReturn(10L);
+ mockQueueWindow(consumer, queue, 100L, 200L, 10L,
10L, 11L, 11L);
when(consumer.pull(eq(queue), eq("*"), eq(10L),
eq(32))).thenReturn(stalledResult);
doNothing().when(consumer).shutdown();
})) {
@@ -262,7 +291,8 @@ class RocketMQMessageProviderTest {
doNothing().when(consumer).start();
when(consumer.fetchSubscribeMessageQueues("TopicA"))
.thenReturn(new
LinkedHashSet<>(List.of(olderQueue, newerQueue)));
- when(consumer.searchOffset(any(MessageQueue.class),
anyLong())).thenReturn(10L);
+ mockQueueWindow(consumer, olderQueue, 100L, 300L,
10L, 10L, 11L, 11L);
+ mockQueueWindow(consumer, newerQueue, 100L, 300L,
10L, 10L, 11L, 11L);
when(consumer.pull(olderQueue, "*", 10L,
32)).thenReturn(olderPullResult);
when(consumer.pull(newerQueue, "*", 10L,
32)).thenReturn(newerPullResult);
doNothing().when(consumer).shutdown();
@@ -290,8 +320,7 @@ class RocketMQMessageProviderTest {
mockConstruction(DefaultMQPullConsumer.class, (consumer,
context) -> {
doNothing().when(consumer).start();
when(consumer.fetchSubscribeMessageQueues("TopicA")).thenReturn(Set.of(queue));
- when(consumer.searchOffset(queue,
100L)).thenReturn(10L);
- when(consumer.searchOffset(queue,
200L)).thenReturn(50L);
+ mockQueueWindow(consumer, queue, 100L, 200L, 10L,
10L, 50L, 50L);
when(consumer.pull(eq(queue), eq("*"), eq(10L),
eq(32))).thenReturn(illegalOffset);
when(consumer.pull(eq(queue), eq("*"), eq(20L),
eq(32))).thenReturn(foundAfterCorrection);
when(consumer.pull(eq(queue), eq("*"), eq(40L),
eq(32))).thenReturn(endOfQueue);
@@ -323,7 +352,8 @@ class RocketMQMessageProviderTest {
doNothing().when(consumer).start();
when(consumer.fetchSubscribeMessageQueues("TopicA"))
.thenReturn(new
LinkedHashSet<>(List.of(olderQueue, newerQueue)));
- when(consumer.searchOffset(any(MessageQueue.class),
anyLong())).thenReturn(10L);
+ mockQueueWindow(consumer, olderQueue, 100L, 300L,
10L, 10L, 11L, 11L);
+ mockQueueWindow(consumer, newerQueue, 100L, 300L,
10L, 10L, 11L, 11L);
when(consumer.pull(olderQueue, "*", 10L,
32)).thenReturn(olderPullResult);
when(consumer.pull(newerQueue, "*", 10L,
32)).thenReturn(newerPullResult);
doNothing().when(consumer).shutdown();
@@ -351,7 +381,8 @@ class RocketMQMessageProviderTest {
doNothing().when(consumer).start();
when(consumer.fetchSubscribeMessageQueues("TopicA"))
.thenReturn(new
LinkedHashSet<>(List.of(firstQueue, secondQueue)));
- when(consumer.searchOffset(any(MessageQueue.class),
anyLong())).thenReturn(10L);
+ mockQueueWindow(consumer, firstQueue, 100L, 300L,
10L, 10L, 11L, 11L);
+ mockQueueWindow(consumer, secondQueue, 100L, 300L,
10L, 10L, 11L, 11L);
when(consumer.pull(firstQueue, "*", 10L,
32)).thenReturn(firstPullResult);
when(consumer.pull(secondQueue, "*", 10L,
32)).thenReturn(secondPullResult);
doNothing().when(consumer).shutdown();
@@ -580,6 +611,83 @@ class RocketMQMessageProviderTest {
storeTimestamp + 24 * 60 * 60 * 1000L);
}
+ @Test
+
+ void getMessageTraceDoesNotUseDecodedBrokerOutsideKnownTopology() throws
Exception {
+ String msgId = MessageDecoder.createMessageId(new
InetSocketAddress("10.2.3.4", 10911), 12345L);
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithBrokerAddresses("172.30.10.100:10911"));
+ when(adminExt.viewMessage("TopicA", msgId))
+ .thenThrow(new IllegalStateException("topic lookup failed"));
+ when(adminExt.queryMessage(anyString(), anyString(), anyInt(),
anyLong(), anyLong()))
+ .thenReturn(new QueryResult(0L, List.of()));
+
+ provider.getMessageTrace("instance-a", msgId, "TopicA");
+
+ verify(adminExt, never()).getDefaultMQAdminExtImpl();
+ ArgumentCaptor<Long> beginCaptor = ArgumentCaptor.forClass(Long.class);
+ ArgumentCaptor<Long> endCaptor = ArgumentCaptor.forClass(Long.class);
+ verify(adminExt).queryMessage(eq("RMQ_SYS_TRACE_TOPIC"), eq(msgId),
eq(64),
+ beginCaptor.capture(), endCaptor.capture());
+ assertThat(endCaptor.getValue() -
beginCaptor.getValue()).isBetween(3_660_000L, 3_670_000L);
+ }
+
+ @Test
+ void queryByTopicReturnsLatestMessagesWhenWindowExceedsLegacyPullLimit()
throws Exception {
+ MessageQueue queue = new MessageQueue("TopicA", "broker-a", 0);
+ long begin = 10_000L;
+ long end = 49_999L;
+ long maxOffsetExclusive = 40_000L;
+ try (MockedConstruction<DefaultMQPullConsumer> mockedConsumers =
+ mockConstruction(DefaultMQPullConsumer.class, (consumer,
context) -> {
+ doNothing().when(consumer).start();
+
when(consumer.fetchSubscribeMessageQueues("TopicA")).thenReturn(Set.of(queue));
+ mockQueueWindow(consumer, queue, begin, end, 0L, 0L,
maxOffsetExclusive, maxOffsetExclusive);
+ when(consumer.pull(eq(queue), eq("*"), anyLong(),
eq(32)))
+ .thenAnswer(invocation ->
topicPullBatch(invocation.getArgument(2), maxOffsetExclusive, begin));
+ doNothing().when(consumer).shutdown();
+ })) {
+ List<MessageRecordVO> messages = provider.queryMessages(
+ "instance-a", "TopicA", null, null, null, begin, end);
+
+ assertThat(messages).hasSize(200);
+ assertThat(messages.get(0).getMsgId()).isEqualTo("msg-39999");
+ assertThat(messages.get(199).getMsgId()).isEqualTo("msg-39800");
+ DefaultMQPullConsumer consumer =
mockedConsumers.constructed().get(0);
+ verify(consumer).pull(queue, "*", 8_000L, 32);
+ verify(consumer, never()).pull(queue, "*", 0L, 32);
+ }
+ }
+
+ @Test
+ void
queryByTopicKeepsPullOffsetsInsideTailBudgetWhenWindowIsLargerThanCap() throws
Exception {
+ MessageQueue queue = new MessageQueue("TopicA", "broker-a", 0);
+ long begin = 20_000L;
+ long end = 69_999L;
+ long maxOffsetExclusive = 50_000L;
+ List<Long> pulledOffsets = new ArrayList<>();
+ try (MockedConstruction<DefaultMQPullConsumer> ignored =
+ mockConstruction(DefaultMQPullConsumer.class, (consumer,
context) -> {
+ doNothing().when(consumer).start();
+
when(consumer.fetchSubscribeMessageQueues("TopicA")).thenReturn(Set.of(queue));
+ mockQueueWindow(consumer, queue, begin, end, 0L, 0L,
maxOffsetExclusive, maxOffsetExclusive);
+ when(consumer.pull(eq(queue), eq("*"), anyLong(),
eq(32)))
+ .thenAnswer(invocation -> {
+ long offset = invocation.getArgument(2);
+ pulledOffsets.add(offset);
+ return topicPullBatch(offset,
maxOffsetExclusive, begin);
+ });
+ doNothing().when(consumer).shutdown();
+ })) {
+ List<MessageRecordVO> messages = provider.queryMessages(
+ "instance-a", "TopicA", null, null, null, begin, end);
+
+ assertThat(messages).hasSize(200);
+ assertThat(pulledOffsets).isNotEmpty();
+ assertThat(pulledOffsets.get(0)).isEqualTo(18_000L);
+ assertThat(pulledOffsets).allMatch(offset -> offset >= 18_000L);
+ }
+ }
+
private MQClientAPIImpl mockOffsetLookupClient() {
DefaultMQAdminExtImpl adminExtImpl = mock(DefaultMQAdminExtImpl.class);
MQClientInstance clientInstance = mock(MQClientInstance.class);
@@ -590,6 +698,32 @@ class RocketMQMessageProviderTest {
return clientApi;
}
+
+ 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;
+ }
+
+ private static void mockQueueWindow(DefaultMQPullConsumer consumer,
MessageQueue queue, long begin, long end,
+ long minOffset, long startOffset, long
endOffsetExclusive,
+ long maxOffsetExclusive) throws
Exception {
+ when(consumer.minOffset(queue)).thenReturn(minOffset);
+ when(consumer.maxOffset(queue)).thenReturn(maxOffsetExclusive);
+ when(consumer.searchOffset(queue, begin)).thenReturn(startOffset);
+ when(consumer.searchOffset(queue, end +
1)).thenReturn(endOffsetExclusive);
+ }
+
private static String traceContext(String... fields) {
return String.join(String.valueOf(TraceConstants.CONTENT_SPLITOR),
fields);
}
@@ -607,4 +741,12 @@ class RocketMQMessageProviderTest {
message.setBody(("body-" + msgId).getBytes(StandardCharsets.UTF_8));
return message;
}
+
+ private static PullResult topicPullBatch(long offset, long
endOffsetExclusive, long storeTimeBase) {
+ long batchEnd = Math.min(offset + 32, endOffsetExclusive);
+ List<MessageExt> messages = LongStream.range(offset, batchEnd)
+ .mapToObj(index -> topicMessage("msg-" + index, storeTimeBase
+ index))
+ .toList();
+ return new PullResult(PullStatus.FOUND, batchEnd, 0L,
endOffsetExclusive, messages);
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsControllerTest.java
index c27ab7021..85630ecd0 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsControllerTest.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.studio.settings;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.studio.auth.AuthenticatedUserContext;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
@@ -88,6 +89,41 @@ class SettingsControllerTest {
.andExpect(jsonPath("$.data.model", is("gpt-4")));
}
+ @Test
+ void getGeneralSettingsShouldRedactNotificationWebhooksForReaders() throws
Exception {
+ AuthenticatedUserContext.setUser("reader", false);
+ try {
+ GeneralSettingsVO settings = GeneralSettingsVO.builder()
+ .theme("dark")
+ .compact(true)
+ .desktopNotify(true)
+ .notifySound(false)
+ .sessionTimeout(30)
+ .requireLogin(true)
+ .llmProvider("openai")
+
.dingtalkWebhook("https://oapi.dingtalk.com/robot/send?access_token=secret";)
+ .emailRecipients("[email protected]")
+ .smsWebhook("https://sms.example.test/notify";)
+ .model("gpt-4")
+ .baseUrl("https://api.openai.com";)
+ .build();
+
when(settingsService.getGeneralSettings()).thenReturn(settings.toBuilder()
+ .dingtalkWebhook("******")
+ .smsWebhook("******")
+ .build());
+
+ mockMvc.perform(get("/api/settings/general"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data.dingtalkWebhook",
is("******")))
+ .andExpect(jsonPath("$.data.dingtalkWebhookConfigured",
is(true)))
+ .andExpect(jsonPath("$.data.emailRecipients",
is("[email protected]")))
+ .andExpect(jsonPath("$.data.smsWebhook", is("******")))
+ .andExpect(jsonPath("$.data.smsWebhookConfigured",
is(true)));
+ } finally {
+ AuthenticatedUserContext.clear();
+ }
+ }
+
@Test
void saveGeneralSettingsShouldReturnSuccess() throws Exception {
doNothing().when(settingsService).saveGeneralSettings(any(GeneralSettingsVO.class));
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsServiceTest.java
index fa01e6b3d..2ff5fdc39 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsServiceTest.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.studio.settings;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.studio.auth.AuthenticatedUserContext;
import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.junit.jupiter.api.AfterEach;
@@ -117,6 +118,28 @@ class SettingsServiceTest {
assertThat(result.getModel()).isEqualTo("gpt-4");
}
+ @Test
+ void getGeneralSettingsShouldRedactNotificationWebhooksForReaderSessions()
{
+ GeneralSettingsVO settings = GeneralSettingsVO.builder()
+ .theme("dark")
+
.dingtalkWebhook("https://oapi.dingtalk.com/robot/send?access_token=secret";)
+ .smsWebhook("https://sms.example.test/notify";)
+ .build();
+ when(settingsRepository.loadGeneralSettings()).thenReturn(settings);
+ AuthenticatedUserContext.setUser("reader", false);
+
+ try {
+ GeneralSettingsVO result = settingsService.getGeneralSettings();
+
+ assertThat(result.getDingtalkWebhook()).isEqualTo("******");
+ assertThat(result.isDingtalkWebhookConfigured()).isTrue();
+ assertThat(result.getSmsWebhook()).isEqualTo("******");
+ assertThat(result.isSmsWebhookConfigured()).isTrue();
+ } finally {
+ AuthenticatedUserContext.clear();
+ }
+ }
+
@Test
void saveGeneralSettingsShouldPreserveExistingApiKeyWhenOmitted() {
GeneralSettingsVO existing = GeneralSettingsVO.builder()
diff --git a/web/src/App.test.tsx b/web/src/App.test.tsx
index aef65392a..506d43fdc 100644
--- a/web/src/App.test.tsx
+++ b/web/src/App.test.tsx
@@ -114,6 +114,7 @@ describe('AuthGate', () => {
renderGate();
expect(await screen.findByText('protected content')).toBeInTheDocument();
+ expect(useAuthStore.getState()).toMatchObject({ user: null, userId: null,
admin: null });
});
it('allows protected routes for an authenticated session', async () => {
diff --git a/web/src/App.tsx b/web/src/App.tsx
index 2fa92870f..c74e21692 100644
--- a/web/src/App.tsx
+++ b/web/src/App.tsx
@@ -71,7 +71,12 @@ export function AuthGate() {
if (status.authenticated && status.user) {
syncAuth(status.user.username, status.user.userId,
status.user.admin);
}
- if (!status.loginRequired || status.authenticated) {
+ if (!status.loginRequired) {
+ clearAuth();
+ setGateState('allowed');
+ return;
+ }
+ if (status.authenticated) {
setGateState('allowed');
return;
}
diff --git a/web/src/api/generalSettings.test.ts
b/web/src/api/generalSettings.test.ts
index b8928fb09..e59711e2a 100644
--- a/web/src/api/generalSettings.test.ts
+++ b/web/src/api/generalSettings.test.ts
@@ -33,6 +33,11 @@ const settings: GeneralSettings = {
apiKeyConfigured: true,
model: 'gpt-5',
baseUrl: 'https://api.example.com/v1',
+ dingtalkWebhook: '******',
+ dingtalkWebhookConfigured: true,
+ emailRecipients: '[email protected]',
+ smsWebhook: '******',
+ smsWebhookConfigured: true,
};
const editableSettings: GeneralSettingsUpdate = {
theme: settings.theme,
@@ -82,6 +87,8 @@ describe('general settings API', () => {
const body = JSON.parse(config.data);
expect(body).not.toHaveProperty('apiKey');
expect(body).not.toHaveProperty('apiKeyConfigured');
+ expect(body).not.toHaveProperty('dingtalkWebhookConfigured');
+ expect(body).not.toHaveProperty('smsWebhookConfigured');
return [200, { code: 200, data: null }];
});
diff --git a/web/src/api/settings.ts b/web/src/api/settings.ts
index 246a0a5e0..b221cf937 100644
--- a/web/src/api/settings.ts
+++ b/web/src/api/settings.ts
@@ -30,11 +30,16 @@ export interface GeneralSettings {
model: string;
baseUrl: string;
dingtalkWebhook?: string;
+ dingtalkWebhookConfigured?: boolean;
emailRecipients?: string;
smsWebhook?: string;
+ smsWebhookConfigured?: boolean;
}
-export type GeneralSettingsUpdate = Omit<GeneralSettings, 'apiKeyConfigured'>
& {
+export type GeneralSettingsUpdate = Omit<
+ GeneralSettings,
+ 'apiKeyConfigured' | 'dingtalkWebhookConfigured' | 'smsWebhookConfigured'
+> & {
apiKey?: string;
clearApiKey?: boolean;
};
@@ -59,8 +64,16 @@ export async function getGeneralSettings() {
}
export async function saveGeneralSettings(data: GeneralSettingsUpdate) {
- const payload = { ...data } as GeneralSettingsUpdate & { apiKeyConfigured?:
boolean };
+ const payload = {
+ ...data,
+ } as GeneralSettingsUpdate & {
+ apiKeyConfigured?: boolean;
+ dingtalkWebhookConfigured?: boolean;
+ smsWebhookConfigured?: boolean;
+ };
delete payload.apiKeyConfigured;
+ delete payload.dingtalkWebhookConfigured;
+ delete payload.smsWebhookConfigured;
if (!payload.apiKey?.trim()) delete payload.apiKey;
await client.post('/settings/general/save', payload);
}
diff --git a/web/src/pages/studio/UserManagement.tsx
b/web/src/pages/studio/UserManagement.tsx
index d1bffe348..98d3933d8 100644
--- a/web/src/pages/studio/UserManagement.tsx
+++ b/web/src/pages/studio/UserManagement.tsx
@@ -18,6 +18,7 @@ import { useCallback, useEffect, useState } from 'react';
import { Button, Card, Form, Input, Modal, Space, Switch, Table, Tag, message
} from 'antd';
import type { ColumnsType } from 'antd/es/table';
import { Key, Plus } from '@phosphor-icons/react';
+import { useNavigate } from 'react-router-dom';
import PageHeader from '../../components/PageHeader';
import InfoBanner from '../../components/InfoBanner';
import { changePassword } from '../../api/auth';
@@ -44,8 +45,10 @@ interface PasswordFormValues {
const dateTime = (value?: string) => (value ? new Date(value).toLocaleString()
: '-');
const UserManagementPage = () => {
+ const navigate = useNavigate();
const admin = useAuthStore((state) => state.admin);
const userId = useAuthStore((state) => state.userId);
+ const clearAuth = useAuthStore((state) => state.logout);
const [users, setUsers] = useState<StudioUser[]>([]);
const [loading, setLoading] = useState(false);
const [createOpen, setCreateOpen] = useState(false);
@@ -101,6 +104,8 @@ const UserManagementPage = () => {
try {
if (passwordTarget.id === userId) {
await changePassword(values.currentPassword ?? '', values.newPassword);
+ clearAuth();
+ navigate('/login', { replace: true });
message.success('密码已修改，请使用新密码重新登录');
} else {
await resetStudioUserPassword(passwordTarget.id, values.newPassword);
@@ -124,7 +129,8 @@ const UserManagementPage = () => {
{
title: '状态',
dataIndex: 'enabled',
- render: (value: boolean) => (value ? <Tag color="green">已启用</Tag> : <Tag
color="default">已禁用</Tag>),
+ render: (value: boolean) =>
+ value ? <Tag color="green">已启用</Tag> : <Tag color="default">已禁用</Tag>,
},
{ title: '创建时间', dataIndex: 'gmtCreate', render: dateTime },
{
@@ -186,12 +192,21 @@ const UserManagementPage = () => {
</Card>
{admin && <Table rowKey="id" loading={loading} columns={columns}
dataSource={users} />}
- <Modal title="新建 Studio 用户" open={createOpen} onOk={() => void
createUser()} onCancel={() => setCreateOpen(false)}>
+ <Modal
+ title="新建 Studio 用户"
+ open={createOpen}
+ onOk={() => void createUser()}
+ onCancel={() => setCreateOpen(false)}
+ >
<Form form={createForm} layout="vertical" initialValues={{ admin:
false }}>
<Form.Item name="username" label="用户名" rules={[{ required: true }, {
max: 128 }]}>
<Input autoComplete="username" />
</Form.Item>
- <Form.Item name="password" label="初始密码" rules={[{ required: true },
{ min: 8, message: '密码至少 8 位' }]}>
+ <Form.Item
+ name="password"
+ label="初始密码"
+ rules={[{ required: true }, { min: 8, message: '密码至少 8 位' }]}
+ >
<Input.Password autoComplete="new-password" />
</Form.Item>
<Form.Item name="admin" label="管理员权限" valuePropName="checked">
@@ -201,7 +216,11 @@ const UserManagementPage = () => {
</Modal>
<Modal
- title={passwordTarget?.id === userId ? '修改我的密码' : `重置
${passwordTarget?.username ?? ''} 的密码`}
+ title={
+ passwordTarget?.id === userId
+ ? '修改我的密码'
+ : `重置 ${passwordTarget?.username ?? ''} 的密码`
+ }
open={passwordTarget !== null}
onOk={() => void updatePassword()}
onCancel={() => {
@@ -215,7 +234,11 @@ const UserManagementPage = () => {
<Input.Password autoComplete="current-password" />
</Form.Item>
)}
- <Form.Item name="newPassword" label="新密码" rules={[{ required: true
}, { min: 8, message: '密码至少 8 位' }]}>
+ <Form.Item
+ name="newPassword"
+ label="新密码"
+ rules={[{ required: true }, { min: 8, message: '密码至少 8 位' }]}
+ >
<Input.Password autoComplete="new-password" />
</Form.Item>
</Form>
