From 1fd7fcc7acdbd9611e9bb4afca8ccbe79b7a77a4 Mon Sep 17 00:00:00 2001 From: liuhy Date: Wed, 29 Jul 2026 01:06:34 -0700 Subject: [PATCH 1/3] [ISSUE #10690] Tolerate malformed SimpleChannel address ports --- .../proxy/service/channel/SimpleChannel.java | 9 +++- .../service/channel/SimpleChannelTest.java | 49 +++++++++++++++++++ 2 files changed, 56 insertions(+), 2 deletions(-) create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/service/channel/SimpleChannelTest.java diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java index 65c1fd40665..ecf7ca750d6 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java @@ -90,7 +90,12 @@ private static SocketAddress parseSocketAddress(String address) { String[] segments = address.split(":"); if (2 == segments.length) { - return new InetSocketAddress(segments[0], Integer.parseInt(segments[1])); + try { + return new InetSocketAddress(segments[0], Integer.parseInt(segments[1])); + } catch (NumberFormatException e) { + log.warn("parse socket address failed. address:{}", address); + return null; + } } return null; @@ -207,4 +212,4 @@ public String getLocalAddress() { public ChannelHandlerContext getChannelHandlerContext() { return channelHandlerContext; } -} \ No newline at end of file +} diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/channel/SimpleChannelTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/channel/SimpleChannelTest.java new file mode 100644 index 00000000000..cc1541d4ff7 --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/channel/SimpleChannelTest.java @@ -0,0 +1,49 @@ +/* + * 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.service.channel; + +import java.net.InetSocketAddress; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +public class SimpleChannelTest { + + @Test + public void testParseValidSocketAddress() { + SimpleChannel channel = new SimpleChannel("127.0.0.1:10911", "127.0.0.1:8080"); + + InetSocketAddress remoteAddress = (InetSocketAddress) channel.remoteAddress(); + assertEquals("127.0.0.1", remoteAddress.getHostString()); + assertEquals(10911, remoteAddress.getPort()); + + InetSocketAddress localAddress = (InetSocketAddress) channel.localAddress(); + assertEquals("127.0.0.1", localAddress.getHostString()); + assertEquals(8080, localAddress.getPort()); + } + + @Test + public void testParseInvalidSocketAddressReturnsNull() { + assertNull(new SimpleChannel(null, null).remoteAddress()); + assertNull(new SimpleChannel("", "").remoteAddress()); + assertNull(new SimpleChannel("127.0.0.1", "127.0.0.1").remoteAddress()); + assertNull(new SimpleChannel("127.0.0.1:not-a-port", "127.0.0.1:not-a-port").remoteAddress()); + assertNull(new SimpleChannel("127.0.0.1:not-a-port", "127.0.0.1:not-a-port").localAddress()); + } +} From 8ca55cbf49ea21e63382ba789ec57c4b87f7c839 Mon Sep 17 00:00:00 2001 From: liuhy Date: Thu, 30 Jul 2026 18:26:48 -0700 Subject: [PATCH 2/3] [ISSUE #10690] Validate SimpleChannel port range --- .../proxy/service/channel/SimpleChannel.java | 9 ++++++- .../service/channel/SimpleChannelTest.java | 26 +++++++++++++++++-- .../relay/LocalProxyRelayServiceTest.java | 10 +++++-- 3 files changed, 40 insertions(+), 5 deletions(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java index ecf7ca750d6..d834f9691e4 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java @@ -43,6 +43,8 @@ */ public class SimpleChannel extends AbstractChannel { protected static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); + private static final int MIN_PORT = 0; + private static final int MAX_PORT = 65535; protected final String remoteAddress; protected final String localAddress; @@ -91,7 +93,12 @@ private static SocketAddress parseSocketAddress(String address) { String[] segments = address.split(":"); if (2 == segments.length) { try { - return new InetSocketAddress(segments[0], Integer.parseInt(segments[1])); + int port = Integer.parseInt(segments[1]); + if (port < MIN_PORT || port > MAX_PORT) { + log.warn("socket address port out of range. address:{}", address); + return null; + } + return new InetSocketAddress(segments[0], port); } catch (NumberFormatException e) { log.warn("parse socket address failed. address:{}", address); return null; diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/channel/SimpleChannelTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/channel/SimpleChannelTest.java index cc1541d4ff7..4edf855c9b9 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/service/channel/SimpleChannelTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/channel/SimpleChannelTest.java @@ -18,6 +18,7 @@ package org.apache.rocketmq.proxy.service.channel; import java.net.InetSocketAddress; +import org.apache.rocketmq.proxy.common.ProxyContext; import org.junit.Test; import static org.junit.Assert.assertEquals; @@ -43,7 +44,28 @@ public void testParseInvalidSocketAddressReturnsNull() { assertNull(new SimpleChannel(null, null).remoteAddress()); assertNull(new SimpleChannel("", "").remoteAddress()); assertNull(new SimpleChannel("127.0.0.1", "127.0.0.1").remoteAddress()); - assertNull(new SimpleChannel("127.0.0.1:not-a-port", "127.0.0.1:not-a-port").remoteAddress()); - assertNull(new SimpleChannel("127.0.0.1:not-a-port", "127.0.0.1:not-a-port").localAddress()); + assertInvalidRemoteAndLocalAddress("127.0.0.1:not-a-port"); + assertInvalidRemoteAndLocalAddress("127.0.0.1:-1"); + assertInvalidRemoteAndLocalAddress("127.0.0.1:65536"); + assertInvalidRemoteAndLocalAddress("127.0.0.1:2147483648"); + assertInvalidRemoteAndLocalAddress("127.0.0.1:"); + assertInvalidRemoteAndLocalAddress("127.0.0.1: "); + } + + @Test + public void testChannelManagerConsumesInvalidSocketAddress() { + ChannelManager channelManager = new ChannelManager(); + SimpleChannel channel = channelManager.createChannel( + ProxyContext.create() + .setRemoteAddress("127.0.0.1:65536") + .setLocalAddress("127.0.0.1:-1")); + + assertNull(channel.remoteAddress()); + assertNull(channel.localAddress()); + } + + private void assertInvalidRemoteAndLocalAddress(String address) { + assertNull(new SimpleChannel(address, address).remoteAddress()); + assertNull(new SimpleChannel(address, address).localAddress()); } } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/relay/LocalProxyRelayServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/relay/LocalProxyRelayServiceTest.java index 4ec797d1aaf..5e3943c6a1b 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/service/relay/LocalProxyRelayServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/relay/LocalProxyRelayServiceTest.java @@ -68,15 +68,21 @@ public void testProcessGetConsumerRunningInfo() { GetConsumerRunningInfoRequestHeader requestHeader = new GetConsumerRunningInfoRequestHeader(); requestHeader.setJstackEnable(true); ArgumentCaptor argumentCaptor = ArgumentCaptor.forClass(RemotingCommand.class); + ArgumentCaptor contextCaptor = + ArgumentCaptor.forClass(SimpleChannelHandlerContext.class); CompletableFuture> future = - localProxyRelayService.processGetConsumerRunningInfo(ProxyContext.create(), remotingCommand, requestHeader); + localProxyRelayService.processGetConsumerRunningInfo(ProxyContext.create() + .setRemoteAddress("127.0.0.1:65536") + .setLocalAddress("127.0.0.1:-1"), remotingCommand, requestHeader); future.complete(new ProxyRelayResult<>(ResponseCode.SUCCESS, remark, runningInfo)); Mockito.verify(nettyRemotingServerMock, Mockito.times(1)) - .processResponseCommand(Mockito.any(SimpleChannelHandlerContext.class), argumentCaptor.capture()); + .processResponseCommand(contextCaptor.capture(), argumentCaptor.capture()); RemotingCommand remotingCommand1 = argumentCaptor.getValue(); assertThat(remotingCommand1.getCode()).isEqualTo(ResponseCode.SUCCESS); assertThat(remotingCommand1.getRemark()).isEqualTo(remark); assertThat(remotingCommand1.getBody()).isEqualTo(runningInfo.encode()); + assertThat(contextCaptor.getValue().channel().remoteAddress()).isNull(); + assertThat(contextCaptor.getValue().channel().localAddress()).isNull(); } @Test From 34196e4d384319786e96ee07fc142f6039718dc0 Mon Sep 17 00:00:00 2001 From: liuhy Date: Fri, 31 Jul 2026 08:01:03 -0700 Subject: [PATCH 3/3] [ISSUE #10690] Cover malformed channel address consumers --- .../remoting/channel/RemotingChannel.java | 9 ++-- .../pipeline/ContextInitPipeline.java | 14 +++++- .../proxy/service/relay/ProxyChannel.java | 32 ++++++++++-- .../remoting/channel/RemotingChannelTest.java | 18 ++++++- .../pipeline/ContextInitPipelineTest.java | 50 +++++++++++++++++++ 5 files changed, 112 insertions(+), 11 deletions(-) create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/remoting/pipeline/ContextInitPipelineTest.java diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/channel/RemotingChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/channel/RemotingChannel.java index 2bdb6eb9bb6..b97d7815bbe 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/channel/RemotingChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/channel/RemotingChannel.java @@ -33,7 +33,6 @@ import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.common.utils.ExceptionUtils; import org.apache.rocketmq.common.utils.FutureUtils; -import org.apache.rocketmq.common.utils.NetworkUtil; import org.apache.rocketmq.logging.org.slf4j.Logger; import org.apache.rocketmq.logging.org.slf4j.LoggerFactory; import org.apache.rocketmq.proxy.common.channel.ChannelHelper; @@ -73,12 +72,12 @@ public RemotingChannel(RemotingProxyOutClient remotingProxyOutClient, ProxyRelay Channel parent, String clientId, Set subscriptionData) { super(proxyRelayService, parent, parent.id(), - NetworkUtil.socketAddress2String(parent.remoteAddress()), - NetworkUtil.socketAddress2String(parent.localAddress())); + socketAddress2String(parent.remoteAddress()), + socketAddress2String(parent.localAddress())); this.remotingProxyOutClient = remotingProxyOutClient; this.clientId = clientId; - this.remoteAddress = NetworkUtil.socketAddress2String(parent.remoteAddress()); - this.localAddress = NetworkUtil.socketAddress2String(parent.localAddress()); + this.remoteAddress = socketAddress2String(parent.remoteAddress()); + this.localAddress = socketAddress2String(parent.localAddress()); this.subscriptionData = subscriptionData; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/pipeline/ContextInitPipeline.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/pipeline/ContextInitPipeline.java index 0a30f620aed..ba5a8decdea 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/pipeline/ContextInitPipeline.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/pipeline/ContextInitPipeline.java @@ -18,6 +18,7 @@ import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; +import java.net.SocketAddress; import org.apache.rocketmq.common.MQVersion; import org.apache.rocketmq.common.utils.NetworkUtil; import org.apache.rocketmq.proxy.common.ProxyContext; @@ -38,7 +39,7 @@ public void execute(ChannelHandlerContext ctx, RemotingCommand request, ProxyCon context.setAction(RemotingHelper.getRequestCodeDesc(request.getCode())) .setProtocolType(ChannelProtocolType.REMOTING.getName()) .setChannel(channel) - .setLocalAddress(NetworkUtil.socketAddress2String(ctx.channel().localAddress())) + .setLocalAddress(socketAddress2String(ctx.channel().localAddress())) .setRemoteAddress(RemotingHelper.parseChannelRemoteAddr(ctx.channel())); if (languageCode != null) { context.setLanguage(languageCode.name()); @@ -50,4 +51,15 @@ public void execute(ChannelHandlerContext ctx, RemotingCommand request, ProxyCon context.setClientVersion(MQVersion.getVersionDesc(version)); } } + + private static String socketAddress2String(SocketAddress address) { + if (address == null) { + return null; + } + try { + return NetworkUtil.socketAddress2String(address); + } catch (RuntimeException ignored) { + return null; + } + } } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java index 72fdfd0259a..b3a3ab5e25d 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java @@ -59,16 +59,40 @@ protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, Stri String localAddress) { super(parent, remoteAddress, localAddress); this.proxyRelayService = proxyRelayService; - this.remoteSocketAddress = NetworkUtil.string2SocketAddress(remoteAddress); - this.localSocketAddress = NetworkUtil.string2SocketAddress(localAddress); + this.remoteSocketAddress = parseSocketAddress(remoteAddress); + this.localSocketAddress = parseSocketAddress(localAddress); } protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, ChannelId id, String remoteAddress, String localAddress) { super(parent, id, remoteAddress, localAddress); this.proxyRelayService = proxyRelayService; - this.remoteSocketAddress = NetworkUtil.string2SocketAddress(remoteAddress); - this.localSocketAddress = NetworkUtil.string2SocketAddress(localAddress); + this.remoteSocketAddress = parseSocketAddress(remoteAddress); + this.localSocketAddress = parseSocketAddress(localAddress); + } + + protected static String socketAddress2String(SocketAddress address) { + if (address == null) { + return null; + } + try { + return NetworkUtil.socketAddress2String(address); + } catch (RuntimeException e) { + log.warn("convert socket address failed. address:{}", address); + return null; + } + } + + private static SocketAddress parseSocketAddress(String address) { + if (address == null || address.isEmpty()) { + return null; + } + try { + return NetworkUtil.string2SocketAddress(address); + } catch (RuntimeException e) { + log.warn("parse proxy channel socket address failed. address:{}", address); + return null; + } } @Override diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/channel/RemotingChannelTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/channel/RemotingChannelTest.java index d947fa5d533..2ade4857fb2 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/channel/RemotingChannelTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/channel/RemotingChannelTest.java @@ -77,4 +77,20 @@ public void testChannelExtendAttributeParse() { assertEquals(subscriptionData, RemotingChannel.parseChannelExtendAttribute(this.remotingChannel)); assertNull(RemotingChannel.parseChannelExtendAttribute(mock(GrpcClientChannel.class))); } -} \ No newline at end of file + + @Test + public void testNullParentSocketAddressIsConsumedSafely() { + when(parent.remoteAddress()).thenReturn(null); + when(parent.localAddress()).thenReturn(null); + + RemotingChannel channel = new RemotingChannel(remotingProxyOutClient, proxyRelayService, + parent, clientId, subscriptionData); + + assertNull(channel.remoteAddress()); + assertNull(channel.localAddress()); + assertNull(channel.getRemoteAddress()); + assertNull(channel.getLocalAddress()); + assertNull(channel.toRemoteChannel().getRemoteAddress()); + assertNull(channel.toRemoteChannel().getLocalAddress()); + } +} diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/pipeline/ContextInitPipelineTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/pipeline/ContextInitPipelineTest.java new file mode 100644 index 00000000000..68b6c956ff5 --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/pipeline/ContextInitPipelineTest.java @@ -0,0 +1,50 @@ +/* + * 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.pipeline; + +import io.netty.channel.ChannelHandlerContext; +import org.apache.rocketmq.proxy.common.ProxyContext; +import org.apache.rocketmq.proxy.processor.channel.ChannelProtocolType; +import org.apache.rocketmq.proxy.service.channel.SimpleChannel; +import org.apache.rocketmq.remoting.protocol.RemotingCommand; +import org.apache.rocketmq.remoting.protocol.RequestCode; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class ContextInitPipelineTest { + + @Test + public void testMalformedSimpleChannelAddressIsConsumedSafely() throws Exception { + SimpleChannel channel = new SimpleChannel("127.0.0.1:65536", "127.0.0.1:-1"); + ChannelHandlerContext ctx = mock(ChannelHandlerContext.class); + when(ctx.channel()).thenReturn(channel); + RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.HEART_BEAT, null); + ProxyContext context = ProxyContext.create(); + + new ContextInitPipeline().execute(ctx, request, context); + + assertSame(channel, context.getChannel()); + assertEquals(ChannelProtocolType.REMOTING.getName(), context.getProtocolType()); + assertEquals("", context.getRemoteAddress()); + assertNull(context.getLocalAddress()); + } +}