opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-10-24 02:15:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

Improve ACL 2.0 authorization context handling (#10925)

Browse Source
This commit is contained in:
dingshuangxi888
2026-08-14 09:58:51 +08:00
committed by GitHub
parent 7eee0fc366
commit 293f588571
31 changed files with 2402 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
auth/src/main/java/org/apache/rocketmq/auth/authorization/AuthorizationCompatibility.java
+116
View File
@@ -0,0 +1,116 @@
/*
* 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.authorization;
import apache.rocketmq.v2.ClientType;
import apache.rocketmq.v2.HeartbeatRequest;
import apache.rocketmq.v2.NotifyClientTerminationRequest;
import apache.rocketmq.v2.TelemetryCommand;
import com.google.protobuf.GeneratedMessageV3;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.heartbeat.HeartbeatData;
import org.apache.rocketmq.remoting.protocol.heartbeat.ProducerData;
final class AuthorizationCompatibility {
private AuthorizationCompatibility() {
}
static boolean matches(RemotingCommand request) {
if (request == null) {
return false;
}
try {
switch (request.getCode()) {
case RequestCode.HEART_BEAT:
return isProducerHeartbeat(request);
case RequestCode.UNREGISTER_CLIENT:
return isProducerUnregister(request);
case RequestCode.END_TRANSACTION:
case RequestCode.VIEW_MESSAGE_BY_ID:
return isHistoricalTopicAbsent(request);
default:
return false;
}
} catch (Throwable ignored) {
return false;
}
}
static boolean matches(GeneratedMessageV3 request) {
if (request instanceof HeartbeatRequest) {
HeartbeatRequest heartbeat = (HeartbeatRequest) request;
return StringUtils.isBlank(heartbeat.getGroup().getName())
&& (heartbeat.getClientType() == ClientType.PRODUCER
|| heartbeat.getClientType() == ClientType.CLIENT_TYPE_UNSPECIFIED);
}
if (request instanceof NotifyClientTerminationRequest) {
NotifyClientTerminationRequest termination = (NotifyClientTerminationRequest) request;
return StringUtils.isBlank(termination.getGroup().getName());
}
if (request instanceof TelemetryCommand) {
TelemetryCommand telemetry = (TelemetryCommand) request;
switch (telemetry.getCommandCase()) {
case SETTINGS:
return telemetry.getSettings().hasPublishing()
&& telemetry.getSettings().getPublishing().getTopicsCount() == 0;
case THREAD_STACK_TRACE:
case VERIFY_MESSAGE_RESULT:
return true;
default:
return false;
}
}
return false;
}
private static boolean isProducerHeartbeat(RemotingCommand request) {
if (request.getBody() == null) {
return false;
}
HeartbeatData heartbeat = HeartbeatData.decode(request.getBody(), HeartbeatData.class);
if (heartbeat == null || CollectionUtils.isNotEmpty(heartbeat.getConsumerDataSet())
|| CollectionUtils.isEmpty(heartbeat.getProducerDataSet())) {
return false;
}
for (ProducerData producer : heartbeat.getProducerDataSet()) {
if (producer == null || producer.getGroupName() == null) {
return false;
}
}
return true;
}
private static boolean isProducerUnregister(RemotingCommand request) {
return StringUtils.isNotBlank(getExtField(request, "producerGroup"))
&& StringUtils.isBlank(getExtField(request, "consumerGroup"));
}
/**
* Historical END_TRANSACTION and VIEW_MESSAGE_BY_ID requests carry no topic field.
*/
private static boolean isHistoricalTopicAbsent(RemotingCommand request) {
return request.getExtFields() != null && StringUtils.isBlank(getExtField(request, "topic"));
}
private static String getExtField(RemotingCommand request, String name) {
return request.getExtFields() == null ? null : request.getExtFields().get(name);
}
}
auth/src/main/java/org/apache/rocketmq/auth/authorization/AuthorizationEvaluator.java
+32 -2
View File
@@ -16,13 +16,16 @@
*/
package org.apache.rocketmq.auth.authorization;
import com.google.protobuf.GeneratedMessageV3;
import java.util.List;
import java.util.function.Supplier;
import org.apache.commons.collections.CollectionUtils;
import org.apache.rocketmq.auth.authorization.context.AuthorizationContext;
import org.apache.rocketmq.auth.authorization.exception.AuthorizationException;
import org.apache.rocketmq.auth.authorization.factory.AuthorizationFactory;
import org.apache.rocketmq.auth.authorization.strategy.AuthorizationStrategy;
import org.apache.rocketmq.auth.config.AuthConfig;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
public class AuthorizationEvaluator {
@@ -36,10 +39,37 @@ public class AuthorizationEvaluator {
this.authorizationStrategy = AuthorizationFactory.getStrategy(authConfig, metadataService);
}
public void evaluate(List<AuthorizationContext> contexts) {
/**
* Visible for testing: allows injecting a stub strategy.
*/
AuthorizationEvaluator(AuthorizationStrategy authorizationStrategy) {
this.authorizationStrategy = authorizationStrategy;
}
public void evaluate(List<? extends AuthorizationContext> contexts) {
if (CollectionUtils.isEmpty(contexts)) {
return;
throw new AuthorizationException("authorization context is empty.");
}
contexts.forEach(this.authorizationStrategy::evaluate);
}
public void evaluate(RemotingCommand request, List<? extends AuthorizationContext> contexts) {
if (CollectionUtils.isNotEmpty(contexts)) {
contexts.forEach(this.authorizationStrategy::evaluate);
return;
}
if (!AuthorizationCompatibility.matches(request)) {
throw new AuthorizationException("authorization context is empty.");
}
}
public void evaluate(GeneratedMessageV3 request, List<? extends AuthorizationContext> contexts) {
if (CollectionUtils.isNotEmpty(contexts)) {
contexts.forEach(this.authorizationStrategy::evaluate);
return;
}
if (!AuthorizationCompatibility.matches(request)) {
throw new AuthorizationException("authorization context is empty.");
}
}
}
auth/src/main/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilder.java
+368 -81
View File
@@ -36,14 +36,16 @@ import com.google.protobuf.GeneratedMessageV3;
import io.grpc.Metadata;
import io.netty.channel.ChannelHandlerContext;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.acl.common.AclException;
@@ -54,24 +56,38 @@ import org.apache.rocketmq.auth.authorization.context.DefaultAuthorizationContex
import org.apache.rocketmq.auth.authorization.exception.AuthorizationException;
import org.apache.rocketmq.auth.authorization.model.Resource;
import org.apache.rocketmq.auth.config.AuthConfig;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.action.Action;
import org.apache.rocketmq.common.action.RocketMQAction;
import org.apache.rocketmq.common.constant.CommonConstants;
import org.apache.rocketmq.common.constant.GrpcConstants;
import org.apache.rocketmq.common.lite.LiteSubscriptionDTO;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.resource.ResourcePattern;
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.common.RemotingHelper;
import org.apache.rocketmq.remoting.protocol.NamespaceUtil;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.RequestHeaderRegistry;
import org.apache.rocketmq.remoting.protocol.body.BatchAck;
import org.apache.rocketmq.remoting.protocol.body.BatchAckMessageRequestBody;
import org.apache.rocketmq.remoting.protocol.body.CheckClientRequestBody;
import org.apache.rocketmq.remoting.protocol.body.CreateTopicListRequestBody;
import org.apache.rocketmq.remoting.protocol.body.DeleteSubscriptionGroupListRequestBody;
import org.apache.rocketmq.remoting.protocol.body.DeleteTopicListRequestBody;
import org.apache.rocketmq.remoting.protocol.body.LockBatchRequestBody;
import org.apache.rocketmq.remoting.protocol.body.LiteSubscriptionCtlRequestBody;
import org.apache.rocketmq.remoting.protocol.body.QueryAssignmentRequestBody;
import org.apache.rocketmq.remoting.protocol.body.SetMessageRequestModeRequestBody;
import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupList;
import org.apache.rocketmq.remoting.protocol.body.UnlockBatchRequestBody;
import org.apache.rocketmq.remoting.protocol.header.CreateTopicRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.GetConsumerListByGroupRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.QueryConsumerOffsetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.UnregisterClientRequestHeader;
@@ -79,18 +95,16 @@ import org.apache.rocketmq.remoting.protocol.header.UpdateConsumerOffsetRequestH
import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumerData;
import org.apache.rocketmq.remoting.protocol.heartbeat.HeartbeatData;
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingDetail;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
public class DefaultAuthorizationContextBuilder implements AuthorizationContextBuilder {
private static final String TOPIC = "topic";
private static final String GROUP = "group";
private static final String A = "a";
private static final String B = "b";
private static final String CONSUMER_GROUP = "consumerGroup";
private final AuthConfig authConfig;
private static final EnumSet<ClientType> CONSUMER_CLIENT_TYPES =
EnumSet.of(ClientType.PUSH_CONSUMER, ClientType.SIMPLE_CONSUMER, ClientType.PULL_CONSUMER);
private final RequestHeaderRegistry requestHeaderRegistry;
public DefaultAuthorizationContextBuilder(AuthConfig authConfig) {
@@ -118,10 +132,12 @@ public class DefaultAuthorizationContextBuilder implements AuthorizationContextB
}
if (message instanceof HeartbeatRequest) {
HeartbeatRequest request = (HeartbeatRequest) message;
if (!isConsumerClientType(request.getClientType())) {
return null;
if (StringUtils.isNotBlank(request.getGroup().getName())) {
if (request.getClientType() == ClientType.PRODUCER) {
throw new AuthorizationException("group is not allowed for producer heartbeat.");
}
result = newGroupSubContexts(metadata, request.getGroup());
}
result = newGroupSubContexts(metadata, request.getGroup());
}
if (message instanceof ReceiveMessageRequest) {
ReceiveMessageRequest request = (ReceiveMessageRequest) message;
@@ -132,9 +148,6 @@ public class DefaultAuthorizationContextBuilder implements AuthorizationContextB
}
if (message instanceof SyncLiteSubscriptionRequest) {
SyncLiteSubscriptionRequest request = (SyncLiteSubscriptionRequest) message;
if (request.getLiteTopicSetCount() <= 0) {
return null;
}
result = newSubContexts(metadata, request.getGroup(), request.getTopic());
}
if (message instanceof AckMessageRequest) {
@@ -153,7 +166,7 @@ public class DefaultAuthorizationContextBuilder implements AuthorizationContextB
}
if (message instanceof ChangeInvisibleDurationRequest) {
ChangeInvisibleDurationRequest request = (ChangeInvisibleDurationRequest) message;
result = newGroupSubContexts(metadata, request.getGroup());
result = newSubContexts(metadata, request.getGroup(), request.getTopic());
}
if (message instanceof QueryRouteRequest) {
QueryRouteRequest request = (QueryRouteRequest) message;
@@ -181,8 +194,8 @@ public class DefaultAuthorizationContextBuilder implements AuthorizationContextB
List<DefaultAuthorizationContext> result = new ArrayList<>();
try {
HashMap<String, String> fields = command.getExtFields();
if (MapUtils.isEmpty(fields)) {
return result;
if (fields == null) {
fields = new HashMap<>();
}
Subject subject = null;
if (fields.containsKey(SessionCredentials.ACCESS_KEY)) {
@@ -195,35 +208,46 @@ public class DefaultAuthorizationContextBuilder implements AuthorizationContextB
Resource group;
switch (command.getCode()) {
case RequestCode.GET_ROUTEINFO_BY_TOPIC:
if (NamespaceUtil.isRetryTopic(fields.get(TOPIC))) {
group = Resource.ofGroup(fields.get(TOPIC));
String routeTopic = requireResource(fields.get(TOPIC), "topic");
if (NamespaceUtil.isRetryTopic(routeTopic)) {
group = Resource.ofGroup(routeTopic);
result.add(DefaultAuthorizationContext.of(subject, group, Arrays.asList(Action.SUB, Action.GET), sourceIp));
} else {
topic = Resource.ofTopic(fields.get(TOPIC));
topic = Resource.ofTopic(routeTopic);
result.add(DefaultAuthorizationContext.of(subject, topic, Arrays.asList(Action.PUB, Action.SUB, Action.GET), sourceIp));
}
break;
case RequestCode.SEND_MESSAGE:
if (NamespaceUtil.isRetryTopic(fields.get(TOPIC))) {
group = Resource.ofGroup(fields.get(TOPIC));
String sendTopic = requireResource(fields.get(TOPIC), "topic");
if (NamespaceUtil.isRetryTopic(sendTopic)) {
group = Resource.ofGroup(sendTopic);
result.add(DefaultAuthorizationContext.of(subject, group, Action.SUB, sourceIp));
} else {
topic = Resource.ofTopic(fields.get(TOPIC));
topic = Resource.ofTopic(sendTopic);
result.add(DefaultAuthorizationContext.of(subject, topic, Action.PUB, sourceIp));
}
break;
case RequestCode.SEND_MESSAGE_V2:
case RequestCode.SEND_BATCH_MESSAGE:
if (NamespaceUtil.isRetryTopic(fields.get(B))) {
group = Resource.ofGroup(fields.get(B));
String compactSendTopic = requireResource(fields.get(B), "topic");
if (NamespaceUtil.isRetryTopic(compactSendTopic)) {
group = Resource.ofGroup(compactSendTopic);
result.add(DefaultAuthorizationContext.of(subject, group, Action.SUB, sourceIp));
} else {
topic = Resource.ofTopic(fields.get(B));
topic = Resource.ofTopic(compactSendTopic);
result.add(DefaultAuthorizationContext.of(subject, topic, Action.PUB, sourceIp));
}
break;
case RequestCode.SEND_REPLY_MESSAGE:
topic = Resource.ofTopic(requireResource(fields.get(TOPIC), "topic"));
result.add(DefaultAuthorizationContext.of(subject, topic, Action.PUB, sourceIp));
break;
case RequestCode.SEND_REPLY_MESSAGE_V2:
topic = Resource.ofTopic(requireResource(fields.get(B), "topic"));
result.add(DefaultAuthorizationContext.of(subject, topic, Action.PUB, sourceIp));
break;
case RequestCode.RECALL_MESSAGE:
topic = Resource.ofTopic(fields.get(TOPIC));
topic = Resource.ofTopic(requireResource(fields.get(TOPIC), "topic"));
result.add(DefaultAuthorizationContext.of(subject, topic, Action.PUB, sourceIp));
break;
case RequestCode.END_TRANSACTION:
@@ -232,32 +256,106 @@ public class DefaultAuthorizationContextBuilder implements AuthorizationContextB
result.add(DefaultAuthorizationContext.of(subject, topic, Action.PUB, sourceIp));
}
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));
}
break;
case RequestCode.CONSUMER_SEND_MSG_BACK:
group = Resource.ofGroup(fields.get(GROUP));
group = Resource.ofGroup(requireResource(fields.get(GROUP), "consumer group"));
result.add(DefaultAuthorizationContext.of(subject, group, Action.SUB, sourceIp));
break;
case RequestCode.PULL_MESSAGE:
if (!NamespaceUtil.isRetryTopic(fields.get(TOPIC))) {
topic = Resource.ofTopic(fields.get(TOPIC));
case RequestCode.LITE_PULL_MESSAGE:
String pullTopic = requireResource(fields.get(TOPIC), "topic");
String pullGroup = requireResource(fields.get(CONSUMER_GROUP), "consumer group");
if (NamespaceUtil.isRetryTopic(pullTopic)) {
if (!StringUtils.equals(pullTopic, MixAll.getRetryTopic(pullGroup))) {
throw new AuthorizationException("retry topic does not match consumer group.");
}
} else {
topic = Resource.ofTopic(pullTopic);
result.add(DefaultAuthorizationContext.of(subject, topic, Action.SUB, sourceIp));
}
group = Resource.ofGroup(fields.get(CONSUMER_GROUP));
group = Resource.ofGroup(pullGroup);
result.add(DefaultAuthorizationContext.of(subject, group, Action.SUB, sourceIp));
break;
case RequestCode.BATCH_ACK_MESSAGE:
BatchAckMessageRequestBody batchAckBody = decodeRequiredBody(
command, BatchAckMessageRequestBody.class, "batch ack");
if (CollectionUtils.isEmpty(batchAckBody.getAcks())) {
throw new AuthorizationException("batch ack is empty.");
}
Set<String> ackResources = new LinkedHashSet<>();
for (BatchAck ack : batchAckBody.getAcks()) {
if (ack == null) {
throw new AuthorizationException("batch ack entry is null.");
}
addUniqueContext(result, ackResources, subject,
Resource.ofTopic(requireResource(ack.getTopic(), "topic")),
Action.SUB, sourceIp);
addUniqueContext(result, ackResources, subject,
Resource.ofGroup(requireResource(ack.getConsumerGroup(), "consumer group")),
Action.SUB, sourceIp);
}
break;
case RequestCode.QUERY_ASSIGNMENT:
QueryAssignmentRequestBody assignmentBody = decodeRequiredBody(
command, QueryAssignmentRequestBody.class, "query assignment");
result.add(DefaultAuthorizationContext.of(subject,
Resource.ofTopic(requireResource(assignmentBody.getTopic(), "topic")),
Action.SUB, sourceIp));
result.add(DefaultAuthorizationContext.of(subject,
Resource.ofGroup(requireResource(assignmentBody.getConsumerGroup(), "consumer group")),
Action.SUB, sourceIp));
break;
case RequestCode.SET_MESSAGE_REQUEST_MODE:
SetMessageRequestModeRequestBody modeBody = decodeRequiredBody(
command, SetMessageRequestModeRequestBody.class, "message request mode");
result.add(DefaultAuthorizationContext.of(subject,
Resource.ofTopic(requireResource(modeBody.getTopic(), "topic")),
Action.SUB, sourceIp));
result.add(DefaultAuthorizationContext.of(subject,
Resource.ofGroup(requireResource(modeBody.getConsumerGroup(), "consumer group")),
Action.UPDATE, sourceIp));
break;
case RequestCode.CHECK_CLIENT_CONFIG:
CheckClientRequestBody checkClientBody = decodeRequiredBody(
command, CheckClientRequestBody.class, "client config");
if (checkClientBody.getSubscriptionData() == null) {
throw new AuthorizationException("subscription is null.");
}
result.add(DefaultAuthorizationContext.of(subject,
Resource.ofTopic(requireResource(
checkClientBody.getSubscriptionData().getTopic(), "topic")),
Action.SUB, sourceIp));
result.add(DefaultAuthorizationContext.of(subject,
Resource.ofGroup(requireResource(checkClientBody.getGroup(), "consumer group")),
Action.SUB, sourceIp));
break;
case RequestCode.QUERY_MESSAGE:
topic = Resource.ofTopic(fields.get(TOPIC));
topic = Resource.ofTopic(requireResource(fields.get(TOPIC), "topic"));
result.add(DefaultAuthorizationContext.of(subject, topic, Arrays.asList(Action.SUB, Action.GET), sourceIp));
break;
case RequestCode.HEART_BEAT:
HeartbeatData heartbeatData = HeartbeatData.decode(command.getBody(), HeartbeatData.class);
HeartbeatData heartbeatData = decodeRequiredBody(command, HeartbeatData.class, "heartbeat");
for (ConsumerData data : heartbeatData.getConsumerDataSet()) {
group = Resource.ofGroup(data.getGroupName());
if (data == null) {
throw new AuthorizationException("consumer data is null.");
}
group = Resource.ofGroup(requireResource(data.getGroupName(), "consumer group"));
result.add(DefaultAuthorizationContext.of(subject, group, Action.SUB, sourceIp));
for (SubscriptionData subscriptionData : data.getSubscriptionDataSet()) {
if (NamespaceUtil.isRetryTopic(subscriptionData.getTopic())) {
if (subscriptionData == null) {
throw new AuthorizationException("subscription is null.");
}
String subscriptionTopic =
requireResource(subscriptionData.getTopic(), "topic");
if (NamespaceUtil.isRetryTopic(subscriptionTopic)) {
continue;
}
topic = Resource.ofTopic(subscriptionData.getTopic());
topic = Resource.ofTopic(subscriptionTopic);
result.add(DefaultAuthorizationContext.of(subject, topic, Action.SUB, sourceIp));
}
}
@@ -273,88 +371,232 @@ public class DefaultAuthorizationContextBuilder implements AuthorizationContextB
case RequestCode.GET_CONSUMER_LIST_BY_GROUP:
final GetConsumerListByGroupRequestHeader getConsumerListByGroupRequestHeader =
command.decodeCommandCustomHeader(GetConsumerListByGroupRequestHeader.class);
group = Resource.ofGroup(getConsumerListByGroupRequestHeader.getConsumerGroup());
group = Resource.ofGroup(requireResource(
getConsumerListByGroupRequestHeader.getConsumerGroup(), "consumer group"));
result.add(DefaultAuthorizationContext.of(subject, group, Arrays.asList(Action.SUB, Action.GET), sourceIp));
break;
case RequestCode.QUERY_CONSUMER_OFFSET:
final QueryConsumerOffsetRequestHeader queryConsumerOffsetRequestHeader =
command.decodeCommandCustomHeader(QueryConsumerOffsetRequestHeader.class);
if (!NamespaceUtil.isRetryTopic(queryConsumerOffsetRequestHeader.getTopic())) {
topic = Resource.ofTopic(queryConsumerOffsetRequestHeader.getTopic());
String queryOffsetTopic = requireResource(
queryConsumerOffsetRequestHeader.getTopic(), "topic");
String queryOffsetGroup = requireResource(
queryConsumerOffsetRequestHeader.getConsumerGroup(), "consumer group");
if (!NamespaceUtil.isRetryTopic(queryOffsetTopic)) {
topic = Resource.ofTopic(queryOffsetTopic);
result.add(DefaultAuthorizationContext.of(subject, topic, Arrays.asList(Action.SUB, Action.GET), sourceIp));
}
group = Resource.ofGroup(queryConsumerOffsetRequestHeader.getConsumerGroup());
group = Resource.ofGroup(queryOffsetGroup);
result.add(DefaultAuthorizationContext.of(subject, group, Arrays.asList(Action.SUB, Action.GET), sourceIp));
break;
case RequestCode.UPDATE_CONSUMER_OFFSET:
final UpdateConsumerOffsetRequestHeader updateConsumerOffsetRequestHeader =
command.decodeCommandCustomHeader(UpdateConsumerOffsetRequestHeader.class);
if (!NamespaceUtil.isRetryTopic(updateConsumerOffsetRequestHeader.getTopic())) {
topic = Resource.ofTopic(updateConsumerOffsetRequestHeader.getTopic());
String updateOffsetTopic = requireResource(
updateConsumerOffsetRequestHeader.getTopic(), "topic");
String updateOffsetGroup = requireResource(
updateConsumerOffsetRequestHeader.getConsumerGroup(), "consumer group");
if (!NamespaceUtil.isRetryTopic(updateOffsetTopic)) {
topic = Resource.ofTopic(updateOffsetTopic);
result.add(DefaultAuthorizationContext.of(subject, topic, Arrays.asList(Action.SUB, Action.UPDATE), sourceIp));
}
group = Resource.ofGroup(updateConsumerOffsetRequestHeader.getConsumerGroup());
group = Resource.ofGroup(updateOffsetGroup);
result.add(DefaultAuthorizationContext.of(subject, group, Arrays.asList(Action.SUB, Action.UPDATE), sourceIp));
break;
case RequestCode.LOCK_BATCH_MQ:
LockBatchRequestBody lockBatchRequestBody = LockBatchRequestBody.decode(command.getBody(), LockBatchRequestBody.class);
group = Resource.ofGroup(lockBatchRequestBody.getConsumerGroup());
group = Resource.ofGroup(requireResource(
lockBatchRequestBody.getConsumerGroup(), "consumer group"));
result.add(DefaultAuthorizationContext.of(subject, group, Action.SUB, sourceIp));
if (CollectionUtils.isNotEmpty(lockBatchRequestBody.getMqSet())) {
for (MessageQueue messageQueue : lockBatchRequestBody.getMqSet()) {
if (NamespaceUtil.isRetryTopic(messageQueue.getTopic())) {
String lockTopic = requireResource(messageQueue.getTopic(), "topic");
if (NamespaceUtil.isRetryTopic(lockTopic)) {
continue;
}
topic = Resource.ofTopic(messageQueue.getTopic());
topic = Resource.ofTopic(lockTopic);
result.add(DefaultAuthorizationContext.of(subject, topic, Action.SUB, sourceIp));
}
}
break;
case RequestCode.UNLOCK_BATCH_MQ:
UnlockBatchRequestBody unlockBatchRequestBody = UnlockBatchRequestBody.decode(command.getBody(), UnlockBatchRequestBody.class);
group = Resource.ofGroup(unlockBatchRequestBody.getConsumerGroup());
UnlockBatchRequestBody unlockBatchRequestBody = UnlockBatchRequestBody.decode(
command.getBody(), UnlockBatchRequestBody.class);
group = Resource.ofGroup(requireResource(
unlockBatchRequestBody.getConsumerGroup(), "consumer group"));
result.add(DefaultAuthorizationContext.of(subject, group, Action.SUB, sourceIp));
if (CollectionUtils.isNotEmpty(unlockBatchRequestBody.getMqSet())) {
for (MessageQueue messageQueue : unlockBatchRequestBody.getMqSet()) {
if (NamespaceUtil.isRetryTopic(messageQueue.getTopic())) {
String unlockTopic = requireResource(messageQueue.getTopic(), "topic");
if (NamespaceUtil.isRetryTopic(unlockTopic)) {
continue;
}
topic = Resource.ofTopic(messageQueue.getTopic());
topic = Resource.ofTopic(unlockTopic);
result.add(DefaultAuthorizationContext.of(subject, topic, Action.SUB, sourceIp));
}
}
break;
case RequestCode.DELETE_TOPIC_IN_BROKER_LIST:
// Batch APIs carry their target list in the request body, not in an annotated
// CommandCustomHeader, so the annotation-based path in
// RequestHeaderRegistry would otherwise produce an empty context list and let
// the request through without a DELETE permission check. Decode the body and
// emit one DELETE context per topic instead.
DeleteTopicListRequestBody deleteTopicListRequestBody =
DeleteTopicListRequestBody.decode(command.getBody(), DeleteTopicListRequestBody.class);
if (CollectionUtils.isNotEmpty(deleteTopicListRequestBody.getTopicList())) {
for (String topicName : deleteTopicListRequestBody.getTopicList()) {
if (StringUtils.isBlank(topicName)) {
continue;
}
topic = Resource.ofTopic(topicName);
result.add(DefaultAuthorizationContext.of(subject, topic, Action.DELETE, sourceIp));
case RequestCode.LITE_SUBSCRIPTION_CTL:
LiteSubscriptionCtlRequestBody liteSubscriptionBody = decodeRequiredBody(
command, LiteSubscriptionCtlRequestBody.class, "lite subscription");
if (CollectionUtils.isEmpty(liteSubscriptionBody.getSubscriptionSet())) {
throw new AuthorizationException("lite subscription is empty.");
}
Set<String> liteSubscriptionResources = new LinkedHashSet<>();
for (LiteSubscriptionDTO subscription : liteSubscriptionBody.getSubscriptionSet()) {
if (subscription == null) {
throw new AuthorizationException("lite subscription is null.");
}
addUniqueContext(result, liteSubscriptionResources, subject,
Resource.ofGroup(requireResource(subscription.getGroup(), "consumer group")),
Action.SUB, sourceIp);
addUniqueContext(result, liteSubscriptionResources, subject,
Resource.ofTopic(requireResource(subscription.getTopic(), "topic")),
Action.SUB, sourceIp);
}
break;
case RequestCode.UPDATE_BROKER_CONFIG:
result.add(DefaultAuthorizationContext.of(subject,
Resource.ofCluster(authConfig.getClusterName()), Action.UPDATE, sourceIp));
break;
case RequestCode.UPDATE_AND_CREATE_TOPIC_LIST:
CreateTopicListRequestBody topicListBody = decodeRequiredBody(
command, CreateTopicListRequestBody.class, "topic list");
if (CollectionUtils.isEmpty(topicListBody.getTopicConfigList())) {
throw new AuthorizationException("topic list is empty.");
}
Set<String> topicListResources = new LinkedHashSet<>();
for (TopicConfig topicConfig : topicListBody.getTopicConfigList()) {
if (topicConfig == null) {
throw new AuthorizationException("topic config is null.");
}
String topicName = requireResource(topicConfig.getTopicName(), "topic");
Resource resource = NamespaceUtil.isRetryTopic(topicName)
? Resource.ofGroup(topicName) : Resource.ofTopic(topicName);
addUniqueContext(result, topicListResources, subject, resource, Action.CREATE, sourceIp);
}
break;
case RequestCode.UPDATE_COLD_DATA_FLOW_CTR_CONFIG:
Properties properties = MixAll.string2Properties(
decodeRequiredText(command, "cold data flow config"));
if (properties == null || properties.isEmpty()) {
throw new AuthorizationException("cold data flow config is empty.");
}
Set<String> coldDataResources = new LinkedHashSet<>();
for (String consumerGroup : properties.stringPropertyNames()) {
addUniqueContext(result, coldDataResources, subject,
Resource.ofGroup(requireResource(consumerGroup, "consumer group")),
Action.UPDATE, sourceIp);
}
break;
case RequestCode.REMOVE_COLD_DATA_FLOW_CTR_CONFIG:
group = Resource.ofGroup(requireResource(
decodeRequiredText(command, "consumer group"), "consumer group"));
result.add(DefaultAuthorizationContext.of(subject, group, Action.UPDATE, sourceIp));
break;
case RequestCode.UPDATE_AND_CREATE_SUBSCRIPTIONGROUP:
SubscriptionGroupConfig subscriptionGroupConfig =
RemotingSerializable.decode(command.getBody(), SubscriptionGroupConfig.class);
if (subscriptionGroupConfig == null
|| StringUtils.isBlank(subscriptionGroupConfig.getGroupName())) {
throw new AuthorizationException("subscription group is null.");
}
result.add(DefaultAuthorizationContext.of(subject,
Resource.ofGroup(subscriptionGroupConfig.getGroupName()), Action.CREATE, sourceIp));
break;
case RequestCode.UPDATE_AND_CREATE_SUBSCRIPTIONGROUP_LIST:
SubscriptionGroupList subscriptionGroupList = decodeRequiredBody(
command, SubscriptionGroupList.class, "subscription group list");
if (CollectionUtils.isEmpty(subscriptionGroupList.getGroupConfigList())) {
throw new AuthorizationException("subscription group list is empty.");
}
Set<String> subscriptionGroupResources = new LinkedHashSet<>();
for (SubscriptionGroupConfig groupConfig : subscriptionGroupList.getGroupConfigList()) {
if (groupConfig == null) {
throw new AuthorizationException("subscription group config is null.");
}
addUniqueContext(result, subscriptionGroupResources, subject,
Resource.ofGroup(requireResource(groupConfig.getGroupName(), "consumer group")),
Action.CREATE, sourceIp);
}
break;
case RequestCode.UPDATE_AND_CREATE_STATIC_TOPIC:
CreateTopicRequestHeader createTopicRequestHeader =
command.decodeCommandCustomHeader(CreateTopicRequestHeader.class);
if (createTopicRequestHeader == null) {
throw new AuthorizationException("topic header is null.");
}
String staticTopic = requireResource(createTopicRequestHeader.getTopic(), "topic");
TopicQueueMappingDetail mappingDetail = decodeRequiredBody(
command, TopicQueueMappingDetail.class, "topic queue mapping");
if (!StringUtils.equals(
staticTopic, requireResource(mappingDetail.getTopic(), "mapping topic"))) {
throw new AuthorizationException("mapping topic does not match topic header.");
}
topic = Resource.ofTopic(staticTopic);
result.add(DefaultAuthorizationContext.of(subject, topic, Action.CREATE, sourceIp));
break;
case RequestCode.GET_BROKER_CONFIG:
case RequestCode.GET_BROKER_RUNTIME_INFO:
case RequestCode.GET_ALL_CONSUMER_OFFSET:
case RequestCode.GET_TIMER_CHECK_POINT:
case RequestCode.GET_ALL_DELAY_OFFSET:
case RequestCode.GET_BROKER_HA_STATUS:
case RequestCode.GET_BROKER_EPOCH_CACHE:
case RequestCode.GET_BROKER_LITE_INFO:
result.add(DefaultAuthorizationContext.of(subject,
Resource.ofCluster(authConfig.getClusterName()), Action.GET, sourceIp));
break;
case RequestCode.GET_ALL_TOPIC_CONFIG:
case RequestCode.GET_TIMER_METRICS:
case RequestCode.GET_SYSTEM_TOPIC_LIST_FROM_BROKER:
result.add(DefaultAuthorizationContext.of(subject,
Resource.of(ResourceType.TOPIC, null, ResourcePattern.ANY), Action.LIST, sourceIp));
break;
case RequestCode.GET_COLD_DATA_FLOW_CTR_INFO:
case RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG:
result.add(DefaultAuthorizationContext.of(subject,
Resource.of(ResourceType.GROUP, null, ResourcePattern.ANY), Action.LIST, sourceIp));
break;
case RequestCode.GET_ALL_MESSAGE_REQUEST_MODE:
result.add(DefaultAuthorizationContext.of(subject,
Resource.of(ResourceType.TOPIC, null, ResourcePattern.ANY), Action.LIST, sourceIp));
result.add(DefaultAuthorizationContext.of(subject,
Resource.of(ResourceType.GROUP, null, ResourcePattern.ANY), Action.LIST, sourceIp));
break;
case RequestCode.SET_COMMITLOG_READ_MODE:
case RequestCode.CLEAN_EXPIRED_CONSUMEQUEUE:
case RequestCode.DELETE_EXPIRED_COMMITLOG:
case RequestCode.CLEAN_UNUSED_TOPIC:
case RequestCode.POP_ROLLBACK:
case RequestCode.SWITCH_TIMER_ENGINE:
result.add(DefaultAuthorizationContext.of(subject,
Resource.ofCluster(authConfig.getClusterName()), Action.UPDATE, sourceIp));
break;
case RequestCode.DELETE_TOPIC_IN_BROKER_LIST:
DeleteTopicListRequestBody deleteTopicListRequestBody = decodeRequiredBody(
command, DeleteTopicListRequestBody.class, "topic list");
if (CollectionUtils.isEmpty(deleteTopicListRequestBody.getTopicList())) {
throw new AuthorizationException("topic list is empty.");
}
Set<String> deleteTopicResources = new LinkedHashSet<>();
for (String topicName : deleteTopicListRequestBody.getTopicList()) {
String requiredTopic = requireResource(topicName, "topic");
Resource resource = NamespaceUtil.isRetryTopic(requiredTopic)
? Resource.ofGroup(requiredTopic) : Resource.ofTopic(requiredTopic);
addUniqueContext(result, deleteTopicResources, subject, resource, Action.DELETE, sourceIp);
}
break;
case RequestCode.DELETE_SUBSCRIPTION_GROUP_LIST:
// See DELETE_TOPIC_IN_BROKER_LIST: emit one DELETE context per group from the
// request body so authorization can enforce per-group DELETE permission.
DeleteSubscriptionGroupListRequestBody deleteGroupListRequestBody =
DeleteSubscriptionGroupListRequestBody.decode(command.getBody(), DeleteSubscriptionGroupListRequestBody.class);
if (CollectionUtils.isNotEmpty(deleteGroupListRequestBody.getGroupNameList())) {
for (String groupName : deleteGroupListRequestBody.getGroupNameList()) {
if (StringUtils.isBlank(groupName)) {
continue;
}
group = Resource.ofGroup(groupName);
result.add(DefaultAuthorizationContext.of(subject, group, Action.DELETE, sourceIp));
}
DeleteSubscriptionGroupListRequestBody deleteGroupListRequestBody = decodeRequiredBody(
command, DeleteSubscriptionGroupListRequestBody.class, "subscription group list");
if (CollectionUtils.isEmpty(deleteGroupListRequestBody.getGroupNameList())) {
throw new AuthorizationException("subscription group list is empty.");
}
Set<String> deleteGroupResources = new LinkedHashSet<>();
for (String groupName : deleteGroupListRequestBody.getGroupNameList()) {
group = Resource.ofGroup(requireResource(groupName, "consumer group"));
addUniqueContext(result, deleteGroupResources, subject, group, Action.DELETE, sourceIp);
}
break;
default:
@@ -375,6 +617,38 @@ public class DefaultAuthorizationContextBuilder implements AuthorizationContextB
return result;
}
private static <T> T decodeRequiredBody(RemotingCommand command, Class<T> bodyClass, String bodyName) {
if (command.getBody() == null || command.getBody().length == 0) {
throw new AuthorizationException(bodyName + " is null.");
}
T body = RemotingSerializable.decode(command.getBody(), bodyClass);
if (body == null) {
throw new AuthorizationException(bodyName + " is null.");
}
return body;
}
private static String decodeRequiredText(RemotingCommand command, String bodyName) {
if (command.getBody() == null || command.getBody().length == 0) {
throw new AuthorizationException(bodyName + " is null.");
}
return new String(command.getBody(), StandardCharsets.UTF_8);
}
private static String requireResource(String resource, String resourceName) {
if (StringUtils.isBlank(resource)) {
throw new AuthorizationException(resourceName + " is null.");
}
return resource;
}
private static void addUniqueContext(List<DefaultAuthorizationContext> contexts,
Set<String> resources, Subject subject, Resource resource, Action action, String sourceIp) {
if (resources.add(resource.getResourceKey())) {
contexts.add(DefaultAuthorizationContext.of(subject, resource, action, sourceIp));
}
}
private List<DefaultAuthorizationContext> buildContextByAnnotation(Subject subject, RemotingCommand request,
String sourceIp) throws Exception {
List<DefaultAuthorizationContext> result = new ArrayList<>();
@@ -406,15 +680,32 @@ public class DefaultAuthorizationContextBuilder implements AuthorizationContextB
String splitter = rocketMQResource.splitter();
Object value = field.get(header);
if (value == null) {
if (field.getAnnotation(CFNotNull.class) != null) {
throw new AuthorizationException(field.getName() + " is null.");
}
continue;
}
boolean resourceRequired = field.getAnnotation(CFNotNull.class) != null;
String fieldValue = value.toString();
if (StringUtils.isBlank(fieldValue)) {
if (resourceRequired) {
requireResource(fieldValue, field.getName());
}
continue;
}
String[] resourceValues;
if (StringUtils.isNotBlank(splitter)) {
resourceValues = StringUtils.split(value.toString(), splitter);
resourceValues = StringUtils.split(fieldValue, splitter);
} else {
resourceValues = new String[] {value.toString()};
resourceValues = new String[] {fieldValue};
}
for (String resourceValue : resourceValues) {
if (StringUtils.isBlank(resourceValue)) {
if (resourceRequired) {
requireResource(resourceValue, field.getName());
}
continue;
}
if (resourceType == ResourceType.TOPIC && NamespaceUtil.isRetryTopic(resourceValue)) {
resource = Resource.ofGroup(resourceValue);
result.add(DefaultAuthorizationContext.of(subject, resource, Arrays.asList(actions), sourceIp));
@@ -475,10 +766,6 @@ public class DefaultAuthorizationContextBuilder implements AuthorizationContextB
return result;
}
private boolean isConsumerClientType(ClientType clientType) {
return CONSUMER_CLIENT_TYPES.contains(clientType);
}
private static List<DefaultAuthorizationContext> newPubContext(Metadata metadata, apache.rocketmq.v2.Resource topic) {
if (topic == null || StringUtils.isBlank(topic.getName())) {
throw new AuthorizationException("topic is null.");
auth/src/test/java/org/apache/rocketmq/auth/authorization/AuthorizationEvaluatorTest.java
+276 -1
View File
@@ -16,7 +16,17 @@
*/
package org.apache.rocketmq.auth.authorization;
import apache.rocketmq.v2.ClientType;
import apache.rocketmq.v2.HeartbeatRequest;
import apache.rocketmq.v2.NotifyClientTerminationRequest;
import apache.rocketmq.v2.Publishing;
import apache.rocketmq.v2.QueryRouteRequest;
import apache.rocketmq.v2.Settings;
import apache.rocketmq.v2.TelemetryCommand;
import apache.rocketmq.v2.ThreadStackTrace;
import apache.rocketmq.v2.VerifyMessageResult;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.apache.commons.collections.CollectionUtils;
@@ -33,15 +43,31 @@ import org.apache.rocketmq.auth.authorization.factory.AuthorizationFactory;
import org.apache.rocketmq.auth.authorization.manager.AuthorizationMetadataManager;
import org.apache.rocketmq.auth.authorization.model.Acl;
import org.apache.rocketmq.auth.authorization.model.Resource;
import org.apache.rocketmq.auth.authorization.strategy.AuthorizationStrategy;
import org.apache.rocketmq.auth.config.AuthConfig;
import org.apache.rocketmq.auth.helper.AuthTestHelper;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.action.Action;
import org.apache.rocketmq.common.resource.ResourcePattern;
import org.apache.rocketmq.common.resource.ResourceType;
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.UnregisterClientRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ViewMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumerData;
import org.apache.rocketmq.remoting.protocol.heartbeat.HeartbeatData;
import org.apache.rocketmq.remoting.protocol.heartbeat.ProducerData;
import org.junit.After;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Before;
import org.junit.Test;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
public class AuthorizationEvaluatorTest {
private AuthConfig authConfig;
@@ -384,6 +410,255 @@ public class AuthorizationEvaluatorTest {
});
}
@Test
public void evaluateTypedAnyListResources() {
Assume.assumeFalse(MixAll.isMac());
User listUser = User.of("list-user", "test");
User getUser = User.of("get-user", "test");
User literalUser = User.of("literal-user", "test");
User topicUser = User.of("topic-user", "test");
this.authenticationMetadataManager.createUser(listUser).join();
this.authenticationMetadataManager.createUser(getUser).join();
this.authenticationMetadataManager.createUser(literalUser).join();
this.authenticationMetadataManager.createUser(topicUser).join();
this.authorizationMetadataManager.createAcl(AuthTestHelper.buildAcl(
"User:list-user", "Topic:*,Group:*", "List", null, Decision.ALLOW)).join();
this.authorizationMetadataManager.createAcl(AuthTestHelper.buildAcl(
"User:get-user", "Topic:*,Group:*", "Get", null, Decision.ALLOW)).join();
this.authorizationMetadataManager.createAcl(AuthTestHelper.buildAcl(
"User:literal-user", "Topic:orders,Group:consumers", "List", null, Decision.ALLOW)).join();
this.authorizationMetadataManager.createAcl(AuthTestHelper.buildAcl(
"User:topic-user", "Topic:*", "List", null, Decision.ALLOW)).join();
this.evaluator.evaluate(Arrays.asList(
typedAnyListContext("list-user", ResourceType.TOPIC),
typedAnyListContext("list-user", ResourceType.GROUP)));
Assert.assertThrows(AuthorizationException.class, () -> this.evaluator.evaluate(
Collections.singletonList(typedAnyListContext("get-user", ResourceType.TOPIC))));
Assert.assertThrows(AuthorizationException.class, () -> this.evaluator.evaluate(
Collections.singletonList(typedAnyListContext("literal-user", ResourceType.TOPIC))));
this.evaluator.evaluate(
Collections.singletonList(typedAnyListContext("topic-user", ResourceType.TOPIC)));
Assert.assertThrows(AuthorizationException.class, () -> this.evaluator.evaluate(
Collections.singletonList(typedAnyListContext("topic-user", ResourceType.GROUP))));
}
@Test
public void rejectsEmptyContextWithoutRequest() {
AuthorizationEvaluator requestEvaluator = new AuthorizationEvaluator(mock(AuthorizationStrategy.class));
Assert.assertThrows(AuthorizationException.class, () -> requestEvaluator.evaluate(null));
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(Collections.emptyList()));
}
@Test
public void evaluatesNonEmptyContextBeforeCompatibilityMatching() {
AuthorizationStrategy strategy = mock(AuthorizationStrategy.class);
AuthorizationEvaluator requestEvaluator = new AuthorizationEvaluator(strategy);
DefaultAuthorizationContext context = DefaultAuthorizationContext.of(
Subject.of("User:test"), Resource.ofTopic("topic"), Action.PUB, "127.0.0.1");
requestEvaluator.evaluate(RemotingCommand.createRequestCommand(-1, null),
Collections.singletonList(context));
verify(strategy).evaluate(context);
}
@Test
public void acceptsOnlyResourceLessRemotingCompatibilityShapes() {
AuthorizationEvaluator requestEvaluator = new AuthorizationEvaluator(mock(AuthorizationStrategy.class));
requestEvaluator.evaluate(producerHeartbeat("producerGroup"), Collections.emptyList());
requestEvaluator.evaluate(producerHeartbeat(" "), Collections.emptyList());
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(producerHeartbeat(null), Collections.emptyList()));
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(remotingRequest(
RequestCode.HEART_BEAT, new HeartbeatData().encode()), Collections.emptyList()));
HeartbeatData invalidHeartbeat = new HeartbeatData();
invalidHeartbeat.setProducerDataSet(Collections.singleton(null));
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(remotingRequest(
RequestCode.HEART_BEAT, invalidHeartbeat.encode()), Collections.emptyList()));
HeartbeatData mixedHeartbeat = new HeartbeatData();
ProducerData producerData = new ProducerData();
producerData.setGroupName("producerGroup");
mixedHeartbeat.setProducerDataSet(Collections.singleton(producerData));
ConsumerData consumerData = new ConsumerData();
consumerData.setGroupName("consumerGroup");
mixedHeartbeat.setConsumerDataSet(Collections.singleton(consumerData));
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(remotingRequest(RequestCode.HEART_BEAT, mixedHeartbeat.encode()),
Collections.emptyList()));
requestEvaluator.evaluate(unregister("producerGroup", null), Collections.emptyList());
requestEvaluator.evaluate(unregister(null, "producerGroup", null), Collections.emptyList());
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(unregister(null, null), Collections.emptyList()));
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(unregister("producerGroup", "consumerGroup"),
Collections.emptyList()));
requestEvaluator.evaluate(endTransaction(true), Collections.emptyList());
requestEvaluator.evaluate(endTransaction(false), Collections.emptyList());
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(
RemotingCommand.createRequestCommand(RequestCode.END_TRANSACTION, null),
Collections.emptyList()));
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(endTransaction(
"topic", 1L, 2L, MessageSysFlag.TRANSACTION_COMMIT_TYPE, "messageId"),
Collections.emptyList()));
requestEvaluator.evaluate(endTransaction(
null, -1L, 2L, MessageSysFlag.TRANSACTION_COMMIT_TYPE, "messageId"), Collections.emptyList());
requestEvaluator.evaluate(endTransaction(
null, 1L, -1L, MessageSysFlag.TRANSACTION_COMMIT_TYPE, "messageId"), Collections.emptyList());
requestEvaluator.evaluate(endTransaction(
null, 1L, 2L, 99, "messageId"), Collections.emptyList());
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(
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(RemotingCommand.createRequestCommand(-1, null),
Collections.emptyList()));
}
@Test
public void acceptsOnlyResourceLessGrpcCompatibilityShapes() {
AuthorizationEvaluator requestEvaluator = new AuthorizationEvaluator(mock(AuthorizationStrategy.class));
requestEvaluator.evaluate(HeartbeatRequest.newBuilder()
.setClientType(ClientType.PRODUCER)
.build(), Collections.emptyList());
requestEvaluator.evaluate(HeartbeatRequest.newBuilder()
.setClientType(ClientType.CLIENT_TYPE_UNSPECIFIED)
.build(), Collections.emptyList());
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(HeartbeatRequest.newBuilder()
.setClientType(ClientType.PRODUCER)
.setGroup(apache.rocketmq.v2.Resource.newBuilder().setName("consumerGroup"))
.build(), Collections.emptyList()));
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(HeartbeatRequest.newBuilder()
.setClientType(ClientType.PUSH_CONSUMER)
.build(), Collections.emptyList()));
requestEvaluator.evaluate(NotifyClientTerminationRequest.getDefaultInstance(), Collections.emptyList());
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(NotifyClientTerminationRequest.newBuilder()
.setGroup(apache.rocketmq.v2.Resource.newBuilder().setName("consumerGroup"))
.build(), Collections.emptyList()));
requestEvaluator.evaluate(TelemetryCommand.newBuilder()
.setThreadStackTrace(ThreadStackTrace.newBuilder().setNonce("nonce"))
.build(), Collections.emptyList());
requestEvaluator.evaluate(TelemetryCommand.newBuilder()
.setVerifyMessageResult(VerifyMessageResult.newBuilder().setNonce("nonce"))
.build(), Collections.emptyList());
requestEvaluator.evaluate(TelemetryCommand.newBuilder()
.setThreadStackTrace(ThreadStackTrace.getDefaultInstance())
.build(), Collections.emptyList());
requestEvaluator.evaluate(TelemetryCommand.newBuilder()
.setVerifyMessageResult(VerifyMessageResult.getDefaultInstance())
.build(), Collections.emptyList());
requestEvaluator.evaluate(TelemetryCommand.newBuilder()
.setSettings(Settings.newBuilder().setPublishing(Publishing.getDefaultInstance()))
.build(), Collections.emptyList());
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(TelemetryCommand.newBuilder()
.setSettings(Settings.newBuilder().setPublishing(Publishing.newBuilder()
.addTopics(apache.rocketmq.v2.Resource.newBuilder().setName("topic"))))
.build(), Collections.emptyList()));
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(TelemetryCommand.getDefaultInstance(), Collections.emptyList()));
Assert.assertThrows(AuthorizationException.class,
() -> requestEvaluator.evaluate(QueryRouteRequest.getDefaultInstance(), Collections.emptyList()));
}
private RemotingCommand producerHeartbeat(String producerGroup) {
HeartbeatData heartbeatData = new HeartbeatData();
ProducerData producerData = new ProducerData();
producerData.setGroupName(producerGroup);
heartbeatData.setProducerDataSet(Collections.singleton(producerData));
return remotingRequest(RequestCode.HEART_BEAT, heartbeatData.encode());
}
private RemotingCommand unregister(String producerGroup, String consumerGroup) {
return unregister("clientId", producerGroup, consumerGroup);
}
private RemotingCommand unregister(String clientId, String producerGroup, String consumerGroup) {
UnregisterClientRequestHeader header = new UnregisterClientRequestHeader();
header.setClientID(clientId);
header.setProducerGroup(producerGroup);
header.setConsumerGroup(consumerGroup);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UNREGISTER_CLIENT, header);
request.makeCustomHeaderToNet();
return request;
}
private RemotingCommand endTransaction(boolean complete) {
return endTransaction(null, 1L, 2L, MessageSysFlag.TRANSACTION_COMMIT_TYPE,
complete ? "messageId" : null);
}
private RemotingCommand endTransaction(String topic, Long transactionOffset,
Long commitLogOffset, Integer state, String messageId) {
EndTransactionRequestHeader header = new EndTransactionRequestHeader();
header.setTopic(topic);
header.setProducerGroup("producerGroup");
header.setTranStateTableOffset(transactionOffset);
header.setCommitLogOffset(commitLogOffset);
header.setCommitOrRollback(state);
header.setMsgId(messageId);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.END_TRANSACTION, header);
request.makeCustomHeaderToNet();
return request;
}
private RemotingCommand viewMessage(Long offset) {
return viewMessage(null, offset);
}
private RemotingCommand viewMessage(String topic, Long offset) {
ViewMessageRequestHeader header = new ViewMessageRequestHeader();
header.setTopic(topic);
header.setOffset(offset);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.VIEW_MESSAGE_BY_ID, header);
request.makeCustomHeaderToNet();
return request;
}
private RemotingCommand remotingRequest(int requestCode, byte[] body) {
RemotingCommand request = RemotingCommand.createRequestCommand(requestCode, null);
request.setBody(body);
return request;
}
private DefaultAuthorizationContext typedAnyListContext(String username, ResourceType resourceType) {
DefaultAuthorizationContext context = DefaultAuthorizationContext.of(
Subject.of("User:" + username),
Resource.of(resourceType, null, ResourcePattern.ANY),
Action.LIST,
"192.168.0.1");
context.setRpcCode(String.valueOf(RequestCode.GET_ALL_TOPIC_CONFIG));
return context;
}
private void clearAllUsers() {
List<User> users = this.authenticationMetadataManager.listUser(null).join();
if (CollectionUtils.isEmpty(users)) {
@@ -399,4 +674,4 @@ public class AuthorizationEvaluatorTest {
}
acls.forEach(acl -> this.authorizationMetadataManager.deleteAcl(acl.getSubject(), null, null).join());
}
}
}
auth/src/test/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilderTest.java
+1026 -84
View File
File diff suppressed because it is too large Load Diff
auth/src/test/java/org/apache/rocketmq/auth/authorization/model/ResourceTest.java
+30 -1
View File
@@ -16,6 +16,9 @@
*/
package org.apache.rocketmq.auth.authorization.model;
import java.util.Collections;
import org.apache.rocketmq.auth.authorization.enums.Decision;
import org.apache.rocketmq.common.action.Action;
import org.apache.rocketmq.common.resource.ResourcePattern;
import org.apache.rocketmq.common.resource.ResourceType;
import org.junit.Assert;
@@ -48,6 +51,32 @@ public class ResourceTest {
@Test
public void isMatch() {
Resource topicAny = Resource.of("Topic:*");
Resource groupAny = Resource.of("Group:*");
Resource topicListResource = Resource.of(ResourceType.TOPIC, null, ResourcePattern.ANY);
Resource groupListResource = Resource.of(ResourceType.GROUP, null, ResourcePattern.ANY);
Assert.assertTrue(topicAny.isMatch(topicListResource));
Assert.assertTrue(groupAny.isMatch(groupListResource));
Assert.assertFalse(topicAny.isMatch(groupListResource));
Assert.assertFalse(Resource.ofTopic("orders").isMatch(topicListResource));
Assert.assertFalse(Resource.ofCluster("DefaultCluster").isMatch(topicListResource));
}
}
@Test
public void typedAnyListPolicyMatch() {
Resource topicListResource = Resource.of(ResourceType.TOPIC, null, ResourcePattern.ANY);
Resource groupListResource = Resource.of(ResourceType.GROUP, null, ResourcePattern.ANY);
PolicyEntry topicListPolicy = PolicyEntry.of(
Resource.of("Topic:*"), Collections.singletonList(Action.LIST), null, Decision.ALLOW);
PolicyEntry literalTopicPolicy = PolicyEntry.of(
Resource.ofTopic("orders"), Collections.singletonList(Action.LIST), null, Decision.ALLOW);
Assert.assertTrue(topicListPolicy.isMatchResource(topicListResource));
Assert.assertTrue(topicListPolicy.isMatchAction(Collections.singletonList(Action.LIST)));
Assert.assertTrue(topicListPolicy.isMatchResource(Resource.ofTopic("orders")));
Assert.assertFalse(topicListPolicy.isMatchResource(groupListResource));
Assert.assertFalse(literalTopicPolicy.isMatchResource(topicListResource));
Assert.assertFalse(topicListPolicy.isMatchAction(Collections.singletonList(Action.GET)));
}
}
broker/src/main/java/org/apache/rocketmq/broker/auth/pipeline/AuthorizationPipeline.java
+1 -1
View File
@@ -50,7 +50,7 @@ public class AuthorizationPipeline implements RequestPipeline {
}
try {
List<AuthorizationContext> contexts = newContexts(ctx, request);
evaluator.evaluate(contexts);
evaluator.evaluate(request, contexts);
} catch (AuthorizationException | AuthenticationException ex) {
throw new AbortProcessException(ResponseCode.NO_PERMISSION, ex.getMessage());
} catch (Throwable ex) {
broker/src/main/java/org/apache/rocketmq/broker/processor/QueryMessageProcessor.java
+52
View File
@@ -20,6 +20,10 @@ import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.FileRegion;
import io.opentelemetry.api.common.Attributes;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.broker.BrokerController;
@@ -28,6 +32,9 @@ import org.apache.rocketmq.broker.pagecache.QueryMessageTransfer;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
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.topic.TopicValidator;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingHelper;
@@ -42,6 +49,7 @@ import org.apache.rocketmq.remoting.protocol.header.QueryMessageResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.ViewMessageRequestHeader;
import org.apache.rocketmq.store.QueryMessageResult;
import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.timer.TimerMessageStore;
import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.LABEL_REQUEST_CODE;
import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.LABEL_RESPONSE_CODE;
@@ -49,6 +57,11 @@ import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.LABEL
public class QueryMessageProcessor implements NettyRequestProcessor {
private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
private static final Set<String> REAL_TOPIC_STORAGE_TOPICS = new HashSet<>(Arrays.asList(
TopicValidator.RMQ_SYS_SCHEDULE_TOPIC,
TimerMessageStore.TIMER_TOPIC,
TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC,
TopicValidator.RMQ_SYS_TRANS_CHECK_MAX_TIME_TOPIC));
private final BrokerController brokerController;
public QueryMessageProcessor(final BrokerController brokerController) {
@@ -146,6 +159,22 @@ public class QueryMessageProcessor implements NettyRequestProcessor {
final SelectMappedBufferResult selectMappedBufferResult =
this.brokerController.getMessageStore().selectOneMessageByOffset(requestHeader.getOffset());
if (selectMappedBufferResult != null) {
if (StringUtils.isNotBlank(requestHeader.getTopic())) {
MessageExt message = MessageDecoder.decode(
selectMappedBufferResult.getByteBuffer().duplicate(), false, false);
if (message == null) {
selectMappedBufferResult.release();
response.setCode(ResponseCode.SYSTEM_ERROR);
response.setRemark("decode message by the offset failed");
return response;
}
if (!matchesRequestTopic(requestHeader.getTopic(), message)) {
selectMappedBufferResult.release();
response.setCode(ResponseCode.NO_PERMISSION);
response.setRemark("The topic does not match the message");
return response;
}
}
response.setCode(ResponseCode.SUCCESS);
response.setRemark(null);
@@ -181,4 +210,27 @@ public class QueryMessageProcessor implements NettyRequestProcessor {
return response;
}
private boolean matchesRequestTopic(String requestTopic, MessageExt message) {
String logicalTopic = message.getTopic();
if (REAL_TOPIC_STORAGE_TOPICS.contains(logicalTopic)) {
logicalTopic = message.getProperty(MessageConst.PROPERTY_REAL_TOPIC);
}
if (Objects.equals(requestTopic, logicalTopic)) {
return true;
}
if (MixAll.isLmq(requestTopic)
&& StringUtils.isNotBlank(logicalTopic)
&& MixAll.topicAllowsLMQ(logicalTopic)) {
String multiDispatch = message.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
if (StringUtils.isNotBlank(multiDispatch)) {
for (String dispatchTopic : StringUtils.split(multiDispatch, MixAll.LMQ_DISPATCH_SEPARATOR)) {
if (Objects.equals(requestTopic, dispatchTopic)) {
return true;
}
}
}
}
return false;
}
}
broker/src/test/java/org/apache/rocketmq/broker/auth/pipeline/AuthorizationPipelineTest.java
+72
View File
@@ -0,0 +1,72 @@
/*
* 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.pipeline;
import io.netty.channel.ChannelHandlerContext;
import java.util.Collections;
import java.util.List;
import org.apache.rocketmq.auth.authorization.context.AuthorizationContext;
import org.apache.rocketmq.auth.config.AuthConfig;
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.heartbeat.HeartbeatData;
import org.apache.rocketmq.remoting.protocol.heartbeat.ProducerData;
import org.junit.Assert;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThatCode;
public class AuthorizationPipelineTest {
@Test
public void allowsCompatibleRequestWithEmptyContexts() throws Exception {
AuthorizationPipeline pipeline = createPipeline();
ProducerData producerData = new ProducerData();
producerData.setGroupName("producerGroup");
HeartbeatData heartbeatData = new HeartbeatData();
heartbeatData.getProducerDataSet().add(producerData);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.HEART_BEAT, null);
request.setBody(heartbeatData.encode());
assertThatCode(() -> pipeline.execute(null, request)).doesNotThrowAnyException();
}
@Test
public void rejectsUnsupportedRequestWithEmptyContexts() {
AuthorizationPipeline pipeline = createPipeline();
RemotingCommand request = RemotingCommand.createRequestCommand(-1, null);
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");
authConfig.setAuthorizationEnabled(true);
return new AuthorizationPipeline(authConfig) {
@Override
protected List<AuthorizationContext> newContexts(ChannelHandlerContext ctx,
RemotingCommand request) {
return Collections.emptyList();
}
};
}
}
broker/src/test/java/org/apache/rocketmq/broker/processor/ClientManageProcessorTest.java
+19
View File
@@ -42,6 +42,7 @@ import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumerData;
import org.apache.rocketmq.remoting.protocol.heartbeat.HeartbeatData;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.protocol.heartbeat.ProducerData;
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.junit.Before;
@@ -146,6 +147,24 @@ public class ClientManageProcessorTest {
assertThat(consumerGroupInfoSimple).isEqualTo(consumerGroupInfo);
}
@Test
public void processRequest_producerHeartbeat() throws RemotingCommandException {
ProducerData producerData = new ProducerData();
producerData.setGroupName(group);
HeartbeatData heartbeatData = new HeartbeatData();
heartbeatData.setClientID(clientId);
heartbeatData.getProducerDataSet().add(producerData);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.HEART_BEAT, null);
request.setLanguage(LanguageCode.JAVA);
request.setVersion(100);
request.setBody(heartbeatData.encode());
RemotingCommand response = clientManageProcessor.processRequest(handlerContext, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(brokerController.getProducerManager().getGroupChannelTable().get(group)).containsKey(channel);
}
@Test
public void test_heartbeat_costTime() {
String topic = "TOPIC_TEST";
broker/src/test/java/org/apache/rocketmq/broker/processor/EndTransactionProcessorTest.java
+19 -1
View File
@@ -154,6 +154,24 @@ public class EndTransactionProcessorTest {
assertThat(response.getCode()).isEqualTo(ResponseCode.ILLEGAL_OPERATION);
}
@Test
public void testProcessRequestAllowsMissingTopicForCompatibility() throws RemotingCommandException {
when(transactionMsgService.commitMessage(any(EndTransactionRequestHeader.class)))
.thenReturn(createResponse(ResponseCode.SUCCESS));
when(messageStore.putMessage(any(MessageExtBrokerInner.class)))
.thenReturn(new PutMessageResult(PutMessageStatus.PUT_OK,
createAppendMessageResult(AppendMessageStatus.PUT_OK)));
EndTransactionRequestHeader header = createEndTransactionRequestHeader(
MessageSysFlag.TRANSACTION_COMMIT_TYPE, false);
header.setTopic(null);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.END_TRANSACTION, header);
request.makeCustomHeaderToNet();
RemotingCommand response = endTransactionProcessor.processRequest(handlerContext, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
private MessageExt createDefaultMessageExt() {
MessageExt messageExt = new MessageExt();
messageExt.setMsgId("12345678");
@@ -169,7 +187,7 @@ public class EndTransactionProcessorTest {
private EndTransactionRequestHeader createEndTransactionRequestHeader(int status, boolean isCheckMsg) {
EndTransactionRequestHeader header = new EndTransactionRequestHeader();
header.setTopic("topic");
header.setTopic(TOPIC);
header.setCommitLogOffset(123456789L);
header.setFromTransactionCheck(isCheckMsg);
header.setCommitOrRollback(status);
broker/src/test/java/org/apache/rocketmq/broker/processor/QueryMessageProcessorTest.java
+134 -4
View File
@@ -19,9 +19,19 @@ package org.apache.rocketmq.broker.processor;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashMap;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.MixAll;
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.topic.TopicValidator;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyClientConfig;
import org.apache.rocketmq.remoting.netty.NettyServerConfig;
@@ -34,6 +44,7 @@ import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.QueryMessageResult;
import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.timer.TimerMessageStore;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -42,12 +53,14 @@ import org.mockito.Mock;
import org.mockito.Spy;
import org.mockito.junit.MockitoJUnitRunner;
import java.util.HashMap;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
@@ -95,7 +108,7 @@ public class QueryMessageProcessorTest {
}
@Test
public void testViewMessageById() throws RemotingCommandException {
public void testViewMessageById() throws Exception {
ViewMessageRequestHeader viewMessageRequestHeader = new ViewMessageRequestHeader();
viewMessageRequestHeader.setTopic("topic");
viewMessageRequestHeader.setOffset(0L);
@@ -107,11 +120,128 @@ public class QueryMessageProcessorTest {
RemotingCommand response = queryMessageProcessor.processRequest(handlerContext, request);
Assert.assertEquals(response.getCode(), ResponseCode.SYSTEM_ERROR);
when(messageStore.selectOneMessageByOffset(anyLong())).thenReturn(new SelectMappedBufferResult(0, null, 0, null));
when(messageStore.selectOneMessageByOffset(anyLong())).thenReturn(messageResult("topic", null));
response = queryMessageProcessor.processRequest(handlerContext, request);
Assert.assertNull(response);
}
@Test
public void testViewMessageByIdAllowsMissingTopicForCompatibility() throws Exception {
ViewMessageRequestHeader header = new ViewMessageRequestHeader();
header.setOffset(0L);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.VIEW_MESSAGE_BY_ID, header);
request.makeCustomHeaderToNet();
when(messageStore.selectOneMessageByOffset(0L)).thenReturn(
new SelectMappedBufferResult(0L, ByteBuffer.allocate(Integer.BYTES), Integer.BYTES, null));
RemotingCommand response = queryMessageProcessor.processRequest(handlerContext, request);
Assert.assertNull(response);
verify(messageStore).selectOneMessageByOffset(0L);
verify(channel).writeAndFlush(any());
}
@Test
public void testViewMessageByIdRejectsMismatchedTopic() throws Exception {
RemotingCommand request = createViewMessageRequest("allowedTopic");
SelectMappedBufferResult result = spy(messageResult("actualTopic", null));
when(messageStore.selectOneMessageByOffset(0L)).thenReturn(result);
RemotingCommand response = queryMessageProcessor.processRequest(handlerContext, request);
Assert.assertEquals(ResponseCode.NO_PERMISSION, response.getCode());
verify(result).release();
verify(channel, never()).writeAndFlush(any());
}
@Test
public void testViewMessageByIdRejectsForgedRealTopicOnNormalMessage() throws Exception {
RemotingCommand request = createViewMessageRequest("allowedTopic");
SelectMappedBufferResult result = spy(messageResult("actualTopic", "allowedTopic"));
when(messageStore.selectOneMessageByOffset(0L)).thenReturn(result);
RemotingCommand response = queryMessageProcessor.processRequest(handlerContext, request);
Assert.assertEquals(ResponseCode.NO_PERMISSION, response.getCode());
verify(result).release();
verify(channel, never()).writeAndFlush(any());
}
@Test
public void testViewMessageByIdRejectsUndecodableMessage() throws Exception {
RemotingCommand request = createViewMessageRequest("topic");
SelectMappedBufferResult result = spy(
new SelectMappedBufferResult(0L, ByteBuffer.allocate(Integer.BYTES), Integer.BYTES, null));
when(messageStore.selectOneMessageByOffset(0L)).thenReturn(result);
RemotingCommand response = queryMessageProcessor.processRequest(handlerContext, request);
Assert.assertEquals(ResponseCode.SYSTEM_ERROR, response.getCode());
verify(result).release();
verify(channel, never()).writeAndFlush(any());
}
@Test
public void testViewMessageByIdMatchesLogicalTopicOfInternalMessages() throws Exception {
RemotingCommand request = createViewMessageRequest("actualTopic");
for (String storedTopic : Arrays.asList(
TopicValidator.RMQ_SYS_SCHEDULE_TOPIC,
TimerMessageStore.TIMER_TOPIC,
TopicValidator.RMQ_SYS_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());
}
@Test
public void testViewMessageByIdMatchesLmqDispatchTopic() throws Exception {
RemotingCommand request = createViewMessageRequest("%LMQ%target");
when(messageStore.selectOneMessageByOffset(0L))
.thenReturn(messageResult("parentTopic", null, "%LMQ%other,%LMQ%target"));
Assert.assertNull(queryMessageProcessor.processRequest(handlerContext, request));
verify(channel).writeAndFlush(any());
}
private RemotingCommand createViewMessageRequest(String topic) {
ViewMessageRequestHeader header = new ViewMessageRequestHeader();
header.setTopic(topic);
header.setOffset(0L);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.VIEW_MESSAGE_BY_ID, header);
request.makeCustomHeaderToNet();
return request;
}
private SelectMappedBufferResult messageResult(String topic, String realTopic) throws Exception {
return messageResult(topic, realTopic, null);
}
private SelectMappedBufferResult messageResult(String topic, String realTopic, String multiDispatch)
throws Exception {
MessageExt message = new MessageExt();
message.setBody("body".getBytes(StandardCharsets.UTF_8));
message.setTopic(topic);
message.setQueueId(0);
message.setQueueOffset(0L);
message.setCommitLogOffset(0L);
message.setBornHost(new InetSocketAddress("127.0.0.1", 10911));
message.setStoreHost(new InetSocketAddress("127.0.0.1", 10911));
if (realTopic != null) {
MessageAccessor.putProperty(message, MessageConst.PROPERTY_REAL_TOPIC, realTopic);
}
if (multiDispatch != null) {
MessageAccessor.putProperty(message, MessageConst.PROPERTY_INNER_MULTI_DISPATCH, multiDispatch);
}
ByteBuffer buffer = ByteBuffer.wrap(MessageDecoder.encode(message, false));
return new SelectMappedBufferResult(0L, buffer, buffer.remaining(), null);
}
private RemotingCommand createQueryMessageRequest(String topic, String key, int maxNum, long beginTimestamp, long endTimestamp,String flag) {
QueryMessageRequestHeader requestHeader = new QueryMessageRequestHeader();
requestHeader.setTopic(topic);
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/pipeline/AuthorizationPipeline.java
+1 -1
View File
@@ -48,7 +48,7 @@ public class AuthorizationPipeline implements RequestPipeline {
}
try {
List<AuthorizationContext> contexts = newContexts(context, headers, request);
authorizationEvaluator.evaluate(contexts);
authorizationEvaluator.evaluate(request, contexts);
} catch (AuthorizationException | AuthenticationException ex) {
throw ex;
} catch (Throwable ex) {
proxy/src/main/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServer.java
+6
View File
@@ -62,6 +62,7 @@ import org.apache.rocketmq.remoting.netty.ResponseFuture;
import org.apache.rocketmq.remoting.netty.TlsSystemConfig;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.RequestHeaderRegistry;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
public class RemotingProtocolServer implements StartAndShutdown, RemotingProxyOutClient {
@@ -239,6 +240,11 @@ public class RemotingProtocolServer implements StartAndShutdown, RemotingProxyOu
remotingServer.registerProcessor(RequestCode.UNLOCK_BATCH_MQ, consumerManagerActivity, this.defaultExecutor);
remotingServer.registerProcessor(RequestCode.GET_ROUTEINFO_BY_TOPIC, getTopicRouteActivity, this.topicRouteExecutor);
/*
* Initialize the mapping of request codes to request headers.
*/
RequestHeaderRegistry.getInstance().initialize();
}
@Override
proxy/src/main/java/org/apache/rocketmq/proxy/remoting/pipeline/AuthorizationPipeline.java
+1 -1
View File
@@ -49,7 +49,7 @@ public class AuthorizationPipeline implements RequestPipeline {
}
try {
List<AuthorizationContext> contexts = newContexts(request, ctx, context);
authorizationEvaluator.evaluate(contexts);
authorizationEvaluator.evaluate(request, contexts);
} catch (AuthorizationException | AuthenticationException ex) {
throw ex;
} catch (Throwable ex) {
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/pipeline/AuthorizationPipelineTest.java
+71
View File
@@ -0,0 +1,71 @@
/*
* 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.ClientType;
import apache.rocketmq.v2.HeartbeatRequest;
import apache.rocketmq.v2.TelemetryCommand;
import com.google.protobuf.GeneratedMessageV3;
import io.grpc.Metadata;
import java.util.Collections;
import java.util.List;
import org.apache.rocketmq.auth.authorization.context.AuthorizationContext;
import org.apache.rocketmq.auth.authorization.exception.AuthorizationException;
import org.apache.rocketmq.auth.config.AuthConfig;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
import org.junit.Assert;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.mockito.Mockito.mock;
public class AuthorizationPipelineTest {
@Test
public void allowsCompatibleRequestWithEmptyContexts() {
AuthorizationPipeline pipeline = createPipeline();
HeartbeatRequest request = HeartbeatRequest.newBuilder()
.setClientType(ClientType.PRODUCER)
.build();
assertThatCode(() -> pipeline.execute(ProxyContext.create(), new Metadata(), request))
.doesNotThrowAnyException();
}
@Test
public void rejectsUnsupportedRequestWithEmptyContexts() {
AuthorizationPipeline pipeline = createPipeline();
Assert.assertThrows(AuthorizationException.class,
() -> pipeline.execute(ProxyContext.create(), new Metadata(),
TelemetryCommand.getDefaultInstance()));
}
private AuthorizationPipeline createPipeline() {
AuthConfig authConfig = new AuthConfig();
authConfig.setConfigName("grpc-authorization-pipeline-test");
authConfig.setAuthorizationEnabled(true);
return new AuthorizationPipeline(authConfig, mock(MessagingProcessor.class)) {
@Override
protected List<AuthorizationContext> newContexts(ProxyContext context, Metadata headers,
GeneratedMessageV3 request) {
return Collections.emptyList();
}
};
}
}
proxy/src/test/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServerTest.java
+47
View File
@@ -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.proxy.remoting;
import org.apache.rocketmq.remoting.RemotingServer;
import org.apache.rocketmq.remoting.protocol.RequestHeaderRegistry;
import org.junit.Test;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.verify;
public class RemotingProtocolServerTest {
@Test
public void shouldInitializeRequestHeaderRegistryWhenRegisteringProcessors() {
RemotingProtocolServer protocolServer =
mock(RemotingProtocolServer.class, Mockito.CALLS_REAL_METHODS);
RemotingServer remotingServer = mock(RemotingServer.class);
RequestHeaderRegistry requestHeaderRegistry = mock(RequestHeaderRegistry.class);
try (MockedStatic<RequestHeaderRegistry> registry = mockStatic(RequestHeaderRegistry.class)) {
registry.when(RequestHeaderRegistry::getInstance).thenReturn(requestHeaderRegistry);
protocolServer.registerRemotingServer(remotingServer);
verify(requestHeaderRegistry).initialize();
}
}
}
proxy/src/test/java/org/apache/rocketmq/proxy/remoting/pipeline/AuthorizationPipelineTest.java
+74
View File
@@ -0,0 +1,74 @@
/*
* 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.remoting.pipeline;
import io.netty.channel.ChannelHandlerContext;
import java.util.Collections;
import java.util.List;
import org.apache.rocketmq.auth.authorization.context.AuthorizationContext;
import org.apache.rocketmq.auth.authorization.exception.AuthorizationException;
import org.apache.rocketmq.auth.config.AuthConfig;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.heartbeat.HeartbeatData;
import org.apache.rocketmq.remoting.protocol.heartbeat.ProducerData;
import org.junit.Assert;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.mockito.Mockito.mock;
public class AuthorizationPipelineTest {
@Test
public void allowsCompatibleRequestWithEmptyContexts() throws Exception {
AuthorizationPipeline pipeline = createPipeline();
ProducerData producerData = new ProducerData();
producerData.setGroupName("producerGroup");
HeartbeatData heartbeatData = new HeartbeatData();
heartbeatData.getProducerDataSet().add(producerData);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.HEART_BEAT, null);
request.setBody(heartbeatData.encode());
assertThatCode(() -> pipeline.execute(null, request, ProxyContext.create()))
.doesNotThrowAnyException();
}
@Test
public void rejectsUnsupportedRequestWithEmptyContexts() {
AuthorizationPipeline pipeline = createPipeline();
RemotingCommand request = RemotingCommand.createRequestCommand(-1, null);
Assert.assertThrows(AuthorizationException.class,
() -> pipeline.execute(null, request, ProxyContext.create()));
}
private AuthorizationPipeline createPipeline() {
AuthConfig authConfig = new AuthConfig();
authConfig.setConfigName("remoting-authorization-pipeline-test");
authConfig.setAuthorizationEnabled(true);
return new AuthorizationPipeline(authConfig, mock(MessagingProcessor.class)) {
@Override
protected List<AuthorizationContext> newContexts(RemotingCommand request,
ChannelHandlerContext ctx, ProxyContext context) {
return Collections.emptyList();
}
};
}
}
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/CreateTopicListRequestHeader.java
+6 -4
View File
@@ -16,13 +16,15 @@
*/
package org.apache.rocketmq.remoting.protocol.header;
import org.apache.rocketmq.common.action.Action;
import org.apache.rocketmq.common.action.RocketMQAction;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.rpc.RpcRequestHeader;
@RocketMQAction(value = RequestCode.UPDATE_AND_CREATE_TOPIC_LIST, action = Action.CREATE)
/**
* Header metadata for a batch topic-create request. Topic names are carried in the request body.
*
* <p>This header must not declare {@code @RocketMQAction}: it has no concrete topic resource.
* Authorization resources are resolved from the request body by {@code DefaultAuthorizationContextBuilder}.
*/
public class CreateTopicListRequestHeader extends RpcRequestHeader {
@Override
public void checkFields() throws RemotingCommandException {
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllSubscriptionGroupRequestHeader.java
+6 -5
View File
@@ -17,15 +17,16 @@
package org.apache.rocketmq.remoting.protocol.header;
import org.apache.rocketmq.common.action.Action;
import org.apache.rocketmq.common.action.RocketMQAction;
import org.apache.rocketmq.common.resource.ResourceType;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.annotation.CFNotNull;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RequestCode;
@RocketMQAction(value = RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, resource = ResourceType.GROUP, action = Action.GET)
/**
* This request targets the complete subscription-group configuration set and does not identify a single group.
*
* <p>This header must not declare {@code @RocketMQAction}: it carries no resource fields, and the
* typed {@code Group:ANY + LIST} resource is constructed by {@code DefaultAuthorizationContextBuilder}.
*/
public class GetAllSubscriptionGroupRequestHeader implements CommandCustomHeader {
@Override
public void checkFields() throws RemotingCommandException {
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllSubscriptionGroupResponseHeader.java
-5
View File
@@ -17,15 +17,10 @@
package org.apache.rocketmq.remoting.protocol.header;
import org.apache.rocketmq.common.action.Action;
import org.apache.rocketmq.common.action.RocketMQAction;
import org.apache.rocketmq.common.resource.ResourceType;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.annotation.CFNotNull;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RequestCode;
@RocketMQAction(value = RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, resource = ResourceType.GROUP, action = Action.LIST)
public class GetAllSubscriptionGroupResponseHeader implements CommandCustomHeader {
@Override
public void checkFields() throws RemotingCommandException {
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllTopicConfigRequestHeader.java
+6 -5
View File
@@ -17,15 +17,16 @@
package org.apache.rocketmq.remoting.protocol.header;
import org.apache.rocketmq.common.action.Action;
import org.apache.rocketmq.common.action.RocketMQAction;
import org.apache.rocketmq.common.resource.ResourceType;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.annotation.CFNotNull;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RequestCode;
@RocketMQAction(value = RequestCode.GET_ALL_TOPIC_CONFIG, resource = ResourceType.TOPIC, action = Action.GET)
/**
* This request targets the complete topic configuration set and does not identify a single topic.
*
* <p>This header must not declare {@code @RocketMQAction}: it carries no resource fields, and the
* typed {@code Topic:ANY + LIST} resource is constructed by {@code DefaultAuthorizationContextBuilder}.
*/
public class GetAllTopicConfigRequestHeader implements CommandCustomHeader {
@Override
public void checkFields() throws RemotingCommandException {
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllTopicConfigResponseHeader.java
-5
View File
@@ -20,14 +20,9 @@
*/
package org.apache.rocketmq.remoting.protocol.header;
import org.apache.rocketmq.common.action.Action;
import org.apache.rocketmq.common.action.RocketMQAction;
import org.apache.rocketmq.common.resource.ResourceType;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RequestCode;
@RocketMQAction(value = RequestCode.GET_ALL_TOPIC_CONFIG, resource = ResourceType.TOPIC, action = Action.LIST)
public class GetAllTopicConfigResponseHeader implements CommandCustomHeader {
@Override
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetLiteClientInfoRequestHeader.java
+12
View File
@@ -17,12 +17,24 @@
package org.apache.rocketmq.remoting.protocol.header;
import org.apache.rocketmq.common.action.Action;
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.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RequestCode;
@RocketMQAction(value = RequestCode.GET_LITE_CLIENT_INFO, action = Action.GET)
public class GetLiteClientInfoRequestHeader implements CommandCustomHeader {
@CFNotNull
@RocketMQResource(ResourceType.TOPIC)
private String parentTopic;
@CFNotNull
@RocketMQResource(ResourceType.GROUP)
private String group;
private String clientId;
private int maxCount = 1000;
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetLiteGroupInfoRequestHeader.java
+4
View File
@@ -16,12 +16,16 @@
*/
package org.apache.rocketmq.remoting.protocol.header;
import org.apache.rocketmq.common.action.Action;
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.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RequestCode;
@RocketMQAction(value = RequestCode.GET_LITE_GROUP_INFO, action = Action.GET)
public class GetLiteGroupInfoRequestHeader implements CommandCustomHeader {
@CFNotNull
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetLiteTopicInfoRequestHeader.java
+9
View File
@@ -17,11 +17,20 @@
package org.apache.rocketmq.remoting.protocol.header;
import org.apache.rocketmq.common.action.Action;
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.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RequestCode;
@RocketMQAction(value = RequestCode.GET_LITE_TOPIC_INFO, action = Action.GET)
public class GetLiteTopicInfoRequestHeader implements CommandCustomHeader {
@CFNotNull
@RocketMQResource(ResourceType.TOPIC)
private String parentTopic;
private String liteTopic;
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetMaxOffsetRequestHeader.java
+1 -1
View File
@@ -31,7 +31,7 @@ import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.rpc.TopicQueueRequestHeader;
@RocketMQAction(value = RequestCode.GET_MAX_OFFSET, action = Action.GET)
@RocketMQAction(value = RequestCode.GET_MAX_OFFSET, action = {Action.SUB, Action.GET})
public class GetMaxOffsetRequestHeader extends TopicQueueRequestHeader {
@CFNotNull
@RocketMQResource(ResourceType.TOPIC)
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetParentTopicInfoRequestHeader.java
+4
View File
@@ -17,12 +17,16 @@
package org.apache.rocketmq.remoting.protocol.header;
import org.apache.rocketmq.common.action.Action;
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.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RequestCode;
@RocketMQAction(value = RequestCode.GET_PARENT_TOPIC_INFO, action = Action.GET)
public class GetParentTopicInfoRequestHeader implements CommandCustomHeader {
@CFNotNull
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/PopLiteMessageRequestHeader.java
+4
View File
@@ -18,12 +18,16 @@
package org.apache.rocketmq.remoting.protocol.header;
import com.google.common.base.MoreObjects;
import org.apache.rocketmq.common.action.Action;
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.annotation.CFNotNull;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.rpc.RpcRequestHeader;
@RocketMQAction(value = RequestCode.POP_LITE_MESSAGE, action = Action.SUB)
public class PopLiteMessageRequestHeader extends RpcRequestHeader {
@CFNotNull
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/SearchOffsetRequestHeader.java
+1 -1
View File
@@ -31,7 +31,7 @@ import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.rpc.TopicQueueRequestHeader;
@RocketMQAction(value = RequestCode.SEARCH_OFFSET_BY_TIMESTAMP, action = Action.GET)
@RocketMQAction(value = RequestCode.SEARCH_OFFSET_BY_TIMESTAMP, action = {Action.SUB, Action.GET})
public class SearchOffsetRequestHeader extends TopicQueueRequestHeader {
@CFNotNull
@RocketMQResource(ResourceType.TOPIC)
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/TriggerLiteDispatchRequestHeader.java
+4
View File
@@ -16,12 +16,16 @@
*/
package org.apache.rocketmq.remoting.protocol.header;
import org.apache.rocketmq.common.action.Action;
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.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RequestCode;
@RocketMQAction(value = RequestCode.TRIGGER_LITE_DISPATCH, action = Action.UPDATE)
public class TriggerLiteDispatchRequestHeader implements CommandCustomHeader {
@CFNotNull