fix: bound the number of entries in a compression table advertisement (#3510)

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.<table>.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)

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])