[ISSUE #10690] Tolerate malformed SimpleChannel address ports

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> 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;
     }
 

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;
+        }
+    }
 }

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;
@@ -90,7 +92,17 @@ private static SocketAddress parseSocketAddress(String address) {
 
         String[] segments = address.split(":");
         if (2 == segments.length) {
-            return new InetSocketAddress(segments[0], Integer.parseInt(segments[1]));
+            try {
+                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;
+            }
         }
 
         return null;
@@ -207,4 +219,4 @@ public String getLocalAddress() {
     public ChannelHandlerContext getChannelHandlerContext() {
         return channelHandlerContext;
     }
-}
+}

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

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)));
     }
-}
+
+    @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());
+    }
+}

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());
+    }
+}

proxy/src/test/java/org/apache/rocketmq/proxy/service/channel/SimpleChannelTest.java

@@ -0,0 +1,71 @@
+/*
+ * 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.apache.rocketmq.proxy.common.ProxyContext;
+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());
+        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());
+    }
+}

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<RemotingCommand> argumentCaptor = ArgumentCaptor.forClass(RemotingCommand.class);
+        ArgumentCaptor<SimpleChannelHandlerContext> contextCaptor =
+            ArgumentCaptor.forClass(SimpleChannelHandlerContext.class);
         CompletableFuture<ProxyRelayResult<ConsumerRunningInfo>> 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