[ISSUE #10795] Validate proxy check client config

proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java

@@ -120,6 +120,10 @@ public class ProxyConfig implements ConfigFile {
      * if true, proxy will check message body size and reject msg if it's body is empty
      */
     private boolean enableMessageBodyEmptyCheck = true;
+    /**
+     * if true, proxy allows SQL/property filter expressions in client config checks.
+     */
+    private boolean enablePropertyFilter = false;
     /**
      * max user property size, 0 or negative number means no limit for proxy
      */
@@ -1579,6 +1583,14 @@ public void setEnableMessageBodyEmptyCheck(boolean enableMessageBodyEmptyCheck)
         this.enableMessageBodyEmptyCheck = enableMessageBodyEmptyCheck;
     }
 
+    public boolean isEnablePropertyFilter() {
+        return enablePropertyFilter;
+    }
+
+    public void setEnablePropertyFilter(boolean enablePropertyFilter) {
+        this.enablePropertyFilter = enablePropertyFilter;
+    }
+
     public int getMaxLiteTopicSize() {
         return maxLiteTopicSize;
     }

proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java

@@ -24,7 +24,10 @@
 import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener;
 import org.apache.rocketmq.broker.client.ProducerChangeListener;
 import org.apache.rocketmq.broker.client.ProducerGroupEvent;
+import org.apache.rocketmq.common.filter.ExpressionType;
+import org.apache.rocketmq.filter.FilterFactory;
 import org.apache.rocketmq.proxy.common.ProxyContext;
+import org.apache.rocketmq.proxy.config.ConfigurationManager;
 import org.apache.rocketmq.proxy.processor.MessagingProcessor;
 import org.apache.rocketmq.proxy.remoting.channel.RemotingChannel;
 import org.apache.rocketmq.proxy.remoting.channel.RemotingChannelManager;
@@ -35,11 +38,13 @@
 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.body.CheckClientRequestBody;
 import org.apache.rocketmq.remoting.protocol.header.UnregisterClientRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.UnregisterClientResponseHeader;
 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.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
 
 import java.util.Set;
 
@@ -158,8 +163,50 @@ protected RemotingCommand unregisterClient(ChannelHandlerContext ctx, RemotingCo
     protected RemotingCommand checkClientConfig(ChannelHandlerContext ctx, RemotingCommand request,
         ProxyContext context) {
         final RemotingCommand response = RemotingCommand.createResponseCommand(null);
+        CheckClientRequestBody requestBody;
+        try {
+            if (request.getBody() == null) {
+                return invalidCheckClientConfigResponse(response, "Request body is required");
+            }
+            requestBody = CheckClientRequestBody.decode(request.getBody(), CheckClientRequestBody.class);
+        } catch (Exception e) {
+            log.warn("Failed to decode check client config request", e);
+            return invalidCheckClientConfigResponse(response, "Failed to decode request body");
+        }
+        if (requestBody != null && requestBody.getSubscriptionData() != null) {
+            SubscriptionData subscriptionData = requestBody.getSubscriptionData();
+            if (ExpressionType.isTagType(subscriptionData.getExpressionType())) {
+                response.setCode(ResponseCode.SUCCESS);
+                response.setRemark(null);
+                return response;
+            }
+
+            if (!ConfigurationManager.getProxyConfig().isEnablePropertyFilter()) {
+                response.setCode(ResponseCode.SYSTEM_ERROR);
+                response.setRemark("Property filter is disabled; enablePropertyFilter must be true to use "
+                    + subscriptionData.getExpressionType());
+                return response;
+            }
+
+            try {
+                FilterFactory.INSTANCE.get(subscriptionData.getExpressionType()).compile(subscriptionData.getSubString());
+            } catch (Exception e) {
+                log.warn("Client {}@{} failed to compile filter expression: {}",
+                    requestBody.getClientId(), requestBody.getGroup(), requestBody.getSubscriptionData(), e);
+                response.setCode(ResponseCode.SUBSCRIPTION_PARSE_FAILED);
+                response.setRemark(e.getMessage());
+                return response;
+            }
+        }
+
         response.setCode(ResponseCode.SUCCESS);
-        response.setRemark("");
+        response.setRemark(null);
+        return response;
+    }
+
+    private RemotingCommand invalidCheckClientConfigResponse(RemotingCommand response, String remark) {
+        response.setCode(ResponseCode.SUBSCRIPTION_PARSE_FAILED);
+        response.setRemark(remark);
         return response;
     }
 

proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivityTest.java

@@ -0,0 +1,122 @@
+/*
+ * 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.activity;
+
+import org.apache.rocketmq.common.filter.ExpressionType;
+import org.apache.rocketmq.proxy.common.ProxyContext;
+import org.apache.rocketmq.proxy.config.ConfigurationManager;
+import org.apache.rocketmq.proxy.config.InitConfigTest;
+import org.apache.rocketmq.proxy.processor.MessagingProcessor;
+import org.apache.rocketmq.proxy.remoting.channel.RemotingChannelManager;
+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.body.CheckClientRequestBody;
+import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ClientManagerActivityTest extends InitConfigTest {
+
+    private ClientManagerActivity clientManagerActivity;
+    private boolean originalEnablePropertyFilter;
+    @Mock
+    private MessagingProcessor messagingProcessor;
+    @Mock
+    private RemotingChannelManager remotingChannelManager;
+
+    @Before
+    public void setUp() {
+        this.clientManagerActivity = new ClientManagerActivity(null, messagingProcessor, remotingChannelManager);
+        this.originalEnablePropertyFilter = ConfigurationManager.getProxyConfig().isEnablePropertyFilter();
+    }
+
+    @After
+    public void tearDown() {
+        ConfigurationManager.getProxyConfig().setEnablePropertyFilter(originalEnablePropertyFilter);
+    }
+
+    @Test
+    public void testCheckClientConfigWithTagExpression() {
+        RemotingCommand response = clientManagerActivity.checkClientConfig(null,
+            createRequest(ExpressionType.TAG, "tagA || tagB"), ProxyContext.create());
+
+        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+    }
+
+    @Test
+    public void testCheckClientConfigRejectsPropertyFilterWhenDisabled() {
+        RemotingCommand response = clientManagerActivity.checkClientConfig(null,
+            createRequest(ExpressionType.SQL92, "a is not null"), ProxyContext.create());
+
+        assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);
+        assertThat(response.getRemark()).contains("enablePropertyFilter", ExpressionType.SQL92);
+    }
+
+    @Test
+    public void testCheckClientConfigRejectsInvalidPropertyFilterExpression() {
+        ConfigurationManager.getProxyConfig().setEnablePropertyFilter(true);
+
+        RemotingCommand response = clientManagerActivity.checkClientConfig(null,
+            createRequest(ExpressionType.SQL92, "a = "), ProxyContext.create());
+
+        assertThat(response.getCode()).isEqualTo(ResponseCode.SUBSCRIPTION_PARSE_FAILED);
+    }
+
+    @Test
+    public void testCheckClientConfigAcceptsValidPropertyFilterExpression() {
+        ConfigurationManager.getProxyConfig().setEnablePropertyFilter(true);
+
+        RemotingCommand response = clientManagerActivity.checkClientConfig(null,
+            createRequest(ExpressionType.SQL92, "a is not null"), ProxyContext.create());
+
+        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+    }
+
+    @Test
+    public void testCheckClientConfigRejectsRequestWithoutBody() {
+        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CHECK_CLIENT_CONFIG, null);
+
+        RemotingCommand response = clientManagerActivity.checkClientConfig(null, request, ProxyContext.create());
+
+        assertThat(response.getCode()).isEqualTo(ResponseCode.SUBSCRIPTION_PARSE_FAILED);
+        assertThat(response.getRemark()).contains("body");
+    }
+
+    private RemotingCommand createRequest(String expressionType, String expression) {
+        SubscriptionData subscriptionData = new SubscriptionData();
+        subscriptionData.setTopic("topic");
+        subscriptionData.setExpressionType(expressionType);
+        subscriptionData.setSubString(expression);
+
+        CheckClientRequestBody requestBody = new CheckClientRequestBody();
+        requestBody.setClientId("clientId");
+        requestBody.setGroup("group");
+        requestBody.setSubscriptionData(subscriptionData);
+
+        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CHECK_CLIENT_CONFIG, null);
+        request.setBody(requestBody.encode());
+        return request;
+    }
+}