fix: bound Jackson payload decompression size, unlimited by default (#3491, #3515)

serialization-jackson/src/main/resources/reference.conf

@@ -205,6 +205,16 @@ pekko.serialization.jackson {
     # If compression is enabled with the `algorithm` setting the payload is compressed
     # when it's larger than this value.
     compress-larger-than = 0 KiB
+
+    # Maximum size of a payload after decompression. A compressed (gzip or lz4)
+    # payload that decompresses to more than this is rejected rather than
+    # allocated, guarding against a small message that inflates without bound.
+    # This applies on deserialization regardless of the `algorithm` setting above.
+    # The default of `unlimited` applies no limit, preserving the behaviour of
+    # earlier releases; a negative number such as -1 also means unlimited. Set a
+    # size such as `256 MiB` to bound decompression, choosing a value larger than
+    # any payload the system legitimately exchanges.
+    max-decompressed-size = unlimited
   }
 
   # Whether the type should be written to the manifest.

serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonSerializer.scala

@@ -207,6 +207,19 @@ import pekko.util.OptionVal
           """"off" or "gzip"""")
     }
   }
+  // "unlimited" or a negative number means no limit; getBytes refuses both, so read them
+  // first. Not toLongOption, which Scala 2.12 does not have.
+  private val maxDecompressedSize: Long = {
+    val raw = conf.getString("compression.max-decompressed-size")
+    if (raw == "unlimited") -1L
+    else
+      try {
+        val n = raw.toLong
+        if (n < 0) n else conf.getBytes("compression.max-decompressed-size")
+      } catch {
+        case _: NumberFormatException => conf.getBytes("compression.max-decompressed-size")
+      }
+  }
   private val migrations: Map[String, JacksonMigration] = {
     import pekko.util.ccompat.JavaConverters._
     conf.getConfig("migrations").root.unwrapped.asScala.toMap.map {
@@ -535,27 +548,45 @@ import pekko.util.OptionVal
   def decompress(bytes: Array[Byte]): Array[Byte] = {
     if (isGZipped(bytes)) {
       val in = new GZIPInputStream(new UnsynchronizedByteArrayInputStream(bytes))
-      val out = new ByteArrayOutputStream()
-      val buffer = new Array[Byte](BufferSize)
-
-      @tailrec def readChunk(): Unit = in.read(buffer) match {
-        case -1 => ()
-        case n  =>
-          out.write(buffer, 0, n)
-          readChunk()
-      }
-
-      try readChunk()
+      try gunzip(in)
       finally in.close()
-      out.toByteArray
     } else {
       LZ4Meta.get(bytes) match {
         case OptionVal.Some(meta) =>
+          // meta.length is the decompressed size declared on the wire; a small
+          // message can declare a huge (or negative) size and drive a large
+          // allocation, so bound it before decompressing.
+          if (meta.length < 0)
+            throw new IllegalArgumentException(
+              s"Compressed message declares a negative decompressed size [${meta.length}] bytes")
+          if (maxDecompressedSize >= 0 && meta.length > maxDecompressedSize)
+            throw new IllegalArgumentException(
+              s"Compressed message declares decompressed size [${meta.length}] bytes, which exceeds the maximum " +
+              s"of [$maxDecompressedSize] bytes (pekko.serialization.jackson.compression.max-decompressed-size)")
           val srcLen = bytes.length - meta.offset
           lz4Decompressor.decompress(bytes, meta.offset, srcLen, meta.length)
         case _ => bytes
       }
     }
   }
 
+  // gunzip with a bound on the decompressed size, so a small gzip payload cannot
+  // inflate without limit (a "zip bomb"). A negative maximum applies no bound.
+  private def gunzip(in: GZIPInputStream): Array[Byte] = {
+    val out = new ByteArrayOutputStream()
+    val buffer = new Array[Byte](BufferSize)
+    var total = 0L
+    var n = in.read(buffer)
+    while (n != -1) {
+      total += n
+      if (maxDecompressedSize >= 0 && total > maxDecompressedSize)
+        throw new IllegalArgumentException(
+          s"Decompressed message exceeds the maximum of [$maxDecompressedSize] bytes " +
+          "(pekko.serialization.jackson.compression.max-decompressed-size)")
+      out.write(buffer, 0, n)
+      n = in.read(buffer)
+    }
+    out.toByteArray
+  }
+
 }

serialization-jackson/src/test/scala/org/apache/pekko/serialization/jackson/JacksonSerializerSpec.scala

@@ -649,6 +649,76 @@ class JacksonJsonSerializerSpec extends JacksonSerializerSpec("jackson-json") {
       check(SimpleCommand("Bob"), false)
       check(new SimpleCommandNotCaseClass("Bob"), false)
     }
+
+    "reject a gzip payload that decompresses beyond max-decompressed-size" in withSystem("""
+        pekko.serialization.jackson.jackson-json.compression {
+          algorithm = gzip
+          compress-larger-than = 0 KiB
+          max-decompressed-size = 1 KiB
+        }
+      """) { sys =>
+      val msg = SimpleCommand("0" * (8 * 1024))
+      val serializer = serializerFor(msg, sys)
+      val blob = serializeToBinary(msg, sys)
+      JacksonSerializer.isGZipped(blob) should ===(true)
+      val ex = intercept[IllegalArgumentException] {
+        deserializeFromBinary(blob, serializer.identifier, serializer.manifest(msg), sys)
+      }
+      ex.getMessage should include("max-decompressed-size")
+    }
+
+    "reject an lz4 payload that declares a size beyond max-decompressed-size" in withSystem("""
+        pekko.serialization.jackson.jackson-json.compression {
+          algorithm = lz4
+          compress-larger-than = 0 KiB
+          max-decompressed-size = 1 KiB
+        }
+      """) { sys =>
+      val msg = SimpleCommand("0" * (8 * 1024))
+      val serializer = serializerFor(msg, sys)
+      val blob = serializeToBinary(msg, sys)
+      JacksonSerializer.isLZ4(blob) should ===(true)
+      val ex = intercept[IllegalArgumentException] {
+        deserializeFromBinary(blob, serializer.identifier, serializer.manifest(msg), sys)
+      }
+      ex.getMessage should include("max-decompressed-size")
+    }
+
+    "apply no gzip decompression limit when max-decompressed-size is -1" in withSystem("""
+        pekko.serialization.jackson.jackson-json.compression {
+          algorithm = gzip
+          compress-larger-than = 0 KiB
+          max-decompressed-size = -1
+        }
+      """) { sys =>
+      val msg = SimpleCommand("0" * (8 * 1024))
+      JacksonSerializer.isGZipped(serializeToBinary(msg, sys)) should ===(true)
+      checkSerialization(msg, sys)
+    }
+
+    "apply no lz4 decompression limit when max-decompressed-size is -1" in withSystem("""
+        pekko.serialization.jackson.jackson-json.compression {
+          algorithm = lz4
+          compress-larger-than = 0 KiB
+          max-decompressed-size = -1
+        }
+      """) { sys =>
+      val msg = SimpleCommand("0" * (8 * 1024))
+      JacksonSerializer.isLZ4(serializeToBinary(msg, sys)) should ===(true)
+      checkSerialization(msg, sys)
+    }
+
+    "apply no decompression limit when max-decompressed-size is unlimited" in withSystem("""
+        pekko.serialization.jackson.jackson-json.compression {
+          algorithm = gzip
+          compress-larger-than = 0 KiB
+          max-decompressed-size = unlimited
+        }
+      """) { sys =>
+      val msg = SimpleCommand("0" * (8 * 1024))
+      JacksonSerializer.isGZipped(serializeToBinary(msg, sys)) should ===(true)
+      checkSerialization(msg, sys)
+    }
   }
 
   "JacksonJsonSerializer without type in manifest" should {