diff --git a/serialization-jackson/src/main/resources/reference.conf b/serialization-jackson/src/main/resources/reference.conf index 5db0ac82b3e..593a4ad0aa8 100644 --- a/serialization-jackson/src/main/resources/reference.conf +++ b/serialization-jackson/src/main/resources/reference.conf @@ -205,6 +205,16 @@ 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. + # The default of `unlimited` applies no limit, preserving the behaviour of + # earlier releases; a negative number such as -1 also means unlimited. Set a + # size such as `256 MiB` to bound decompression, choosing a value larger than + # any payload the system legitimately exchanges. + max-decompressed-size = unlimited } # 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 ebadef6a7ca..9530aeb74b1 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,19 @@ import pekko.util.OptionVal """"off" or "gzip"""") } } + // "unlimited" or a negative number means no limit; getBytes refuses both, so read them + // first. Not toLongOption, which Scala 2.12 does not have. + private val maxDecompressedSize: Long = { + val raw = conf.getString("compression.max-decompressed-size") + if (raw == "unlimited") -1L + else + try { + val n = raw.toLong + if (n < 0) n else conf.getBytes("compression.max-decompressed-size") + } catch { + case _: NumberFormatException => conf.getBytes("compression.max-decompressed-size") + } + } private val migrations: Map[String, JacksonMigration] = { import pekko.util.ccompat.JavaConverters._ conf.getConfig("migrations").root.unwrapped.asScala.toMap.map { @@ -535,22 +548,21 @@ 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() - val buffer = new Array[Byte](BufferSize) - - @tailrec def readChunk(): Unit = in.read(buffer) match { - case -1 => () - case n => - out.write(buffer, 0, n) - readChunk() - } - - try readChunk() + 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) + throw new IllegalArgumentException( + s"Compressed message declares a negative decompressed size [${meta.length}] bytes") + if (maxDecompressedSize >= 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 @@ -558,4 +570,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"). A negative maximum applies no bound. + 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 (maxDecompressedSize >= 0 && 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 60fab76c931..30ed2818f33 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 @@ -649,6 +649,76 @@ 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") + } + + "apply no gzip decompression limit when max-decompressed-size is -1" in withSystem(""" + pekko.serialization.jackson.jackson-json.compression { + algorithm = gzip + compress-larger-than = 0 KiB + max-decompressed-size = -1 + } + """) { sys => + val msg = SimpleCommand("0" * (8 * 1024)) + JacksonSerializer.isGZipped(serializeToBinary(msg, sys)) should ===(true) + checkSerialization(msg, sys) + } + + "apply no lz4 decompression limit when max-decompressed-size is -1" in withSystem(""" + pekko.serialization.jackson.jackson-json.compression { + algorithm = lz4 + compress-larger-than = 0 KiB + max-decompressed-size = -1 + } + """) { sys => + val msg = SimpleCommand("0" * (8 * 1024)) + JacksonSerializer.isLZ4(serializeToBinary(msg, sys)) should ===(true) + checkSerialization(msg, sys) + } + + "apply no decompression limit when max-decompressed-size is unlimited" in withSystem(""" + pekko.serialization.jackson.jackson-json.compression { + algorithm = gzip + compress-larger-than = 0 KiB + max-decompressed-size = unlimited + } + """) { sys => + val msg = SimpleCommand("0" * (8 * 1024)) + JacksonSerializer.isGZipped(serializeToBinary(msg, sys)) should ===(true) + checkSerialization(msg, sys) + } } "JacksonJsonSerializer without type in manifest" should {