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 5b645144ea fix: bound nesting depth when deserializing enclosed
payloads (#3500) (#3501)
5b645144ea is described below
commit 5b645144ea46230516085440cb639608a20dc1f1
Author: PJ Fanning <[email protected]>
AuthorDate: Wed Sep 2 12:59:01 2026 +0100
fix: bound nesting depth when deserializing enclosed payloads (#3500)
(#3501)
* fix: bound nesting depth when deserializing enclosed payloads (#3500)
Motivation:
Several of the remoting wire formats enclose a serialized payload whose
enclosed
message is itself a serialized payload (for example Some, Optional,
Status.Failure,
StatusReply, a Throwable cause, or an ActorSelectionMessage). Each level is
parsed
separately, so neither a protobuf parser's own nesting limit nor the size
of the
message bounds how deep the chain can go relative to the stack: the
recursion tracks
the nesting rather than the number of bytes. A sufficiently deeply nested
message
fails with a StackOverflowError rather than a serialization error.
Modification:
Add a per-thread nesting-depth counter (NestedDeserialization). A level is
counted
where a serializer is actually invoked: Serialization.deserializeByteArray
counts
one, and in WrappedPayloadSupport.deserializePayload the two branches that
call a
serializer directly count one each, while the branch that delegates back to
Serialization leaves the counting to it. Nesting deeper than
pekko.actor.serialization-max-nesting-depth (default 32) is rejected with a
NotSerializableException. Ordinary nesting depths are unaffected.
Result:
An over-nested payload is rejected as an ordinary serialization error
instead of
exhausting the stack.
Tests:
- sbt "remote/testOnly
org.apache.pekko.remote.serialization.NestedPayloadDepthSpec" - 5 passed
- sbt "actor-tests/testOnly org.apache.pekko.serialization.SerializeSpec
org.apache.pekko.serialization.WireManifestClassLoadingSpec" - 21 passed
- sbt "actor/mimaReportBinaryIssues" "remote/mimaReportBinaryIssues" - no
issues
- scalafmt on the changed Scala sources
References:
None - robustness of nested payload deserialization
* fix: keep allow-java-serialization comment attached to its setting
* docs: note why the depth counter is a one-element Array[Int]
---
actor/src/main/resources/reference.conf | 7 ++
.../serialization/NestedDeserialization.scala | 65 +++++++++++++
.../apache/pekko/serialization/Serialization.scala | 49 ++++++----
.../serialization/WrappedPayloadSupport.scala | 10 +-
.../serialization/NestedPayloadDepthSpec.scala | 107 +++++++++++++++++++++
5 files changed, 216 insertions(+), 22 deletions(-)
diff --git a/actor/src/main/resources/reference.conf
b/actor/src/main/resources/reference.conf
index ae0ab55625..fe6282851b 100644
--- a/actor/src/main/resources/reference.conf
+++ b/actor/src/main/resources/reference.conf
@@ -777,6 +777,13 @@ pekko {
#
allow-java-serialization = off
+ # Maximum nesting depth when a serialized payload encloses another
serialized
+ # payload (for example Some, Optional, Status.Failure, StatusReply, a
Throwable
+ # cause, or an ActorSelectionMessage). Each level is parsed separately, so
this is
+ # the only bound on the chain. A message nested deeper than this is
rejected with a
+ # NotSerializableException rather than failing with a StackOverflowError.
+ serialization-max-nesting-depth = 32
+
# Log warnings when the Java serialization is used to serialize messages.
# Java serialization is not very performant and should not be used in
production
# environments unless you don't care about performance and security. In
that case
diff --git
a/actor/src/main/scala/org/apache/pekko/serialization/NestedDeserialization.scala
b/actor/src/main/scala/org/apache/pekko/serialization/NestedDeserialization.scala
new file mode 100644
index 0000000000..d4d63dedcb
--- /dev/null
+++
b/actor/src/main/scala/org/apache/pekko/serialization/NestedDeserialization.scala
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.serialization
+
+import java.io.NotSerializableException
+
+import org.apache.pekko.annotation.InternalApi
+
+/**
+ * INTERNAL API
+ *
+ * Bounds how deeply one deserialization may nest.
+ *
+ * Several wire formats carry a nested payload: the enclosed message is an
opaque byte
+ * array that is deserialized in turn, and that payload may itself enclose
another. Each
+ * level is parsed on its own, so a protobuf parser's nesting limit does not
bound the
+ * chain - only the size of the message does. Recursion therefore tracks the
nesting
+ * rather than the size of the message, and a sufficiently deep chain fails
with a
+ * `StackOverflowError` instead of a serialization error.
+ *
+ * The depth is per thread because one message is deserialized synchronously
on one thread.
+ */
+@InternalApi
+private[pekko] object NestedDeserialization {
+
+ // a one-element Array[Int] serves as a mutable int holder, so incrementing
the
+ // depth does not box the way a ThreadLocal[Int] would on every get/set
+ private val depth = new ThreadLocal[Array[Int]] {
+ override def initialValue(): Array[Int] = new Array[Int](1)
+ }
+
+ /**
+ * Runs `body` one level deeper, failing with `NotSerializableException` -
the ordinary
+ * way a message that cannot be deserialized is reported - when the nesting
exceeds `max`.
+ */
+ def atNextLevel[T](max: Int)(body: => T): T = {
+ val counter = depth.get()
+ counter(0) += 1
+ try {
+ if (counter(0) > max)
+ throw new NotSerializableException(
+ s"Message exceeds the maximum deserialization nesting depth of
[$max]. " +
+ "Configure with 'pekko.actor.serialization-max-nesting-depth'.")
+ body
+ } finally counter(0) -= 1
+ }
+
+ /** Current nesting depth, for testing. */
+ def currentDepth: Int = depth.get()(0)
+}
diff --git
a/actor/src/main/scala/org/apache/pekko/serialization/Serialization.scala
b/actor/src/main/scala/org/apache/pekko/serialization/Serialization.scala
index b823597437..b0b9e1e678 100644
--- a/actor/src/main/scala/org/apache/pekko/serialization/Serialization.scala
+++ b/actor/src/main/scala/org/apache/pekko/serialization/Serialization.scala
@@ -156,6 +156,13 @@ class Serialization(val system: ExtendedActorSystem)
extends Extension {
val log: LoggingAdapter = _log
private val manifestCache = new AtomicReference[Map[String,
Option[Class[_]]]](Map.empty[String, Option[Class[_]]])
+ /**
+ * INTERNAL API: maximum nesting depth for a payload that encloses another
serialized payload.
+ */
+ @InternalApi
+ private[pekko] val maxNestingDepth: Int =
+
system.settings.config.getInt("pekko.actor.serialization-max-nesting-depth")
+
/** INTERNAL API */
@InternalApi private[pekko] def serializationInformation:
Serialization.Information =
system.provider.serializationInformation
@@ -227,27 +234,29 @@ class Serialization(val system: ExtendedActorSystem)
extends Extension {
}
withTransportInformation { () =>
- serializer match {
- case s2: SerializerWithStringManifest => s2.fromBinary(bytes, manifest)
- case s1 =>
- if (manifest == "")
- s1.fromBinary(bytes, None)
- else {
- val cache = manifestCache.get
- cache.get(manifest) match {
- case Some(cachedClassManifest) => s1.fromBinary(bytes,
cachedClassManifest)
- case None =>
- system.dynamicAccess.getClassFor[AnyRef](manifest) match {
- case Success(classManifest) =>
- val classManifestOption: Option[Class[_]] =
Some(classManifest)
- updateCache(cache, manifest, classManifestOption)
- s1.fromBinary(bytes, classManifestOption)
- case Failure(_) =>
- throw new NotSerializableException(
- s"Cannot find manifest class [$manifest] for serializer
with id [${serializer.identifier}].")
- }
+ NestedDeserialization.atNextLevel(maxNestingDepth) {
+ serializer match {
+ case s2: SerializerWithStringManifest => s2.fromBinary(bytes,
manifest)
+ case s1 =>
+ if (manifest == "")
+ s1.fromBinary(bytes, None)
+ else {
+ val cache = manifestCache.get
+ cache.get(manifest) match {
+ case Some(cachedClassManifest) => s1.fromBinary(bytes,
cachedClassManifest)
+ case None =>
+ system.dynamicAccess.getClassFor[AnyRef](manifest) match {
+ case Success(classManifest) =>
+ val classManifestOption: Option[Class[_]] =
Some(classManifest)
+ updateCache(cache, manifest, classManifestOption)
+ s1.fromBinary(bytes, classManifestOption)
+ case Failure(_) =>
+ throw new NotSerializableException(
+ s"Cannot find manifest class [$manifest] for
serializer with id [${serializer.identifier}].")
+ }
+ }
}
- }
+ }
}
}
}
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/serialization/WrappedPayloadSupport.scala
b/remote/src/main/scala/org/apache/pekko/remote/serialization/WrappedPayloadSupport.scala
index eadcafc96e..47948b774b 100644
---
a/remote/src/main/scala/org/apache/pekko/remote/serialization/WrappedPayloadSupport.scala
+++
b/remote/src/main/scala/org/apache/pekko/remote/serialization/WrappedPayloadSupport.scala
@@ -22,6 +22,7 @@ import pekko.remote.ContainerFormats
import pekko.serialization.ByteBufferSerializer
import pekko.serialization.{ SerializationExtension, Serializers }
import pekko.serialization.DisabledJavaSerializer
+import pekko.serialization.NestedDeserialization
import pekko.serialization.Serialization
import pekko.serialization.SerializerWithStringManifest
@@ -93,15 +94,20 @@ private[pekko] object WrappedPayloadSupport {
def deserializePayload(payload: ContainerFormats.Payload, serialization:
Serialization): Any = {
val manifest = if (payload.hasMessageManifest)
payload.getMessageManifest.toStringUtf8 else ""
+ // A payload may enclose another payload; bound how deep that can go. The
level is counted
+ // where a serializer is actually invoked: the two branches below call one
directly and count
+ // it here, while the third delegates to `Serialization`, which counts it
itself.
+ val max = serialization.maxNestingDepth
serialization.serializerByIdentity(payload.getSerializerId) match {
case serializer: ByteBufferSerializer =>
// may avoid one copy of the serialized payload if the proto byte is
the right kind and the
// underlying payload serializer handles byte buffers
val buffer = payload.getEnclosedMessage.asReadOnlyByteBuffer()
buffer.order(ByteOrder.LITTLE_ENDIAN)
- serializer.fromBinary(buffer, manifest)
+ NestedDeserialization.atNextLevel(max)(serializer.fromBinary(buffer,
manifest))
case serializer: SerializerWithStringManifest =>
- serializer.fromBinary(payload.getEnclosedMessage.toByteArray, manifest)
+ NestedDeserialization.atNextLevel(max)(
+ serializer.fromBinary(payload.getEnclosedMessage.toByteArray,
manifest))
case _ =>
// only old class based manifest serializers?
serialization.deserialize(payload.getEnclosedMessage.toByteArray,
payload.getSerializerId, manifest).get
diff --git
a/remote/src/test/scala/org/apache/pekko/remote/serialization/NestedPayloadDepthSpec.scala
b/remote/src/test/scala/org/apache/pekko/remote/serialization/NestedPayloadDepthSpec.scala
new file mode 100644
index 0000000000..de85fab68b
--- /dev/null
+++
b/remote/src/test/scala/org/apache/pekko/remote/serialization/NestedPayloadDepthSpec.scala
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.remote.serialization
+
+import java.io.NotSerializableException
+
+import org.apache.pekko
+import pekko.actor.Status
+import pekko.protobufv3.internal.{ ByteString => ProtoByteString }
+import pekko.remote.ContainerFormats
+import pekko.serialization.SerializationExtension
+import pekko.serialization.Serializers
+import pekko.testkit.PekkoSpec
+
+class NestedPayloadDepthSpec extends PekkoSpec("""
+ pekko.actor.allow-java-serialization = off
+ """) {
+
+ private val serialization = SerializationExtension(system)
+ private val MiscSerializerId = 16
+ private val OptionManifest = "C"
+
+ private def payloadFor(obj: AnyRef): ContainerFormats.Payload = {
+ val ser = serialization.findSerializerFor(obj)
+ ContainerFormats.Payload
+ .newBuilder()
+
.setEnclosedMessage(ProtoByteString.copyFrom(serialization.serialize(obj).get))
+ .setSerializerId(ser.identifier)
+
.setMessageManifest(ProtoByteString.copyFromUtf8(Serializers.manifestFor(ser,
obj)))
+ .build()
+ }
+
+ /** Wraps a payload in one more `Some(...)` layer, as the wire format
encodes it. */
+ private def wrapInOption(inner: ContainerFormats.Payload):
ContainerFormats.Payload = {
+ val optionBytes =
ContainerFormats.Option.newBuilder().setValue(inner).build().toByteArray
+ ContainerFormats.Payload
+ .newBuilder()
+ .setEnclosedMessage(ProtoByteString.copyFrom(optionBytes))
+ .setSerializerId(MiscSerializerId)
+ .setMessageManifest(ProtoByteString.copyFromUtf8(OptionManifest))
+ .build()
+ }
+
+ private def nestedOption(depth: Int): ContainerFormats.Payload = (1 to
depth).foldLeft(payloadFor(pekko.Done))(
+ (acc, _) => wrapInOption(acc))
+
+ private def deserialize(p: ContainerFormats.Payload): AnyRef =
+ serialization.deserialize(p.getEnclosedMessage.toByteArray,
p.getSerializerId, OptionManifest).get
+
+ "Deserialization of a nested payload" must {
+
+ "accept nesting within the configured depth" in {
+ serialization.maxNestingDepth should ===(32)
+ deserialize(nestedOption(4)) should
===(Some(Some(Some(Some(pekko.Done)))))
+ }
+
+ "reject nesting beyond the configured depth with NotSerializableException"
in {
+ // Deeper than the limit, but still a tiny message: without a bound the
depth of
+ // this recursion is limited only by the stack.
+ val ex = intercept[NotSerializableException] {
+ deserialize(nestedOption(200))
+ }
+ ex.getMessage should include("nesting depth")
+ }
+
+ "reject deep nesting that would otherwise exhaust the stack" in {
+ val deep = nestedOption(20000)
+ withClue(s"payload is only ${deep.getEnclosedMessage.size()} bytes: ") {
+ intercept[NotSerializableException] {
+ deserialize(deep)
+ }
+ }
+ }
+
+ "not leak depth between messages" in {
+ intercept[NotSerializableException](deserialize(nestedOption(200)))
+ // a later, well-formed message must still be accepted
+ deserialize(nestedOption(4)) should
===(Some(Some(Some(Some(pekko.Done)))))
+ }
+
+ "still deserialize ordinary wrapped messages" in {
+ val failure = Status.Failure(new IllegalArgumentException("boom"))
+ val roundTripped = serialization
+ .deserialize(
+ serialization.serialize(failure).get,
+ serialization.findSerializerFor(failure).identifier,
+ Serializers.manifestFor(serialization.findSerializerFor(failure),
failure))
+ .get
+ roundTripped.getClass should ===(classOf[Status.Failure])
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
