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 fd19b01453 fix: bound Jackson payload decompression size (#3491)
fd19b01453 is described below
commit fd19b0145339837fa0eaac5c724d44fd8d70717b
Author: PJ Fanning <[email protected]>
AuthorDate: Mon Aug 31 17:52:37 2026 +0100
fix: bound Jackson payload decompression size (#3491)
Motivation:
JacksonSerializer.decompress inflated gzip payloads with an unbounded
transferTo, and passed the lz4 decompressed length declared on the wire
straight to the decompressor as the allocation size. A small, well-formed
message could therefore declare (or expand to) an arbitrarily large size and
drive an OutOfMemoryError on deserialization.
Modification:
Add a `compression.max-decompressed-size` setting (default 256 MiB) to the
jackson and jackson3 modules. On deserialization the gzip path copies
through
a bounded loop and the lz4 path rejects a declared length that is negative
or
over the cap, before allocating. Applies regardless of the `algorithm`
setting, since decompression is chosen by the payload's magic bytes.
Result:
A payload that would decompress beyond the cap is rejected with an
IllegalArgumentException instead of exhausting the heap.
Tests:
- sbt "serialization-jackson/testOnly *JacksonJsonSerializerSpec"
"serialization-jackson3/testOnly *JacksonJsonSerializerSpec" - 71 passed each,
incl. new gzip/lz4 cap tests
- sbt "serialization-jackson/mimaReportBinaryIssues" - no issues (changed
symbols are @InternalApi/private)
- sbt scalafmt for changed main and test sources
References:
None - found while reviewing the draft threat model in #3478
---
.../src/main/resources/reference.conf | 6 ++++
.../serialization/jackson/JacksonSerializer.scala | 31 ++++++++++++++++++--
.../jackson/JacksonSerializerSpec.scala | 34 ++++++++++++++++++++++
.../src/main/resources/reference.conf | 6 ++++
.../serialization/jackson3/JacksonSerializer.scala | 31 ++++++++++++++++++--
.../jackson3/JacksonSerializerSpec.scala | 34 ++++++++++++++++++++++
6 files changed, 136 insertions(+), 6 deletions(-)
diff --git a/serialization-jackson/src/main/resources/reference.conf
b/serialization-jackson/src/main/resources/reference.conf
index 5db0ac82b3..5ad52a7110 100644
--- a/serialization-jackson/src/main/resources/reference.conf
+++ b/serialization-jackson/src/main/resources/reference.conf
@@ -205,6 +205,12 @@ pekko.serialization.jackson {
# If compression is enabled with the `algorithm` setting the payload is
compressed
# when it's larger than this value.
compress-larger-than = 0 KiB
+
+ # Maximum size of a payload after decompression. A compressed (gzip or lz4)
+ # payload that decompresses to more than this is rejected rather than
+ # allocated, guarding against a small message that inflates without bound.
+ # This applies on deserialization regardless of the `algorithm` setting
above.
+ max-decompressed-size = 256 MiB
}
# Whether the type should be written to the manifest.
diff --git
a/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonSerializer.scala
b/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonSerializer.scala
index f749005a2f..82fb9cd3f8 100644
---
a/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonSerializer.scala
+++
b/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonSerializer.scala
@@ -207,6 +207,7 @@ import pekko.util.OptionVal
""""off" or "gzip"""")
}
}
+ private val maxDecompressedSize: Long =
conf.getBytes("compression.max-decompressed-size")
private val migrations: Map[String, JacksonMigration] = {
import scala.jdk.CollectionConverters._
conf.getConfig("migrations").root.unwrapped.asScala.toMap.map {
@@ -535,13 +536,18 @@ import pekko.util.OptionVal
def decompress(bytes: Array[Byte]): Array[Byte] = {
if (isGZipped(bytes)) {
val in = new GZIPInputStream(new
UnsynchronizedByteArrayInputStream(bytes))
- val out = new ByteArrayOutputStream()
- try in.transferTo(out)
+ try gunzip(in)
finally in.close()
- out.toByteArray
} else {
LZ4Meta.get(bytes) match {
case OptionVal.Some(meta) =>
+ // meta.length is the decompressed size declared on the wire; a small
+ // message can declare a huge (or negative) size and drive a large
+ // allocation, so bound it before decompressing.
+ if (meta.length < 0 || meta.length > maxDecompressedSize)
+ throw new IllegalArgumentException(
+ s"Compressed message declares decompressed size [${meta.length}]
bytes, which exceeds the maximum " +
+ s"of [$maxDecompressedSize] bytes
(pekko.serialization.jackson.compression.max-decompressed-size)")
val srcLen = bytes.length - meta.offset
lz4Decompressor.decompress(bytes, meta.offset, srcLen, meta.length)
case _ => bytes
@@ -549,4 +555,23 @@ import pekko.util.OptionVal
}
}
+ // gunzip with a bound on the decompressed size, so a small gzip payload
cannot
+ // inflate without limit (a "zip bomb").
+ private def gunzip(in: GZIPInputStream): Array[Byte] = {
+ val out = new ByteArrayOutputStream()
+ val buffer = new Array[Byte](BufferSize)
+ var total = 0L
+ var n = in.read(buffer)
+ while (n != -1) {
+ total += n
+ if (total > maxDecompressedSize)
+ throw new IllegalArgumentException(
+ s"Decompressed message exceeds the maximum of [$maxDecompressedSize]
bytes " +
+ "(pekko.serialization.jackson.compression.max-decompressed-size)")
+ out.write(buffer, 0, n)
+ n = in.read(buffer)
+ }
+ out.toByteArray
+ }
+
}
diff --git
a/serialization-jackson/src/test/scala/org/apache/pekko/serialization/jackson/JacksonSerializerSpec.scala
b/serialization-jackson/src/test/scala/org/apache/pekko/serialization/jackson/JacksonSerializerSpec.scala
index e4c15686d4..b2b6c2eb2c 100644
---
a/serialization-jackson/src/test/scala/org/apache/pekko/serialization/jackson/JacksonSerializerSpec.scala
+++
b/serialization-jackson/src/test/scala/org/apache/pekko/serialization/jackson/JacksonSerializerSpec.scala
@@ -653,6 +653,40 @@ class JacksonJsonSerializerSpec extends
JacksonSerializerSpec("jackson-json") {
check(SimpleCommand("Bob"), false)
check(new SimpleCommandNotCaseClass("Bob"), false)
}
+
+ "reject a gzip payload that decompresses beyond max-decompressed-size" in
withSystem("""
+ pekko.serialization.jackson.jackson-json.compression {
+ algorithm = gzip
+ compress-larger-than = 0 KiB
+ max-decompressed-size = 1 KiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val serializer = serializerFor(msg, sys)
+ val blob = serializeToBinary(msg, sys)
+ JacksonSerializer.isGZipped(blob) should ===(true)
+ val ex = intercept[IllegalArgumentException] {
+ deserializeFromBinary(blob, serializer.identifier,
serializer.manifest(msg), sys)
+ }
+ ex.getMessage should include("max-decompressed-size")
+ }
+
+ "reject an lz4 payload that declares a size beyond max-decompressed-size"
in withSystem("""
+ pekko.serialization.jackson.jackson-json.compression {
+ algorithm = lz4
+ compress-larger-than = 0 KiB
+ max-decompressed-size = 1 KiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val serializer = serializerFor(msg, sys)
+ val blob = serializeToBinary(msg, sys)
+ JacksonSerializer.isLZ4(blob) should ===(true)
+ val ex = intercept[IllegalArgumentException] {
+ deserializeFromBinary(blob, serializer.identifier,
serializer.manifest(msg), sys)
+ }
+ ex.getMessage should include("max-decompressed-size")
+ }
}
"JacksonJsonSerializer without type in manifest" should {
diff --git a/serialization-jackson3/src/main/resources/reference.conf
b/serialization-jackson3/src/main/resources/reference.conf
index ba406acc12..0a94af43f2 100644
--- a/serialization-jackson3/src/main/resources/reference.conf
+++ b/serialization-jackson3/src/main/resources/reference.conf
@@ -186,6 +186,12 @@ pekko.serialization.jackson3 {
# If compression is enabled with the `algorithm` setting the payload is
compressed
# when it's larger than this value.
compress-larger-than = 0 KiB
+
+ # Maximum size of a payload after decompression. A compressed (gzip or lz4)
+ # payload that decompresses to more than this is rejected rather than
+ # allocated, guarding against a small message that inflates without bound.
+ # This applies on deserialization regardless of the `algorithm` setting
above.
+ max-decompressed-size = 256 MiB
}
# Whether the type should be written to the manifest.
diff --git
a/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonSerializer.scala
b/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonSerializer.scala
index 43a5633be5..6a9db1da2c 100644
---
a/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonSerializer.scala
+++
b/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonSerializer.scala
@@ -208,6 +208,7 @@ import pekko.util.OptionVal
""""off" or "gzip"""")
}
}
+ private val maxDecompressedSize: Long =
conf.getBytes("compression.max-decompressed-size")
private val migrations: Map[String, JacksonMigration] = {
import scala.jdk.CollectionConverters._
conf.getConfig("migrations").root.unwrapped.asScala.toMap.map {
@@ -536,13 +537,18 @@ import pekko.util.OptionVal
def decompress(bytes: Array[Byte]): Array[Byte] = {
if (isGZipped(bytes)) {
val in = new GZIPInputStream(new
UnsynchronizedByteArrayInputStream(bytes))
- val out = new ByteArrayOutputStream()
- try in.transferTo(out)
+ try gunzip(in)
finally in.close()
- out.toByteArray
} else {
LZ4Meta.get(bytes) match {
case OptionVal.Some(meta) =>
+ // meta.length is the decompressed size declared on the wire; a small
+ // message can declare a huge (or negative) size and drive a large
+ // allocation, so bound it before decompressing.
+ if (meta.length < 0 || meta.length > maxDecompressedSize)
+ throw new IllegalArgumentException(
+ s"Compressed message declares decompressed size [${meta.length}]
bytes, which exceeds the maximum " +
+ s"of [$maxDecompressedSize] bytes
(pekko.serialization.jackson3.compression.max-decompressed-size)")
val srcLen = bytes.length - meta.offset
lz4Decompressor.decompress(bytes, meta.offset, srcLen, meta.length)
case _ => bytes
@@ -550,4 +556,23 @@ import pekko.util.OptionVal
}
}
+ // gunzip with a bound on the decompressed size, so a small gzip payload
cannot
+ // inflate without limit (a "zip bomb").
+ private def gunzip(in: GZIPInputStream): Array[Byte] = {
+ val out = new ByteArrayOutputStream()
+ val buffer = new Array[Byte](BufferSize)
+ var total = 0L
+ var n = in.read(buffer)
+ while (n != -1) {
+ total += n
+ if (total > maxDecompressedSize)
+ throw new IllegalArgumentException(
+ s"Decompressed message exceeds the maximum of [$maxDecompressedSize]
bytes " +
+ "(pekko.serialization.jackson3.compression.max-decompressed-size)")
+ out.write(buffer, 0, n)
+ n = in.read(buffer)
+ }
+ out.toByteArray
+ }
+
}
diff --git
a/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonSerializerSpec.scala
b/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonSerializerSpec.scala
index 3b3193fb2b..8338f13c40 100644
---
a/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonSerializerSpec.scala
+++
b/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonSerializerSpec.scala
@@ -598,6 +598,40 @@ class JacksonJsonSerializerSpec extends
JacksonSerializerSpec("jackson-json") {
check(SimpleCommand("Bob"), false)
check(new SimpleCommandNotCaseClass("Bob"), false)
}
+
+ "reject a gzip payload that decompresses beyond max-decompressed-size" in
withSystem("""
+ pekko.serialization.jackson3.jackson-json.compression {
+ algorithm = gzip
+ compress-larger-than = 0 KiB
+ max-decompressed-size = 1 KiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val serializer = serializerFor(msg, sys)
+ val blob = serializeToBinary(msg, sys)
+ JacksonSerializer.isGZipped(blob) should ===(true)
+ val ex = intercept[IllegalArgumentException] {
+ deserializeFromBinary(blob, serializer.identifier,
serializer.manifest(msg), sys)
+ }
+ ex.getMessage should include("max-decompressed-size")
+ }
+
+ "reject an lz4 payload that declares a size beyond max-decompressed-size"
in withSystem("""
+ pekko.serialization.jackson3.jackson-json.compression {
+ algorithm = lz4
+ compress-larger-than = 0 KiB
+ max-decompressed-size = 1 KiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val serializer = serializerFor(msg, sys)
+ val blob = serializeToBinary(msg, sys)
+ JacksonSerializer.isLZ4(blob) should ===(true)
+ val ex = intercept[IllegalArgumentException] {
+ deserializeFromBinary(blob, serializer.identifier,
serializer.manifest(msg), sys)
+ }
+ ex.getMessage should include("max-decompressed-size")
+ }
}
"JacksonJsonSerializer without type in manifest" should {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
