This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new bee586bcd0 Improve ACL 2.0 authentication and authorization handling
(#10997)
bee586bcd0 is described below
commit bee586bcd08833e969aaa7b81171569c02102a7a
Author: dingshuangxi888 <[email protected]>
AuthorDate: Thu Sep 3 15:36:26 2026 +0800
Improve ACL 2.0 authentication and authorization handling (#10997)
* fix(auth): require topic for view message authorization
* fix(auth): restrict user password responses
* fix(auth): trust only authenticated gRPC subject
* fix(auth): complete ACL follow-up handling
---
.../strategy/AbstractAuthenticationStrategy.java | 12 +-
.../strategy/StatefulAuthenticationStrategy.java | 4 +
.../authorization/AuthorizationCompatibility.java | 3 +-
.../DefaultAuthorizationContextBuilder.java | 6 +-
.../apache/rocketmq/auth/config/AuthConfig.java | 54 +++++-
.../StatefulAuthenticationStrategyTest.java | 58 +++++++
.../authorization/AuthorizationEvaluatorTest.java | 29 +++-
.../DefaultAuthorizationContextBuilderTest.java | 20 ++-
.../rocketmq/auth/config/AuthConfigTest.java | 73 ++++++++
.../broker/auth/converter/UserConverter.java | 1 -
.../broker/processor/AdminBrokerProcessor.java | 66 +++++++-
.../broker/processor/EndTransactionProcessor.java | 8 +
.../broker/processor/QueryMessageProcessor.java | 1 +
.../AbstractTransactionalMessageCheckListener.java | 2 +-
.../TransactionalMessageRocksDBService.java | 2 +-
.../broker/auth/converter/UserConverterTest.java | 47 ++++++
.../auth/pipeline/AuthorizationPipelineTest.java | 14 ++
.../AdminBrokerProcessorConfigSanitizeTest.java | 55 +++++++
.../broker/processor/AdminBrokerProcessorTest.java | 183 ++++++++++++++++++++-
.../processor/EndTransactionProcessorTest.java | 39 ++++-
.../processor/QueryMessageProcessorTest.java | 3 +-
...faultTransactionalMessageCheckListenerTest.java | 30 ++++
.../TransactionalMessageRocksDBServiceTest.java | 100 +++++++++++
.../selector/DefaultMQProducerImplTest.java | 42 ++++-
.../container/BrokerContainerProcessor.java | 34 +++-
...BrokerContainerProcessorConfigSanitizeTest.java | 55 +++++++
.../rocketmq/proxy/common/utils/GrpcUtils.java | 13 ++
.../proxy/grpc/interceptor/HeaderInterceptor.java | 11 +-
.../grpc/pipeline/AuthenticationPipeline.java | 23 ++-
.../activity/AbstractRemotingActivity.java | 5 +-
.../grpc/interceptor/HeaderInterceptorTest.java | 150 +++++++++++++++++
.../grpc/pipeline/AuthenticationPipelineTest.java | 111 +++++++++++++
.../activity/AbstractRemotingActivityTest.java | 31 +++-
.../metadata/ClusterMetadataServiceTest.java | 13 ++
.../ResumeCheckHalfMessageRequestHeader.java | 2 +
.../tools/command/auth/ListUserSubCommand.java | 7 +-
36 files changed, 1245 insertions(+), 62 deletions(-)
diff --git
a/auth/src/main/java/org/apache/rocketmq/auth/authentication/strategy/AbstractAuthenticationStrategy.java
b/auth/src/main/java/org/apache/rocketmq/auth/authentication/strategy/AbstractAuthenticationStrategy.java
index bc7052014d..ab185909c1 100644
---
a/auth/src/main/java/org/apache/rocketmq/auth/authentication/strategy/AbstractAuthenticationStrategy.java
+++
b/auth/src/main/java/org/apache/rocketmq/auth/authentication/strategy/AbstractAuthenticationStrategy.java
@@ -16,10 +16,7 @@
*/
package org.apache.rocketmq.auth.authentication.strategy;
-import java.util.HashSet;
-import java.util.Set;
import java.util.function.Supplier;
-import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.auth.authentication.context.AuthenticationContext;
import
org.apache.rocketmq.auth.authentication.exception.AuthenticationException;
import org.apache.rocketmq.auth.authentication.factory.AuthenticationFactory;
@@ -30,7 +27,6 @@ import org.apache.rocketmq.common.utils.ExceptionUtils;
public abstract class AbstractAuthenticationStrategy implements
AuthenticationStrategy {
protected final AuthConfig authConfig;
- protected final Set<String> authenticationWhiteSet = new HashSet<>();
protected final AuthenticationProvider<AuthenticationContext>
authenticationProvider;
public AbstractAuthenticationStrategy(AuthConfig authConfig, Supplier<?>
metadataService) {
@@ -39,12 +35,6 @@ public abstract class AbstractAuthenticationStrategy
implements AuthenticationSt
if (this.authenticationProvider != null) {
this.authenticationProvider.initialize(authConfig,
metadataService);
}
- if (StringUtils.isNotBlank(authConfig.getAuthenticationWhitelist())) {
- String[] whitelist =
StringUtils.split(authConfig.getAuthenticationWhitelist(), ",");
- for (String rpcCode : whitelist) {
- this.authenticationWhiteSet.add(StringUtils.trim(rpcCode));
- }
- }
}
protected void doEvaluate(AuthenticationContext context) {
@@ -57,7 +47,7 @@ public abstract class AbstractAuthenticationStrategy
implements AuthenticationSt
if (this.authenticationProvider == null) {
return;
}
- if (this.authenticationWhiteSet.contains(context.getRpcCode())) {
+ if (!authConfig.isAuthenticationRequired(context.getRpcCode())) {
return;
}
try {
diff --git
a/auth/src/main/java/org/apache/rocketmq/auth/authentication/strategy/StatefulAuthenticationStrategy.java
b/auth/src/main/java/org/apache/rocketmq/auth/authentication/strategy/StatefulAuthenticationStrategy.java
index 914d99ac2e..74fc179147 100644
---
a/auth/src/main/java/org/apache/rocketmq/auth/authentication/strategy/StatefulAuthenticationStrategy.java
+++
b/auth/src/main/java/org/apache/rocketmq/auth/authentication/strategy/StatefulAuthenticationStrategy.java
@@ -42,6 +42,10 @@ public class StatefulAuthenticationStrategy extends
AbstractAuthenticationStrate
@Override
public void evaluate(AuthenticationContext context) {
+ if (!this.authConfig.isAuthenticationRequired(context.getRpcCode())) {
+ this.doEvaluate(context);
+ return;
+ }
if (StringUtils.isBlank(context.getChannelId())) {
this.doEvaluate(context);
return;
diff --git
a/auth/src/main/java/org/apache/rocketmq/auth/authorization/AuthorizationCompatibility.java
b/auth/src/main/java/org/apache/rocketmq/auth/authorization/AuthorizationCompatibility.java
index 75fe11bf74..803ecf63ae 100644
---
a/auth/src/main/java/org/apache/rocketmq/auth/authorization/AuthorizationCompatibility.java
+++
b/auth/src/main/java/org/apache/rocketmq/auth/authorization/AuthorizationCompatibility.java
@@ -44,7 +44,6 @@ final class AuthorizationCompatibility {
case RequestCode.UNREGISTER_CLIENT:
return isProducerUnregister(request);
case RequestCode.END_TRANSACTION:
- case RequestCode.VIEW_MESSAGE_BY_ID:
return isHistoricalTopicAbsent(request);
default:
return false;
@@ -104,7 +103,7 @@ final class AuthorizationCompatibility {
}
/**
- * Historical END_TRANSACTION and VIEW_MESSAGE_BY_ID requests carry no
topic field.
+ * Historical END_TRANSACTION requests carry no topic field.
*/
private static boolean isHistoricalTopicAbsent(RemotingCommand request) {
return request.getExtFields() != null &&
StringUtils.isBlank(getExtField(request, "topic"));
diff --git
a/auth/src/main/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilder.java
b/auth/src/main/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilder.java
index 61aac9af28..af090b5f0f 100644
---
a/auth/src/main/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilder.java
+++
b/auth/src/main/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilder.java
@@ -257,10 +257,8 @@ public class DefaultAuthorizationContextBuilder implements
AuthorizationContextB
}
break;
case RequestCode.VIEW_MESSAGE_BY_ID:
- if (StringUtils.isNotBlank(fields.get(TOPIC))) {
- topic = Resource.ofTopic(fields.get(TOPIC));
- result.add(DefaultAuthorizationContext.of(subject,
topic, Action.GET, sourceIp));
- }
+ topic =
Resource.ofTopic(requireResource(fields.get(TOPIC), "topic"));
+ result.add(DefaultAuthorizationContext.of(subject, topic,
Action.GET, sourceIp));
break;
case RequestCode.CONSUMER_SEND_MSG_BACK:
group =
Resource.ofGroup(requireResource(fields.get(GROUP), "consumer group"));
diff --git a/auth/src/main/java/org/apache/rocketmq/auth/config/AuthConfig.java
b/auth/src/main/java/org/apache/rocketmq/auth/config/AuthConfig.java
index ed294c8ecb..239972b454 100644
--- a/auth/src/main/java/org/apache/rocketmq/auth/config/AuthConfig.java
+++ b/auth/src/main/java/org/apache/rocketmq/auth/config/AuthConfig.java
@@ -16,6 +16,11 @@
*/
package org.apache.rocketmq.auth.config;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Set;
+import org.apache.commons.lang3.StringUtils;
+
public class AuthConfig implements Cloneable {
private String configName;
@@ -32,7 +37,7 @@ public class AuthConfig implements Cloneable {
private String authenticationStrategy;
- private String authenticationWhitelist;
+ private volatile WhitelistSnapshot authenticationWhitelist;
private String initAuthenticationUser;
@@ -79,6 +84,48 @@ public class AuthConfig implements Cloneable {
}
}
+ public boolean isAuthenticationRequired(String rpcCode) {
+ return authenticationEnabled && !contains(authenticationWhitelist,
rpcCode);
+ }
+
+ private static boolean contains(WhitelistSnapshot snapshot, String
rpcCode) {
+ return snapshot != null && StringUtils.isNotEmpty(rpcCode)
+ && snapshot.entries.contains(rpcCode);
+ }
+
+ private static Set<String> parseWhitelist(String whitelist) {
+ if (StringUtils.isBlank(whitelist)) {
+ return Collections.emptySet();
+ }
+ Set<String> result = new LinkedHashSet<>();
+ for (String rpcCode : StringUtils.split(whitelist, ',')) {
+ String value = StringUtils.trim(rpcCode);
+ if (StringUtils.isNotEmpty(value)) {
+ result.add(value);
+ }
+ }
+ return Collections.unmodifiableSet(result);
+ }
+
+ private static final class WhitelistSnapshot {
+ private final String value;
+ private final Set<String> entries;
+
+ private WhitelistSnapshot(String value) {
+ this.value = value;
+ this.entries = parseWhitelist(value);
+ }
+
+ private static WhitelistSnapshot of(String value) {
+ return value == null ? null : new WhitelistSnapshot(value);
+ }
+
+ @Override
+ public String toString() {
+ return value;
+ }
+ }
+
public String getConfigName() {
return configName;
}
@@ -136,11 +183,12 @@ public class AuthConfig implements Cloneable {
}
public String getAuthenticationWhitelist() {
- return authenticationWhitelist;
+ WhitelistSnapshot snapshot = authenticationWhitelist;
+ return snapshot == null ? null : snapshot.value;
}
public void setAuthenticationWhitelist(String authenticationWhitelist) {
- this.authenticationWhitelist = authenticationWhitelist;
+ this.authenticationWhitelist =
WhitelistSnapshot.of(authenticationWhitelist);
}
public String getInitAuthenticationUser() {
diff --git
a/auth/src/test/java/org/apache/rocketmq/auth/authentication/strategy/StatefulAuthenticationStrategyTest.java
b/auth/src/test/java/org/apache/rocketmq/auth/authentication/strategy/StatefulAuthenticationStrategyTest.java
new file mode 100644
index 0000000000..c9480a22f7
--- /dev/null
+++
b/auth/src/test/java/org/apache/rocketmq/auth/authentication/strategy/StatefulAuthenticationStrategyTest.java
@@ -0,0 +1,58 @@
+/*
+ * 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.auth.authentication.strategy;
+
+import org.apache.rocketmq.auth.authentication.context.AuthenticationContext;
+import
org.apache.rocketmq.auth.authentication.context.DefaultAuthenticationContext;
+import
org.apache.rocketmq.auth.authentication.exception.AuthenticationException;
+import org.apache.rocketmq.auth.config.AuthConfig;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+
+public class StatefulAuthenticationStrategyTest {
+
+ @Test
+ public void authenticationWhitelistBypassIsNotCachedAsSuccess() {
+ AuthConfig authConfig = new AuthConfig();
+ authConfig.setConfigName("stateful-whitelist-" + System.nanoTime());
+ authConfig.setAuthenticationEnabled(true);
+ authConfig.setAuthenticationWhitelist("PUBLIC_RPC");
+
+ StatefulAuthenticationStrategy strategy = spy(
+ new StatefulAuthenticationStrategy(authConfig, null));
+ strategy.evaluate(context("PUBLIC_RPC"));
+
+ doThrow(new AuthenticationException("signature is invalid"))
+ .when(strategy).doEvaluate(any(AuthenticationContext.class));
+
+ assertThatThrownBy(() -> strategy.evaluate(context("PROTECTED_RPC")))
+ .isInstanceOf(AuthenticationException.class)
+ .hasMessageContaining("signature is invalid");
+ }
+
+ private DefaultAuthenticationContext context(String rpcCode) {
+ DefaultAuthenticationContext context = new
DefaultAuthenticationContext();
+ context.setChannelId("channel-id");
+ context.setRpcCode(rpcCode);
+ context.setUsername("claimed-super-user");
+ return context;
+ }
+}
diff --git
a/auth/src/test/java/org/apache/rocketmq/auth/authorization/AuthorizationEvaluatorTest.java
b/auth/src/test/java/org/apache/rocketmq/auth/authorization/AuthorizationEvaluatorTest.java
index 8d49e965cb..231b5e8529 100644
---
a/auth/src/test/java/org/apache/rocketmq/auth/authorization/AuthorizationEvaluatorTest.java
+++
b/auth/src/test/java/org/apache/rocketmq/auth/authorization/AuthorizationEvaluatorTest.java
@@ -54,6 +54,7 @@ import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import
org.apache.rocketmq.remoting.protocol.header.EndTransactionRequestHeader;
+import
org.apache.rocketmq.remoting.protocol.header.ResumeCheckHalfMessageRequestHeader;
import
org.apache.rocketmq.remoting.protocol.header.UnregisterClientRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ViewMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumerData;
@@ -523,15 +524,28 @@ public class AuthorizationEvaluatorTest {
requestEvaluator.evaluate(endTransaction(
null, 1L, 2L, null, "messageId"), Collections.emptyList());
- requestEvaluator.evaluate(viewMessage(0L), Collections.emptyList());
- requestEvaluator.evaluate(viewMessage(-1L), Collections.emptyList());
- requestEvaluator.evaluate(viewMessage(null), Collections.emptyList());
+ Assert.assertThrows(AuthorizationException.class,
+ () -> requestEvaluator.evaluate(viewMessage(0L),
Collections.emptyList()));
+ Assert.assertThrows(AuthorizationException.class,
+ () -> requestEvaluator.evaluate(viewMessage(-1L),
Collections.emptyList()));
+ Assert.assertThrows(AuthorizationException.class,
+ () -> requestEvaluator.evaluate(viewMessage(null),
Collections.emptyList()));
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(
RemotingCommand.createRequestCommand(RequestCode.VIEW_MESSAGE_BY_ID, null),
Collections.emptyList()));
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(viewMessage("topic", 0L),
Collections.emptyList()));
+
+ Assert.assertThrows(AuthorizationException.class,
+ () -> requestEvaluator.evaluate(resumeCheckHalfMessage(null,
"messageId"),
+ Collections.emptyList()));
+ Assert.assertThrows(AuthorizationException.class,
+ () -> requestEvaluator.evaluate(resumeCheckHalfMessage(" ",
"messageId"),
+ Collections.emptyList()));
+ Assert.assertThrows(AuthorizationException.class,
+ () -> requestEvaluator.evaluate(resumeCheckHalfMessage("topic",
"messageId"),
+ Collections.emptyList()));
Assert.assertThrows(AuthorizationException.class,
() ->
requestEvaluator.evaluate(RemotingCommand.createRequestCommand(-1, null),
Collections.emptyList()));
@@ -643,6 +657,15 @@ public class AuthorizationEvaluatorTest {
return request;
}
+ private RemotingCommand resumeCheckHalfMessage(String topic, String
messageId) {
+ ResumeCheckHalfMessageRequestHeader header = new
ResumeCheckHalfMessageRequestHeader();
+ header.setTopic(topic);
+ header.setMsgId(messageId);
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.RESUME_CHECK_HALF_MESSAGE,
header);
+ request.makeCustomHeaderToNet();
+ return request;
+ }
+
private RemotingCommand remotingRequest(int requestCode, byte[] body) {
RemotingCommand request =
RemotingCommand.createRequestCommand(requestCode, null);
request.setBody(body);
diff --git
a/auth/src/test/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilderTest.java
b/auth/src/test/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilderTest.java
index dbad9e9d01..f90352ef09 100644
---
a/auth/src/test/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilderTest.java
+++
b/auth/src/test/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilderTest.java
@@ -99,6 +99,7 @@ import
org.apache.rocketmq.remoting.protocol.header.PullMessageRequestHeader;
import
org.apache.rocketmq.remoting.protocol.header.QueryConsumerOffsetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.QueryMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.RecallMessageRequestHeader;
+import
org.apache.rocketmq.remoting.protocol.header.ResumeCheckHalfMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.SearchOffsetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeaderV2;
@@ -409,7 +410,6 @@ public class DefaultAuthorizationContextBuilderTest {
request.makeCustomHeaderToNet();
RemotingCommand endTransactionWithoutTopic = request;
Assert.assertTrue(builder.build(channelHandlerContext,
endTransactionWithoutTopic).isEmpty());
-
ConsumerSendMsgBackRequestHeader consumerSendMsgBackRequestHeader =
new ConsumerSendMsgBackRequestHeader();
consumerSendMsgBackRequestHeader.setGroup("group");
request =
RemotingCommand.createRequestCommand(RequestCode.CONSUMER_SEND_MSG_BACK,
consumerSendMsgBackRequestHeader);
@@ -604,6 +604,14 @@ public class DefaultAuthorizationContextBuilderTest {
() -> builder.build(channelHandlerContext, request));
}
+ ResumeCheckHalfMessageRequestHeader resumeHeader = new
ResumeCheckHalfMessageRequestHeader();
+ resumeHeader.setMsgId("messageId");
+ Assert.assertThrows(AuthorizationException.class, () ->
builder.build(channelHandlerContext,
+ remotingRequest(RequestCode.RESUME_CHECK_HALF_MESSAGE,
resumeHeader, null)));
+ resumeHeader.setTopic(" ");
+ Assert.assertThrows(AuthorizationException.class, () ->
builder.build(channelHandlerContext,
+ remotingRequest(RequestCode.RESUME_CHECK_HALF_MESSAGE,
resumeHeader, null)));
+
RemotingCommand sendBackRequest =
remotingRequest(RequestCode.CONSUMER_SEND_MSG_BACK, null, null);
sendBackRequest.addExtField("group", " ");
Assert.assertThrows(AuthorizationException.class,
@@ -1044,11 +1052,13 @@ public class DefaultAuthorizationContextBuilderTest {
ViewMessageRequestHeader viewMessageHeader = new
ViewMessageRequestHeader();
viewMessageHeader.setOffset(0L);
- Assert.assertTrue(builder.build(channelHandlerContext,
- remotingRequest(RequestCode.VIEW_MESSAGE_BY_ID, viewMessageHeader,
null)).isEmpty());
+ Assert.assertThrows(AuthorizationException.class,
+ () -> builder.build(channelHandlerContext,
+ remotingRequest(RequestCode.VIEW_MESSAGE_BY_ID,
viewMessageHeader, null)));
viewMessageHeader.setTopic(" ");
- Assert.assertTrue(builder.build(channelHandlerContext,
- remotingRequest(RequestCode.VIEW_MESSAGE_BY_ID, viewMessageHeader,
null)).isEmpty());
+ Assert.assertThrows(AuthorizationException.class,
+ () -> builder.build(channelHandlerContext,
+ remotingRequest(RequestCode.VIEW_MESSAGE_BY_ID,
viewMessageHeader, null)));
}
@Test
diff --git
a/auth/src/test/java/org/apache/rocketmq/auth/config/AuthConfigTest.java
b/auth/src/test/java/org/apache/rocketmq/auth/config/AuthConfigTest.java
new file mode 100644
index 0000000000..1257fdc1ff
--- /dev/null
+++ b/auth/src/test/java/org/apache/rocketmq/auth/config/AuthConfigTest.java
@@ -0,0 +1,73 @@
+/*
+ * 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.auth.config;
+
+import com.alibaba.fastjson2.JSON;
+import java.util.Properties;
+import org.apache.rocketmq.common.MixAll;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class AuthConfigTest {
+
+ @Test
+ public void authenticationRequiredTracksEnabledStateAndWhitelistUpdates() {
+ AuthConfig authConfig = new AuthConfig();
+
+ assertThat(authConfig.isAuthenticationRequired("10")).isFalse();
+
+ authConfig.setAuthenticationEnabled(true);
+ authConfig.setAuthenticationWhitelist(" 10, 11,10 ");
+ assertThat(authConfig.isAuthenticationRequired("10")).isFalse();
+ assertThat(authConfig.isAuthenticationRequired("11")).isFalse();
+ assertThat(authConfig.isAuthenticationRequired("12")).isTrue();
+ assertThat(authConfig.isAuthenticationRequired("")).isTrue();
+ assertThat(authConfig.isAuthenticationRequired(null)).isTrue();
+
+ authConfig.setAuthenticationWhitelist("12");
+ assertThat(authConfig.isAuthenticationRequired("10")).isTrue();
+ assertThat(authConfig.isAuthenticationRequired("12")).isFalse();
+ }
+
+ @Test
+ public void
authenticationWhitelistPreservesConfigBindingAndCloneIsolation() {
+ String whitelist = " AUTH_A,AUTH_B,AUTH_A ";
+ AuthConfig authConfig = new AuthConfig();
+ authConfig.setAuthenticationEnabled(true);
+ authConfig.setAuthenticationWhitelist(whitelist);
+
+ Properties properties = MixAll.object2Properties(authConfig);
+ assertThat(properties)
+ .containsEntry("authenticationWhitelist", whitelist)
+ .doesNotContainKeys("entries", "value");
+
+ AuthConfig propertiesCopy = new AuthConfig();
+ MixAll.properties2Object(properties, propertiesCopy);
+
assertThat(propertiesCopy.getAuthenticationWhitelist()).isEqualTo(whitelist.trim());
+
assertThat(propertiesCopy.isAuthenticationRequired("AUTH_A")).isFalse();
+
+ AuthConfig jsonCopy = JSON.parseObject(JSON.toJSONString(authConfig),
AuthConfig.class);
+ assertThat(jsonCopy.getAuthenticationWhitelist()).isEqualTo(whitelist);
+ assertThat(jsonCopy.isAuthenticationRequired("AUTH_B")).isFalse();
+
+ AuthConfig cloned = authConfig.clone();
+ authConfig.setAuthenticationWhitelist("AUTH_C");
+ assertThat(cloned.isAuthenticationRequired("AUTH_A")).isFalse();
+ assertThat(cloned.isAuthenticationRequired("AUTH_C")).isTrue();
+ }
+}
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/auth/converter/UserConverter.java
b/broker/src/main/java/org/apache/rocketmq/broker/auth/converter/UserConverter.java
index 12756d8fd3..587c473c50 100644
---
a/broker/src/main/java/org/apache/rocketmq/broker/auth/converter/UserConverter.java
+++
b/broker/src/main/java/org/apache/rocketmq/broker/auth/converter/UserConverter.java
@@ -33,7 +33,6 @@ public class UserConverter {
public static UserInfo convertUser(User user) {
UserInfo result = new UserInfo();
result.setUsername(user.getUsername());
- result.setPassword(user.getPassword());
if (user.getUserType() != null) {
result.setUserType(user.getUserType().getName());
}
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
index 876a72fa29..de90396064 100644
---
a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+++
b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
@@ -35,6 +35,7 @@ import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
@@ -1177,8 +1178,15 @@ public class AdminBrokerProcessor implements
NettyRequestProcessor {
final RemotingCommand response =
RemotingCommand.createResponseCommand(GetBrokerConfigResponseHeader.class);
final GetBrokerConfigResponseHeader responseHeader =
(GetBrokerConfigResponseHeader) response.readCustomHeader();
- String content =
this.brokerController.getConfiguration().getAllConfigsFormatString();
- if (content != null && content.length() > 0) {
+ String content =
sanitizeConfigForResponse(this.brokerController.getConfiguration().getAllConfigsFormatString());
+ if (content == null) {
+ LOGGER.error("AdminBrokerProcessor#getBrokerConfig: failed to
sanitize broker config, caller={}",
+ RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
+ response.setCode(ResponseCode.SYSTEM_ERROR);
+ response.setRemark("Failed to sanitize broker config");
+ return response;
+ }
+ if (content.length() > 0) {
try {
content = MixAll.adjustConfigForPlatform(content);
response.setBody(content.getBytes(MixAll.DEFAULT_CHARSET));
@@ -1199,6 +1207,30 @@ public class AdminBrokerProcessor implements
NettyRequestProcessor {
return response;
}
+ // sensitive config keys that must never be returned by config query
interfaces
+ private static final String[] SENSITIVE_CONFIG_KEYS = new String[] {
+ "initAuthenticationUser", "innerClientAuthenticationCredentials"
+ };
+
+ /**
+ * Remove sensitive entries from the exported config content. Returns null
when the
+ * content cannot be parsed, so callers must fail closed instead of
returning the
+ * original content.
+ */
+ static String sanitizeConfigForResponse(String content) {
+ if (content == null || content.isEmpty()) {
+ return content;
+ }
+ Properties properties = MixAll.string2Properties(content);
+ if (properties == null) {
+ return null;
+ }
+ for (String key : SENSITIVE_CONFIG_KEYS) {
+ properties.remove(key);
+ }
+ return MixAll.properties2String(properties, true);
+ }
+
private RemotingCommand
rewriteRequestForStaticTopic(SearchOffsetRequestHeader requestHeader,
TopicQueueMappingContext mappingContext) {
try {
@@ -3031,6 +3063,12 @@ public class AdminBrokerProcessor implements
NettyRequestProcessor {
selectMappedBufferResult = this.brokerController.getMessageStore()
.selectOneMessageByOffset(messageId.getOffset());
MessageExt msg =
MessageDecoder.decode(selectMappedBufferResult.getByteBuffer(), true, false);
+ if (!Objects.equals(requestHeader.getTopic(),
+ msg.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC))) {
+ response.setCode(ResponseCode.NO_PERMISSION);
+ response.setRemark("The topic does not match the transaction
message");
+ return response;
+ }
msg.putUserProperty(MessageConst.PROPERTY_TRANSACTION_CHECK_TIMES,
String.valueOf(0));
PutMessageResult putMessageResult =
this.brokerController.getMessageStore()
.putMessage(toMessageExtBrokerInner(msg));
@@ -3349,6 +3387,9 @@ public class AdminBrokerProcessor implements
NettyRequestProcessor {
response.setCode(ResponseCode.SUCCESS);
if (user != null) {
UserInfo userInfo = UserConverter.convertUser(user);
+ if (canReadUserPassword(request,
requestHeader.getUsername())) {
+ userInfo.setPassword(user.getPassword());
+ }
response.setBody(JSON.toJSONString(userInfo).getBytes(StandardCharsets.UTF_8));
}
})
@@ -3524,6 +3565,27 @@ public class AdminBrokerProcessor implements
NettyRequestProcessor {
.isSuperUser(accessKey).join();
}
+ private boolean canReadUserPassword(RemotingCommand request, String
username) {
+ if (this.brokerController.getAuthConfig() == null
+ || !this.brokerController.getAuthConfig()
+ .isAuthenticationRequired(String.valueOf(request.getCode()))) {
+ return false;
+ }
+ String accessKey = getAccessKey(request);
+ if (StringUtils.isEmpty(accessKey)) {
+ return false;
+ }
+ if (StringUtils.equals(accessKey, username)) {
+ return true;
+ }
+ return this.brokerController.getAuthenticationMetadataManager()
+ .isSuperUser(accessKey).join();
+ }
+
+ private String getAccessKey(RemotingCommand request) {
+ return request.getExtFields() == null ? null :
request.getExtFields().get("AccessKey");
+ }
+
private Void handleAuthException(RemotingCommand response, Throwable ex) {
Throwable throwable = ExceptionUtils.getRealException(ex);
if (throwable instanceof AuthenticationException || throwable
instanceof AuthorizationException) {
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/processor/EndTransactionProcessor.java
b/broker/src/main/java/org/apache/rocketmq/broker/processor/EndTransactionProcessor.java
index a792f80456..c568f1e4bf 100644
---
a/broker/src/main/java/org/apache/rocketmq/broker/processor/EndTransactionProcessor.java
+++
b/broker/src/main/java/org/apache/rocketmq/broker/processor/EndTransactionProcessor.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.broker.processor;
import io.netty.channel.ChannelHandlerContext;
+import java.util.Objects;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.broker.BrokerController;
@@ -236,6 +237,13 @@ public class EndTransactionProcessor implements
NettyRequestProcessor {
private RemotingCommand checkPrepareMessage(MessageExt msgExt,
EndTransactionRequestHeader requestHeader) {
final RemotingCommand response =
RemotingCommand.createResponseCommand(null);
if (msgExt != null) {
+ if (StringUtils.isNotBlank(requestHeader.getTopic())
+ && !Objects.equals(requestHeader.getTopic(),
+ msgExt.getProperty(MessageConst.PROPERTY_REAL_TOPIC))) {
+ response.setCode(ResponseCode.NO_PERMISSION);
+ response.setRemark("The topic does not match the transaction
message");
+ return response;
+ }
final String pgroupRead =
msgExt.getProperty(MessageConst.PROPERTY_PRODUCER_GROUP);
if (!pgroupRead.equals(requestHeader.getProducerGroup())) {
response.setCode(ResponseCode.SYSTEM_ERROR);
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/processor/QueryMessageProcessor.java
b/broker/src/main/java/org/apache/rocketmq/broker/processor/QueryMessageProcessor.java
index cb1c4287ce..1791782685 100644
---
a/broker/src/main/java/org/apache/rocketmq/broker/processor/QueryMessageProcessor.java
+++
b/broker/src/main/java/org/apache/rocketmq/broker/processor/QueryMessageProcessor.java
@@ -61,6 +61,7 @@ public class QueryMessageProcessor implements
NettyRequestProcessor {
TopicValidator.RMQ_SYS_SCHEDULE_TOPIC,
TimerMessageStore.TIMER_TOPIC,
TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC,
+ TopicValidator.RMQ_SYS_ROCKSDB_TRANS_HALF_TOPIC,
TopicValidator.RMQ_SYS_TRANS_CHECK_MAX_TIME_TOPIC));
private final BrokerController brokerController;
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/transaction/AbstractTransactionalMessageCheckListener.java
b/broker/src/main/java/org/apache/rocketmq/broker/transaction/AbstractTransactionalMessageCheckListener.java
index d1b77355b0..1084ac2081 100644
---
a/broker/src/main/java/org/apache/rocketmq/broker/transaction/AbstractTransactionalMessageCheckListener.java
+++
b/broker/src/main/java/org/apache/rocketmq/broker/transaction/AbstractTransactionalMessageCheckListener.java
@@ -50,7 +50,7 @@ public abstract class
AbstractTransactionalMessageCheckListener {
public void sendCheckMessage(MessageExt msgExt) throws Exception {
CheckTransactionStateRequestHeader checkTransactionStateRequestHeader
= new CheckTransactionStateRequestHeader();
- checkTransactionStateRequestHeader.setTopic(msgExt.getTopic());
+
checkTransactionStateRequestHeader.setTopic(msgExt.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC));
checkTransactionStateRequestHeader.setCommitLogOffset(msgExt.getCommitLogOffset());
checkTransactionStateRequestHeader.setOffsetMsgId(msgExt.getMsgId());
checkTransactionStateRequestHeader.setMsgId(msgExt.getUserProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX));
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/transaction/rocksdb/TransactionalMessageRocksDBService.java
b/broker/src/main/java/org/apache/rocketmq/broker/transaction/rocksdb/TransactionalMessageRocksDBService.java
index dbd3575d69..d7940a726a 100644
---
a/broker/src/main/java/org/apache/rocketmq/broker/transaction/rocksdb/TransactionalMessageRocksDBService.java
+++
b/broker/src/main/java/org/apache/rocketmq/broker/transaction/rocksdb/TransactionalMessageRocksDBService.java
@@ -223,7 +223,7 @@ public class TransactionalMessageRocksDBService {
}
try {
CheckTransactionStateRequestHeader
checkTransactionStateRequestHeader = new CheckTransactionStateRequestHeader();
- checkTransactionStateRequestHeader.setTopic(msgExt.getTopic());
+
checkTransactionStateRequestHeader.setTopic(msgExt.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC));
checkTransactionStateRequestHeader.setCommitLogOffset(msgExt.getCommitLogOffset());
checkTransactionStateRequestHeader.setOffsetMsgId(msgExt.getMsgId());
checkTransactionStateRequestHeader.setMsgId(MessageClientIDSetter.getUniqID(msgExt));
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/auth/converter/UserConverterTest.java
b/broker/src/test/java/org/apache/rocketmq/broker/auth/converter/UserConverterTest.java
new file mode 100644
index 0000000000..a514248bda
--- /dev/null
+++
b/broker/src/test/java/org/apache/rocketmq/broker/auth/converter/UserConverterTest.java
@@ -0,0 +1,47 @@
+/*
+ * 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.broker.auth.converter;
+
+import org.apache.rocketmq.auth.authentication.enums.UserType;
+import org.apache.rocketmq.auth.authentication.model.User;
+import org.apache.rocketmq.remoting.protocol.body.UserInfo;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class UserConverterTest {
+
+ @Test
+ public void outboundConversionOmitsPasswordByDefault() {
+ UserInfo result = UserConverter.convertUser(
+ User.of("user", "secret", UserType.NORMAL));
+
+ assertThat(result.getUsername()).isEqualTo("user");
+ assertThat(result.getPassword()).isNull();
+ assertThat(result.getUserType()).isEqualTo("Normal");
+ }
+
+ @Test
+ public void inboundConversionRetainsPasswordForAuthentication() {
+ User result = UserConverter.convertUser(
+ UserInfo.of("user", "secret", "Normal"));
+
+ assertThat(result.getUsername()).isEqualTo("user");
+ assertThat(result.getPassword()).isEqualTo("secret");
+ assertThat(result.getUserType()).isEqualTo(UserType.NORMAL);
+ }
+}
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/auth/pipeline/AuthorizationPipelineTest.java
b/broker/src/test/java/org/apache/rocketmq/broker/auth/pipeline/AuthorizationPipelineTest.java
index 0fa2ceb49e..b452e1f93e 100644
---
a/broker/src/test/java/org/apache/rocketmq/broker/auth/pipeline/AuthorizationPipelineTest.java
+++
b/broker/src/test/java/org/apache/rocketmq/broker/auth/pipeline/AuthorizationPipelineTest.java
@@ -25,6 +25,7 @@ import org.apache.rocketmq.common.AbortProcessException;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.header.ViewMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.heartbeat.HeartbeatData;
import org.apache.rocketmq.remoting.protocol.heartbeat.ProducerData;
import org.junit.Assert;
@@ -57,6 +58,19 @@ public class AuthorizationPipelineTest {
Assert.assertEquals(ResponseCode.NO_PERMISSION,
exception.getResponseCode());
}
+ @Test
+ public void rejectsTopicLessViewMessageWithEmptyContexts() {
+ AuthorizationPipeline pipeline = createPipeline();
+ ViewMessageRequestHeader header = new ViewMessageRequestHeader();
+ header.setOffset(0L);
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.VIEW_MESSAGE_BY_ID, header);
+ request.makeCustomHeaderToNet();
+
+ AbortProcessException exception =
Assert.assertThrows(AbortProcessException.class,
+ () -> pipeline.execute(null, request));
+ Assert.assertEquals(ResponseCode.NO_PERMISSION,
exception.getResponseCode());
+ }
+
private AuthorizationPipeline createPipeline() {
AuthConfig authConfig = new AuthConfig();
authConfig.setConfigName("broker-authorization-pipeline-test");
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorConfigSanitizeTest.java
b/broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorConfigSanitizeTest.java
new file mode 100644
index 0000000000..be78a8908f
--- /dev/null
+++
b/broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorConfigSanitizeTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.broker.processor;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+public class AdminBrokerProcessorConfigSanitizeTest {
+
+ @Test
+ public void testSanitizeRemovesSensitiveKeys() {
+ String content = "brokerName=broker-a\n"
+ +
"initAuthenticationUser={\"username\":\"rocketmq\",\"password\":\"secret\"}\n"
+ +
"innerClientAuthenticationCredentials={\"accessKey\":\"ak\",\"secretKey\":\"sk\"}\n"
+ + "listenPort=10911\n";
+
+ String sanitized =
AdminBrokerProcessor.sanitizeConfigForResponse(content);
+
+ assertFalse(sanitized.contains("initAuthenticationUser"));
+
assertFalse(sanitized.contains("innerClientAuthenticationCredentials"));
+ assertFalse(sanitized.contains("secret"));
+ assertTrue(sanitized.contains("brokerName=broker-a"));
+ assertTrue(sanitized.contains("listenPort=10911"));
+ }
+
+ @Test
+ public void testSanitizeFailsClosedOnUnparsableContent() {
+ // malformed unicode escape makes Properties.load throw
+
assertNull(AdminBrokerProcessor.sanitizeConfigForResponse("key=\\uZZZZ\n"));
+ }
+
+ @Test
+ public void testSanitizePassesThroughNullOrEmpty() {
+ assertNull(AdminBrokerProcessor.sanitizeConfigForResponse(null));
+ assertEquals("", AdminBrokerProcessor.sanitizeConfigForResponse(""));
+ }
+}
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java
b/broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java
index 6a32adbe1d..ddadfae410 100644
---
a/broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java
+++
b/broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java
@@ -33,6 +33,7 @@ import
org.apache.rocketmq.auth.authorization.manager.AuthorizationMetadataManag
import org.apache.rocketmq.auth.authorization.model.Acl;
import org.apache.rocketmq.auth.authorization.model.Environment;
import org.apache.rocketmq.auth.authorization.model.Resource;
+import org.apache.rocketmq.auth.config.AuthConfig;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
@@ -62,7 +63,9 @@ import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.lite.LiteUtil;
import org.apache.rocketmq.common.message.MessageAccessor;
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.MessageExtBrokerInner;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
@@ -131,6 +134,8 @@ import
org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfi
import org.apache.rocketmq.store.CommitLog;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.MessageStore;
+import org.apache.rocketmq.store.PutMessageResult;
+import org.apache.rocketmq.store.PutMessageStatus;
import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.logfile.DefaultMappedFile;
@@ -144,6 +149,7 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.Spy;
import org.mockito.junit.MockitoJUnitRunner;
@@ -157,6 +163,7 @@ import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -184,6 +191,7 @@ import static org.mockito.ArgumentMatchers.anySet;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -199,9 +207,11 @@ public class AdminBrokerProcessorTest {
@Mock
private Channel channel;
+ private final AuthConfig authConfig = new AuthConfig();
+
@Spy
private BrokerController brokerController = new BrokerController(new
BrokerConfig(), new NettyServerConfig(), new NettyClientConfig(),
- new MessageStoreConfig(), null);
+ new MessageStoreConfig(), authConfig);
@Mock
private MessageStore messageStore;
@@ -254,6 +264,8 @@ public class AdminBrokerProcessorTest {
@Before
public void init() throws Exception {
+ authConfig.setAuthenticationEnabled(true);
+ authConfig.setAuthenticationWhitelist(null);
brokerController.setMessageStore(messageStore);
brokerController.setAuthenticationMetadataManager(authenticationMetadataManager);
brokerController.setAuthorizationMetadataManager(authorizationMetadataManager);
@@ -338,6 +350,36 @@ public class AdminBrokerProcessorTest {
assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);
}
+ @Test
+ public void testResumeCheckHalfMessageRejectsMismatchedTopic() throws
Exception {
+ RemotingCommand request = createResumeCheckHalfMessageCommand();
+ when(messageStore.selectOneMessageByOffset(any(Long.class)))
+ .thenReturn(createSelectMappedBufferResult("otherTopic"));
+
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+
+ assertThat(response.getCode()).isEqualTo(ResponseCode.NO_PERMISSION);
+ assertThat(response.getRemark()).isEqualTo("The topic does not match
the transaction message");
+ verify(messageStore, never()).putMessage(any());
+ }
+
+ @Test
+ public void testResumeCheckHalfMessageAcceptsMatchingTopic() throws
Exception {
+ RemotingCommand request = createResumeCheckHalfMessageCommand();
+ when(messageStore.selectOneMessageByOffset(any(Long.class)))
+ .thenReturn(createSelectMappedBufferResult("topic"));
+ when(messageStore.putMessage(any(MessageExtBrokerInner.class)))
+ .thenReturn(new PutMessageResult(PutMessageStatus.PUT_OK, null));
+
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+ ArgumentCaptor<MessageExtBrokerInner> messageCaptor =
ArgumentCaptor.forClass(MessageExtBrokerInner.class);
+ verify(messageStore).putMessage(messageCaptor.capture());
+
assertThat(messageCaptor.getValue().getProperty(MessageConst.PROPERTY_REAL_TOPIC)).isEqualTo("topic");
+
assertThat(messageCaptor.getValue().getProperty(MessageConst.PROPERTY_TRANSACTION_CHECK_TIMES)).isEqualTo("0");
+ }
+
@Test
public void testUpdateAndCreateTopicInRocksdb() throws Exception {
if (notToBeExecuted()) {
@@ -1282,7 +1324,7 @@ public class AdminBrokerProcessorTest {
getUserRequestHeader.setUsername("abc");
RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.AUTH_GET_USER,
getUserRequestHeader);
request.setVersion(441);
- request.addExtField("AccessKey", "rocketmq");
+ request.addExtField("AccessKey", "abc");
request.makeCustomHeaderToNet();
RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
@@ -1292,9 +1334,54 @@ public class AdminBrokerProcessorTest {
assertThat(userInfo.getUserType()).isEqualTo("Normal");
}
+ @Test
+ public void testGetUserBySuperUser() throws RemotingCommandException {
+ when(authenticationMetadataManager.isSuperUser(eq("rocketmq")))
+ .thenReturn(CompletableFuture.completedFuture(true));
+ when(authenticationMetadataManager.getUser(eq("abc")))
+ .thenReturn(CompletableFuture.completedFuture(
+ User.of("abc", "normal-user-secret", UserType.NORMAL)));
+
+ GetUserRequestHeader requestHeader = new GetUserRequestHeader("abc");
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.AUTH_GET_USER, requestHeader);
+ request.addExtField("AccessKey", "rocketmq");
+ request.makeCustomHeaderToNet();
+
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+ UserInfo userInfo = JSON.parseObject(
+ new String(response.getBody(), StandardCharsets.UTF_8),
UserInfo.class);
+ assertThat(userInfo.getPassword()).isEqualTo("normal-user-secret");
+ }
+
+ @Test
+ public void testGetUserByOtherNormalUserDoesNotReturnPassword() throws
RemotingCommandException {
+ when(authenticationMetadataManager.isSuperUser(eq("other")))
+ .thenReturn(CompletableFuture.completedFuture(false));
+ when(authenticationMetadataManager.getUser(eq("abc")))
+ .thenReturn(CompletableFuture.completedFuture(
+ User.of("abc", "normal-user-secret", UserType.NORMAL)));
+
+ GetUserRequestHeader requestHeader = new GetUserRequestHeader("abc");
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.AUTH_GET_USER, requestHeader);
+ request.addExtField("AccessKey", "other");
+ request.makeCustomHeaderToNet();
+
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+ String content = new String(response.getBody(),
StandardCharsets.UTF_8);
+ UserInfo userInfo = JSON.parseObject(content, UserInfo.class);
+ assertThat(userInfo.getUsername()).isEqualTo("abc");
+ assertThat(userInfo.getPassword()).isNull();
+ assertThat(content).doesNotContain("password", "normal-user-secret");
+ }
+
@Test
public void testListUser() throws RemotingCommandException {
-
when(authenticationMetadataManager.listUser(eq("abc"))).thenReturn(CompletableFuture.completedFuture(Arrays.asList(User.of("abc",
"123", UserType.NORMAL))));
+ when(authenticationMetadataManager.listUser(eq("abc"))).thenReturn(
+ CompletableFuture.completedFuture(Arrays.asList(
+ User.of("abc", "normal-user-secret", UserType.NORMAL),
+ User.of("super", "super-user-secret", UserType.SUPER))));
ListUsersRequestHeader listUserRequestHeader = new
ListUsersRequestHeader();
listUserRequestHeader.setFilter("abc");
@@ -1304,10 +1391,81 @@ public class AdminBrokerProcessorTest {
request.makeCustomHeaderToNet();
RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
- List<UserInfo> userInfo = JSON.parseArray(new
String(response.getBody()), UserInfo.class);
- assertThat(userInfo.get(0).getUsername()).isEqualTo("abc");
- assertThat(userInfo.get(0).getPassword()).isEqualTo("123");
- assertThat(userInfo.get(0).getUserType()).isEqualTo("Normal");
+ String content = new String(response.getBody(),
StandardCharsets.UTF_8);
+ List<UserInfo> users = JSON.parseArray(content, UserInfo.class);
+ assertThat(users.get(0).getUsername()).isEqualTo("abc");
+ assertThat(users.get(0).getUserType()).isEqualTo("Normal");
+ assertThat(users).allMatch(user -> user.getPassword() == null);
+ assertThat(content).doesNotContain("password", "normal-user-secret",
"super-user-secret");
+ }
+
+ @Test
+ public void testListUserByNormalUserDoesNotReturnPassword() throws
RemotingCommandException {
+ when(authenticationMetadataManager.listUser(eq("abc")))
+
.thenReturn(CompletableFuture.completedFuture(Collections.singletonList(
+ User.of("abc", "normal-user-secret", UserType.NORMAL))));
+
+ ListUsersRequestHeader requestHeader = new
ListUsersRequestHeader("abc");
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.AUTH_LIST_USER, requestHeader);
+ request.addExtField("AccessKey", "abc");
+ request.makeCustomHeaderToNet();
+
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+ String content = new String(response.getBody(),
StandardCharsets.UTF_8);
+ List<UserInfo> users = JSON.parseArray(content, UserInfo.class);
+ assertThat(users).hasSize(1);
+ assertThat(users.get(0).getPassword()).isNull();
+ assertThat(content).doesNotContain("password", "normal-user-secret");
+ verify(authenticationMetadataManager,
never()).isSuperUser(anyString());
+ }
+
+ @Test
+ public void testGetUserWithoutAuthenticatedIdentityDoesNotReturnPassword()
throws RemotingCommandException {
+ when(authenticationMetadataManager.getUser(eq("abc")))
+ .thenReturn(CompletableFuture.completedFuture(
+ User.of("abc", "normal-user-secret", UserType.NORMAL)));
+ GetUserRequestHeader requestHeader = new GetUserRequestHeader("abc");
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.AUTH_GET_USER, requestHeader);
+ request.makeCustomHeaderToNet();
+
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+ String content = new String(response.getBody(),
StandardCharsets.UTF_8);
+ assertThat(JSON.parseObject(content,
UserInfo.class).getPassword()).isNull();
+ assertThat(content).doesNotContain("password", "normal-user-secret");
+ }
+
+ @Test
+ public void testGetUserDoesNotTrustAccessKeyWhenAuthenticationDisabled()
throws RemotingCommandException {
+ authConfig.setAuthenticationEnabled(false);
+ when(authenticationMetadataManager.getUser(eq("abc")))
+ .thenReturn(CompletableFuture.completedFuture(
+ User.of("abc", "normal-user-secret", UserType.NORMAL)));
+ GetUserRequestHeader requestHeader = new GetUserRequestHeader("abc");
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.AUTH_GET_USER, requestHeader);
+ request.addExtField("AccessKey", "abc");
+ request.makeCustomHeaderToNet();
+
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+ String content = new String(response.getBody(),
StandardCharsets.UTF_8);
+ assertThat(JSON.parseObject(content,
UserInfo.class).getPassword()).isNull();
+ assertThat(content).doesNotContain("password", "normal-user-secret");
+ }
+
+ @Test
+ public void
testGetUserDoesNotTrustAccessKeyWhenAuthenticationIsWhitelisted() throws
RemotingCommandException {
+
authConfig.setAuthenticationWhitelist(String.valueOf(RequestCode.AUTH_GET_USER));
+ when(authenticationMetadataManager.getUser(eq("abc")))
+ .thenReturn(CompletableFuture.completedFuture(
+ User.of("abc", "normal-user-secret", UserType.NORMAL)));
+ GetUserRequestHeader requestHeader = new GetUserRequestHeader("abc");
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.AUTH_GET_USER, requestHeader);
+ request.addExtField("AccessKey", "abc");
+ request.makeCustomHeaderToNet();
+
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+ String content = new String(response.getBody(),
StandardCharsets.UTF_8);
+ assertThat(JSON.parseObject(content,
UserInfo.class).getPassword()).isNull();
+ assertThat(content).doesNotContain("password", "normal-user-secret");
}
@Test
@@ -1895,6 +2053,17 @@ public class AdminBrokerProcessorTest {
return result;
}
+ private SelectMappedBufferResult createSelectMappedBufferResult(String
realTopic) throws Exception {
+ MessageExt message = createDefaultMessageExt();
+ message.setBody("body".getBytes(StandardCharsets.UTF_8));
+ message.setTopic(TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC);
+ message.setBornHost(new InetSocketAddress("127.0.0.1", 10911));
+ message.setStoreHost(new InetSocketAddress("127.0.0.1", 10911));
+ MessageAccessor.putProperty(message, MessageConst.PROPERTY_REAL_TOPIC,
realTopic);
+ ByteBuffer buffer = ByteBuffer.wrap(MessageDecoder.encode(message,
false));
+ return new SelectMappedBufferResult(0, buffer, buffer.remaining(),
null);
+ }
+
private ResumeCheckHalfMessageRequestHeader
createResumeCheckHalfMessageRequestHeader() {
ResumeCheckHalfMessageRequestHeader header = new
ResumeCheckHalfMessageRequestHeader();
header.setTopic("topic");
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/processor/EndTransactionProcessorTest.java
b/broker/src/test/java/org/apache/rocketmq/broker/processor/EndTransactionProcessorTest.java
index 42af2dbe47..7b2d4bdf26 100644
---
a/broker/src/test/java/org/apache/rocketmq/broker/processor/EndTransactionProcessorTest.java
+++
b/broker/src/test/java/org/apache/rocketmq/broker/processor/EndTransactionProcessorTest.java
@@ -29,6 +29,7 @@ import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.common.stats.Stats;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
+import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyClientConfig;
import org.apache.rocketmq.remoting.netty.NettyServerConfig;
@@ -53,6 +54,8 @@ import java.nio.charset.StandardCharsets;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
@@ -123,6 +126,33 @@ public class EndTransactionProcessorTest {
assertThat(brokerController.getBrokerStatsManager().getStatsItem(Stats.TOPIC_PUT_SIZE,
TOPIC).getValue().sum()).isEqualTo(1L);
}
+ @Test
+ public void testProcessRequestRejectsMismatchedTopic() throws
RemotingCommandException {
+
when(transactionMsgService.commitMessage(any(EndTransactionRequestHeader.class)))
+ .thenReturn(createResponse(ResponseCode.SUCCESS));
+ RemotingCommand request = createEndTransactionMsgCommand(
+ MessageSysFlag.TRANSACTION_COMMIT_TYPE, false, "allowedTopic");
+
+ RemotingCommand response =
endTransactionProcessor.processRequest(handlerContext, request);
+
+ assertThat(response.getCode()).isEqualTo(ResponseCode.NO_PERMISSION);
+ verify(messageStore,
never()).putMessage(any(MessageExtBrokerInner.class));
+ verify(transactionMsgService,
never()).deletePrepareMessage(any(MessageExt.class));
+ }
+
+ @Test
+ public void testProcessRequestRejectsMismatchedTopicOnRollback() throws
RemotingCommandException {
+
when(transactionMsgService.rollbackMessage(any(EndTransactionRequestHeader.class)))
+ .thenReturn(createResponse(ResponseCode.SUCCESS));
+ RemotingCommand request = createEndTransactionMsgCommand(
+ MessageSysFlag.TRANSACTION_ROLLBACK_TYPE, false, "allowedTopic");
+
+ RemotingCommand response =
endTransactionProcessor.processRequest(handlerContext, request);
+
+ assertThat(response.getCode()).isEqualTo(ResponseCode.NO_PERMISSION);
+ verify(transactionMsgService,
never()).deletePrepareMessage(any(MessageExt.class));
+ }
+
@Test
public void testProcessRequest_NotType() throws RemotingCommandException {
RemotingCommand request =
createEndTransactionMsgCommand(MessageSysFlag.TRANSACTION_NOT_TYPE, true);
@@ -175,6 +205,7 @@ public class EndTransactionProcessorTest {
private MessageExt createDefaultMessageExt() {
MessageExt messageExt = new MessageExt();
messageExt.setMsgId("12345678");
+ messageExt.setTopic(TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC);
messageExt.setQueueId(0);
messageExt.setCommitLogOffset(123456789L);
messageExt.setQueueOffset(1234);
@@ -199,7 +230,12 @@ public class EndTransactionProcessorTest {
}
private RemotingCommand createEndTransactionMsgCommand(int status, boolean
isCheckMsg) {
+ return createEndTransactionMsgCommand(status, isCheckMsg, TOPIC);
+ }
+
+ private RemotingCommand createEndTransactionMsgCommand(int status, boolean
isCheckMsg, String topic) {
EndTransactionRequestHeader header =
createEndTransactionRequestHeader(status, isCheckMsg);
+ header.setTopic(topic);
RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.END_TRANSACTION, header);
request.makeCustomHeaderToNet();
return request;
@@ -215,6 +251,7 @@ public class EndTransactionProcessorTest {
private MessageExt createRejectMessageExt() {
MessageExt messageExt = new MessageExt();
messageExt.setMsgId("12345678");
+ messageExt.setTopic(TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC);
messageExt.setQueueId(0);
messageExt.setCommitLogOffset(123456789L);
messageExt.setQueueOffset(1234);
@@ -223,7 +260,7 @@ public class EndTransactionProcessorTest {
MessageAccessor.putProperty(messageExt,
MessageConst.PROPERTY_REAL_QUEUE_ID, "0");
MessageAccessor.putProperty(messageExt,
MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
MessageAccessor.putProperty(messageExt,
MessageConst.PROPERTY_PRODUCER_GROUP, "testTransactionGroup");
- MessageAccessor.putProperty(messageExt,
MessageConst.PROPERTY_REAL_TOPIC, "TEST");
+ MessageAccessor.putProperty(messageExt,
MessageConst.PROPERTY_REAL_TOPIC, TOPIC);
MessageAccessor.putProperty(messageExt,
MessageConst.PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS, "60");
return messageExt;
}
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/processor/QueryMessageProcessorTest.java
b/broker/src/test/java/org/apache/rocketmq/broker/processor/QueryMessageProcessorTest.java
index f842e8da6b..bc77a3a0cb 100644
---
a/broker/src/test/java/org/apache/rocketmq/broker/processor/QueryMessageProcessorTest.java
+++
b/broker/src/test/java/org/apache/rocketmq/broker/processor/QueryMessageProcessorTest.java
@@ -189,13 +189,14 @@ public class QueryMessageProcessorTest {
TopicValidator.RMQ_SYS_SCHEDULE_TOPIC,
TimerMessageStore.TIMER_TOPIC,
TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC,
+ TopicValidator.RMQ_SYS_ROCKSDB_TRANS_HALF_TOPIC,
TopicValidator.RMQ_SYS_TRANS_CHECK_MAX_TIME_TOPIC)) {
when(messageStore.selectOneMessageByOffset(0L))
.thenReturn(messageResult(storedTopic, "actualTopic"));
Assert.assertNull(queryMessageProcessor.processRequest(handlerContext,
request));
}
- verify(channel, times(4)).writeAndFlush(any());
+ verify(channel, times(5)).writeAndFlush(any());
}
@Test
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/transaction/queue/DefaultTransactionalMessageCheckListenerTest.java
b/broker/src/test/java/org/apache/rocketmq/broker/transaction/queue/DefaultTransactionalMessageCheckListenerTest.java
index 986b15aa09..3b5cf09282 100644
---
a/broker/src/test/java/org/apache/rocketmq/broker/transaction/queue/DefaultTransactionalMessageCheckListenerTest.java
+++
b/broker/src/test/java/org/apache/rocketmq/broker/transaction/queue/DefaultTransactionalMessageCheckListenerTest.java
@@ -16,8 +16,11 @@
*/
package org.apache.rocketmq.broker.transaction.queue;
+import io.netty.channel.Channel;
import java.net.InetSocketAddress;
import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.client.ProducerManager;
+import org.apache.rocketmq.broker.client.net.Broker2Client;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
@@ -26,22 +29,35 @@ import
org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.remoting.netty.NettyClientConfig;
import org.apache.rocketmq.remoting.netty.NettyServerConfig;
+import
org.apache.rocketmq.remoting.protocol.header.CheckTransactionStateRequestHeader;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.Spy;
import org.mockito.junit.MockitoJUnitRunner;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
@RunWith(MockitoJUnitRunner.class)
public class DefaultTransactionalMessageCheckListenerTest {
private DefaultTransactionalMessageCheckListener listener;
@Mock
private MessageStore messageStore;
+ @Mock
+ private ProducerManager producerManager;
+ @Mock
+ private Broker2Client broker2Client;
+ @Mock
+ private Channel channel;
@Spy
private BrokerController brokerController = new BrokerController(new
BrokerConfig(),
@@ -69,7 +85,21 @@ public class DefaultTransactionalMessageCheckListenerTest {
@Test
public void testSendCheckMessage() throws Exception {
MessageExt messageExt = createMessageExt();
+ String producerGroup = "producerGroup";
+ messageExt.setTopic(TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC);
+ MessageAccessor.putProperty(messageExt,
MessageConst.PROPERTY_PRODUCER_GROUP, producerGroup);
+
when(brokerController.getProducerManager()).thenReturn(producerManager);
+
when(producerManager.getAvailableChannel(producerGroup)).thenReturn(channel);
+ when(brokerController.getBroker2Client()).thenReturn(broker2Client);
+
listener.sendCheckMessage(messageExt);
+
+ ArgumentCaptor<CheckTransactionStateRequestHeader> headerCaptor =
+ ArgumentCaptor.forClass(CheckTransactionStateRequestHeader.class);
+ verify(broker2Client).checkProducerTransactionState(
+ eq(producerGroup), eq(channel), headerCaptor.capture(),
eq(messageExt));
+ assertThat(headerCaptor.getValue().getTopic()).isEqualTo("realTopic");
+ assertThat(messageExt.getTopic()).isEqualTo("realTopic");
}
@Test
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/transaction/rocksdb/TransactionalMessageRocksDBServiceTest.java
b/broker/src/test/java/org/apache/rocketmq/broker/transaction/rocksdb/TransactionalMessageRocksDBServiceTest.java
new file mode 100644
index 0000000000..c8940fc85d
--- /dev/null
+++
b/broker/src/test/java/org/apache/rocketmq/broker/transaction/rocksdb/TransactionalMessageRocksDBServiceTest.java
@@ -0,0 +1,100 @@
+/*
+ * 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.broker.transaction.rocksdb;
+
+import io.netty.channel.Channel;
+import java.lang.reflect.Method;
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.client.ProducerManager;
+import org.apache.rocketmq.broker.client.net.Broker2Client;
+import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.message.MessageAccessor;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.topic.TopicValidator;
+import
org.apache.rocketmq.remoting.protocol.header.CheckTransactionStateRequestHeader;
+import org.apache.rocketmq.store.MessageStore;
+import org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage;
+import org.apache.rocketmq.store.transaction.TransMessageRocksDBStore;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class TransactionalMessageRocksDBServiceTest {
+ @Mock
+ private MessageStore messageStore;
+ @Mock
+ private TransMessageRocksDBStore transMessageRocksDBStore;
+ @Mock
+ private MessageRocksDBStorage messageRocksDBStorage;
+ @Mock
+ private BrokerController brokerController;
+ @Mock
+ private BrokerConfig brokerConfig;
+ @Mock
+ private ProducerManager producerManager;
+ @Mock
+ private Broker2Client broker2Client;
+ @Mock
+ private Channel channel;
+
+ private TransactionalMessageRocksDBService service;
+
+ @Before
+ public void setUp() {
+
when(messageStore.getTransMessageRocksDBStore()).thenReturn(transMessageRocksDBStore);
+
when(transMessageRocksDBStore.getMessageRocksDBStorage()).thenReturn(messageRocksDBStorage);
+ service = new TransactionalMessageRocksDBService(messageStore,
brokerController);
+ }
+
+ @Test
+ public void testSendCheckMessageUsesRealTopicInRequestHeader() throws
Exception {
+ String realTopic = "realTopic";
+ String producerGroup = "producerGroup";
+ MessageExt messageExt = new MessageExt();
+ messageExt.setTopic(TopicValidator.RMQ_SYS_ROCKSDB_TRANS_HALF_TOPIC);
+ MessageAccessor.putProperty(messageExt,
MessageConst.PROPERTY_REAL_TOPIC, realTopic);
+ MessageAccessor.putProperty(messageExt,
MessageConst.PROPERTY_REAL_QUEUE_ID, "1");
+ MessageAccessor.putProperty(messageExt,
MessageConst.PROPERTY_PRODUCER_GROUP, producerGroup);
+ MessageAccessor.putProperty(messageExt,
MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, "messageId");
+ when(brokerController.getBrokerConfig()).thenReturn(brokerConfig);
+
when(brokerController.getProducerManager()).thenReturn(producerManager);
+
when(producerManager.getAvailableChannel(producerGroup)).thenReturn(channel);
+ when(brokerController.getBroker2Client()).thenReturn(broker2Client);
+
+ Method sendCheckMessage = TransactionalMessageRocksDBService.class
+ .getDeclaredMethod("sendCheckMessage", MessageExt.class);
+ sendCheckMessage.setAccessible(true);
+ sendCheckMessage.invoke(service, messageExt);
+
+ ArgumentCaptor<CheckTransactionStateRequestHeader> headerCaptor =
+ ArgumentCaptor.forClass(CheckTransactionStateRequestHeader.class);
+ verify(broker2Client).checkProducerTransactionState(
+ eq(producerGroup), eq(channel), headerCaptor.capture(),
eq(messageExt));
+ assertThat(headerCaptor.getValue().getTopic()).isEqualTo(realTopic);
+ assertThat(messageExt.getTopic()).isEqualTo(realTopic);
+ }
+}
diff --git
a/client/src/test/java/org/apache/rocketmq/client/producer/selector/DefaultMQProducerImplTest.java
b/client/src/test/java/org/apache/rocketmq/client/producer/selector/DefaultMQProducerImplTest.java
index 77a83af19c..bb61f9afd9 100644
---
a/client/src/test/java/org/apache/rocketmq/client/producer/selector/DefaultMQProducerImplTest.java
+++
b/client/src/test/java/org/apache/rocketmq/client/producer/selector/DefaultMQProducerImplTest.java
@@ -27,6 +27,7 @@ import org.apache.rocketmq.client.impl.MQClientAPIImpl;
import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
+import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.RequestCallback;
import org.apache.rocketmq.client.producer.SendCallback;
@@ -36,15 +37,18 @@ import
org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.ServiceState;
import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.producer.RecallMessageHandle;
import org.apache.rocketmq.remoting.exception.RemotingException;
import
org.apache.rocketmq.remoting.protocol.header.CheckTransactionStateRequestHeader;
+import
org.apache.rocketmq.remoting.protocol.header.EndTransactionRequestHeader;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
@@ -66,10 +70,13 @@ import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertThrows;
import static org.mockito.AdditionalMatchers.or;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
@@ -145,6 +152,39 @@ public class DefaultMQProducerImplTest {
defaultMQProducerImpl.checkTransactionState(defaultBrokerAddr,
mock(MessageExt.class), mock(CheckTransactionStateRequestHeader.class));
}
+ @Test
+ public void testCheckTransactionStateEchoesTopicToEndTransaction() throws
Exception {
+ String realTopic = "realTopic";
+ MessageExt messageExt = new MessageExt();
+ messageExt.setMsgId("messageId");
+ MessageAccessor.putProperty(
+ messageExt, MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX,
"messageId");
+ CheckTransactionStateRequestHeader checkHeader = new
CheckTransactionStateRequestHeader();
+ checkHeader.setTopic(realTopic);
+ checkHeader.setCommitLogOffset(123L);
+ checkHeader.setTranStateTableOffset(456L);
+ TransactionListener transactionListener =
mock(TransactionListener.class);
+ when(transactionListener.checkLocalTransaction(messageExt))
+ .thenReturn(LocalTransactionState.COMMIT_MESSAGE);
+ ((TransactionMQProducer) defaultMQProducerImpl.getDefaultMQProducer())
+ .setTransactionListener(transactionListener);
+ ExecutorService checkExecutor = mock(ExecutorService.class);
+ doAnswer(invocation -> {
+ invocation.<Runnable>getArgument(0).run();
+ return null;
+ }).when(checkExecutor).submit(any(Runnable.class));
+ setField(defaultMQProducerImpl, "checkExecutor", checkExecutor);
+
+ defaultMQProducerImpl.checkTransactionState(defaultBrokerAddr,
messageExt, checkHeader);
+
+ ArgumentCaptor<EndTransactionRequestHeader> endHeaderCaptor =
+ ArgumentCaptor.forClass(EndTransactionRequestHeader.class);
+ verify(mQClientAPIImpl).endTransactionOneway(
+ eq(defaultBrokerAddr), endHeaderCaptor.capture(), isNull(),
eq(3000L));
+ assertEquals(realTopic, endHeaderCaptor.getValue().getTopic());
+ assertTrue(endHeaderCaptor.getValue().getFromTransactionCheck());
+ }
+
@Test
public void testCreateTopic() throws MQClientException {
defaultMQProducerImpl.createTopic("key", defaultTopic, 0);
diff --git
a/container/src/main/java/org/apache/rocketmq/container/BrokerContainerProcessor.java
b/container/src/main/java/org/apache/rocketmq/container/BrokerContainerProcessor.java
index cfedb24aee..5d534965fe 100644
---
a/container/src/main/java/org/apache/rocketmq/container/BrokerContainerProcessor.java
+++
b/container/src/main/java/org/apache/rocketmq/container/BrokerContainerProcessor.java
@@ -280,6 +280,30 @@ public class BrokerContainerProcessor implements
NettyRequestProcessor {
return response;
}
+ // sensitive config keys that must never be returned by config query
interfaces
+ private static final String[] SENSITIVE_CONFIG_KEYS = new String[] {
+ "initAuthenticationUser", "innerClientAuthenticationCredentials"
+ };
+
+ /**
+ * Remove sensitive entries from the exported config content. Returns null
when the
+ * content cannot be parsed, so callers must fail closed instead of
returning the
+ * original content.
+ */
+ static String sanitizeConfigForResponse(String content) {
+ if (content == null || content.isEmpty()) {
+ return content;
+ }
+ Properties properties = MixAll.string2Properties(content);
+ if (properties == null) {
+ return null;
+ }
+ for (String key : SENSITIVE_CONFIG_KEYS) {
+ properties.remove(key);
+ }
+ return MixAll.properties2String(properties, true);
+ }
+
private boolean validateBlackListConfigExist(Properties properties) {
for (String blackConfig : configBlackList) {
if (properties.containsKey(blackConfig)) {
@@ -294,8 +318,14 @@ public class BrokerContainerProcessor implements
NettyRequestProcessor {
final RemotingCommand response =
RemotingCommand.createResponseCommand(GetBrokerConfigResponseHeader.class);
final GetBrokerConfigResponseHeader responseHeader =
(GetBrokerConfigResponseHeader) response.readCustomHeader();
- String content =
this.brokerContainer.getConfiguration().getAllConfigsFormatString();
- if (content != null && content.length() > 0) {
+ String content =
sanitizeConfigForResponse(this.brokerContainer.getConfiguration().getAllConfigsFormatString());
+ if (content == null) {
+ LOGGER.error("BrokerContainerProcessor#getBrokerConfig: failed to
sanitize broker config");
+ response.setCode(ResponseCode.SYSTEM_ERROR);
+ response.setRemark("Failed to sanitize broker config");
+ return response;
+ }
+ if (content.length() > 0) {
try {
content = MixAll.adjustConfigForPlatform(content);
response.setBody(content.getBytes(MixAll.DEFAULT_CHARSET));
diff --git
a/container/src/test/java/org/apache/rocketmq/container/BrokerContainerProcessorConfigSanitizeTest.java
b/container/src/test/java/org/apache/rocketmq/container/BrokerContainerProcessorConfigSanitizeTest.java
new file mode 100644
index 0000000000..0aaa6d2c69
--- /dev/null
+++
b/container/src/test/java/org/apache/rocketmq/container/BrokerContainerProcessorConfigSanitizeTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.container;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+public class BrokerContainerProcessorConfigSanitizeTest {
+
+ @Test
+ public void testSanitizeRemovesSensitiveKeys() {
+ String content = "brokerName=broker-a\n"
+ +
"initAuthenticationUser={\"username\":\"rocketmq\",\"password\":\"secret\"}\n"
+ +
"innerClientAuthenticationCredentials={\"accessKey\":\"ak\",\"secretKey\":\"sk\"}\n"
+ + "listenPort=10911\n";
+
+ String sanitized =
BrokerContainerProcessor.sanitizeConfigForResponse(content);
+
+ assertFalse(sanitized.contains("initAuthenticationUser"));
+
assertFalse(sanitized.contains("innerClientAuthenticationCredentials"));
+ assertFalse(sanitized.contains("secret"));
+ assertTrue(sanitized.contains("brokerName=broker-a"));
+ assertTrue(sanitized.contains("listenPort=10911"));
+ }
+
+ @Test
+ public void testSanitizeFailsClosedOnUnparsableContent() {
+ // malformed unicode escape makes Properties.load throw
+
assertNull(BrokerContainerProcessor.sanitizeConfigForResponse("key=\\uZZZZ\n"));
+ }
+
+ @Test
+ public void testSanitizePassesThroughNullOrEmpty() {
+ assertNull(BrokerContainerProcessor.sanitizeConfigForResponse(null));
+ assertEquals("",
BrokerContainerProcessor.sanitizeConfigForResponse(""));
+ }
+}
diff --git
a/proxy/src/main/java/org/apache/rocketmq/proxy/common/utils/GrpcUtils.java
b/proxy/src/main/java/org/apache/rocketmq/proxy/common/utils/GrpcUtils.java
index 5c50de4426..0e4a72d26f 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/common/utils/GrpcUtils.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/common/utils/GrpcUtils.java
@@ -35,6 +35,19 @@ public class GrpcUtils {
}
}
+ /**
+ * Replaces the current value for a metadata key. A null value clears the
key.
+ */
+ public static <T> void putHeader(Metadata headers, Metadata.Key<T> key, T
value) {
+ if (headers == null) {
+ return;
+ }
+ headers.discardAll(key);
+ if (value != null) {
+ headers.put(key, value);
+ }
+ }
+
public static <R, W, T> T getAttribute(ServerCall<R, W> call,
Attributes.Key<T> key) {
Attributes attributes = call.getAttributes();
if (attributes == null) {
diff --git
a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/interceptor/HeaderInterceptor.java
b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/interceptor/HeaderInterceptor.java
index e3e7884155..cd3db34f73 100644
---
a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/interceptor/HeaderInterceptor.java
+++
b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/interceptor/HeaderInterceptor.java
@@ -40,16 +40,19 @@ public class HeaderInterceptor implements ServerInterceptor
{
Metadata headers,
ServerCallHandler<R, W> next
) {
+ // The authorization subject must only come from verified
authentication context.
+ GrpcUtils.putHeader(headers, GrpcConstants.AUTHORIZATION_AK, null);
+
String remoteAddress = getProxyProtocolAddress(call.getAttributes());
if (StringUtils.isBlank(remoteAddress)) {
SocketAddress remoteSocketAddress =
call.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR);
remoteAddress = parseSocketAddress(remoteSocketAddress);
}
- GrpcUtils.putHeaderIfNotExist(headers, GrpcConstants.REMOTE_ADDRESS,
remoteAddress);
+ GrpcUtils.putHeader(headers, GrpcConstants.REMOTE_ADDRESS,
remoteAddress);
SocketAddress localSocketAddress =
call.getAttributes().get(Grpc.TRANSPORT_ATTR_LOCAL_ADDR);
String localAddress = parseSocketAddress(localSocketAddress);
- GrpcUtils.putHeaderIfNotExist(headers, GrpcConstants.LOCAL_ADDRESS,
localAddress);
+ GrpcUtils.putHeader(headers, GrpcConstants.LOCAL_ADDRESS,
localAddress);
for (Attributes.Key<?> key : call.getAttributes().keys()) {
if (!StringUtils.startsWith(key.toString(),
HAProxyConstants.PROXY_PROTOCOL_PREFIX)) {
@@ -58,12 +61,12 @@ public class HeaderInterceptor implements ServerInterceptor
{
Metadata.Key<String> headerKey
= Metadata.Key.of(key.toString(),
Metadata.ASCII_STRING_MARSHALLER);
String headerValue = String.valueOf(call.getAttributes().get(key));
- GrpcUtils.putHeaderIfNotExist(headers, headerKey, headerValue);
+ GrpcUtils.putHeader(headers, headerKey, headerValue);
}
String channelId = call.getAttributes().get(AttributeKeys.CHANNEL_ID);
if (StringUtils.isNotBlank(channelId)) {
- GrpcUtils.putHeaderIfNotExist(headers, GrpcConstants.CHANNEL_ID,
channelId);
+ GrpcUtils.putHeader(headers, GrpcConstants.CHANNEL_ID, channelId);
}
return next.startCall(call, headers);
diff --git
a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/pipeline/AuthenticationPipeline.java
b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/pipeline/AuthenticationPipeline.java
index e317b48f1e..2720d19796 100644
---
a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/pipeline/AuthenticationPipeline.java
+++
b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/pipeline/AuthenticationPipeline.java
@@ -17,7 +17,6 @@
package org.apache.rocketmq.proxy.grpc.pipeline;
import com.google.protobuf.GeneratedMessageV3;
-import io.grpc.Context;
import io.grpc.Metadata;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.auth.authentication.AuthenticationEvaluator;
@@ -46,13 +45,17 @@ public class AuthenticationPipeline implements
RequestPipeline {
@Override
public void execute(ProxyContext context, Metadata headers,
GeneratedMessageV3 request) {
+ GrpcUtils.putHeader(headers, GrpcConstants.AUTHORIZATION_AK, null);
if (!authConfig.isAuthenticationEnabled()) {
return;
}
try {
- Metadata metadata = GrpcConstants.METADATA.get(Context.current());
- AuthenticationContext authenticationContext = newContext(context,
metadata, request);
+ AuthenticationContext authenticationContext = newContext(context,
headers, request);
authenticationEvaluator.evaluate(authenticationContext);
+ if (authenticationContext != null
+ &&
authConfig.isAuthenticationRequired(authenticationContext.getRpcCode())) {
+ putAuthenticatedIdentity(headers, authenticationContext);
+ }
} catch (AuthenticationException ex) {
throw ex;
} catch (Throwable ex) {
@@ -70,13 +73,17 @@ public class AuthenticationPipeline implements
RequestPipeline {
* @return
*/
protected AuthenticationContext newContext(ProxyContext context, Metadata
headers, GeneratedMessageV3 request) {
- AuthenticationContext result =
AuthenticationFactory.newContext(authConfig, headers, request);
- if (result instanceof DefaultAuthenticationContext) {
- DefaultAuthenticationContext defaultAuthenticationContext =
(DefaultAuthenticationContext) result;
+ return AuthenticationFactory.newContext(authConfig, headers, request);
+ }
+
+ private void putAuthenticatedIdentity(Metadata metadata,
AuthenticationContext authenticationContext) {
+ if (authenticationContext instanceof DefaultAuthenticationContext) {
+ DefaultAuthenticationContext defaultAuthenticationContext =
+ (DefaultAuthenticationContext) authenticationContext;
if
(StringUtils.isNotBlank(defaultAuthenticationContext.getUsername())) {
- GrpcUtils.putHeaderIfNotExist(headers,
GrpcConstants.AUTHORIZATION_AK, defaultAuthenticationContext.getUsername());
+ GrpcUtils.putHeader(metadata, GrpcConstants.AUTHORIZATION_AK,
+ defaultAuthenticationContext.getUsername());
}
}
- return result;
}
}
diff --git
a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivity.java
b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivity.java
index 2d09c39429..ae574994a3 100644
---
a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivity.java
+++
b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivity.java
@@ -21,6 +21,8 @@ import io.netty.channel.ChannelHandlerContext;
import java.util.HashMap;
import java.util.Map;
import org.apache.rocketmq.acl.common.AclException;
+import
org.apache.rocketmq.auth.authentication.exception.AuthenticationException;
+import org.apache.rocketmq.auth.authorization.exception.AuthorizationException;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.constant.LoggerName;
@@ -129,7 +131,8 @@ public abstract class AbstractRemotingActivity implements
NettyRequestProcessor
} else if (t instanceof MQBrokerException) {
MQBrokerException e = (MQBrokerException) t;
writeResponse(ctx, context, request,
RemotingCommand.createResponseCommand(e.getResponseCode(),
e.getErrorMessage()), t);
- } else if (t instanceof AclException) {
+ } else if (t instanceof AclException || t instanceof
AuthenticationException
+ || t instanceof AuthorizationException) {
writeResponse(ctx, context, request,
RemotingCommand.createResponseCommand(ResponseCode.NO_PERMISSION,
t.getMessage()), t);
} else {
writeResponse(ctx, context, request,
diff --git
a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/interceptor/HeaderInterceptorTest.java
b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/interceptor/HeaderInterceptorTest.java
new file mode 100644
index 0000000000..f22df0a349
--- /dev/null
+++
b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/interceptor/HeaderInterceptorTest.java
@@ -0,0 +1,150 @@
+/*
+ * 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.proxy.grpc.interceptor;
+
+import io.grpc.Attributes;
+import io.grpc.Grpc;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
+import io.grpc.ServerCall;
+import io.grpc.ServerCallHandler;
+import java.net.InetSocketAddress;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.rocketmq.common.constant.GrpcConstants;
+import org.apache.rocketmq.common.constant.HAProxyConstants;
+import org.apache.rocketmq.proxy.grpc.constant.AttributeKeys;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class HeaderInterceptorTest {
+
+ private static final InetSocketAddress REAL_REMOTE = new
InetSocketAddress("10.0.0.7", 5000);
+ private static final InetSocketAddress REAL_LOCAL = new
InetSocketAddress("10.0.0.1", 8081);
+
+ private Metadata intercept(Metadata inboundHeaders) {
+ final Attributes attributes = Attributes.newBuilder()
+ .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, REAL_REMOTE)
+ .set(Grpc.TRANSPORT_ATTR_LOCAL_ADDR, REAL_LOCAL)
+ .build();
+ return intercept(inboundHeaders, attributes);
+ }
+
+ private Metadata intercept(Metadata inboundHeaders, Attributes attributes)
{
+ HeaderInterceptor interceptor = new HeaderInterceptor();
+
+ ServerCall<Object, Object> call = new ServerCall<Object, Object>() {
+ @Override
+ public Attributes getAttributes() {
+ return attributes;
+ }
+
+ @Override
+ public void request(int numMessages) {
+ }
+
+ @Override
+ public void sendHeaders(Metadata headers) {
+ }
+
+ @Override
+ public void sendMessage(Object message) {
+ }
+
+ @Override
+ public void close(io.grpc.Status status, Metadata trailers) {
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return false;
+ }
+
+ @Override
+ public MethodDescriptor<Object, Object> getMethodDescriptor() {
+ return null;
+ }
+ };
+
+ final AtomicReference<Metadata> forwarded = new AtomicReference<>();
+ ServerCallHandler<Object, Object> next = (c, headers) -> {
+ forwarded.set(headers);
+ return null;
+ };
+
+ interceptor.interceptCall(call, inboundHeaders, next);
+ return forwarded.get();
+ }
+
+ @Test
+ public void shouldSetConnectionAddresses() {
+ Metadata forwarded = intercept(new Metadata());
+
+
assertThat(forwarded.get(GrpcConstants.REMOTE_ADDRESS)).isEqualTo("10.0.0.7:5000");
+
assertThat(forwarded.get(GrpcConstants.LOCAL_ADDRESS)).isEqualTo("10.0.0.1:8081");
+ }
+
+ @Test
+ public void shouldReplaceExistingAddressHeaders() {
+ Metadata headers = new Metadata();
+ headers.put(GrpcConstants.REMOTE_ADDRESS, "127.0.0.1:9999");
+ headers.put(GrpcConstants.LOCAL_ADDRESS, "127.0.0.1:1");
+
+ Metadata forwarded = intercept(headers);
+
+
assertThat(forwarded.getAll(GrpcConstants.REMOTE_ADDRESS)).containsExactly("10.0.0.7:5000");
+
assertThat(forwarded.getAll(GrpcConstants.LOCAL_ADDRESS)).containsExactly("10.0.0.1:8081");
+ }
+
+ @Test
+ public void shouldRemoveClientAuthorizationSubject() {
+ Metadata headers = new Metadata();
+ headers.put(GrpcConstants.AUTHORIZATION_AK, "admin");
+ headers.put(GrpcConstants.AUTHORIZATION_AK, "another-admin");
+
+ Metadata forwarded = intercept(headers);
+
+
assertThat(forwarded.containsKey(GrpcConstants.AUTHORIZATION_AK)).isFalse();
+ }
+
+ @Test
+ public void shouldUseProxyProtocolSourceAndReplaceConnectionMetadata() {
+ Metadata.Key<String> proxyAddress = Metadata.Key.of(
+ HAProxyConstants.PROXY_PROTOCOL_ADDR,
Metadata.ASCII_STRING_MARSHALLER);
+ Metadata headers = new Metadata();
+ headers.put(GrpcConstants.REMOTE_ADDRESS, "127.0.0.1:9999");
+ headers.put(GrpcConstants.LOCAL_ADDRESS, "127.0.0.1:1");
+ headers.put(GrpcConstants.CHANNEL_ID, "client-channel");
+ headers.put(proxyAddress, "127.0.0.1");
+
+ Attributes attributes = Attributes.newBuilder()
+ .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, REAL_REMOTE)
+ .set(Grpc.TRANSPORT_ATTR_LOCAL_ADDR, REAL_LOCAL)
+ .set(AttributeKeys.PROXY_PROTOCOL_ADDR, "203.0.113.7")
+ .set(AttributeKeys.PROXY_PROTOCOL_PORT, "7000")
+ .set(AttributeKeys.CHANNEL_ID, "server-channel")
+ .build();
+
+ Metadata forwarded = intercept(headers, attributes);
+
+
assertThat(forwarded.getAll(GrpcConstants.REMOTE_ADDRESS)).containsExactly("203.0.113.7:7000");
+
assertThat(forwarded.getAll(GrpcConstants.LOCAL_ADDRESS)).containsExactly("10.0.0.1:8081");
+
assertThat(forwarded.getAll(GrpcConstants.CHANNEL_ID)).containsExactly("server-channel");
+
assertThat(forwarded.getAll(proxyAddress)).containsExactly("203.0.113.7");
+ }
+}
diff --git
a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/pipeline/AuthenticationPipelineTest.java
b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/pipeline/AuthenticationPipelineTest.java
new file mode 100644
index 0000000000..f31a268169
--- /dev/null
+++
b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/pipeline/AuthenticationPipelineTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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.proxy.grpc.pipeline;
+
+import apache.rocketmq.v2.TelemetryCommand;
+import io.grpc.Metadata;
+import java.lang.reflect.Field;
+import org.apache.rocketmq.auth.authentication.AuthenticationEvaluator;
+import org.apache.rocketmq.auth.authentication.context.AuthenticationContext;
+import
org.apache.rocketmq.auth.authentication.context.DefaultAuthenticationContext;
+import org.apache.rocketmq.auth.config.AuthConfig;
+import org.apache.rocketmq.common.constant.GrpcConstants;
+import org.apache.rocketmq.proxy.common.ProxyContext;
+import org.apache.rocketmq.proxy.processor.MessagingProcessor;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+public class AuthenticationPipelineTest {
+
+ @Test
+ public void removesAuthorizationSubjectWhenAuthenticationIsDisabled() {
+ AuthConfig authConfig = new AuthConfig();
+ authConfig.setConfigName("grpc-authentication-disabled-test");
+ AuthenticationPipeline pipeline = new AuthenticationPipeline(
+ authConfig, mock(MessagingProcessor.class));
+ Metadata metadata = new Metadata();
+ metadata.put(GrpcConstants.AUTHORIZATION_AK, "forged-user");
+
+ pipeline.execute(ProxyContext.create(), metadata,
TelemetryCommand.getDefaultInstance());
+
+
assertThat(metadata.containsKey(GrpcConstants.AUTHORIZATION_AK)).isFalse();
+ }
+
+ @Test
+ public void replacesAuthorizationSubjectAfterAuthentication() throws
Exception {
+ AuthConfig authConfig = new AuthConfig();
+ authConfig.setConfigName("grpc-authentication-pipeline-test");
+ authConfig.setAuthenticationEnabled(true);
+ DefaultAuthenticationContext authenticationContext = new
DefaultAuthenticationContext();
+ authenticationContext.setRpcCode("authenticated-rpc");
+ authenticationContext.setUsername("verified-user");
+ AuthenticationPipeline pipeline = new AuthenticationPipeline(
+ authConfig, mock(MessagingProcessor.class)) {
+ @Override
+ protected AuthenticationContext newContext(ProxyContext context,
Metadata headers,
+ com.google.protobuf.GeneratedMessageV3 request) {
+ return authenticationContext;
+ }
+ };
+ AuthenticationEvaluator authenticationEvaluator =
mock(AuthenticationEvaluator.class);
+ Field evaluatorField =
AuthenticationPipeline.class.getDeclaredField("authenticationEvaluator");
+ evaluatorField.setAccessible(true);
+ evaluatorField.set(pipeline, authenticationEvaluator);
+
+ Metadata metadata = new Metadata();
+ metadata.put(GrpcConstants.AUTHORIZATION_AK, "forged-user");
+ metadata.put(GrpcConstants.AUTHORIZATION_AK, "another-forged-user");
+ pipeline.execute(ProxyContext.create(), metadata,
TelemetryCommand.getDefaultInstance());
+
+ verify(authenticationEvaluator).evaluate(authenticationContext);
+
assertThat(metadata.getAll(GrpcConstants.AUTHORIZATION_AK)).containsExactly("verified-user");
+ }
+
+ @Test
+ public void doesNotPublishAuthorizationSubjectForWhitelistedRequest()
throws Exception {
+ String rpcCode = TelemetryCommand.getDescriptor().getFullName();
+ AuthConfig authConfig = new AuthConfig();
+ authConfig.setConfigName("grpc-authentication-whitelist-test");
+ authConfig.setAuthenticationEnabled(true);
+ authConfig.setAuthenticationWhitelist("other-rpc, " + rpcCode);
+ DefaultAuthenticationContext authenticationContext = new
DefaultAuthenticationContext();
+ authenticationContext.setRpcCode(rpcCode);
+ authenticationContext.setUsername("unverified-user");
+ AuthenticationPipeline pipeline = new AuthenticationPipeline(
+ authConfig, mock(MessagingProcessor.class)) {
+ @Override
+ protected AuthenticationContext newContext(ProxyContext context,
Metadata headers,
+ com.google.protobuf.GeneratedMessageV3 request) {
+ return authenticationContext;
+ }
+ };
+ AuthenticationEvaluator authenticationEvaluator =
mock(AuthenticationEvaluator.class);
+ Field evaluatorField =
AuthenticationPipeline.class.getDeclaredField("authenticationEvaluator");
+ evaluatorField.setAccessible(true);
+ evaluatorField.set(pipeline, authenticationEvaluator);
+
+ Metadata metadata = new Metadata();
+ metadata.put(GrpcConstants.AUTHORIZATION_AK, "forged-user");
+ pipeline.execute(ProxyContext.create(), metadata,
TelemetryCommand.getDefaultInstance());
+
+ verify(authenticationEvaluator).evaluate(authenticationContext);
+
assertThat(metadata.containsKey(GrpcConstants.AUTHORIZATION_AK)).isFalse();
+ }
+}
diff --git
a/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivityTest.java
b/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivityTest.java
index 11dd6bc40c..a698ef343b 100644
---
a/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivityTest.java
+++
b/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivityTest.java
@@ -23,6 +23,8 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import java.util.concurrent.CompletableFuture;
import org.apache.rocketmq.acl.common.AclException;
+import
org.apache.rocketmq.auth.authentication.exception.AuthenticationException;
+import org.apache.rocketmq.auth.authorization.exception.AuthorizationException;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.MQVersion;
@@ -184,6 +186,33 @@ public class AbstractRemotingActivityTest extends
InitConfigTest {
assertThat(captor.getValue().getCode()).isEqualTo(ResponseCode.NO_PERMISSION);
}
+ @Test
+ public void testRequestAuthExceptions() throws Exception {
+ ArgumentCaptor<RemotingCommand> captor =
ArgumentCaptor.forClass(RemotingCommand.class);
+ String brokerName = "broker";
+ CompletableFuture<RemotingCommand> authenticationFuture = new
CompletableFuture<>();
+ authenticationFuture.completeExceptionally(new
AuthenticationException("authentication failed"));
+ CompletableFuture<RemotingCommand> authorizationFuture = new
CompletableFuture<>();
+ authorizationFuture.completeExceptionally(new
AuthorizationException("authorization failed"));
+ when(messagingProcessorMock.request(any(), eq(brokerName), any(),
anyLong()))
+ .thenReturn(authenticationFuture, authorizationFuture);
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, null);
+ request.setOpaque(123);
+ request.addExtField(AbstractRemotingActivity.BROKER_NAME_FIELD,
brokerName);
+
+ remotingActivity.request(ctx, request, null, 10000);
+ remotingActivity.request(ctx, request, null, 10000);
+
+ verify(ctx, times(2)).writeAndFlush(captor.capture());
+ assertThat(captor.getAllValues()).allSatisfy(response -> {
+
assertThat(response.getCode()).isEqualTo(ResponseCode.NO_PERMISSION);
+ assertThat(response.getOpaque()).isEqualTo(123);
+ assertThat(response.isResponseType()).isTrue();
+ });
+
assertThat(captor.getAllValues().get(0).getRemark()).isEqualTo("authentication
failed");
+
assertThat(captor.getAllValues().get(1).getRemark()).isEqualTo("authorization
failed");
+ }
+
@Test
public void testRequestDefaultException() throws Exception {
ArgumentCaptor<RemotingCommand> captor =
ArgumentCaptor.forClass(RemotingCommand.class);
@@ -199,4 +228,4 @@ public class AbstractRemotingActivityTest extends
InitConfigTest {
verify(ctx, times(1)).writeAndFlush(captor.capture());
assertThat(captor.getValue().getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);
}
-}
\ No newline at end of file
+}
diff --git
a/proxy/src/test/java/org/apache/rocketmq/proxy/service/metadata/ClusterMetadataServiceTest.java
b/proxy/src/test/java/org/apache/rocketmq/proxy/service/metadata/ClusterMetadataServiceTest.java
index 5894f87199..04f4cd6a79 100644
---
a/proxy/src/test/java/org/apache/rocketmq/proxy/service/metadata/ClusterMetadataServiceTest.java
+++
b/proxy/src/test/java/org/apache/rocketmq/proxy/service/metadata/ClusterMetadataServiceTest.java
@@ -21,6 +21,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.Optional;
import java.util.Set;
+import org.apache.rocketmq.auth.authentication.model.User;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.proxy.common.ProxyContext;
@@ -28,6 +29,7 @@ import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.service.BaseServiceTest;
import org.apache.rocketmq.proxy.service.route.MessageQueueView;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.body.UserInfo;
import
org.apache.rocketmq.remoting.protocol.statictopic.TopicConfigAndQueueMapping;
import
org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.junit.Before;
@@ -90,6 +92,17 @@ public class ClusterMetadataServiceTest extends
BaseServiceTest {
assertEquals(1,
this.clusterMetadataService.subscriptionGroupConfigCache.asMap().size());
}
+ @Test
+ public void testGetUserRetainsCredentialForAuthentication() throws
Exception {
+ when(this.mqClientAPIExt.getUser(anyString(), eq("abc"), anyLong()))
+ .thenReturn(UserInfo.of("abc", "user-secret-for-proxy", "Normal"));
+
+ User user = this.clusterMetadataService.getUser(ProxyContext.create(),
"abc").join();
+
+ assertNotNull(user);
+ assertEquals("user-secret-for-proxy", user.getPassword());
+ }
+
@Test
public void findOneBroker() {
diff --git
a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/ResumeCheckHalfMessageRequestHeader.java
b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/ResumeCheckHalfMessageRequestHeader.java
index 923fd37ea6..eff55847b8 100644
---
a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/ResumeCheckHalfMessageRequestHeader.java
+++
b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/ResumeCheckHalfMessageRequestHeader.java
@@ -22,6 +22,7 @@ import org.apache.rocketmq.common.action.RocketMQAction;
import org.apache.rocketmq.common.resource.ResourceType;
import org.apache.rocketmq.common.resource.RocketMQResource;
import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
import org.apache.rocketmq.remoting.annotation.CFNullable;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RequestCode;
@@ -30,6 +31,7 @@ import org.apache.rocketmq.remoting.protocol.RequestCode;
public class ResumeCheckHalfMessageRequestHeader implements
CommandCustomHeader {
@RocketMQResource(ResourceType.TOPIC)
+ @CFNotNull
private String topic;
@CFNullable
private String msgId;
diff --git
a/tools/src/main/java/org/apache/rocketmq/tools/command/auth/ListUserSubCommand.java
b/tools/src/main/java/org/apache/rocketmq/tools/command/auth/ListUserSubCommand.java
index 24eb62427f..d84c43d01e 100644
---
a/tools/src/main/java/org/apache/rocketmq/tools/command/auth/ListUserSubCommand.java
+++
b/tools/src/main/java/org/apache/rocketmq/tools/command/auth/ListUserSubCommand.java
@@ -34,7 +34,7 @@ import org.apache.rocketmq.tools.command.SubCommandException;
public class ListUserSubCommand implements SubCommand {
- private static final String FORMAT = "%-16s %-22s %-22s %-22s%n";
+ private static final String FORMAT = "%-16s %-22s %-22s%n";
@Override
public String commandName() {
@@ -115,7 +115,8 @@ public class ListUserSubCommand implements SubCommand {
}
private void printUsers(List<UserInfo> users) {
- System.out.printf(FORMAT, "#UserName", "#Password", "#UserType",
"#UserStatus");
- users.forEach(user -> System.out.printf(FORMAT, user.getUsername(),
user.getPassword(), user.getUserType(), user.getUserStatus()));
+ System.out.printf(FORMAT, "#UserName", "#UserType", "#UserStatus");
+ users.forEach(user -> System.out.printf(FORMAT,
+ user.getUsername(), user.getUserType(), user.getUserStatus()));
}
}
