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 291e3d0fd7..e2d72b3c98 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 d956d110cf..33394ef5a9 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 7810e6acc9..5ef8f18b73 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 60a0f6c91d..a650770362 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]