From 9aae7bc253f57c766f712eadf7a2267d42aa10b4 Mon Sep 17 00:00:00 2001 From: liuhy Date: Mon, 3 Aug 2026 07:20:19 -0700 Subject: [PATCH 1/2] [ISSUE #10795] Validate proxy check client config --- .../rocketmq/proxy/config/ProxyConfig.java | 12 +++ .../activity/ClientManagerActivity.java | 35 ++++++- .../activity/ClientManagerActivityTest.java | 94 +++++++++++++++++++ 3 files changed, 140 insertions(+), 1 deletion(-) create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivityTest.java diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java b/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java index a7896c11e07..432b67f59da 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java +++ b/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; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java index 05d8e5fbe13..bfc5358f839 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java +++ b/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,36 @@ protected RemotingCommand unregisterClient(ChannelHandlerContext ctx, RemotingCo protected RemotingCommand checkClientConfig(ChannelHandlerContext ctx, RemotingCommand request, ProxyContext context) { final RemotingCommand response = RemotingCommand.createResponseCommand(null); + CheckClientRequestBody requestBody = CheckClientRequestBody.decode(request.getBody(), + CheckClientRequestBody.class); + 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("The proxy does not support consumer to filter message by " + + subscriptionData.getExpressionType()); + return response; + } + + try { + FilterFactory.INSTANCE.get(subscriptionData.getExpressionType()).compile(subscriptionData.getSubString()); + } catch (Exception e) { + log.warn("Client {}@{} filter message, but failed to compile expression! sub={}, error={}", + requestBody.getClientId(), requestBody.getGroup(), requestBody.getSubscriptionData(), e.getMessage()); + response.setCode(ResponseCode.SUBSCRIPTION_PARSE_FAILED); + response.setRemark(e.getMessage()); + return response; + } + } + response.setCode(ResponseCode.SUCCESS); - response.setRemark(""); + response.setRemark(null); return response; } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivityTest.java new file mode 100644 index 00000000000..09e17d714fe --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivityTest.java @@ -0,0 +1,94 @@ +/* + * 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.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; + @Mock + private MessagingProcessor messagingProcessor; + @Mock + private RemotingChannelManager remotingChannelManager; + + @Before + public void setUp() { + this.clientManagerActivity = new ClientManagerActivity(null, messagingProcessor, remotingChannelManager); + } + + @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(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); + } + + 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; + } +} From f5f6f3b8f0ab48cc156c7f6db671e1dc4670c5d8 Mon Sep 17 00:00:00 2001 From: liuhy Date: Tue, 4 Aug 2026 03:35:34 -0700 Subject: [PATCH 2/2] fix(proxy): harden check client config validation --- .../activity/ClientManagerActivity.java | 24 +++++++++++---- .../activity/ClientManagerActivityTest.java | 30 ++++++++++++++++++- 2 files changed, 48 insertions(+), 6 deletions(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java index bfc5358f839..db510ce3e2f 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java @@ -163,8 +163,16 @@ protected RemotingCommand unregisterClient(ChannelHandlerContext ctx, RemotingCo protected RemotingCommand checkClientConfig(ChannelHandlerContext ctx, RemotingCommand request, ProxyContext context) { final RemotingCommand response = RemotingCommand.createResponseCommand(null); - CheckClientRequestBody requestBody = CheckClientRequestBody.decode(request.getBody(), - CheckClientRequestBody.class); + 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())) { @@ -175,7 +183,7 @@ protected RemotingCommand checkClientConfig(ChannelHandlerContext ctx, RemotingC if (!ConfigurationManager.getProxyConfig().isEnablePropertyFilter()) { response.setCode(ResponseCode.SYSTEM_ERROR); - response.setRemark("The proxy does not support consumer to filter message by " + response.setRemark("Property filter is disabled; enablePropertyFilter must be true to use " + subscriptionData.getExpressionType()); return response; } @@ -183,8 +191,8 @@ protected RemotingCommand checkClientConfig(ChannelHandlerContext ctx, RemotingC try { FilterFactory.INSTANCE.get(subscriptionData.getExpressionType()).compile(subscriptionData.getSubString()); } catch (Exception e) { - log.warn("Client {}@{} filter message, but failed to compile expression! sub={}, error={}", - requestBody.getClientId(), requestBody.getGroup(), requestBody.getSubscriptionData(), e.getMessage()); + 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; @@ -196,6 +204,12 @@ protected RemotingCommand checkClientConfig(ChannelHandlerContext ctx, RemotingC return response; } + private RemotingCommand invalidCheckClientConfigResponse(RemotingCommand response, String remark) { + response.setCode(ResponseCode.SUBSCRIPTION_PARSE_FAILED); + response.setRemark(remark); + return response; + } + public void doChannelCloseEvent(String remoteAddr, Channel channel) { Set remotingChannelSet = this.remotingChannelManager.removeChannel(channel); for (RemotingChannel remotingChannel : remotingChannelSet) { diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivityTest.java index 09e17d714fe..0ecb1602591 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivityTest.java @@ -27,6 +27,7 @@ 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; @@ -39,6 +40,7 @@ public class ClientManagerActivityTest extends InitConfigTest { private ClientManagerActivity clientManagerActivity; + private boolean originalEnablePropertyFilter; @Mock private MessagingProcessor messagingProcessor; @Mock @@ -47,6 +49,12 @@ public class ClientManagerActivityTest extends InitConfigTest { @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 @@ -63,7 +71,7 @@ public void testCheckClientConfigRejectsPropertyFilterWhenDisabled() { createRequest(ExpressionType.SQL92, "a is not null"), ProxyContext.create()); assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR); - assertThat(response.getRemark()).contains(ExpressionType.SQL92); + assertThat(response.getRemark()).contains("enablePropertyFilter", ExpressionType.SQL92); } @Test @@ -76,6 +84,26 @@ public void testCheckClientConfigRejectsInvalidPropertyFilterExpression() { 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");