This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch 1.7.x
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/1.7.x by this push:
new 08d5b70097 fix: bound the size a compressed payload may expand to
(#3502) (#3527)
08d5b70097 is described below
commit 08d5b7009778e0500dfcb21949451e9a98521934
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Sep 4 11:36:46 2026 +0100
fix: bound the size a compressed payload may expand to (#3502) (#3527)
* fix: bound the size a compressed payload may expand to
Motivation:
Five serializers gzip their payload and decompress it on the way back in,
each with the same unbounded loop: read the whole GZIPInputStream into a
ByteArrayOutputStream. 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.
Modification:
Add Decompression (@InternalApi) with a gunzip that stops once the
decompressed size passes pekko.serialization.max-decompressed-size
(default 256 MiB) and reports it as a NotSerializableException, and route
all twelve call sites through it. The Jackson serializers already bound
decompression and keep their own
pekko.serialization.jackson.compression.max-decompressed-size.
Result:
An over-expanding payload is rejected as an ordinary serialization
failure. No behaviour change for payloads within the limit.
* change the default max-decompressed-size to unlimited
Motivation:
A bounded default could reject a payload an existing cluster legitimately
exchanges, so a patch release carrying a 256 MiB default could break
running clusters on upgrade. The bound should be opt-in.
Modification:
Default pekko.serialization.max-decompressed-size to -1, meaning no limit
and matching the behaviour of earlier releases. A negative maximum skips
the size check in gunzip. Config's getBytes refuses negative numbers, so
the setting is read as a plain long first and as a memory size only when
that is not a negative number.
Result:
Decompression is unbounded by default; configuring a size such as 256 MiB
bounds it.
Tests:
- sbt "actor-tests/testOnly
org.apache.pekko.serialization.DecompressionSpec" - 8 passed
- sbt "cluster/testOnly
org.apache.pekko.cluster.protobuf.ClusterMessageSerializerDecompressionSpec" -
4 passed
- sbt "distributed-data/testOnly
org.apache.pekko.cluster.ddata.protobuf.SerializationSupportDecompressionSpec"
- 3 passed
- sbt "actor/scalafmtCheckAll" "actor-tests/scalafmtCheckAll" - clean
References:
Refs #3502
* also accept "unlimited" for max-decompressed-size
Motivation:
Review on #3515 noted that an explicit keyword is clearer than a magic
number. Keep the two sibling settings consistent: accept both spellings
here as well.
Modification:
pekko.serialization.max-decompressed-size reads "unlimited" or any
negative number as no limit; the reference.conf default is written as
`unlimited`. New tests cover the keyword default and an explicit -1.
Result:
`max-decompressed-size = unlimited` and `= -1` both disable the bound.
Tests:
- sbt "actor-tests/testOnly
org.apache.pekko.serialization.DecompressionSpec" - 9 passed
- sbt "actor/scalafmtCheckAll" "actor-tests/scalafmtCheckAll" - clean
References:
Refs #3515, Refs #3502
---
.../pekko/serialization/DecompressionSpec.scala | 104 +++++++++++++++++++++
actor/src/main/resources/reference.conf | 13 +++
.../apache/pekko/serialization/Decompression.scala | 82 ++++++++++++++++
.../metrics/protobuf/MessageSerializer.scala | 27 ++----
.../ClusterShardingMessageSerializer.scala | 20 +---
.../DistributedPubSubMessageSerializer.scala | 19 +---
.../protobuf/ClusterMessageSerializer.scala | 20 +---
...ClusterMessageSerializerDecompressionSpec.scala | 83 ++++++++++++++++
.../ddata/protobuf/SerializationSupport.scala | 24 +----
.../SerializationSupportDecompressionSpec.scala | 66 +++++++++++++
10 files changed, 371 insertions(+), 87 deletions(-)
diff --git
a/actor-tests/src/test/scala/org/apache/pekko/serialization/DecompressionSpec.scala
b/actor-tests/src/test/scala/org/apache/pekko/serialization/DecompressionSpec.scala
new file mode 100644
index 0000000000..2273f154d0
--- /dev/null
+++
b/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)
+ }
+ }
+}
diff --git a/actor/src/main/resources/reference.conf
b/actor/src/main/resources/reference.conf
index fe6282851b..f6af1ad7f4 100644
--- a/actor/src/main/resources/reference.conf
+++ b/actor/src/main/resources/reference.conf
@@ -849,6 +849,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 = [
diff --git
a/actor/src/main/scala/org/apache/pekko/serialization/Decompression.scala
b/actor/src/main/scala/org/apache/pekko/serialization/Decompression.scala
new file mode 100644
index 0000000000..e8976cb45b
--- /dev/null
+++ b/actor/src/main/scala/org/apache/pekko/serialization/Decompression.scala
@@ -0,0 +1,82 @@
+/*
+ * 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 scala.util.Try
+
+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 =>
+ Try(raw.trim.toLong).toOption 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()
+ }
+}
diff --git
a/cluster-metrics/src/main/scala/org/apache/pekko/cluster/metrics/protobuf/MessageSerializer.scala
b/cluster-metrics/src/main/scala/org/apache/pekko/cluster/metrics/protobuf/MessageSerializer.scala
index 5f6dada57f..1e94f3efa3 100644
---
a/cluster-metrics/src/main/scala/org/apache/pekko/cluster/metrics/protobuf/MessageSerializer.scala
+++
b/cluster-metrics/src/main/scala/org/apache/pekko/cluster/metrics/protobuf/MessageSerializer.scala
@@ -15,8 +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 scala.annotation.tailrec
+import java.util.zip.GZIPOutputStream
import scala.collection.immutable
import org.apache.pekko
import pekko.actor.{ Address, ExtendedActorSystem }
@@ -26,7 +25,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
+}
import pekko.util.ccompat._
import pekko.util.ccompat.JavaConverters._
@@ -46,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
@@ -78,20 +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()
- 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()
- finally in.close()
- out.toByteArray
+ Decompression.gunzip(bytes, maxDecompressedSize)
}
override def fromBinary(bytes: Array[Byte], manifest: String): AnyRef =
manifest match {
diff --git
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/protobuf/ClusterShardingMessageSerializer.scala
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/protobuf/ClusterShardingMessageSerializer.scala
index 00d630ecae..cf8a7408e8 100644
---
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/protobuf/ClusterShardingMessageSerializer.scala
+++
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/protobuf/ClusterShardingMessageSerializer.scala
@@ -15,10 +15,8 @@ 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.annotation.tailrec
import scala.collection.immutable
import scala.concurrent.duration._
@@ -37,9 +35,9 @@ import
pekko.cluster.sharding.internal.EventSourcedRememberEntitiesShardStore.{
import
pekko.cluster.sharding.internal.EventSourcedRememberEntitiesShardStore.{
EntitiesStarted, EntitiesStopped }
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
import pekko.util.ccompat._
@@ -57,6 +55,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"
@@ -626,20 +625,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()
- 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()
- finally in.close()
- out.toByteArray
+ Decompression.gunzip(bytes, maxDecompressedSize)
}
}
diff --git
a/cluster-tools/src/main/scala/org/apache/pekko/cluster/pubsub/protobuf/DistributedPubSubMessageSerializer.scala
b/cluster-tools/src/main/scala/org/apache/pekko/cluster/pubsub/protobuf/DistributedPubSubMessageSerializer.scala
index 0dcc8141ed..b6f80f4940 100644
---
a/cluster-tools/src/main/scala/org/apache/pekko/cluster/pubsub/protobuf/DistributedPubSubMessageSerializer.scala
+++
b/cluster-tools/src/main/scala/org/apache/pekko/cluster/pubsub/protobuf/DistributedPubSubMessageSerializer.scala
@@ -15,9 +15,7 @@ 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.annotation.tailrec
import scala.collection.immutable.TreeMap
import org.apache.pekko
import pekko.actor.{ Address, ExtendedActorSystem }
@@ -25,7 +23,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._
@@ -41,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
@@ -98,20 +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()
- 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()
- finally in.close()
- out.toByteArray
+ Decompression.gunzip(bytes, maxDecompressedSize)
}
private def addressToProto(address: Address): dm.Address.Builder = address
match {
diff --git
a/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala
b/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala
index 557f9a274b..aa6332785f 100644
---
a/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala
+++
b/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala
@@ -14,8 +14,7 @@
package org.apache.pekko.cluster.protobuf
import java.io.ByteArrayOutputStream
-import java.util.zip.{ GZIPInputStream, GZIPOutputStream }
-import scala.annotation.tailrec
+import java.util.zip.GZIPOutputStream
import scala.collection.immutable
import scala.concurrent.duration.Deadline
import scala.annotation.nowarn
@@ -27,7 +26,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
@@ -85,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
@@ -166,20 +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()
- 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()
- finally in.close()
- out.toByteArray
+ Decompression.gunzip(bytes, maxDecompressedSize)
}
private def heartbeatToProtoByteArray(hb: ClusterHeartbeatSender.Heartbeat):
Array[Byte] = {
diff --git
a/cluster/src/test/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializerDecompressionSpec.scala
b/cluster/src/test/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializerDecompressionSpec.scala
new file mode 100644
index 0000000000..26a26036fd
--- /dev/null
+++
b/cluster/src/test/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializerDecompressionSpec.scala
@@ -0,0 +1,83 @@
+/*
+ * 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.cluster.protobuf
+
+import java.io.{ ByteArrayOutputStream, NotSerializableException }
+import java.util.zip.GZIPOutputStream
+
+import org.apache.pekko
+import pekko.actor.ExtendedActorSystem
+import pekko.cluster.GossipEnvelope
+import pekko.cluster.protobuf.msg.{ ClusterMessages => cm }
+import pekko.protobufv3.internal.ByteString
+import pekko.testkit.PekkoSpec
+
+class ClusterMessageSerializerDecompressionSpec
+ extends PekkoSpec("""
+ pekko.actor.provider = cluster
+ pekko.serialization.max-decompressed-size = 4 KiB
+ """) {
+
+ private val serializer = new
ClusterMessageSerializer(system.asInstanceOf[ExtendedActorSystem])
+
+ // 8 MiB of zeros gzips to a few KiB, so this is well inside any frame limit
but expands
+ // far past the 4 KiB configured above.
+ private val bomb: Array[Byte] = {
+ val bos = new ByteArrayOutputStream()
+ val zip = new GZIPOutputStream(bos)
+ try zip.write(new Array[Byte](8 * 1024 * 1024))
+ finally zip.close()
+ bos.toByteArray
+ }
+
+ "ClusterMessageSerializer" must {
+
+ "reject a Welcome whose payload expands past the maximum" in {
+ intercept[NotSerializableException] {
+ serializer.fromBinary(bomb, "W")
+ }.getMessage should include("max-decompressed-size")
+ }
+
+ "reject a GossipEnvelope whose gossip expands past the maximum" in {
+ // GossipEnvelope defers decompression until the gossip is read, so the
failure
+ // surfaces from `gossip` rather than from fromBinary.
+ val envelope = cm.GossipEnvelope
+ .newBuilder()
+
.setFrom(serializer.uniqueAddressToProto(pekko.cluster.Cluster(system).selfUniqueAddress))
+
.setTo(serializer.uniqueAddressToProto(pekko.cluster.Cluster(system).selfUniqueAddress))
+ .setSerializedGossip(ByteString.copyFrom(bomb))
+ .build()
+
+ val msg = serializer.fromBinary(envelope.toByteArray,
"GE").asInstanceOf[GossipEnvelope]
+ intercept[NotSerializableException] {
+ msg.gossip
+ }.getMessage should include("max-decompressed-size")
+ }
+
+ "still round trip a Welcome that stays within the maximum" in {
+ val welcome = pekko.cluster.InternalClusterAction
+ .Welcome(pekko.cluster.Cluster(system).selfUniqueAddress,
pekko.cluster.Gossip.empty)
+ serializer.fromBinary(serializer.toBinary(welcome), "W") should
===(welcome)
+ }
+
+ "bound the compressed size at a small fraction of the decompressed size"
in {
+ // guards the premise of the test above: the rejected payload really is
tiny on the wire
+ bomb.length should be < (64 * 1024)
+ }
+ }
+}
diff --git
a/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/protobuf/SerializationSupport.scala
b/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/protobuf/SerializationSupport.scala
index d4679a34d3..d30bac8f4b 100644
---
a/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/protobuf/SerializationSupport.scala
+++
b/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/protobuf/SerializationSupport.scala
@@ -14,9 +14,7 @@
package org.apache.pekko.cluster.ddata.protobuf
import java.io.ByteArrayOutputStream
-import java.util.zip.GZIPInputStream
import java.util.zip.GZIPOutputStream
-import scala.annotation.tailrec
import scala.collection.immutable.TreeMap
import org.apache.pekko
import pekko.actor.ActorRef
@@ -25,7 +23,6 @@ import pekko.actor.ExtendedActorSystem
import pekko.cluster.UniqueAddress
import pekko.cluster.ddata.VersionVector
import pekko.cluster.ddata.protobuf.msg.{ ReplicatorMessages => dm }
-import pekko.io.UnsynchronizedByteArrayInputStream
import pekko.protobufv3.internal.ByteString
import pekko.protobufv3.internal.MessageLite
import pekko.remote.ByteStringUtils
@@ -75,22 +72,11 @@ trait SerializationSupport {
bos.toByteArray
}
- def decompress(bytes: Array[Byte]): Array[Byte] = {
- 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()
- finally in.close()
- out.toByteArray
- }
+ def decompress(bytes: Array[Byte]): Array[Byte] =
+ // `system` is a constructor parameter of the serializers mixing this in,
so it is not yet
+ // assigned when trait fields initialize; read the maximum per call rather
than adding a
+ // field to this public trait. The lookup is negligible next to the
decompression itself.
+ Decompression.gunzip(bytes, Decompression.maxDecompressedSize(system))
def addressToProto(address: Address): dm.Address.Builder = address match {
case Address(_, _, Some(host), Some(port)) =>
diff --git
a/distributed-data/src/test/scala/org/apache/pekko/cluster/ddata/protobuf/SerializationSupportDecompressionSpec.scala
b/distributed-data/src/test/scala/org/apache/pekko/cluster/ddata/protobuf/SerializationSupportDecompressionSpec.scala
new file mode 100644
index 0000000000..e0fd1f686b
--- /dev/null
+++
b/distributed-data/src/test/scala/org/apache/pekko/cluster/ddata/protobuf/SerializationSupportDecompressionSpec.scala
@@ -0,0 +1,66 @@
+/*
+ * 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.cluster.ddata.protobuf
+
+import java.io.{ ByteArrayOutputStream, NotSerializableException }
+import java.util.zip.GZIPOutputStream
+
+import org.apache.pekko
+import pekko.actor.ExtendedActorSystem
+import pekko.cluster.ddata.ORSet
+import pekko.testkit.PekkoSpec
+
+class SerializationSupportDecompressionSpec
+ extends PekkoSpec("""
+ pekko.actor.provider = cluster
+ pekko.remote.artery.canonical.port = 0
+ pekko.serialization.max-decompressed-size = 4 KiB
+ """) {
+
+ // `SerializationSupport` is a public trait mixed into serializers whose
`system` is a
+ // constructor parameter, so the maximum is read per call rather than held
in a field.
+ private val serializer = new
ReplicatedDataSerializer(system.asInstanceOf[ExtendedActorSystem])
+
+ 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
+ }
+
+ "SerializationSupport" must {
+
+ "reject a payload that expands past the maximum" in {
+ intercept[NotSerializableException] {
+ serializer.decompress(gzip(new Array[Byte](8 * 1024 * 1024)))
+ }.getMessage should include("max-decompressed-size")
+ }
+
+ "still round trip a payload within the maximum" in {
+ val payload = Array.tabulate[Byte](1024)(i => (i % 251).toByte)
+ serializer.decompress(gzip(payload)) should ===(payload)
+ }
+
+ "still round trip a compressed ORSet" in {
+ val orset = ORSet().add(pekko.cluster.Cluster(system).selfUniqueAddress,
"a")
+ val manifest = serializer.manifest(orset)
+ serializer.fromBinary(serializer.toBinary(orset), manifest) should
===(orset)
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
