Pipe: Add upper bound check for Pipe request decompression buffer (apache#15699)

XNX02 · SteveYurongSu · XNX02 · commit 9ce8d7bf8b02 · 2025-06-23T21:13:48.000+08:00
Co-authored-by: Steve Yurong Su &lt;rong@apache.org&gt;
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -276,6 +276,8 @@ public class CommonConfig {
   private long pipeReceiverLoginPeriodicVerificationIntervalMs = 300000;
   private double pipeReceiverActualToEstimatedMemoryRatio = 3;
 
+  private int pipeReceiverReqDecompressedMaxLengthInBytes = 1073741824; // 1GB
+
   private int pipeMaxAllowedHistoricalTsFilePerDataRegion = Integer.MAX_VALUE; // Deprecated
   private int pipeMaxAllowedPendingTsFileEpochPerDataRegion = Integer.MAX_VALUE; // Deprecated
   private int pipeMaxAllowedPinnedMemTableCount = Integer.MAX_VALUE; // per data region
@@ -1477,6 +1479,22 @@ public double getPipeReceiverActualToEstimatedMemoryRatio() {
     return pipeReceiverActualToEstimatedMemoryRatio;
   }
 
+  public void setPipeReceiverReqDecompressedMaxLengthInBytes(
+      int pipeReceiverReqDecompressedMaxLengthInBytes) {
+    if (this.pipeReceiverReqDecompressedMaxLengthInBytes
+        == pipeReceiverReqDecompressedMaxLengthInBytes) {
+      return;
+    }
+    this.pipeReceiverReqDecompressedMaxLengthInBytes = pipeReceiverReqDecompressedMaxLengthInBytes;
+    logger.info(
+        "pipeReceiverReqDecompressedMaxLengthInBytes is set to {}.",
+        pipeReceiverReqDecompressedMaxLengthInBytes);
+  }
+
+  public int getPipeReceiverReqDecompressedMaxLengthInBytes() {
+    return pipeReceiverReqDecompressedMaxLengthInBytes;
+  }
+
   public int getPipeMaxAllowedHistoricalTsFilePerDataRegion() {
     return pipeMaxAllowedHistoricalTsFilePerDataRegion;
   }
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
@@ -329,6 +329,10 @@ public double getPipeReceiverActualToEstimatedMemoryRatio() {
     return COMMON_CONFIG.getPipeReceiverActualToEstimatedMemoryRatio();
   }
 
+  public int getPipeReceiverReqDecompressedMaxLengthInBytes() {
+    return COMMON_CONFIG.getPipeReceiverReqDecompressedMaxLengthInBytes();
+  }
+
   /////////////////////////////// Hybrid Mode ///////////////////////////////
 
   public int getPipeMaxAllowedHistoricalTsFilePerDataRegion() {
@@ -614,6 +618,9 @@ public void printAllConfigs() {
     LOGGER.info(
         "PipeReceiverActualToEstimatedMemoryRatio: {}",
         getPipeReceiverActualToEstimatedMemoryRatio());
+    LOGGER.info(
+        "PipeReceiverReqDecompressedMaxLengthInBytes: {}",
+        getPipeReceiverReqDecompressedMaxLengthInBytes());
 
     LOGGER.info(
         "PipeMaxAllowedHistoricalTsFilePerDataRegion: {}",
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
@@ -435,6 +435,11 @@ public static void loadPipeInternalConfig(CommonConfig config, TrimProperties pr
             properties.getProperty(
                 "pipe_receiver_actual_to_estimated_memory_ratio",
                 Double.toString(config.getPipeReceiverActualToEstimatedMemoryRatio()))));
+    config.setPipeReceiverReqDecompressedMaxLengthInBytes(
+        Integer.parseInt(
+            properties.getProperty(
+                "pipe_receiver_req_decompressed_max_length_in_bytes",
+                String.valueOf(config.getPipeReceiverReqDecompressedMaxLengthInBytes()))));
 
     config.setPipeMaxAllowedHistoricalTsFilePerDataRegion(
         Integer.parseInt(
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/payload/thrift/request/PipeTransferCompressedReq.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/payload/thrift/request/PipeTransferCompressedReq.java
@@ -19,6 +19,7 @@
 
 package org.apache.iotdb.commons.pipe.connector.payload.thrift.request;
 
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
 import org.apache.iotdb.commons.pipe.connector.compressor.PipeCompressor;
 import org.apache.iotdb.commons.pipe.connector.compressor.PipeCompressorFactory;
 import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
@@ -91,6 +92,7 @@ public static TPipeTransferReq fromTPipeTransferReq(final TPipeTransferReq trans
       compressors.add(
           PipeCompressorFactory.getCompressor(ReadWriteIOUtils.readByte(compressedBuffer)));
       uncompressedLengths.add(ReadWriteIOUtils.readInt(compressedBuffer));
+      checkDecompressedLength(uncompressedLengths.get(i));
     }
 
     byte[] body = new byte[compressedBuffer.remaining()];
@@ -110,6 +112,19 @@ public static TPipeTransferReq fromTPipeTransferReq(final TPipeTransferReq trans
     return decompressedReq;
   }
 
+  /** This method is used to prevent decompression bomb attacks. */
+  private static void checkDecompressedLength(final int decompressedLength)
+      throws IllegalArgumentException {
+    final int maxDecompressedLength =
+        PipeConfig.getInstance().getPipeReceiverReqDecompressedMaxLengthInBytes();
+    if (decompressedLength < 0 || decompressedLength > maxDecompressedLength) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Decompressed length should be between 0 and %d, but got %d.",
+              maxDecompressedLength, decompressedLength));
+    }
+  }
+
   /**
    * For air-gap connectors. Generate the bytes of a compressed req from the bytes of original req.
    */
