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 3f7bf3118c6..becdce2c803 100644 --- a/actor/src/main/resources/reference.conf +++ b/actor/src/main/resources/reference.conf @@ -897,6 +897,19 @@ pekko { } + # Maximum size a gzipped payload may expand to when a serializer decompresses it. + # gzip expands by up to about three orders of magnitude, so neither the size of the + # compressed bytes nor the transport's frame limit bounds the buffer the decompressed + # bytes are read into. A payload that expands beyond this is rejected with a + # NotSerializableException. The default of `unlimited` applies no limit, preserving + # the behaviour of earlier releases; a negative number such as -1 also means + # unlimited. Set a size such as `256 MiB` to bound decompression, choosing a value + # larger than any payload the cluster legitimately exchanges. + # Applies to the cluster, cluster-metrics, cluster-sharding, cluster-tools and + # distributed-data serializers; the Jackson serializers have their own + # `pekko.serialization.jackson.compression.max-decompressed-size`. + serialization.max-decompressed-size = unlimited + serialization.protobuf { # deprecated, use `allowed-classes` instead whitelist-class = [ 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..458011c9ebe --- /dev/null +++ b/actor/src/main/scala/org/apache/pekko/serialization/Decompression.scala @@ -0,0 +1,80 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * license agreements; and to You under the Apache License, version 2.0: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * This file is part of the Apache Pekko project, which was derived from Akka. + */ + +/* + * Copyright (C) 2009-2022 Lightbend Inc. + */ + +package org.apache.pekko.serialization + +import java.io.{ ByteArrayOutputStream, NotSerializableException } +import java.util.zip.GZIPInputStream + +import org.apache.pekko +import pekko.actor.ActorSystem +import pekko.annotation.InternalApi +import pekko.io.UnsynchronizedByteArrayInputStream + +/** + * INTERNAL API + * + * Several wire formats gzip the serialized payload. Decompression is size amplifying: gzip + * expands by up to about three orders of magnitude, so the size of the compressed bytes is + * not a useful bound on the buffer they are read into, and neither is the transport's frame + * limit. These helpers stop once the decompressed size passes a configured maximum and + * report that as a serialization failure rather than reading the stream to its end. + */ +@InternalApi private[pekko] object Decompression { + + private final val BufferSize = 1024 * 4 + + /** + * The configured `pekko.serialization.max-decompressed-size`, in bytes. + * `unlimited` or a negative number means no limit; `getBytes` refuses both, + * so they are read before interpreting the value as a size. + */ + def maxDecompressedSize(system: ActorSystem): Long = { + val path = "pekko.serialization.max-decompressed-size" + val config = system.settings.config + config.getString(path) match { + case "unlimited" => -1L + case raw => + raw.toLongOption match { + case Some(n) if n < 0 => n + case _ => config.getBytes(path) + } + } + } + + /** + * Gunzip `bytes`, failing with a `NotSerializableException` as soon as more than + * `maxDecompressedSize` bytes have been produced. A negative `maxDecompressedSize` + * applies no limit. + */ + def gunzip(bytes: Array[Byte], maxDecompressedSize: Long): Array[Byte] = { + val in = new GZIPInputStream(new UnsynchronizedByteArrayInputStream(bytes)) + try { + val out = new ByteArrayOutputStream(BufferSize) + val buffer = new Array[Byte](BufferSize) + var total = 0L + var n = in.read(buffer) + while (n != -1) { + total += n + if (maxDecompressedSize >= 0 && total > maxDecompressedSize) + throw new NotSerializableException( + s"Compressed message expands to more than the maximum decompressed size of " + + s"[$maxDecompressedSize] bytes. " + + "Configure with 'pekko.serialization.max-decompressed-size'.") + out.write(buffer, 0, n) + n = in.read(buffer) + } + out.toByteArray + } finally in.close() + } +} 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 95db03623e4..23a6f77ff01 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, NotSerializableException } -import java.util.zip.{ GZIPInputStream, GZIPOutputStream } +import java.util.zip.GZIPOutputStream import scala.collection.immutable import scala.concurrent.duration.Deadline @@ -27,7 +27,6 @@ import pekko.cluster._ import pekko.cluster.InternalClusterAction._ import pekko.cluster.protobuf.msg.{ ClusterMessages => cm } import pekko.cluster.routing.{ ClusterRouterPool, ClusterRouterPoolSettings } -import pekko.io.UnsynchronizedByteArrayInputStream import pekko.protobufv3.internal.MessageLite import pekko.remote.ByteStringUtils import pekko.routing.Pool @@ -84,6 +83,7 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem) with BaseSerializer { import ClusterMessageSerializer._ private lazy val serialization = SerializationExtension(system) + private val maxDecompressedSize: Long = Decompression.maxDecompressedSize(system) // must be lazy because serializer is initialized from Cluster extension constructor private lazy val GossipTimeToLive = Cluster(system).settings.GossipTimeToLive @@ -165,11 +165,7 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem) } def decompress(bytes: Array[Byte]): Array[Byte] = { - val in = new GZIPInputStream(new UnsynchronizedByteArrayInputStream(bytes)) - val out = new ByteArrayOutputStream() - try in.transferTo(out) - finally in.close() - out.toByteArray + Decompression.gunzip(bytes, maxDecompressedSize) } private def heartbeatToProtoByteArray(hb: ClusterHeartbeatSender.Heartbeat): Array[Byte] = { 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) + } + } +}