diff --git a/remote/src/main/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializer.scala b/remote/src/main/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializer.scala index 5fe8b53bd1..73c1ccf37d 100644 --- a/remote/src/main/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializer.scala +++ b/remote/src/main/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializer.scala @@ -27,6 +27,7 @@ import pekko.remote.artery.compress.CompressionProtocol._ import pekko.serialization.{ BaseSerializer, Serialization, SerializationExtension, SerializerWithStringManifest } import pekko.remote.artery.Flush import pekko.remote.artery.FlushAck +import pekko.util.Helpers.toRootLowerCase /** INTERNAL API */ private[pekko] object ArteryMessageSerializer { @@ -60,6 +61,18 @@ private[pekko] final class ArteryMessageSerializer(val system: ExtendedActorSyst private lazy val serialization = SerializationExtension(system) + // `pekko.remote.artery.advanced.compression..max` bounds the number of entries the + // sending side puts in a table, and is normally the same setting across a cluster. Parsed the + // same way ArterySettings parses it, without building the whole settings object here. + private def compressionMax(table: String): Int = { + val path = s"pekko.remote.artery.advanced.compression.$table.max" + if (toRootLowerCase(system.settings.config.getString(path)) == "off") 0 + else system.settings.config.getInt(path) + } + + private val maxActorRefCompressionEntries: Int = compressionMax("actor-refs") + private val maxClassManifestCompressionEntries: Int = compressionMax("manifests") + override def manifest(o: AnyRef): String = o match { // most frequent ones first case _: SystemMessageDelivery.SystemMessageEnvelope => SystemMessageEnvelopeManifest case _: SystemMessageDelivery.Ack => SystemMessageDeliveryAckManifest @@ -123,7 +136,11 @@ private[pekko] final class ArteryMessageSerializer(val system: ExtendedActorSyst case ActorRefCompressionAdvertisementAckManifest => deserializeCompressionTableAdvertisementAck(bytes, ActorRefCompressionAdvertisementAck.apply) case ClassManifestCompressionAdvertisementManifest => - deserializeCompressionAdvertisement(bytes, identity, ClassManifestCompressionAdvertisement.apply) + deserializeCompressionAdvertisement( + bytes, + identity, + maxClassManifestCompressionEntries, + ClassManifestCompressionAdvertisement.apply) case ClassManifestCompressionAdvertisementAckManifest => deserializeCompressionTableAdvertisementAck(bytes, ClassManifestCompressionAdvertisementAck.apply) case ArteryHeartbeatManifest => RemoteWatcher.ArteryHeartbeat @@ -158,7 +175,11 @@ private[pekko] final class ArteryMessageSerializer(val system: ExtendedActorSyst serializeCompressionAdvertisement(adv)(serializeActorRef) def deserializeActorRefCompressionAdvertisement(bytes: Array[Byte]): ActorRefCompressionAdvertisement = - deserializeCompressionAdvertisement(bytes, deserializeActorRef, ActorRefCompressionAdvertisement.apply) + deserializeCompressionAdvertisement( + bytes, + deserializeActorRef, + maxActorRefCompressionEntries, + ActorRefCompressionAdvertisement.apply) def serializeCompressionAdvertisement[T](adv: CompressionAdvertisement[T])( keySerializer: T => String): ArteryControlFormats.CompressionTableAdvertisement = { @@ -179,9 +200,19 @@ private[pekko] final class ArteryMessageSerializer(val system: ExtendedActorSyst def deserializeCompressionAdvertisement[T, U]( bytes: Array[Byte], keyDeserializer: String => T, + maxEntries: Int, create: (UniqueAddress, CompressionTable[T]) => U): U = { val protoAdv = ArteryControlFormats.CompressionTableAdvertisement.parseFrom(bytes) + // Every key is resolved, and for actor refs that means parsing a path and populating the + // resolve cache, so a message with far more entries than a table can hold is work out of + // proportion to its size. `maxEntries` is 0 when compression is switched off here, and then + // there is no configured number to check against. + if (maxEntries > 0 && protoAdv.getKeysCount > maxEntries) + throw new NotSerializableException( + s"Compression table advertisement carries [${protoAdv.getKeysCount}] entries, more than " + + s"the configured maximum of [$maxEntries]") + val kvs = protoAdv.getKeysList.asScala .map(keyDeserializer) diff --git a/remote/src/test/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializerSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializerSpec.scala index bc01ea2e78..88c6167881 100644 --- a/remote/src/test/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializerSpec.scala +++ b/remote/src/test/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializerSpec.scala @@ -19,7 +19,7 @@ import org.apache.pekko import pekko.actor._ import pekko.remote.artery.Flush import pekko.remote.artery.FlushAck -import pekko.remote.{ RemoteWatcher, UniqueAddress } +import pekko.remote.{ ArteryControlFormats, RemoteWatcher, UniqueAddress } import pekko.remote.artery.{ ActorSystemTerminating, ActorSystemTerminatingAck, Quarantined, SystemMessageDelivery } import pekko.remote.artery.OutboundHandshake.{ HandshakeReq, HandshakeRsp } import pekko.remote.artery.compress.CompressionProtocol.{ @@ -74,6 +74,30 @@ class ArteryMessageSerializerSpec extends PekkoSpec { "not support UniqueAddresses without host/port set" in pending + "reject a compression table advertisement with more entries than the configured maximum" in { + val serializer = new ArteryMessageSerializer(system.asInstanceOf[ExtendedActorSystem]) + val max = system.settings.config.getInt("pekko.remote.artery.advanced.compression.actor-refs.max") + + def advertisement(entries: Int): Array[Byte] = { + val builder = ArteryControlFormats.CompressionTableAdvertisement.newBuilder + .setFrom(serializer.serializeUniqueAddress(uniqueAddress())) + .setOriginUid(17L) + .setTableVersion(1) + (0 until entries).foreach { i => + builder.addKeys(s"pekko://sys@host:1234/user/a$i") + builder.addValues(i) + } + builder.build().toByteArray + } + + // a table of exactly the configured size is what a peer legitimately advertises + serializer.fromBinary(advertisement(max), "f") shouldBe a[ActorRefCompressionAdvertisement] + + intercept[NotSerializableException] { + serializer.fromBinary(advertisement(max + 1), "f") + }.getMessage should include(s"more than the configured maximum of [$max]") + } + "reject invalid manifest" in { intercept[IllegalArgumentException] { val serializer = new ArteryMessageSerializer(system.asInstanceOf[ExtendedActorSystem])