From fb0d2d16e3b1b2bc8ba01a554a74c213162a1efa Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Tue, 1 Sep 2026 19:19:26 +0100 Subject: [PATCH] fix: bound the number of entries in a compression table advertisement Motivation: deserializeCompressionAdvertisement resolves every key in the advertised table, and for actor refs that means parsing a path and populating the resolve cache. The key list was unbounded, so the only limit was the transport frame size. Measured, 10000 entries is 369 KB of wire and about 150 ms of CPU on the inbound control stream, against 9 KB for the 256 entry table a peer legitimately advertises. Modification: Reject an advertisement carrying more entries than pekko.remote.artery.advanced.compression..max, the setting that bounds the table on the sending side and is normally the same across a cluster. When it is "off" locally there is no number to check against and no bound is applied. Result: An oversized advertisement is reported as a serialization failure, which the inbound stream logs and drops. Advertisements are resent periodically, so a dropped one costs at most a delay in establishing compression. --- .../ArteryMessageSerializer.scala | 35 +++++++++++++++++-- .../ArteryMessageSerializerSpec.scala | 26 +++++++++++++- 2 files changed, 58 insertions(+), 3 deletions(-) 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 291e3d0fd75..b487f1f2127 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.OutboundHandshake.{ HandshakeReq, HandshakeRsp } import pekko.remote.artery.compress.{ CompressionProtocol, CompressionTable } import pekko.remote.artery.compress.CompressionProtocol._ import pekko.serialization.{ BaseSerializer, Serialization, SerializationExtension, SerializerWithStringManifest } +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 7810e6acc90..8bced341d5d 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 @@ -17,7 +17,7 @@ import java.io.NotSerializableException import org.apache.pekko import pekko.actor._ -import pekko.remote.{ RemoteWatcher, UniqueAddress } +import pekko.remote.{ ArteryControlFormats, RemoteWatcher, UniqueAddress } import pekko.remote.artery.{ ActorSystemTerminating, ActorSystemTerminatingAck, Quarantined, SystemMessageDelivery } import pekko.remote.artery.Flush import pekko.remote.artery.FlushAck @@ -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])