From 6ff0d0b7f9ee5bcb1140c4070d4b3bb6a1631a28 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Fri, 4 Sep 2026 09:47:44 +0100 Subject: [PATCH] fix: bound the size a compressed payload may expand to (#3502) * 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 --- .../serialization/DecompressionSpec.scala | 104 ++++++++++++++++++ actor/src/main/resources/reference.conf | 13 +++ .../pekko/serialization/Decompression.scala | 82 ++++++++++++++ .../metrics/protobuf/MessageSerializer.scala | 27 ++--- .../ClusterShardingMessageSerializer.scala | 20 +--- .../DistributedPubSubMessageSerializer.scala | 19 +--- .../protobuf/ClusterMessageSerializer.scala | 20 +--- ...erMessageSerializerDecompressionSpec.scala | 83 ++++++++++++++ .../ddata/protobuf/SerializationSupport.scala | 24 +--- ...erializationSupportDecompressionSpec.scala | 66 +++++++++++ 10 files changed, 371 insertions(+), 87 deletions(-) create mode 100644 actor-tests/src/test/scala/org/apache/pekko/serialization/DecompressionSpec.scala create mode 100644 actor/src/main/scala/org/apache/pekko/serialization/Decompression.scala create mode 100644 cluster/src/test/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializerDecompressionSpec.scala create mode 100644 distributed-data/src/test/scala/org/apache/pekko/cluster/ddata/protobuf/SerializationSupportDecompressionSpec.scala 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 00000000000..2273f154d0e --- /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 fe6282851b0..f6af1ad7f43 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 00000000000..e8976cb45b1 --- /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. + */ + +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 5f6dada57f2..1e94f3efa32 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 00d630ecaef..cf8a7408e8c 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 0dcc8141edb..b6f80f49408 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 48c0d4a83d5..93d26635aee 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 00000000000..26a26036fda --- /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 d4679a34d30..d30bac8f4b6 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 00000000000..e0fd1f686bd --- /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) + } + } +}