fix: reject messages whose parallel repeated fields disagree in length

remote/src/main/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializer.scala

@@ -182,15 +182,36 @@ private[pekko] final class ArteryMessageSerializer(val system: ExtendedActorSyst
       create: (UniqueAddress, CompressionTable[T]) => U): U = {
     val protoAdv = ArteryControlFormats.CompressionTableAdvertisement.parseFrom(bytes)
 
+    // the two lists are parallel; `zip` on its own would drop the tail of the longer one and
+    // build a table the sender did not advertise, which is then acknowledged as accepted
+    if (protoAdv.getKeysCount != protoAdv.getValuesCount)
+      throw new NotSerializableException(
+        s"Compression table advertisement carries [${protoAdv.getKeysCount}] keys and " +
+        s"[${protoAdv.getValuesCount}] values, which must match")
+
     val kvs =
       protoAdv.getKeysList.asScala
         .map(keyDeserializer)
         .zip(protoAdv.getValuesList.asScala.asInstanceOf[Iterable[Int]] /* to avoid having to call toInt explicitly */ )
 
-    val table = CompressionTable[T](protoAdv.getOriginUid, protoAdv.getTableVersion.byteValue, kvs.toMap)
+    val table =
+      CompressionTable[T](protoAdv.getOriginUid, tableVersion(protoAdv.getTableVersion), kvs.toMap)
     create(deserializeUniqueAddress(protoAdv.getFrom), table)
   }
 
+  /**
+   * A compression table version is a `Byte` on both sides, so a value that does not survive the
+   * narrowing is not one any peer advertised. Narrowing it silently would make versions 256
+   * apart indistinguishable, and the version is echoed back to the sender in an ack.
+   */
+  private def tableVersion(version: Int): Byte = {
+    if (version < Byte.MinValue || version > Byte.MaxValue)
+      throw new NotSerializableException(
+        s"Compression table version [$version] is outside the range " +
+        s"[${Byte.MinValue}, ${Byte.MaxValue}] that a table version can hold")
+    version.toByte
+  }
+
   def serializeCompressionTableAdvertisementAck(from: UniqueAddress, version: Int): MessageLite =
     ArteryControlFormats.CompressionTableAdvertisementAck.newBuilder
       .setFrom(serializeUniqueAddress(from))
