Add maxMessageSize config (#15781)

Add maxMessageSize config
---------

Co-authored-by: earthchen <[email protected]>

Author: RainYuY

dubbo-common/src/main/java/org/apache/dubbo/config/nested/TripleConfig.java

@@ -41,6 +41,9 @@ public class TripleConfig implements Serializable {
     public static final int DEFAULT_CONNECTION_INITIAL_WINDOW_SIZE_KEY = 65_536;
     public static final int DEFAULT_MAX_FRAME_SIZE = 8_388_608;
     public static final int DEFAULT_MAX_HEADER_LIST_SIZE = 32_768;
+    public static final int DEFAULT_MAX_MESSAGE_SIZE = 50 * 1024 * 1024;
+
+    public static final String H2_SETTINGS_MAX_MESSAGE_SIZE_KEY = "dubbo.protocol.triple.max-message-size";
 
     /**
      * Whether enable verbose mode.
@@ -143,6 +146,11 @@ public class TripleConfig implements Serializable {
      */
     private Integer maxHeaderListSize;
 
+    /**
+     * Maximum message size.
+     */
+    private Integer maxMessageSize;
+
     @Nested
     private RestConfig rest;
 
@@ -334,6 +342,19 @@ public void setMaxHeaderListSize(Integer maxHeaderListSize) {
         this.maxHeaderListSize = maxHeaderListSize;
     }
 
+    public Integer getMaxMessageSize() {
+        return maxMessageSize;
+    }
+
+    @Parameter(excluded = true, key = H2_SETTINGS_MAX_MESSAGE_SIZE_KEY)
+    public int getMaxMessageSizeOrDefault() {
+        return maxMessageSize == null ? DEFAULT_MAX_MESSAGE_SIZE : maxMessageSize;
+    }
+
+    public void setMaxMessageSize(Integer maxMessageSize) {
+        this.maxMessageSize = maxMessageSize;
+    }
+
     public RestConfig getRest() {
         return rest;
     }

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/LengthFieldStreamingDecoder.java

@@ -16,8 +16,13 @@
  */
 package org.apache.dubbo.remoting.http12.message;
 
+import org.apache.dubbo.common.config.Configuration;
+import org.apache.dubbo.common.config.ConfigurationUtils;
 import org.apache.dubbo.remoting.http12.CompositeInputStream;
 import org.apache.dubbo.remoting.http12.exception.DecodeException;
+import org.apache.dubbo.rpc.Constants;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.model.ApplicationModel;
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
@@ -43,6 +48,8 @@ public class LengthFieldStreamingDecoder implements StreamingDecoder {
 
     private final int lengthFieldLength;
 
+    private final int maxMessageSize;
+
     private int requiredLength;
 
     public LengthFieldStreamingDecoder() {
@@ -57,6 +64,8 @@ public LengthFieldStreamingDecoder(int lengthFieldOffset, int lengthFieldLength)
         this.lengthFieldOffset = lengthFieldOffset;
         this.lengthFieldLength = lengthFieldLength;
         this.requiredLength = lengthFieldOffset + lengthFieldLength;
+        Configuration conf = ConfigurationUtils.getEnvConfiguration(ApplicationModel.defaultModel());
+        this.maxMessageSize = conf.getInt(Constants.H2_SETTINGS_MAX_MESSAGE_SIZE, 50 * 1024 * 1024);
     }
 
     @Override
@@ -150,6 +159,14 @@ private void processHeader() throws IOException {
         ignore = accumulate.read(lengthBytes);
         requiredLength = bytesToInt(lengthBytes);
 
+        // Validate bounds
+        if (requiredLength < 0) {
+            throw new RpcException("Invalid message length: " + requiredLength);
+        }
+        if (requiredLength > maxMessageSize) {
+            throw new RpcException(String.format("Message size %d exceeds limit %d", requiredLength, maxMessageSize));
+        }
+
         // Continue reading the frame body.
         state = DecodeState.PAYLOAD;
     }

dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java

@@ -109,6 +109,9 @@ public interface Constants {
     String H2_SETTINGS_BUILTIN_SERVICE_INIT = "dubbo.tri.builtin.service.init";
 
     String H2_SETTINGS_JSON_FRAMEWORK_NAME = "dubbo.protocol.triple.rest.json-framework";
+
+    String H2_SETTINGS_MAX_MESSAGE_SIZE = "dubbo.protocol.triple.max-message-size";
+
     String H2_SETTINGS_DISALLOWED_CONTENT_TYPES = "dubbo.protocol.triple.rest.disallowed-content-types";
     String H2_SETTINGS_OPENAPI_PREFIX = "dubbo.protocol.triple.rest.openapi";
 

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Bzip2.java

@@ -16,7 +16,11 @@
  */
 package org.apache.dubbo.rpc.protocol.tri.compressor;
 
+import org.apache.dubbo.common.config.Configuration;
+import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.rpc.Constants;
 import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.model.ApplicationModel;
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
@@ -35,11 +39,18 @@ public class Bzip2 implements Compressor, DeCompressor {
 
     public static final String BZIP2 = "bzip2";
 
+    private final int maxMessageSize;
+
     @Override
     public String getMessageEncoding() {
         return BZIP2;
     }
 
+    public Bzip2() {
+        Configuration conf = ConfigurationUtils.getEnvConfiguration(ApplicationModel.defaultModel());
+        this.maxMessageSize = conf.getInteger(Constants.H2_SETTINGS_MAX_MESSAGE_SIZE, 50 * 1024 * 1024);
+    }
+
     @Override
     public byte[] compress(byte[] payloadByteArr) throws RpcException {
         if (null == payloadByteArr || 0 == payloadByteArr.length) {
@@ -77,10 +88,16 @@ public byte[] decompress(byte[] payloadByteArr) {
         ByteArrayOutputStream out = new ByteArrayOutputStream();
         ByteArrayInputStream in = new ByteArrayInputStream(payloadByteArr);
         try {
+            int totalBytesRead = 0;
             BZip2CompressorInputStream unZip = new BZip2CompressorInputStream(in);
             byte[] buffer = new byte[2048];
             int n;
             while ((n = unZip.read(buffer)) >= 0) {
+                totalBytesRead += n;
+                if (totalBytesRead > maxMessageSize) {
+                    throw new RpcException("Decompressed message size " + totalBytesRead
+                            + " exceeds the maximum configured message size " + maxMessageSize);
+                }
                 out.write(buffer, 0, n);
             }
         } catch (Exception e) {

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Gzip.java

@@ -16,7 +16,11 @@
  */
 package org.apache.dubbo.rpc.protocol.tri.compressor;
 
+import org.apache.dubbo.common.config.Configuration;
+import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.rpc.Constants;
 import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.model.ApplicationModel;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
@@ -32,6 +36,13 @@ public class Gzip implements Compressor, DeCompressor {
 
     public static final String GZIP = "gzip";
 
+    private final int maxMessageSize;
+
+    public Gzip() {
+        Configuration conf = ConfigurationUtils.getEnvConfiguration(ApplicationModel.defaultModel());
+        this.maxMessageSize = conf.getInteger(Constants.H2_SETTINGS_MAX_MESSAGE_SIZE, 50 * 1024 * 1024);
+    }
+
     @Override
     public String getMessageEncoding() {
         return GZIP;
@@ -70,10 +81,17 @@ public byte[] decompress(byte[] payloadByteArr) throws RpcException {
 
         ByteArrayInputStream byteInStream = new ByteArrayInputStream(payloadByteArr);
         ByteArrayOutputStream byteOutStream = new ByteArrayOutputStream();
+
         try (GZIPInputStream gzipInputStream = new GZIPInputStream(byteInStream)) {
             int readByteNum;
+            int totalBytesRead = 0;
             byte[] bufferArr = new byte[256];
             while ((readByteNum = gzipInputStream.read(bufferArr)) >= 0) {
+                totalBytesRead += readByteNum;
+                if (totalBytesRead > maxMessageSize) {
+                    throw new RpcException("Decompressed message size " + totalBytesRead
+                            + " exceeds the maximum configured message size " + maxMessageSize);
+                }
                 byteOutStream.write(bufferArr, 0, readByteNum);
             }
         } catch (Exception exception) {

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/frame/TriDecoder.java

@@ -16,7 +16,11 @@
  */
 package org.apache.dubbo.rpc.protocol.tri.frame;
 
+import org.apache.dubbo.common.config.Configuration;
+import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.rpc.Constants;
 import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.model.ApplicationModel;
 import org.apache.dubbo.rpc.protocol.tri.compressor.DeCompressor;
 
 import io.netty.buffer.ByteBuf;
@@ -31,6 +35,7 @@ public class TriDecoder implements Deframer {
     private final CompositeByteBuf accumulate = Unpooled.compositeBuffer();
     private final Listener listener;
     private final DeCompressor decompressor;
+    private final Integer maxMessageSize;
     private boolean compressedFlag;
     private long pendingDeliveries;
     private boolean inDelivery = false;
@@ -42,6 +47,8 @@ public class TriDecoder implements Deframer {
     private GrpcDecodeState state = GrpcDecodeState.HEADER;
 
     public TriDecoder(DeCompressor decompressor, Listener listener) {
+        Configuration conf = ConfigurationUtils.getEnvConfiguration(ApplicationModel.defaultModel());
+        maxMessageSize = conf.getInteger(Constants.H2_SETTINGS_MAX_MESSAGE_SIZE, 50 * 1024 * 1024);
         this.decompressor = decompressor;
         this.listener = listener;
     }
@@ -123,6 +130,13 @@ private void processHeader() {
 
         requiredLength = accumulate.readInt();
 
+        if (requiredLength < 0) {
+            throw new RpcException("Invalid message length: " + requiredLength);
+        }
+        if (requiredLength > maxMessageSize) {
+            throw new RpcException(String.format("Message size %d exceeds limit %d", requiredLength, maxMessageSize));
+        }
+
         // Continue reading the frame body.
         state = GrpcDecodeState.PAYLOAD;
     }