fix: bound the size a compressed payload may expand to

actor-tests/src/test/scala/org/apache/pekko/serialization/DecompressionSpec.scala

@@ -0,0 +1,104 @@
+/*
+ * 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.pekko.serialization
+
+import java.io.{ ByteArrayOutputStream, NotSerializableException }
+import java.util.zip.GZIPOutputStream
+
+import com.typesafe.config.ConfigFactory
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.testkit.PekkoSpec
+
+class DecompressionSpec extends PekkoSpec {
+
+  private def gzip(bytes: Array[Byte]): Array[Byte] = {
+    val bos = new ByteArrayOutputStream()
+    val zip = new GZIPOutputStream(bos)
+    try zip.write(bytes)
+    finally zip.close()
+    bos.toByteArray
+  }
+
+  "Decompression" must {
+
+    "round trip a payload within the limit" in {
+      val payload = Array.tabulate[Byte](8 * 1024)(i => (i % 251).toByte)
+      Decompression.gunzip(gzip(payload), maxDecompressedSize = 1024 * 1024) should ===(payload)
+    }
+
+    "accept a payload of exactly the limit" in {
+      val payload = new Array[Byte](1024)
+      Decompression.gunzip(gzip(payload), maxDecompressedSize = 1024).length should ===(1024)
+    }
+
+    "reject a payload one byte over the limit" in {
+      val payload = new Array[Byte](1025)
+      intercept[NotSerializableException] {
+        Decompression.gunzip(gzip(payload), maxDecompressedSize = 1024)
+      }
+    }
+
+    "name the setting in the failure so it can be raised" in {
+      intercept[NotSerializableException] {
+        Decompression.gunzip(gzip(new Array[Byte](64)), maxDecompressedSize = 8)
+      }.getMessage should include("pekko.serialization.max-decompressed-size")
+    }
+
+    "reject a highly compressible payload without decompressing all of it" in {
+      // 64 MiB of zeros compresses to roughly 64 KiB. Without the bound this allocates the
+      // full 64 MiB; with it, reading stops just past the 1 KiB limit.
+      val bomb = gzip(new Array[Byte](64 * 1024 * 1024))
+      bomb.length should be < (1024 * 1024)
+      intercept[NotSerializableException] {
+        Decompression.gunzip(bomb, maxDecompressedSize = 1024)
+      }
+    }
+
+    "apply no limit when the maximum is negative" in {
+      // the same payload the boundary tests reject at 1 KiB
+      val payload = new Array[Byte](1025)
+      Decompression.gunzip(gzip(payload), maxDecompressedSize = -1).length should ===(1025)
+    }
+
+    "read the maximum from configuration, unlimited by default" in {
+      Decompression.maxDecompressedSize(system) should ===(-1L)
+    }
+
+    "read a configured maximum as a size" in {
+      val sys = ActorSystem(
+        "DecompressionSpec-configured",
+        ConfigFactory
+          .parseString("pekko.serialization.max-decompressed-size = 16 KiB")
+          .withFallback(system.settings.config))
+      try Decompression.maxDecompressedSize(sys) should ===(16L * 1024)
+      finally shutdown(sys)
+    }
+
+    "read a configured -1 as unlimited" in {
+      val sys = ActorSystem(
+        "DecompressionSpec-negative",
+        ConfigFactory
+          .parseString("pekko.serialization.max-decompressed-size = -1")
+          .withFallback(system.settings.config))
+      try Decompression.maxDecompressedSize(sys) should ===(-1L)
+      finally shutdown(sys)
+    }
+  }
+}

actor/src/main/resources/reference.conf

@@ -897,6 +897,19 @@ pekko {
 
   }
 
+  # Maximum size a gzipped payload may expand to when a serializer decompresses it.
+  # gzip expands by up to about three orders of magnitude, so neither the size of the
+  # compressed bytes nor the transport's frame limit bounds the buffer the decompressed
+  # bytes are read into. A payload that expands beyond this is rejected with a
+  # NotSerializableException. 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 cluster legitimately exchanges.
+  # Applies to the cluster, cluster-metrics, cluster-sharding, cluster-tools and
+  # distributed-data serializers; the Jackson serializers have their own
+  # `pekko.serialization.jackson.compression.max-decompressed-size`.
+  serialization.max-decompressed-size = unlimited
+
   serialization.protobuf {
     # deprecated, use `allowed-classes` instead
     whitelist-class = [

actor/src/main/scala/org/apache/pekko/serialization/Decompression.scala

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.serialization
+
+import java.io.{ ByteArrayOutputStream, NotSerializableException }
+import java.util.zip.GZIPInputStream
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.annotation.InternalApi
+import pekko.io.UnsynchronizedByteArrayInputStream
+
+/**
+ * INTERNAL API
+ *
+ * Several wire formats gzip the serialized payload. Decompression is size amplifying: gzip
+ * expands by up to about three orders of magnitude, so the size of the compressed bytes is
+ * not a useful bound on the buffer they are read into, and neither is the transport's frame
+ * limit. These helpers stop once the decompressed size passes a configured maximum and
+ * report that as a serialization failure rather than reading the stream to its end.
+ */
+@InternalApi private[pekko] object Decompression {
+
+  private final val BufferSize = 1024 * 4
+
+  /**
+   * The configured `pekko.serialization.max-decompressed-size`, in bytes.
+   * `unlimited` or a negative number means no limit; `getBytes` refuses both,
+   * so they are read before interpreting the value as a size.
+   */
+  def maxDecompressedSize(system: ActorSystem): Long = {
+    val path = "pekko.serialization.max-decompressed-size"
+    val config = system.settings.config
+    config.getString(path) match {
+      case "unlimited" => -1L
+      case raw         =>
+        raw.toLongOption match {
+          case Some(n) if n < 0 => n
+          case _                => config.getBytes(path)
+        }
+    }
+  }
+
+  /**
+   * Gunzip `bytes`, failing with a `NotSerializableException` as soon as more than
+   * `maxDecompressedSize` bytes have been produced. A negative `maxDecompressedSize`
+   * applies no limit.
+   */
+  def gunzip(bytes: Array[Byte], maxDecompressedSize: Long): Array[Byte] = {
+    val in = new GZIPInputStream(new UnsynchronizedByteArrayInputStream(bytes))
+    try {
+      val out = new ByteArrayOutputStream(BufferSize)
+      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 NotSerializableException(
+            s"Compressed message expands to more than the maximum decompressed size of " +
+            s"[$maxDecompressedSize] bytes. " +
+            "Configure with 'pekko.serialization.max-decompressed-size'.")
+        out.write(buffer, 0, n)
+        n = in.read(buffer)
+      }
+      out.toByteArray
+    } finally in.close()
+  }
+}

