fix: reject oversized telnet payload before header copy

dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/codec/ExchangeCodec.java

@@ -87,6 +87,13 @@ public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOE
     @Override
     public Object decode(Channel channel, ChannelBuffer buffer) throws IOException {
         int readable = buffer.readableBytes();
+        if (readable > 0
+                && isServerSide(channel)
+                && (readable < 2
+                        || buffer.getByte(buffer.readerIndex()) != MAGIC_HIGH
+                        || buffer.getByte(buffer.readerIndex() + 1) != MAGIC_LOW)) {
+            checkPayload(channel, readable);
+        }
         byte[] header = new byte[Math.min(readable, HEADER_LENGTH)];
         buffer.readBytes(header);
         return decode(channel, buffer, readable, header);

dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/codec/ExchangeCodecTest.java

@@ -32,6 +32,7 @@
 import org.apache.dubbo.remoting.exchange.codec.ExchangeCodec;
 import org.apache.dubbo.remoting.exchange.support.DefaultFuture;
 import org.apache.dubbo.remoting.telnet.codec.TelnetCodec;
+import org.apache.dubbo.remoting.transport.ExceedPayloadLimitException;
 import org.apache.dubbo.rpc.model.FrameworkModel;
 
 import java.io.ByteArrayOutputStream;
@@ -41,6 +42,7 @@
 import java.util.Map;
 
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Assumptions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
@@ -213,6 +215,19 @@ void test_Decode_Check_Payload() throws IOException {
         }
     }
 
+    @Test
+    void testDecodeRejectsCompleteNonMagicInputBeforeCopy() {
+        Assumptions.assumeTrue(codec instanceof ExchangeCodec);
+        int payloadLimit = 16;
+        byte[] oversizedInput = new byte[payloadLimit + 1];
+        ChannelBuffer buffer = ChannelBuffers.wrappedBuffer(oversizedInput);
+        Channel channel = getServerSideChannel(url.addParameter(Constants.PAYLOAD_KEY, payloadLimit));
+
+        Assertions.assertThrows(ExceedPayloadLimitException.class, () -> codec.decode(channel, buffer));
+        Assertions.assertEquals(
+                0, buffer.readerIndex(), "payload validation should reject non-magic input before reading it");
+    }
+
     @Test
     void test_Decode_Header_Need_Readmore() throws IOException {
         byte[] header = new byte[] {MAGIC_HIGH, MAGIC_LOW, 0, 0, 0, 0, 0, 0, 0, 0, 0};