Fix ACL hook routing for internal proxy system resources

proxy/src/main/java/org/apache/rocketmq/proxy/common/InternalContextHolder.java

@@ -0,0 +1,34 @@
+/*
+ * 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.common;
+
+public class InternalContextHolder {
+    private static final ThreadLocal<Boolean> IS_INTERNAL = ThreadLocal.withInitial(() -> false);
+
+    public static void beginInternalScope() {
+        IS_INTERNAL.set(true);
+    }
+
+    public static void clear() {
+        IS_INTERNAL.remove();
+    }
+
+    public static boolean isInternalScope() {
+        return IS_INTERNAL.get();
+    }
+}

proxy/src/main/java/org/apache/rocketmq/proxy/common/SystemResourceAwareRpcHook.java

@@ -0,0 +1,133 @@
+/*
+ * 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.common;
+
+import java.util.Map;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.topic.TopicValidator;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeaderV2;
+import org.apache.rocketmq.remoting.protocol.header.UnregisterClientRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.namesrv.GetRouteInfoRequestHeader;
+
+public class SystemResourceAwareRpcHook implements RPCHook {
+    private final RPCHook userHook;
+    private final RPCHook systemHook;
+
+    public SystemResourceAwareRpcHook(RPCHook userHook, RPCHook systemHook) {
+        this.userHook = userHook;
+        this.systemHook = systemHook;
+    }
+
+    @Override
+    public void doBeforeRequest(String remoteAddr, RemotingCommand request) {
+        if (isTargetingSystemResource(request)) {
+            if (systemHook != null) {
+                systemHook.doBeforeRequest(remoteAddr, request);
+            }
+            return;
+        }
+        if (userHook != null) {
+            userHook.doBeforeRequest(remoteAddr, request);
+        }
+    }
+
+    @Override
+    public void doAfterResponse(String remoteAddr, RemotingCommand request, RemotingCommand response) {
+        if (isTargetingSystemResource(request)) {
+            if (systemHook != null) {
+                systemHook.doAfterResponse(remoteAddr, request, response);
+            }
+            return;
+        }
+        if (userHook != null) {
+            userHook.doAfterResponse(remoteAddr, request, response);
+        }
+    }
+
+    private boolean isTargetingSystemResource(RemotingCommand request) {
+        if (!InternalContextHolder.isInternalScope()) {
+            return false;
+        }
+
+        if (request == null) {
+            return false;
+        }
+
+        int code = request.getCode();
+        try {
+            switch (code) {
+                case RequestCode.GET_ROUTEINFO_BY_TOPIC:
+                    GetRouteInfoRequestHeader routeHeader = request
+                        .decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);
+                    return TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC.equals(routeHeader.getTopic());
+
+                case RequestCode.SEND_MESSAGE:
+                    SendMessageRequestHeader sendHeader = request
+                        .decodeCommandCustomHeader(SendMessageRequestHeader.class);
+                    return TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC.equals(sendHeader.getTopic())
+                        || MixAll.CLIENT_INNER_PRODUCER_GROUP.equals(sendHeader.getProducerGroup());
+
+                case RequestCode.SEND_MESSAGE_V2:
+                    SendMessageRequestHeaderV2 sendHeaderV2 = request
+                        .decodeCommandCustomHeader(SendMessageRequestHeaderV2.class);
+                    SendMessageRequestHeader v1Header =
+                        SendMessageRequestHeaderV2.createSendMessageRequestHeaderV1(sendHeaderV2);
+                    return TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC.equals(v1Header.getTopic())
+                        || MixAll.CLIENT_INNER_PRODUCER_GROUP.equals(v1Header.getProducerGroup());
+
+                case RequestCode.UNREGISTER_CLIENT:
+                    UnregisterClientRequestHeader unregisterHeader = request
+                        .decodeCommandCustomHeader(UnregisterClientRequestHeader.class);
+                    return MixAll.CLIENT_INNER_PRODUCER_GROUP.equals(unregisterHeader.getProducerGroup())
+                        || MixAll.TOOLS_CONSUMER_GROUP.equals(unregisterHeader.getConsumerGroup());
+
+                default:
+                    return checkFallbackExtFields(request.getExtFields());
+            }
+        } catch (Exception e) {
+            return false;
+        }
+    }
+
+    private boolean checkFallbackExtFields(Map<String, String> extFields) {
+        if (extFields == null || extFields.isEmpty()) {
+            return false;
+        }
+        String topic = extFields.get("topic");
+        if (TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC.equals(topic)) {
+            return true;
+        }
+
+        String producerGroup = extFields.get("producerGroup");
+        if (MixAll.CLIENT_INNER_PRODUCER_GROUP.equals(producerGroup)) {
+            return true;
+        }
+
+        String consumerGroup = extFields.get("consumerGroup");
+        if (MixAll.TOOLS_CONSUMER_GROUP.equals(consumerGroup)) {
+            return true;
+        }
+
+        String generalGroup = extFields.get("group");
+        return MixAll.CLIENT_INNER_PRODUCER_GROUP.equals(generalGroup);
+    }
+}

proxy/src/main/java/org/apache/rocketmq/proxy/service/ClusterServiceManager.java

@@ -35,6 +35,7 @@
 import org.apache.rocketmq.logging.org.slf4j.Logger;
 import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
 import org.apache.rocketmq.proxy.common.ProxyContext;
+import org.apache.rocketmq.proxy.common.SystemResourceAwareRpcHook;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
 import org.apache.rocketmq.proxy.config.ProxyConfig;
 import org.apache.rocketmq.proxy.service.admin.AdminService;
@@ -54,6 +55,9 @@
 import org.apache.rocketmq.proxy.service.transaction.TransactionService;
 import org.apache.rocketmq.remoting.RPCHook;
 import org.apache.rocketmq.remoting.RemotingClient;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.commons.lang3.StringUtils;
 
 public class ClusterServiceManager extends AbstractStartAndShutdown implements ServiceManager {
     private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
@@ -74,22 +78,31 @@ public class ClusterServiceManager extends AbstractStartAndShutdown implements S
     protected MQClientAPIFactory transactionClientAPIFactory;
     protected MQClientAPIFactory liteSubscriptionAPIFactory;
 
-    public ClusterServiceManager(RPCHook rpcHook) {
-        this(rpcHook, null);
-    }
-
     public ClusterServiceManager(RPCHook rpcHook, ObjectCreator<RemotingClient> remotingClientCreator) {
         ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
         NameserverAccessConfig nameserverAccessConfig = new NameserverAccessConfig(proxyConfig.getNamesrvAddr(),
             proxyConfig.getNamesrvDomain(), proxyConfig.getNamesrvDomainSubgroup());
         this.scheduledExecutorService = ThreadUtils.newScheduledThreadPool(3);
 
+        String proxyAccessKey = System.getProperty("rocketmq.proxy.accessKey", System.getenv("ROCKETMQ_PROXY_ACCESS_KEY"));
+        String proxySecretKey = System.getProperty("rocketmq.proxy.secretKey", System.getenv("ROCKETMQ_PROXY_SECRET_KEY"));
+
+        RPCHook systemHook = null;
+        if (StringUtils.isNotBlank(proxyAccessKey) && StringUtils.isNotBlank(proxySecretKey)) {
+            systemHook = new AclClientRPCHook(new SessionCredentials(proxyAccessKey, proxySecretKey));
+            log.info("SystemResourceAwareRpcHook initialized with provided Proxy Admin Credentials.");
+        } else {
+            log.warn("No Proxy Admin Credentials found (rocketmq.proxy.accessKey/secretKey). System requests will be anonymous.");
+        }
+
+        RPCHook systemAwareHook = new SystemResourceAwareRpcHook(rpcHook, systemHook);
+
         this.messagingClientAPIFactory = new MQClientAPIFactory(
             nameserverAccessConfig,
             "ClusterMQClient_",
             proxyConfig.getRocketmqMQClientNum(),
             new DoNothingClientRemotingProcessor(null),
-            rpcHook,
+            systemAwareHook,
             scheduledExecutorService,
             remotingClientCreator
         );
@@ -99,7 +112,7 @@ public ClusterServiceManager(RPCHook rpcHook, ObjectCreator<RemotingClient> remo
             "OperationClient_",
             1,
             new DoNothingClientRemotingProcessor(null),
-            rpcHook,
+            systemAwareHook,
             this.scheduledExecutorService,
             remotingClientCreator
         );
@@ -110,14 +123,14 @@ public ClusterServiceManager(RPCHook rpcHook, ObjectCreator<RemotingClient> remo
         this.adminService = new DefaultAdminService(this.operationClientAPIFactory);
 
         this.producerManager = new ProducerManager();
-        this.consumerManager = new ClusterConsumerManager(this.topicRouteService, this.adminService, this.operationClientAPIFactory, new ConsumerIdsChangeListenerImpl(), proxyConfig.getChannelExpiredTimeout(), rpcHook);
+        this.consumerManager = new ClusterConsumerManager(this.topicRouteService, this.adminService, this.operationClientAPIFactory, new ConsumerIdsChangeListenerImpl(), proxyConfig.getChannelExpiredTimeout(), systemAwareHook);
 
         this.transactionClientAPIFactory = new MQClientAPIFactory(
             nameserverAccessConfig,
             "ClusterTransaction_",
             1,
             new ProxyClientRemotingProcessor(producerManager, consumerManager),
-            rpcHook,
+            systemAwareHook,
             scheduledExecutorService,
             remotingClientCreator
         );
@@ -132,7 +145,7 @@ public ClusterServiceManager(RPCHook rpcHook, ObjectCreator<RemotingClient> remo
             "LiteSubscription_",
             1,
             new ProxyClientRemotingProcessor(producerManager, consumerManager),
-            rpcHook,
+            systemAwareHook,
             scheduledExecutorService);
         this.liteSubscriptionService = new LiteSubscriptionService(this.topicRouteService, this.liteSubscriptionAPIFactory);
 

proxy/src/main/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncer.java

@@ -48,6 +48,9 @@
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
+import static org.apache.rocketmq.proxy.common.InternalContextHolder.beginInternalScope;
+import static org.apache.rocketmq.proxy.common.InternalContextHolder.clear;
+
 public class HeartbeatSyncer extends AbstractSystemMessageSyncer {
 
     protected ThreadPoolExecutor threadPoolExecutor;
@@ -113,6 +116,8 @@ public void onConsumerRegister(String consumerGroup, ClientChannelInfo clientCha
         try {
             this.threadPoolExecutor.submit(() -> {
                 try {
+                    beginInternalScope();
+
                     RemoteChannel remoteChannel = RemoteChannel.create(clientChannelInfo.getChannel());
                     if (remoteChannel == null) {
                         return;
@@ -134,13 +139,13 @@ public void onConsumerRegister(String consumerGroup, ClientChannelInfo clientCha
                     log.debug("sync register heart beat. topic:{}, data:{}", this.getBroadcastTopicName(), data);
                     this.sendSystemMessage(data);
                 } catch (Throwable t) {
-                    log.error("heartbeat register broadcast failed. group:{}, clientChannelInfo:{}, consumeType:{}, messageModel:{}, consumeFromWhere:{}, subList:{}",
-                        consumerGroup, clientChannelInfo, consumeType, messageModel, consumeFromWhere, subList, t);
+                    log.error("heartbeat register broadcast failed...", t);
+                } finally {
+                    clear();
                 }
             });
         } catch (Throwable t) {
-            log.error("heartbeat submit register broadcast failed. group:{}, clientChannelInfo:{}, consumeType:{}, messageModel:{}, consumeFromWhere:{}, subList:{}",
-                consumerGroup, clientChannelInfo, consumeType, messageModel, consumeFromWhere, subList, t);
+            log.error("heartbeat submit register broadcast failed...", t);
         }
     }
 
@@ -151,6 +156,8 @@ public void onConsumerUnRegister(String consumerGroup, ClientChannelInfo clientC
         try {
             this.threadPoolExecutor.submit(() -> {
                 try {
+                    beginInternalScope();
+
                     RemoteChannel remoteChannel = RemoteChannel.create(clientChannelInfo.getChannel());
                     if (remoteChannel == null) {
                         return;
@@ -171,13 +178,13 @@ public void onConsumerUnRegister(String consumerGroup, ClientChannelInfo clientC
                     log.debug("sync unregister heart beat. topic:{}, data:{}", this.getBroadcastTopicName(), data);
                     this.sendSystemMessage(data);
                 } catch (Throwable t) {
-                    log.error("heartbeat unregister broadcast failed. group:{}, clientChannelInfo:{}, consumeType:{}",
-                        consumerGroup, clientChannelInfo, t);
+                    log.error("heartbeat unregister broadcast failed...", t);
+                } finally {
+                    clear();
                 }
             });
         } catch (Throwable t) {
-            log.error("heartbeat submit unregister broadcast failed. group:{}, clientChannelInfo:{}, consumeType:{}",
-                consumerGroup, clientChannelInfo, t);
+            log.error("heartbeat submit unregister broadcast failed...", t);
         }
     }