cluster-metrics/src/main/scala/org/apache/pekko/cluster/metrics/protobuf/MessageSerializer.scala

@@ -15,7 +15,7 @@ package org.apache.pekko.cluster.metrics.protobuf
 
 import java.{ lang => jl }
 import java.io.{ ByteArrayOutputStream, NotSerializableException, ObjectOutputStream }
-import java.util.zip.{ GZIPInputStream, GZIPOutputStream }
+import java.util.zip.GZIPOutputStream
 
 import scala.collection.immutable
 import scala.jdk.CollectionConverters._
@@ -28,7 +28,13 @@ import pekko.dispatch.Dispatchers
 import pekko.io.UnsynchronizedByteArrayInputStream
 import pekko.protobufv3.internal.MessageLite
 import pekko.remote.ByteStringUtils
-import pekko.serialization.{ BaseSerializer, SerializationExtension, SerializerWithStringManifest, Serializers }
+import pekko.serialization.{
+  BaseSerializer,
+  Decompression,
+  SerializationExtension,
+  SerializerWithStringManifest,
+  Serializers
+}
 
 /**
  * Protobuf serializer for `ClusterMetricsMessage` types.
@@ -45,6 +51,7 @@ class MessageSerializer(val system: ExtendedActorSystem) extends SerializerWithS
   private val SystemLoadAverageMetricsSelectorManifest = "f"
 
   private lazy val serialization = SerializationExtension(system)
+  private val maxDecompressedSize: Long = Decompression.maxDecompressedSize(system)
 
   override def manifest(obj: AnyRef): String = obj match {
     case _: MetricsGossipEnvelope         => MetricsGossipEnvelopeManifest
@@ -77,11 +84,7 @@ class MessageSerializer(val system: ExtendedActorSystem) extends SerializerWithS
   }
 
   def decompress(bytes: Array[Byte]): Array[Byte] = {
-    val in = new GZIPInputStream(new UnsynchronizedByteArrayInputStream(bytes))
-    val out = new ByteArrayOutputStream()
-    try in.transferTo(out)
-    finally in.close()
-    out.toByteArray
+    Decompression.gunzip(bytes, maxDecompressedSize)
   }
 
   override def fromBinary(bytes: Array[Byte], manifest: String): AnyRef = manifest match {

cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/protobuf/ClusterShardingMessageSerializer.scala

@@ -15,7 +15,6 @@ package org.apache.pekko.cluster.sharding.protobuf
 
 import java.io.ByteArrayOutputStream
 import java.io.NotSerializableException
-import java.util.zip.GZIPInputStream
 import java.util.zip.GZIPOutputStream
 
 import scala.collection.immutable
@@ -37,9 +36,9 @@ import pekko.cluster.sharding.internal.EventSourcedRememberEntitiesShardStore.{
 import pekko.cluster.sharding.internal.EventSourcedRememberEntitiesShardStore.{ State => EntityState }
 import pekko.cluster.sharding.protobuf.msg.{ ClusterShardingMessages => sm }
 import pekko.cluster.sharding.protobuf.msg.ClusterShardingMessages
-import pekko.io.UnsynchronizedByteArrayInputStream
 import pekko.protobufv3.internal.MessageLite
 import pekko.serialization.BaseSerializer
+import pekko.serialization.Decompression
 import pekko.serialization.Serialization
 import pekko.serialization.SerializerWithStringManifest
 
@@ -54,6 +53,7 @@ private[pekko] class ClusterShardingMessageSerializer(val system: ExtendedActorS
   import ShardCoordinator.Internal._
 
   private final val BufferSize = 1024 * 4
+  private val maxDecompressedSize: Long = Decompression.maxDecompressedSize(system)
 
   private val CoordinatorStateManifest = "AA"
   private val ShardRegionRegisteredManifest = "AB"
@@ -641,11 +641,7 @@ private[pekko] class ClusterShardingMessageSerializer(val system: ExtendedActorS
   }
 
   private def decompress(bytes: Array[Byte]): Array[Byte] = {
-    val in = new GZIPInputStream(new UnsynchronizedByteArrayInputStream(bytes))
-    val out = new ByteArrayOutputStream()
-    try in.transferTo(out)
-    finally in.close()
-    out.toByteArray
+    Decompression.gunzip(bytes, maxDecompressedSize)
   }
 
 }

cluster-tools/src/main/scala/org/apache/pekko/cluster/pubsub/protobuf/DistributedPubSubMessageSerializer.scala

@@ -15,7 +15,6 @@ package org.apache.pekko.cluster.pubsub.protobuf
 
 import java.io.ByteArrayOutputStream
 import java.io.NotSerializableException
-import java.util.zip.GZIPInputStream
 import java.util.zip.GZIPOutputStream
 
 import scala.collection.immutable.TreeMap
@@ -27,7 +26,6 @@ import pekko.actor.ActorRef
 import pekko.cluster.pubsub.DistributedPubSubMediator._
 import pekko.cluster.pubsub.DistributedPubSubMediator.Internal._
 import pekko.cluster.pubsub.protobuf.msg.{ DistributedPubSubMessages => dm }
-import pekko.io.UnsynchronizedByteArrayInputStream
 import pekko.protobufv3.internal.{ ByteString, MessageLite }
 import pekko.remote.ByteStringUtils
 import pekko.serialization._
@@ -40,6 +38,7 @@ private[pekko] class DistributedPubSubMessageSerializer(val system: ExtendedActo
     with BaseSerializer {
 
   private lazy val serialization = SerializationExtension(system)
+  private val maxDecompressedSize: Long = Decompression.maxDecompressedSize(system)
 
   private final val BufferSize = 1024 * 4
 
@@ -97,11 +96,7 @@ private[pekko] class DistributedPubSubMessageSerializer(val system: ExtendedActo
   }
 
   private def decompress(bytes: Array[Byte]): Array[Byte] = {
-    val in = new GZIPInputStream(new UnsynchronizedByteArrayInputStream(bytes))
-    val out = new ByteArrayOutputStream()
-    try in.transferTo(out)
-    finally in.close()
-    out.toByteArray
+    Decompression.gunzip(bytes, maxDecompressedSize)
   }
 
   private def addressToProto(address: Address): dm.Address.Builder = address match {

cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala

@@ -14,7 +14,7 @@
 package org.apache.pekko.cluster.protobuf
 
 import java.io.{ ByteArrayOutputStream, NotSerializableException }
-import java.util.zip.{ GZIPInputStream, GZIPOutputStream }
+import java.util.zip.GZIPOutputStream
 
 import scala.collection.immutable
 import scala.concurrent.duration.Deadline
@@ -27,7 +27,6 @@ import pekko.cluster._
 import pekko.cluster.InternalClusterAction._
 import pekko.cluster.protobuf.msg.{ ClusterMessages => cm }
 import pekko.cluster.routing.{ ClusterRouterPool, ClusterRouterPoolSettings }
-import pekko.io.UnsynchronizedByteArrayInputStream
 import pekko.protobufv3.internal.MessageLite
 import pekko.remote.ByteStringUtils
 import pekko.routing.Pool
@@ -84,6 +83,7 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
     with BaseSerializer {
   import ClusterMessageSerializer._
   private lazy val serialization = SerializationExtension(system)
+  private val maxDecompressedSize: Long = Decompression.maxDecompressedSize(system)
 
   // must be lazy because serializer is initialized from Cluster extension constructor
   private lazy val GossipTimeToLive = Cluster(system).settings.GossipTimeToLive
@@ -165,11 +165,7 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
   }
 
   def decompress(bytes: Array[Byte]): Array[Byte] = {
-    val in = new GZIPInputStream(new UnsynchronizedByteArrayInputStream(bytes))
-    val out = new ByteArrayOutputStream()
-    try in.transferTo(out)
-    finally in.close()
-    out.toByteArray
+    Decompression.gunzip(bytes, maxDecompressedSize)
   }
 
   private def heartbeatToProtoByteArray(hb: ClusterHeartbeatSender.Heartbeat): Array[Byte] = {