This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch 1.7.x
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/1.7.x by this push:
new 9a54d1294c fix: bounds check the lookup table indexes in gossip
(#3508) (#3520)
9a54d1294c is described below
commit 9a54d1294c3ac7821bbed3c940769c3d5027d1ff
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Sep 4 15:26:38 2026 +0100
fix: bounds check the lookup table indexes in gossip (#3508) (#3520)
* 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.
* Add imports for StandardCharsets and Files
---
.../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 aa6332785f..4c74ae6a4c 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.GZIPOutputStream
import scala.collection.immutable
import scala.concurrent.duration.Deadline
@@ -390,6 +390,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()
@@ -549,10 +563,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
@@ -564,11 +578,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
@@ -576,7 +592,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
@@ -587,14 +603,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
@@ -603,7 +619,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 6bb4b07707..7df867f874 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
@@ -27,6 +28,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
@@ -181,6 +183,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]