@@ -201,7 +222,7 @@ private[pekko] final class ArteryMessageSerializer(val system: ExtendedActorSyst
       bytes: Array[Byte],
       create: (UniqueAddress, Byte) => AnyRef): AnyRef = {
     val msg = ArteryControlFormats.CompressionTableAdvertisementAck.parseFrom(bytes)
-    create(deserializeUniqueAddress(msg.getFrom), msg.getVersion.toByte)
+    create(deserializeUniqueAddress(msg.getFrom), tableVersion(msg.getVersion))
   }
 
   def serializeSystemMessageEnvelope(

remote/src/main/scala/org/apache/pekko/remote/serialization/DaemonMsgCreateSerializer.scala

@@ -13,6 +13,8 @@
 
 package org.apache.pekko.remote.serialization
 
+import java.io.NotSerializableException
+
 import scala.collection.immutable
 import scala.jdk.CollectionConverters._
 
@@ -189,6 +191,14 @@ private[pekko] final class DaemonMsgCreateSerializer(val system: ExtendedActorSy
       val args: Vector[AnyRef] =
         // message from a newer node always contains serializer ids and possibly a string manifest for each position
         if (protoProps.getSerializerIdsCount > 0) {
+          // `toBinary` writes args, manifests, serializer ids and hasManifest one entry at a time,
+          // so the four are parallel. Indexing them by the serializer id count alone would raise
+          // IndexOutOfBoundsException on a message where they are not.
+          requireSameLength(
+            protoProps.getSerializerIdsCount,
+            "args" -> protoProps.getArgsCount,
+            "manifests" -> protoProps.getManifestsCount,
+            "hasManifest" -> protoProps.getHasManifestCount)
           for {
             idx <- (0 until protoProps.getSerializerIdsCount).toVector
           } yield {
@@ -202,6 +212,7 @@ private[pekko] final class DaemonMsgCreateSerializer(val system: ExtendedActorSy
         } else {
           // message from an older node, which only provides data and class name
           // and never any serializer ids
+          requireSameLength(protoProps.getArgsCount, "manifests" -> protoProps.getManifestsCount)
           proto.getProps.getArgsList.asScala
             .zip(proto.getProps.getManifestsList.asScala)
             .iterator
@@ -218,6 +229,20 @@ private[pekko] final class DaemonMsgCreateSerializer(val system: ExtendedActorSy
       supervisor = deserializeActorRef(system, proto.getSupervisor))
   }
 
+  /**
+   * The repeated fields of `PropsData` are parallel arrays; a message whose lengths disagree is
+   * one no `toBinary` produced. Reject it as a serialization failure rather than indexing past
+   * the end of the shorter one.
+   */
+  private def requireSameLength(expected: Int, counts: (String, Int)*): Unit =
+    counts.foreach {
+      case (name, count) =>
+        if (count != expected)
+          throw new NotSerializableException(
+            s"DaemonMsgCreate has [$expected] constructor arguments but [$count] $name; " +
+            "the repeated fields must all be the same length")
+    }
+
   private def checkAllowedActorClass(actorClass: Class[?]): Unit =
     if (!allowList.isAllowed(actorClass)) {
       val ex = new NotAllowedClassRemoteDeploymentAttemptException(actorClass, allowList.allowedClassNames)

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,58 @@ class ArteryMessageSerializerSpec extends PekkoSpec {
 
     "not support UniqueAddresses without host/port set" in pending
 
+    "reject a compression table advertisement whose keys and values disagree in length" in {
+      val serializer = new ArteryMessageSerializer(system.asInstanceOf[ExtendedActorSystem])
+      val bytes = ArteryControlFormats.CompressionTableAdvertisement.newBuilder
+        .setFrom(serializer.serializeUniqueAddress(uniqueAddress()))
+        .setOriginUid(17L)
+        .setTableVersion(1)
+        .addKeys("a")
+        .addKeys("b")
+        .addValues(0)
+        .build()
+        .toByteArray
+
+      intercept[NotSerializableException] {
+        serializer.fromBinary(bytes, "h") // ClassManifestCompressionAdvertisement
+      }.getMessage should include("must match")
+    }
+
+    "reject a compression table version that does not fit in a byte" in {
+      val serializer = new ArteryMessageSerializer(system.asInstanceOf[ExtendedActorSystem])
+
+      // the version is a Byte on both sides, so 128 is not a version any peer advertised;
+      // narrowing it silently would make it indistinguishable from -128
+      val advertisement = ArteryControlFormats.CompressionTableAdvertisement.newBuilder
+        .setFrom(serializer.serializeUniqueAddress(uniqueAddress()))
+        .setOriginUid(17L)
+        .setTableVersion(128)
+        .build()
+        .toByteArray
+      intercept[NotSerializableException] {
+        serializer.fromBinary(advertisement, "h")
+      }.getMessage should include("outside the range")
+
+      val ack = ArteryControlFormats.CompressionTableAdvertisementAck.newBuilder
+        .setFrom(serializer.serializeUniqueAddress(uniqueAddress()))
+        .setVersion(128)
+        .build()
+        .toByteArray
+      intercept[NotSerializableException] {
+        serializer.fromBinary(ack, "i") // ClassManifestCompressionAdvertisementAck
+      }.getMessage should include("outside the range")
+    }
+
+    "accept the whole byte range of compression table versions" in {
+      val serializer = new ArteryMessageSerializer(system.asInstanceOf[ExtendedActorSystem])
+      Seq[Byte](Byte.MinValue, -1, 0, 1, Byte.MaxValue).foreach { version =>
+        withClue(s"version $version: ") {
+          val msg = ClassManifestCompressionAdvertisementAck(uniqueAddress(), version)
+          serializer.fromBinary(serializer.toBinary(msg), serializer.manifest(msg)) should ===(msg)
+        }
+      }
+    }
+
     "reject invalid manifest" in {
       intercept[IllegalArgumentException] {
         val serializer = new ArteryMessageSerializer(system.asInstanceOf[ExtendedActorSystem])

remote/src/test/scala/org/apache/pekko/remote/serialization/DaemonMsgCreateSerializerAllowListSpec.scala

@@ -17,6 +17,7 @@
 
 package org.apache.pekko.remote.serialization
 
+import java.io.NotSerializableException
 import java.nio.charset.StandardCharsets.UTF_8
 import java.util.concurrent.atomic.AtomicInteger
 
@@ -27,6 +28,7 @@ import pekko.actor.Actor
 import pekko.actor.Deploy
 import pekko.actor.Props
 import pekko.remote.DaemonMsgCreate
+import pekko.remote.WireFormats
 import pekko.remote.NotAllowedClassRemoteDeploymentAttemptException
 import pekko.serialization.SerializationExtension
 import pekko.serialization.SerializerWithStringManifest
@@ -88,6 +90,45 @@ class DaemonMsgCreateSerializerAllowListSpec
 
   "DaemonMsgCreateSerializer with the remote deployment allow list enabled" must {
 
+    "reject props whose repeated fields disagree in length" in {
+      val msg = daemonMsgCreate(classOf[AllowedActor])
+      val serializer = ser.findSerializerFor(msg)
+      val proto = WireFormats.DaemonMsgCreateData.parseFrom(serializer.toBinary(msg))
+      proto.getProps.getSerializerIdsCount should ===(1)
+
+      // one more serializer id than there are args, manifests and hasManifest entries; the loop
+      // is driven by the serializer id count, so this used to index past the end of the others
+      val tampered = proto.toBuilder
+        .setProps(proto.getProps.toBuilder.addSerializerIds(0))
+        .build()
+        .toByteArray
+
+      intercept[NotSerializableException] {
+        serializer.fromBinary(tampered, None)
+      }.getMessage should include("same length")
+    }
+
+    "reject old format props whose args and manifests disagree in length" in {
+      val msg = daemonMsgCreate(classOf[AllowedActor])
+      val serializer = ser.findSerializerFor(msg)
+      val proto = WireFormats.DaemonMsgCreateData.parseFrom(serializer.toBinary(msg))
+
+      // no serializer ids selects the pre-2.4 branch, where args and manifests were zipped and a
+      // longer manifest list was silently dropped
+      val tampered = proto.toBuilder
+        .setProps(
+          proto.getProps.toBuilder
+            .clearSerializerIds()
+            .clearHasManifest()
+            .addManifests(classOf[String].getName))
+        .build()
+        .toByteArray
+
+      intercept[NotSerializableException] {
+        serializer.fromBinary(tampered, None)
+      }.getMessage should include("same length")
+    }
+
     "deserialize a DaemonMsgCreate for an allow-listed class" in {
       val bytes = ser.serialize(daemonMsgCreate(classOf[AllowedActor])).get
       val got = ser.deserialize(bytes, classOf[DaemonMsgCreate]).get.asInstanceOf[DaemonMsgCreate]