From d15e86a222e151e2948f0f2d83eb0c9da63af109 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Tue, 1 Sep 2026 15:45:50 +0100 Subject: [PATCH 1/3] 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. --- .../serialization/DecompressionSpec.scala | 74 +++++++++++++++++ actor/src/main/resources/reference.conf | 9 ++ .../pekko/serialization/Decompression.scala | 67 +++++++++++++++ .../metrics/protobuf/MessageSerializer.scala | 17 ++-- .../ClusterShardingMessageSerializer.scala | 10 +-- .../DistributedPubSubMessageSerializer.scala | 9 +- .../protobuf/ClusterMessageSerializer.scala | 10 +-- ...erMessageSerializerDecompressionSpec.scala | 83 +++++++++++++++++++ .../ddata/protobuf/SerializationSupport.scala | 14 ++-- ...erializationSupportDecompressionSpec.scala | 66 +++++++++++++++ 10 files changed, 322 insertions(+), 37 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..44baddd505f --- /dev/null +++ b/actor-tests/src/test/scala/org/apache/pekko/serialization/DecompressionSpec.scala @@ -0,0 +1,74 @@ +/* + * 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 org.apache.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) + } + } + + "read the maximum from configuration" in { + Decompression.maxDecompressedSize(system) should ===(256L * 1024 * 1024) + } + } +} diff --git a/actor/src/main/resources/reference.conf b/actor/src/main/resources/reference.conf index 1f83d9ed23c..b179af05590 100644 --- a/actor/src/main/resources/reference.conf +++ b/actor/src/main/resources/reference.conf @@ -897,6 +897,15 @@ 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. 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 = 256 MiB + 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..e27f5473214 --- /dev/null +++ b/actor/src/main/scala/org/apache/pekko/serialization/Decompression.scala @@ -0,0 +1,67 @@ +/* + * 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 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. + */ + def maxDecompressedSize(system: ActorSystem): Long = + system.settings.config.getBytes("pekko.serialization.max-decompressed-size") + + /** + * Gunzip `bytes`, failing with a `NotSerializableException` as soon as more than + * `maxDecompressedSize` bytes have been produced. + */ + 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 (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 02144eb92bd..1b0623dce07 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,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 { 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 01254975a68..2bbc2d052eb 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,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) } } 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 b578c1f31cf..926c39fbaf8 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,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 { 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 d25e4d620c9..2af5bb183f6 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,7 +14,7 @@ package org.apache.pekko.cluster.protobuf import java.io.ByteArrayOutputStream -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] = { 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 b14cf6eb2de..c4cf8e1114e 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,7 +14,6 @@ package org.apache.pekko.cluster.ddata.protobuf import java.io.ByteArrayOutputStream -import java.util.zip.GZIPInputStream import java.util.zip.GZIPOutputStream import scala.collection.immutable.TreeMap @@ -27,7 +26,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 @@ -74,13 +72,11 @@ trait SerializationSupport { bos.toByteArray } - 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 - } + 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) + } + } +} From d1655dedaa36e1c41af6e5971aa59d590dde2ca0 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Thu, 3 Sep 2026 07:46:31 +0100 Subject: [PATCH 2/3] 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 --- .../serialization/DecompressionSpec.scala | 26 ++++++++++++++++--- actor/src/main/resources/reference.conf | 11 +++++--- .../pekko/serialization/Decompression.scala | 17 +++++++++--- 3 files changed, 43 insertions(+), 11 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 index 44baddd505f..38bc7183592 100644 --- a/actor-tests/src/test/scala/org/apache/pekko/serialization/DecompressionSpec.scala +++ b/actor-tests/src/test/scala/org/apache/pekko/serialization/DecompressionSpec.scala @@ -20,7 +20,11 @@ package org.apache.pekko.serialization import java.io.{ ByteArrayOutputStream, NotSerializableException } import java.util.zip.GZIPOutputStream -import org.apache.pekko.testkit.PekkoSpec +import com.typesafe.config.ConfigFactory + +import org.apache.pekko +import pekko.actor.ActorSystem +import pekko.testkit.PekkoSpec class DecompressionSpec extends PekkoSpec { @@ -67,8 +71,24 @@ class DecompressionSpec extends PekkoSpec { } } - "read the maximum from configuration" in { - Decompression.maxDecompressedSize(system) should ===(256L * 1024 * 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) } } } diff --git a/actor/src/main/resources/reference.conf b/actor/src/main/resources/reference.conf index b179af05590..edc17bfb48a 100644 --- a/actor/src/main/resources/reference.conf +++ b/actor/src/main/resources/reference.conf @@ -901,10 +901,13 @@ pekko { # 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. 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 = 256 MiB + # NotSerializableException. The default of -1 applies no limit, preserving the + # behaviour of earlier releases; 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 = -1 serialization.protobuf { # deprecated, use `allowed-classes` instead diff --git a/actor/src/main/scala/org/apache/pekko/serialization/Decompression.scala b/actor/src/main/scala/org/apache/pekko/serialization/Decompression.scala index e27f5473214..14b8a84866e 100644 --- a/actor/src/main/scala/org/apache/pekko/serialization/Decompression.scala +++ b/actor/src/main/scala/org/apache/pekko/serialization/Decompression.scala @@ -36,13 +36,22 @@ import pekko.io.UnsynchronizedByteArrayInputStream /** * The configured `pekko.serialization.max-decompressed-size`, in bytes. + * Negative means no limit; `getBytes` refuses negative values, so those are + * read before interpreting the value as a size. */ - def maxDecompressedSize(system: ActorSystem): Long = - system.settings.config.getBytes("pekko.serialization.max-decompressed-size") + def maxDecompressedSize(system: ActorSystem): Long = { + val path = "pekko.serialization.max-decompressed-size" + val config = system.settings.config + config.getString(path).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. + * `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)) @@ -53,7 +62,7 @@ import pekko.io.UnsynchronizedByteArrayInputStream var n = in.read(buffer) while (n != -1) { total += n - if (total > maxDecompressedSize) + if (maxDecompressedSize >= 0 && total > maxDecompressedSize) throw new NotSerializableException( s"Compressed message expands to more than the maximum decompressed size of " + s"[$maxDecompressedSize] bytes. " + From 23e696282b551fd05b61f70514c08d8981e5db9d Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Thu, 3 Sep 2026 13:48:04 +0100 Subject: [PATCH 3/3] 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 | 10 ++++++++++ actor/src/main/resources/reference.conf | 9 +++++---- .../apache/pekko/serialization/Decompression.scala | 14 +++++++++----- 3 files changed, 24 insertions(+), 9 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 index 38bc7183592..2273f154d0e 100644 --- a/actor-tests/src/test/scala/org/apache/pekko/serialization/DecompressionSpec.scala +++ b/actor-tests/src/test/scala/org/apache/pekko/serialization/DecompressionSpec.scala @@ -90,5 +90,15 @@ class DecompressionSpec extends PekkoSpec { 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 353e6259535..becdce2c803 100644 --- a/actor/src/main/resources/reference.conf +++ b/actor/src/main/resources/reference.conf @@ -901,13 +901,14 @@ pekko { # 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 -1 applies no limit, preserving the - # behaviour of earlier releases; set a size such as `256 MiB` to bound decompression, - # choosing a value larger than any payload the cluster legitimately exchanges. + # 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 = -1 + serialization.max-decompressed-size = unlimited serialization.protobuf { # deprecated, use `allowed-classes` instead diff --git a/actor/src/main/scala/org/apache/pekko/serialization/Decompression.scala b/actor/src/main/scala/org/apache/pekko/serialization/Decompression.scala index 14b8a84866e..458011c9ebe 100644 --- a/actor/src/main/scala/org/apache/pekko/serialization/Decompression.scala +++ b/actor/src/main/scala/org/apache/pekko/serialization/Decompression.scala @@ -36,15 +36,19 @@ import pekko.io.UnsynchronizedByteArrayInputStream /** * The configured `pekko.serialization.max-decompressed-size`, in bytes. - * Negative means no limit; `getBytes` refuses negative values, so those are - * read before interpreting the value as a size. + * `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).toLongOption match { - case Some(n) if n < 0 => n - case _ => config.getBytes(path) + config.getString(path) match { + case "unlimited" => -1L + case raw => + raw.toLongOption match { + case Some(n) if n < 0 => n + case _ => config.getBytes(path) + } } }