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 f062f2332b fix: bounds check the lookup table indexes in gossip (#3508)
f062f2332b is described below
commit f062f2332b840bcef173fe14b8b27d92f3a9b264
Author: PJ Fanning <[email protected]>
AuthorDate: Thu Sep 3 10:10:52 2026 +0100
fix: bounds check the lookup table indexes in gossip (#3508)
Motivation:
Gossip interns addresses, roles, hashes and app versions into tables and
refers to them by index. Every index in gossipFromProto came straight off
the wire into Vector.apply with no range check, so a negative or out of
range one raised IndexOutOfBoundsException instead of a serialization
failure.
For a GossipEnvelope this matters more than usual: gossipEnvelopeFromProto
defers the parse into a thunk, so the throw happens inside ClusterCoreDaemon
when the gossip is read rather than on a deserialization thread.
Modification:
Look the indexes up through a helper that range checks first and reports a
NotSerializableException naming the index and the table size. Every index
gossipToProto writes is in range, so nothing a peer legitimately sends is
affected.
Result:
Gossip that refers to a table entry the sender did not include is reported
as a serialization failure.
---
.../protobuf/ClusterMessageSerializer.scala | 34 +++++++---
.../protobuf/ClusterMessageSerializerSpec.scala | 75 ++++++++++++++++++++++
2 files changed, 100 insertions(+), 9 deletions(-)
diff --git
a/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala
b/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala
index 0e5b4fc3a3..95db03623e 100644
---
a/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala
+++
b/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala
@@ -13,7 +13,7 @@
package org.apache.pekko.cluster.protobuf
-import java.io.ByteArrayOutputStream
+import java.io.{ ByteArrayOutputStream, NotSerializableException }
import java.util.zip.{ GZIPInputStream, GZIPOutputStream }
import scala.collection.immutable
@@ -389,6 +389,20 @@ final class ClusterMessageSerializer(val system:
ExtendedActorSystem)
case _ => throw new IllegalArgumentException(s"Unknown $unknown
[$value] in cluster message")
}
+ /**
+ * Gossip interns addresses, roles, hashes and app versions into tables and
refers to them by
+ * index. Every index `gossipToProto` writes is in range, so one that is not
came from a
+ * message no `toBinary` produced; resolve it as a serialization failure
rather than letting
+ * `Vector.apply` raise IndexOutOfBoundsException. For a GossipEnvelope this
matters more than
+ * usual, because the parse is deferred and runs on the cluster daemon
rather than on a
+ * deserialization thread.
+ */
+ private def lookup[T](mapping: immutable.Seq[T], index: Int, what: String):
T =
+ if (index < 0 || index >= mapping.size)
+ throw new NotSerializableException(
+ s"Cluster message refers to $what index [$index], but only
[${mapping.size}] were sent")
+ else mapping(index)
+
private def joinToProto(node: UniqueAddress, roles: Set[String], appVersion:
Version): cm.Join =
cm.Join
.newBuilder()
@@ -548,10 +562,10 @@ final class ClusterMessageSerializer(val system:
ExtendedActorSystem)
val recordBuilder = new immutable.VectorBuilder[Reachability.Record]
val versionsBuilder = Map.newBuilder[UniqueAddress, Long]
for (o <- observerReachability) {
- val observer = addressMapping(o.getAddressIndex)
+ val observer = lookup(addressMapping, o.getAddressIndex, "address")
versionsBuilder += ((observer, o.getVersion))
for (s <- o.getSubjectReachabilityList.asScala) {
- val subject = addressMapping(s.getAddressIndex)
+ val subject = lookup(addressMapping, s.getAddressIndex, "address")
val record =
Reachability.Record(observer, subject,
reachabilityStatusFromInt(s.getStatus.getNumber), s.getVersion)
recordBuilder += record
@@ -563,11 +577,13 @@ final class ClusterMessageSerializer(val system:
ExtendedActorSystem)
def memberFromProto(member: cm.Member) =
new Member(
- addressMapping(member.getAddressIndex),
+ lookup(addressMapping, member.getAddressIndex, "address"),
member.getUpNumber,
memberStatusFromInt(member.getStatus.getNumber),
rolesFromProto(member.getRolesIndexesList.asScala.toSeq),
- if (appVersionMapping.isEmpty) Version.Zero else
appVersionMapping(member.getAppVersionIndex))
+ // an older node sends no app versions at all, which is not the same
as an index it did not send
+ if (appVersionMapping.isEmpty) Version.Zero
+ else lookup(appVersionMapping, member.getAppVersionIndex, "app
version"))
def rolesFromProto(roleIndexes: Seq[Integer]): Set[String] = {
var containsDc = false
@@ -575,7 +591,7 @@ final class ClusterMessageSerializer(val system:
ExtendedActorSystem)
for {
roleIndex <- roleIndexes
- role = roleMapping(roleIndex)
+ role = lookup(roleMapping, roleIndex, "role")
} {
if (role.startsWith(ClusterSettings.DcRolePrefix)) containsDc = true
roles += role
@@ -586,14 +602,14 @@ final class ClusterMessageSerializer(val system:
ExtendedActorSystem)
}
def tombstoneFromProto(tombstone: cm.Tombstone): (UniqueAddress, Long) =
- (addressMapping(tombstone.getAddressIndex), tombstone.getTimestamp)
+ (lookup(addressMapping, tombstone.getAddressIndex, "address"),
tombstone.getTimestamp)
val members: immutable.SortedSet[Member] =
gossip.getMembersList.asScala.iterator.map(memberFromProto).to(immutable.SortedSet)
val reachability =
reachabilityFromProto(gossip.getOverview.getObserverReachabilityList.asScala)
val seen: Set[UniqueAddress] =
-
gossip.getOverview.getSeenList.asScala.iterator.map(addressMapping(_)).to(immutable.Set)
+
gossip.getOverview.getSeenList.asScala.iterator.map(lookup(addressMapping, _,
"address")).to(immutable.Set)
val overview = GossipOverview(seen, reachability)
val tombstones: Map[UniqueAddress, Long] =
gossip.getTombstonesList.asScala.iterator.map(tombstoneFromProto).toMap
@@ -602,7 +618,7 @@ final class ClusterMessageSerializer(val system:
ExtendedActorSystem)
private def vectorClockFromProto(version: cm.VectorClock, hashMapping:
immutable.Seq[String]) = {
VectorClock(scala.collection.immutable.TreeMap.from(version.getVersionsList.asScala.iterator.map(v
=>
- (VectorClock.Node.fromHash(hashMapping(v.getHashIndex)),
v.getTimestamp))))
+ (VectorClock.Node.fromHash(lookup(hashMapping, v.getHashIndex, "hash")),
v.getTimestamp))))
}
private def gossipEnvelopeFromProto(envelope: cm.GossipEnvelope):
GossipEnvelope = {
diff --git
a/cluster/src/test/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializerSpec.scala
b/cluster/src/test/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializerSpec.scala
index b75da537e9..ec993a6ead 100644
---
a/cluster/src/test/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializerSpec.scala
+++
b/cluster/src/test/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializerSpec.scala
@@ -13,6 +13,7 @@
package org.apache.pekko.cluster.protobuf
+import java.io.NotSerializableException
import java.nio.charset.StandardCharsets
import java.nio.file.Files
@@ -26,6 +27,7 @@ import pekko.cluster._
import pekko.cluster.InternalClusterAction.CompatibleConfig
import pekko.cluster.protobuf.msg.{ ClusterMessages => cm }
import pekko.cluster.routing.{ ClusterRouterPool, ClusterRouterPoolSettings }
+import pekko.remote.ByteStringUtils
import pekko.routing.RoundRobinPool
import pekko.testkit.PekkoSpec
import pekko.util.Version
@@ -182,6 +184,79 @@ class ClusterMessageSerializerSpec extends
PekkoSpec("pekko.actor.provider = clu
ClusterMessageSerializer.OldWelcomeManifest)
}
+ "reject gossip that refers to a lookup table entry it did not send" in {
+ // Gossip interns addresses, roles, hashes and app versions and refers
to them by index.
+ // Every index gossipToProto writes is in range, so an out of range one
is a message no
+ // toBinary produced; it must not surface as IndexOutOfBoundsException.
+ val node1 = VectorClock.Node("node1")
+ val gossip = (Gossip(SortedSet(a1, b1)) :+ node1).seen(a1.uniqueAddress)
+ val welcome = InternalClusterAction.Welcome(a1.uniqueAddress, gossip)
+ val proto =
cm.Welcome.parseFrom(serializer.decompress(serializer.toBinary(welcome)))
+
+ def rejects(tamper: cm.Gossip.Builder => Unit): String = {
+ val g = proto.getGossip.toBuilder
+ tamper(g)
+ val bytes = serializer.compress(proto.toBuilder.setGossip(g).build())
+ intercept[NotSerializableException](serializer.fromBinary(bytes,
"W")).getMessage
+ }
+
+ // an address index one past the end of allAddresses
+ rejects(_.setMembers(0,
proto.getGossip.getMembers(0).toBuilder.setAddressIndex(99))) should include(
+ "address index [99]")
+ // a negative index
+ rejects(_.setMembers(0,
proto.getGossip.getMembers(0).toBuilder.setAddressIndex(-1))) should include(
+ "address index [-1]")
+ // a role index out of range
+ rejects(_.setMembers(0,
proto.getGossip.getMembers(0).toBuilder.setRolesIndexes(0, 99))) should include(
+ "role index [99]")
+ // a seen entry pointing nowhere
+ rejects(_.setOverview(proto.getGossip.getOverview.toBuilder.setSeen(0,
99))) should include("address index [99]")
+ // a vector clock hash index pointing nowhere
+ rejects(
+ _.setVersion(
+ proto.getGossip.getVersion.toBuilder
+ .setVersions(0,
proto.getGossip.getVersion.getVersions(0).toBuilder.setHashIndex(99)))) should
include(
+ "hash index [99]")
+ }
+
+ "reject a gossip status that refers to a hash it did not send" in {
+ val node1 = VectorClock.Node("node1")
+ val gossip = Gossip(SortedSet(a1)) :+ node1
+ val status = GossipStatus(a1.uniqueAddress, gossip.version,
gossip.seenDigest)
+ val proto = cm.GossipStatus.parseFrom(serializer.toBinary(status))
+
+ val tampered = proto.toBuilder
+ .setVersion(proto.getVersion.toBuilder.setVersions(0,
+ proto.getVersion.getVersions(0).toBuilder.setHashIndex(7)))
+ .build()
+ .toByteArray
+
+ intercept[NotSerializableException] {
+ serializer.fromBinary(tampered, "GS")
+ }.getMessage should include("hash index [7]")
+ }
+
+ "reject a gossip envelope with a bad index when the gossip is read" in {
+ // GossipEnvelope defers the parse, so the failure surfaces from
`gossip` rather than from
+ // fromBinary - on the cluster daemon's thread rather than a
deserialization thread.
+ val gossip = Gossip(SortedSet(a1, b1))
+ val envelope = GossipEnvelope(a1.uniqueAddress, b1.uniqueAddress, gossip)
+ val proto = cm.GossipEnvelope.parseFrom(serializer.toBinary(envelope))
+ val inner =
cm.Gossip.parseFrom(serializer.decompress(proto.getSerializedGossip.toByteArray))
+ val tamperedInner =
+ inner.toBuilder.setMembers(0,
inner.getMembers(0).toBuilder.setAddressIndex(99)).build()
+
+ val bytes = proto.toBuilder
+
.setSerializedGossip(ByteStringUtils.toProtoByteStringUnsafe(serializer.compress(tamperedInner)))
+ .build()
+ .toByteArray
+
+ val msg = serializer.fromBinary(bytes, "GE").asInstanceOf[GossipEnvelope]
+ intercept[NotSerializableException] {
+ msg.gossip
+ }.getMessage should include("address index [99]")
+ }
+
"add a default data center role to gossip if none is present" in {
val env = roundtrip(GossipEnvelope(a1.uniqueAddress, d1.uniqueAddress,
Gossip(SortedSet(a1, d1))))
env.gossip.members.head.roles should be(Set(ClusterSettings.DcRolePrefix
+ "default"))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
