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

[ISSUE #6968] fix grpc acl bug (#6969)

Browse Source 
* feat(acl): fix acl bug

Signed-off-by: lyx <1419360299@qq.com>

# Conflicts:
#	proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java

* add access test for two client

Signed-off-by: lyx <1419360299@qq.com>

* use specific acl config

Signed-off-by: lyx <1419360299@qq.com>

* Recovering unchange file

Signed-off-by: lyx <1419360299@qq.com>

* let test pass

Signed-off-by: lyx <1419360299@qq.com>

---------

Signed-off-by: lyx <1419360299@qq.com>
This commit is contained in:
lyx
2023-07-15 18:15:57 +08:00
committed by GitHub
parent 70a66eda2c
commit 5914ff8dbb
8 changed files with 255 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAccessResource.java
+13 -8
View File
@@ -223,7 +223,7 @@ public class PlainAccessResource implements AccessResource {
if (!request.hasGroup()) {
throw new AclException("Consumer heartbeat doesn't have group");
} else {
accessResource.addResourceAndPerm(request.getGroup(), Permission.SUB);
accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
}
}
} else if (SendMessageRequest.getDescriptor().getFullName().equals(rpcFullName)) {
@@ -240,15 +240,15 @@ public class PlainAccessResource implements AccessResource {
accessResource.addResourceAndPerm(topic, Permission.PUB);
} else if (ReceiveMessageRequest.getDescriptor().getFullName().equals(rpcFullName)) {
ReceiveMessageRequest request = (ReceiveMessageRequest) messageV3;
accessResource.addResourceAndPerm(request.getGroup(), Permission.SUB);
accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
accessResource.addResourceAndPerm(request.getMessageQueue().getTopic(), Permission.SUB);
} else if (AckMessageRequest.getDescriptor().getFullName().equals(rpcFullName)) {
AckMessageRequest request = (AckMessageRequest) messageV3;
accessResource.addResourceAndPerm(request.getGroup(), Permission.SUB);
accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
accessResource.addResourceAndPerm(request.getTopic(), Permission.SUB);
} else if (ForwardMessageToDeadLetterQueueRequest.getDescriptor().getFullName().equals(rpcFullName)) {
ForwardMessageToDeadLetterQueueRequest request = (ForwardMessageToDeadLetterQueueRequest) messageV3;
accessResource.addResourceAndPerm(request.getGroup(), Permission.SUB);
accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
accessResource.addResourceAndPerm(request.getTopic(), Permission.SUB);
} else if (EndTransactionRequest.getDescriptor().getFullName().equals(rpcFullName)) {
EndTransactionRequest request = (EndTransactionRequest) messageV3;
@@ -264,7 +264,7 @@ public class PlainAccessResource implements AccessResource {
}
if (command.getSettings().hasSubscription()) {
Subscription subscription = command.getSettings().getSubscription();
accessResource.addResourceAndPerm(subscription.getGroup(), Permission.SUB);
accessResource.addGroupResourceAndPerm(subscription.getGroup(), Permission.SUB);
for (SubscriptionEntry entry : subscription.getSubscriptionsList()) {
accessResource.addResourceAndPerm(entry.getTopic(), Permission.SUB);
}
@@ -275,17 +275,17 @@ public class PlainAccessResource implements AccessResource {
}
} else if (NotifyClientTerminationRequest.getDescriptor().getFullName().equals(rpcFullName)) {
NotifyClientTerminationRequest request = (NotifyClientTerminationRequest) messageV3;
accessResource.addResourceAndPerm(request.getGroup(), Permission.SUB);
accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
} else if (QueryRouteRequest.getDescriptor().getFullName().equals(rpcFullName)) {
QueryRouteRequest request = (QueryRouteRequest) messageV3;
accessResource.addResourceAndPerm(request.getTopic(), Permission.ANY);
} else if (QueryAssignmentRequest.getDescriptor().getFullName().equals(rpcFullName)) {
QueryAssignmentRequest request = (QueryAssignmentRequest) messageV3;
accessResource.addResourceAndPerm(request.getGroup(), Permission.SUB);
accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
accessResource.addResourceAndPerm(request.getTopic(), Permission.SUB);
} else if (ChangeInvisibleDurationRequest.getDescriptor().getFullName().equals(rpcFullName)) {
ChangeInvisibleDurationRequest request = (ChangeInvisibleDurationRequest) messageV3;
accessResource.addResourceAndPerm(request.getGroup(), Permission.SUB);
accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
accessResource.addResourceAndPerm(request.getTopic(), Permission.SUB);
}
} catch (Throwable t) {
@@ -299,6 +299,11 @@ public class PlainAccessResource implements AccessResource {
addResourceAndPerm(resourceName, permission);
}
private void addGroupResourceAndPerm(Resource resource, byte permission) {
String resourceName = NamespaceUtil.wrapNamespace(resource.getResourceNamespace(), resource.getName());
addResourceAndPerm(getRetryTopic(resourceName), permission);
}
public static PlainAccessResource build(PlainAccessConfig plainAccessConfig, RemoteAddressStrategy remoteAddressStrategy) {
PlainAccessResource plainAccessResource = new PlainAccessResource();
plainAccessResource.setAccessKey(plainAccessConfig.getAccessKey());
acl/src/test/java/org/apache/rocketmq/acl/RemotingClientAccessTest.java
+189
View File
@@ -0,0 +1,189 @@
/*
* 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.acl;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.AclException;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.acl.plain.AclTestHelper;
import org.apache.rocketmq.acl.plain.PlainAccessResource;
import org.apache.rocketmq.acl.plain.PlainAccessValidator;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.header.PullMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
public class RemotingClientAccessTest {
private PlainAccessValidator plainAccessValidator;
private AclClientRPCHook aclClient;
private SessionCredentials sessionCredentials;
private File confHome;
private String clientAddress = "10.7.1.3";
@Before
public void init() throws IOException {
String folder = "access_acl_conf";
confHome = AclTestHelper.copyResources(folder, true);
System.setProperty("rocketmq.home.dir", confHome.getAbsolutePath());
System.setProperty("rocketmq.acl.plain.file", "/access_acl_conf/acl/plain_acl.yml".replace("/", File.separator));
plainAccessValidator = new PlainAccessValidator();
sessionCredentials = new SessionCredentials();
sessionCredentials.setAccessKey("rocketmq3");
sessionCredentials.setSecretKey("12345678");
aclClient = new AclClientRPCHook(sessionCredentials);
}
@After
public void cleanUp() {
AclTestHelper.recursiveDelete(confHome);
}
@Test(expected = AclException.class)
public void testProduceDenyTopic() {
SendMessageRequestHeader messageRequestHeader = new SendMessageRequestHeader();
messageRequestHeader.setTopic("topicD");
RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, messageRequestHeader);
aclClient.doBeforeRequest(clientAddress, remotingCommand);
ByteBuffer buf = remotingCommand.encodeHeader();
buf.getInt();
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), clientAddress);
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
Assert.fail("Should not throw IOException");
}
}
@Test
public void testProduceAuthorizedTopic() {
SendMessageRequestHeader messageRequestHeader = new SendMessageRequestHeader();
messageRequestHeader.setTopic("topicA");
RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, messageRequestHeader);
aclClient.doBeforeRequest(clientAddress, remotingCommand);
ByteBuffer buf = remotingCommand.encodeHeader();
buf.getInt();
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), clientAddress);
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
Assert.fail("Should not throw IOException");
}
}
@Test(expected = AclException.class)
public void testConsumeDenyTopic() {
PullMessageRequestHeader pullMessageRequestHeader = new PullMessageRequestHeader();
pullMessageRequestHeader.setTopic("topicD");
pullMessageRequestHeader.setConsumerGroup("groupB");
RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, pullMessageRequestHeader);
aclClient.doBeforeRequest("", remotingCommand);
ByteBuffer buf = remotingCommand.encodeHeader();
buf.getInt();
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
Assert.fail("Should not throw IOException");
}
}
@Test
public void testConsumeAuthorizedTopic() {
PullMessageRequestHeader pullMessageRequestHeader = new PullMessageRequestHeader();
pullMessageRequestHeader.setTopic("topicB");
pullMessageRequestHeader.setConsumerGroup("groupB");
RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, pullMessageRequestHeader);
aclClient.doBeforeRequest("", remotingCommand);
ByteBuffer buf = remotingCommand.encodeHeader();
buf.getInt();
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
Assert.fail("Should not throw IOException");
}
}
@Test(expected = AclException.class)
public void testConsumeInDeniedGroup() {
PullMessageRequestHeader pullMessageRequestHeader = new PullMessageRequestHeader();
pullMessageRequestHeader.setTopic("topicB");
pullMessageRequestHeader.setConsumerGroup("groupD");
RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, pullMessageRequestHeader);
aclClient.doBeforeRequest("", remotingCommand);
ByteBuffer buf = remotingCommand.encodeHeader();
buf.getInt();
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
Assert.fail("Should not throw IOException");
}
}
@Test
public void testConsumeInAuthorizedGroup() {
PullMessageRequestHeader pullMessageRequestHeader = new PullMessageRequestHeader();
pullMessageRequestHeader.setTopic("topicB");
pullMessageRequestHeader.setConsumerGroup("groupB");
RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, pullMessageRequestHeader);
aclClient.doBeforeRequest("", remotingCommand);
ByteBuffer buf = remotingCommand.encodeHeader();
buf.getInt();
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
Assert.fail("Should not throw IOException");
}
}
}
acl/src/test/resources/access_acl_conf/acl/plain_acl.yml
+31
View File
@@ -0,0 +1,31 @@
# 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.
accounts:
- accessKey: rocketmq3
secretKey: 12345678
admin: false
defaultTopicPerm: DENY
defaultGroupPerm: DENY
topicPerms:
- topicA=PUB
- topicB=SUB
- topicC=PUB|SUB
- topicD=DENY
groupPerms:
- groupB=SUB
- groupC=PUB|SUB
- groupD=DENY
acl/src/test/resources/conf/acl/plain_acl.yml
-1
View File
@@ -41,4 +41,3 @@ accounts:
whiteRemoteAddress: 192.168.1.*
# if it is admin, it could access all resources
admin: true
pom.xml
+1
View File
@@ -147,6 +147,7 @@
<awaitility.version>4.1.0</awaitility.version>
<truth.version>0.30</truth.version>
<s3mock-junit4.version>2.11.0</s3mock-junit4.version>
<rocketmq-client-java.version>5.0.5</rocketmq-client-java.version>
<!-- Build plugin dependencies -->
<versions-maven-plugin.version>2.2</versions-maven-plugin.version>
proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java
+15 -2
View File
@@ -29,11 +29,14 @@ import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.acl.AccessValidator;
import org.apache.rocketmq.acl.plain.PlainAccessValidator;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.BrokerStartup;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.thread.ThreadPoolMonitor;
import org.apache.rocketmq.common.utils.ServiceProvider;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.common.utils.AbstractStartAndShutdown;
@@ -75,16 +78,17 @@ public class ProxyStartup {
MessagingProcessor messagingProcessor = createMessagingProcessor();
List<AccessValidator> accessValidators = loadAccessValidators();
// create grpcServer
GrpcServer grpcServer = GrpcServerBuilder.newBuilder(executor, ConfigurationManager.getProxyConfig().getGrpcServerPort())
.addService(createServiceProcessor(messagingProcessor))
.addService(ChannelzService.newInstance(100))
.addService(ProtoReflectionService.newInstance())
.configInterceptor()
.configInterceptor(accessValidators)
.build();
PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(grpcServer);
RemotingProtocolServer remotingServer = new RemotingProtocolServer(messagingProcessor);
RemotingProtocolServer remotingServer = new RemotingProtocolServer(messagingProcessor, accessValidators);
PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(remotingServer);
// start servers one by one.
@@ -109,6 +113,15 @@ public class ProxyStartup {
log.info(new Date() + " rocketmq-proxy startup successfully");
}
protected static List<AccessValidator> loadAccessValidators() {
List<AccessValidator> accessValidators = ServiceProvider.load(AccessValidator.class);
if (accessValidators.isEmpty()) {
log.info("ServiceProvider loaded no AccessValidator, using default org.apache.rocketmq.acl.plain.PlainAccessValidator");
accessValidators.add(new PlainAccessValidator());
}
return accessValidators;
}
protected static void initConfiguration(CommandLineArgument commandLineArgument) throws Exception {
if (StringUtils.isNotBlank(commandLineArgument.getProxyConfigPath())) {
System.setProperty(Configuration.CONFIG_PATH_PROPERTY, commandLineArgument.getProxyConfigPath());
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java
+1 -9
View File
@@ -28,9 +28,7 @@ import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.acl.AccessValidator;
import org.apache.rocketmq.acl.plain.PlainAccessValidator;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.utils.ServiceProvider;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
@@ -98,14 +96,8 @@ public class GrpcServerBuilder {
return new GrpcServer(this.serverBuilder.build());
}
public GrpcServerBuilder configInterceptor() {
public GrpcServerBuilder configInterceptor(List<AccessValidator> accessValidators) {
// grpc interceptors, including acl, logging etc.
List<AccessValidator> accessValidators = ServiceProvider.load(AccessValidator.class);
if (accessValidators.isEmpty()) {
log.info("ServiceProvider loaded no AccessValidator, using default org.apache.rocketmq.acl.plain.PlainAccessValidator");
accessValidators.add(new PlainAccessValidator());
}
this.serverBuilder.intercept(new AuthenticationInterceptor(accessValidators));
this.serverBuilder
proxy/src/main/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServer.java
+5 -10
View File
@@ -19,7 +19,6 @@ package org.apache.rocketmq.proxy.remoting;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.netty.channel.Channel;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
@@ -28,15 +27,14 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.acl.AccessValidator;
import org.apache.rocketmq.acl.plain.PlainAccessValidator;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.future.FutureTaskExt;
import org.apache.rocketmq.common.thread.ThreadPoolMonitor;
import org.apache.rocketmq.common.thread.ThreadPoolStatusMonitor;
import org.apache.rocketmq.common.utils.StartAndShutdown;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.common.utils.StartAndShutdown;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.config.ProxyConfig;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
@@ -86,11 +84,11 @@ public class RemotingProtocolServer implements StartAndShutdown, RemotingProxyOu
protected final ThreadPoolExecutor defaultExecutor;
protected final ScheduledExecutorService timerExecutor;
public RemotingProtocolServer(MessagingProcessor messagingProcessor) {
public RemotingProtocolServer(MessagingProcessor messagingProcessor, List<AccessValidator> accessValidators) {
this.messagingProcessor = messagingProcessor;
this.remotingChannelManager = new RemotingChannelManager(this, messagingProcessor.getProxyRelayService());
RequestPipeline pipeline = createRequestPipeline();
RequestPipeline pipeline = createRequestPipeline(accessValidators);
this.getTopicRouteActivity = new GetTopicRouteActivity(pipeline, messagingProcessor);
this.clientManagerActivity = new ClientManagerActivity(pipeline, messagingProcessor, remotingChannelManager);
this.consumerManagerActivity = new ConsumerManagerActivity(pipeline, messagingProcessor);
@@ -254,15 +252,12 @@ public class RemotingProtocolServer implements StartAndShutdown, RemotingProxyOu
return future;
}
protected RequestPipeline createRequestPipeline() {
protected RequestPipeline createRequestPipeline(List<AccessValidator> accessValidators) {
RequestPipeline pipeline = (ctx, request, context) -> {
};
List<AccessValidator> accessValidatorList = new ArrayList<>();
accessValidatorList.add(new PlainAccessValidator());
// add pipeline
// the last pipe add will execute at the first
return pipeline.pipe(new AuthenticationPipeline(accessValidatorList));
return pipeline.pipe(new AuthenticationPipeline(accessValidators));
}
protected class ThreadPoolHeadSlowTimeMillsMonitor implements ThreadPoolStatusMonitor {