This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new ff10ec2cca fix: reject messages whose parallel repeated fields
disagree in length (#3507)
ff10ec2cca is described below
commit ff10ec2ccac43383536de15dd59f49afc3ff34ab
Author: PJ Fanning <[email protected]>
AuthorDate: Mon Sep 7 13:09:46 2026 +0100
fix: reject messages whose parallel repeated fields disagree in length
(#3507)
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.
---
.../serialization/ArteryMessageSerializer.scala | 25 ++++++++++-
.../serialization/DaemonMsgCreateSerializer.scala | 25 +++++++++++
.../ArteryMessageSerializerSpec.scala | 52 ++++++++++++++++++++++
.../DaemonMsgCreateSerializerAllowListSpec.scala | 41 +++++++++++++++++
4 files changed, 141 insertions(+), 2 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 b487f1f212..5c22206ab0 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
@@ -213,15 +213,36 @@ private[pekko] final class ArteryMessageSerializer(val
system: ExtendedActorSyst
s"Compression table advertisement carries [${protoAdv.getKeysCount}]
entries, more than " +
s"the configured maximum of [$maxEntries]")
+ // 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))
@@ -232,7 +253,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 8bced341d5..901a2ac73c 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
@@ -98,6 +98,58 @@ class ArteryMessageSerializerSpec extends PekkoSpec {
}.getMessage should include(s"more than the configured maximum of
[$max]")
}
+ "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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
