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..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 @@ -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; } 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..0ecb1602591 --- /dev/null +++ b/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; + } +}