From 3de71235f8b0a2ae605221db7b795c0cc4eebd07 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Tue, 1 Sep 2026 18:30:39 +0100 Subject: [PATCH] fix: reject messages whose parallel repeated fields disagree in length Motivation: Three places read repeated protobuf fields that are written in lockstep but read as if their lengths were guaranteed to agree. DaemonMsgCreateSerializer drives its loop over the constructor arguments by getSerializerIdsCount and indexes args, manifests and hasManifest with it, so a message where those disagree raises IndexOutOfBoundsException. The pre-2.4 branch zips args with manifests, which silently drops the tail of the longer one. ArteryMessageSerializer zips the keys and values of a compression table advertisement, so a mismatch silently builds a table the sender did not advertise, which is then acknowledged back to the sender as accepted. It also narrows the advertised table version, and the ack's version, from int to byte with byteValue, so versions 256 apart are indistinguishable. Modification: Check the lengths agree before indexing, and check the table version fits in a byte before narrowing it. Report either as NotSerializableException. Both are conditions no toBinary produces. Result: A malformed message is reported as a serialization failure rather than raising IndexOutOfBoundsException or being silently accepted as something other than what it said. --- .../ArteryMessageSerializer.scala | 25 ++++++++- .../DaemonMsgCreateSerializer.scala | 25 +++++++++ .../ArteryMessageSerializerSpec.scala | 54 ++++++++++++++++++- ...emonMsgCreateSerializerAllowListSpec.scala | 41 ++++++++++++++ 4 files changed, 142 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..e2d72b3c988 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 @@ -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( diff --git a/remote/src/main/scala/org/apache/pekko/remote/serialization/DaemonMsgCreateSerializer.scala b/remote/src/main/scala/org/apache/pekko/remote/serialization/DaemonMsgCreateSerializer.scala index d956d110cfc..33394ef5a93 100644 --- a/remote/src/main/scala/org/apache/pekko/remote/serialization/DaemonMsgCreateSerializer.scala +++ b/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) 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..5ef8f18b738 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,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]) diff --git a/remote/src/test/scala/org/apache/pekko/remote/serialization/DaemonMsgCreateSerializerAllowListSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/serialization/DaemonMsgCreateSerializerAllowListSpec.scala index 60a0f6c91d1..a650770362c 100644 --- a/remote/src/test/scala/org/apache/pekko/remote/serialization/DaemonMsgCreateSerializerAllowListSpec.scala +++ b/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]