Limit the depth of all nested values when decoding

build.sbt

@@ -27,7 +27,6 @@ inThisBuild(
     semanticdbVersion := scalafixSemanticdb.revision,
     startYear := Some(2019),
     tlBaseVersion := "2.0",
-    tlFatalWarnings := false,
     tlSiteKeepFiles := false,
     // The website is published from series/1.x until the 2.0.0 release
     tlSitePublishBranch := None,

modules/core/src/main/scala/vulcan/BinaryDecoder.scala

@@ -47,7 +47,7 @@ object BinaryDecoder {
   private[vulcan] val MaxZeroByteItems: Long =
     1000000L
 
-  /** The limit for the number of nested values of recursive schemas, to avoid stack overflows. */
+  /** The limit for nested records, arrays and maps, to avoid stack overflows. */
   private[vulcan] val MaxDepth: Int =
     512
 
@@ -81,26 +81,39 @@ object BinaryDecoder {
           Left(new InfiniteRecord(record))
 
         case None =>
-          val root = new Compilation(definitions, minSizes).compile(writerSchema)
+          val root = new Compilation(minSizes).compile(writerSchema)
           Right((reader: BinaryReader) => root.read(new Input(reader)))
       }
     }
 
   /**
     * The state of reading a value, including the number of zero byte items
-    * read so far, and the number of nested values read for recursive schemas.
+    * read so far, and the number of records, arrays and maps being read.
     */
   private final class Input(val reader: BinaryReader) {
     var zeroByteItems: Long = 0L
 
-    var depth: Int = 0
+    private[this] var depth: Int = 0
+
+    /** Starts reading a nested record, array or map, failing beyond the depth limit. */
+    def enter(): Unit = {
+      if (depth == MaxDepth)
+        throw new InvalidBinary(
+          s"Invalid depth ${MaxDepth + 1} at position ${reader.position}, exceeding the limit of $MaxDepth nested records, arrays and maps"
+        )
+
+      depth += 1
+    }
+
+    def exit(): Unit =
+      depth -= 1
   }
 
   private sealed abstract class Node {
     def read(input: Input): AvroValue
   }
 
-  private final class Compilation(definitions: collection.Map[String, Schema.Named], minSizes: MinSizes) {
+  private final class Compilation(minSizes: MinSizes) {
     private val compiled = mutable.Map.empty[String, Reference]
 
     def compile(schema: Schema): Node =
@@ -149,27 +162,15 @@ object BinaryDecoder {
     }
   }
 
-  /**
-    * A named schema, which is being compiled if references are recursive. Since references
-    * are only used for recursive schemas, reading them limits the depth of nested values.
-    */
+  /** A named schema, which is being compiled if references are recursive. */
   private final class Reference extends Node {
     var node: Node = null
 
     def resolved: Node =
       if (node != null) node else this
 
-    override def read(input: Input): AvroValue = {
-      if (input.depth == MaxDepth)
-        throw new InvalidBinary(
-          s"Invalid depth ${MaxDepth + 1} at position ${input.reader.position}, exceeding the limit of $MaxDepth nested values of recursive schemas"
-        )
-
-      input.depth += 1
-      val value = node.read(input)
-      input.depth -= 1
-      value
-    }
+    override def read(input: Input): AvroValue =
+      node.read(input)
   }
 
   private final class ConstantNode(val value: AvroValue) extends Node {
@@ -217,12 +218,14 @@ object BinaryDecoder {
 
   private final class RecordNode(fields: Array[Node]) extends Node {
     override def read(input: Input): AvroValue = {
+      input.enter()
       val values = new Array[AnyRef](fields.length)
       var index = 0
       while (index < fields.length) {
         values(index) = fields(index).read(input)
         index += 1
       }
+      input.exit()
       AvroValue.Record(Vector.from(arraySeq[AvroValue](values)))
     }
   }
@@ -249,31 +252,36 @@ object BinaryDecoder {
     protected def readItem(input: Input): A
 
     protected final def readItems(input: Input): Vector[A] = {
+      input.enter()
       var count = readBlockCount(input, minItemSize)
-      if (count == 0) Vector.empty
-      else {
-        val first = new Array[AnyRef](count.toInt)
-        var index = 0
-        while (index < first.length) {
-          first(index) = readItem(input)
-          index += 1
-        }
-
-        count = readBlockCount(input, minItemSize)
-        if (count == 0) Vector.from(arraySeq[A](first))
+      val items =
+        if (count == 0) Vector.empty
         else {
-          val items = Vector.newBuilder[A] ++= arraySeq[A](first)
-          while (count > 0) {
-            var index = 0L
-            while (index < count) {
-              items += readItem(input)
-              index += 1
+          val first = new Array[AnyRef](count.toInt)
+          var index = 0
+          while (index < first.length) {
+            first(index) = readItem(input)
+            index += 1
+          }
+
+          count = readBlockCount(input, minItemSize)
+          if (count == 0) Vector.from(arraySeq[A](first))
+          else {
+            val items = Vector.newBuilder[A] ++= arraySeq[A](first)
+            while (count > 0) {
+              var index = 0L
+              while (index < count) {
+                items += readItem(input)
+                index += 1
+              }
+              count = readBlockCount(input, minItemSize)
             }
-            count = readBlockCount(input, minItemSize)
+            items.result()
           }
-          items.result()
         }
-      }
+
+      input.exit()
+      items
     }
   }
 

modules/core/src/main/scala/vulcan/Codec.scala

@@ -87,10 +87,10 @@ sealed abstract class Codec[A] {
     * Returns a `BinaryDecoder` for Avro binary written using the provided
     * writer schema, or an error if the schema or writer schema is invalid.
     *
-    * To protect against invalid input, decoding fails for values nested
-    * more than 512 levels deep in recursive schemas, and for values with
-    * more than 1,000,000 array or map items which are encoded using zero
-    * bytes, like nulls and empty records.
+    * To protect against invalid input, decoding fails for values with
+    * records, arrays and maps nested more than 512 levels deep, and for
+    * values with more than 1,000,000 array or map items which are encoded
+    * using zero bytes, like nulls and empty records.
     */
   final def binaryDecoder(writerSchema: Schema): Either[AvroException, BinaryDecoder[A]] =
     (schema >> BinaryDecoder(writerSchema)).map { compiled =>
@@ -114,8 +114,8 @@ sealed abstract class Codec[A] {
     * Returns a `JsonDecoder` for Avro `Json` written using the provided
     * writer schema, or an error if the schema or writer schema is invalid.
     *
-    * To protect against invalid input, decoding fails for values nested
-    * more than 512 levels deep in recursive schemas.
+    * To protect against invalid input, decoding fails for values with
+    * records, arrays and maps nested more than 512 levels deep.
     */
   final def jsonDecoder(writerSchema: Schema): Either[AvroException, JsonDecoder[A]] =
     (schema >> JsonDecoder(writerSchema)).map { compiled =>

modules/core/src/main/scala/vulcan/JsonDecoder.scala

@@ -44,7 +44,7 @@ object JsonDecoder {
         fa.decode(input)
     }
 
-  /** The limit for the number of nested values of recursive schemas, to avoid stack overflows. */
+  /** The limit for nested records, arrays and maps, to avoid stack overflows. */
   private[vulcan] val MaxDepth: Int =
     512
 
@@ -58,7 +58,7 @@ object JsonDecoder {
       }
     }
 
-  /** Decodes values, where the depth is the number of nested values of recursive schemas. */
+  /** Decodes values, where the depth is the number of records, arrays and maps being decoded. */
   private sealed abstract class Node {
     def decode(json: Json, depth: Int): Either[AvroException, AvroValue]
   }
@@ -129,24 +129,15 @@ object JsonDecoder {
     }
   }
 
-  /**
-    * A named schema, which is being compiled if references are recursive. Since references
-    * are only used for recursive schemas, decoding them limits the depth of nested values.
-    */
+  /** A named schema, which is being compiled if references are recursive. */
   private final class Reference extends Node {
     var node: Node = null
 
     def resolved: Node =
       if (node != null) node else this
 
     override def decode(json: Json, depth: Int): Either[AvroException, AvroValue] =
-      if (depth == MaxDepth)
-        Left(
-          new InvalidJson(
-            s"Invalid depth ${MaxDepth + 1}, exceeding the limit of $MaxDepth nested values of recursive schemas"
-          )
-        )
-      else node.decode(json, depth + 1)
+      node.decode(json, depth)
   }
 
   private object NullNode extends Node {
@@ -213,6 +204,9 @@ object JsonDecoder {
 
     override def decode(json: Json, depth: Int): Either[AvroException, AvroValue] =
       json.asObject match {
+        case Some(_) if depth == MaxDepth =>
+          tooDeep
+
         case Some(obj) =>
           val builder = Codec.vectorBuilder[AvroValue](fields.length)
 
@@ -222,7 +216,7 @@ object JsonDecoder {
               val field = fields(index)
               field.value(obj) match {
                 case Some(fieldJson) =>
-                  field.node.decode(fieldJson, depth) match {
+                  field.node.decode(fieldJson, depth + 1) match {
                     case Right(value) =>
                       builder += value
                       loop(index + 1)
@@ -286,13 +280,16 @@ object JsonDecoder {
   private final class ArrayNode(items: Node) extends Node {
     override def decode(json: Json, depth: Int): Either[AvroException, AvroValue] =
       json.asArray match {
+        case Some(_) if depth == MaxDepth =>
+          tooDeep
+
         case Some(values) =>
           val builder = Codec.vectorBuilder[AvroValue](values.size)
 
           // Decodes values in a loop, so nested values use few stack frames
           @tailrec def loop(index: Int): Either[AvroException, AvroValue] =
             if (index < values.size)
-              items.decode(values(index), depth) match {
+              items.decode(values(index), depth + 1) match {
                 case Right(value) =>
                   builder += value
                   loop(index + 1)
@@ -312,6 +309,9 @@ object JsonDecoder {
   private final class MapNode(values: Node) extends Node {
     override def decode(json: Json, depth: Int): Either[AvroException, AvroValue] =
       json.asObject match {
+        case Some(_) if depth == MaxDepth =>
+          tooDeep
+
         case Some(obj) =>
           val entries = obj.toIterable.iterator
           val builder = Codec.vectorBuilder[(String, AvroValue)](obj.size)
@@ -320,7 +320,7 @@ object JsonDecoder {
           @tailrec def loop(): Either[AvroException, AvroValue] =
             if (entries.hasNext) {
               val entry = entries.next()
-              values.decode(entry._2, depth) match {
+              values.decode(entry._2, depth + 1) match {
                 case Right(value) =>
                   builder += ((entry._1, value))
                   loop()
@@ -380,6 +380,13 @@ object JsonDecoder {
   private def invalid(json: Json, schema: String): Either[AvroException, AvroValue] =
     Left(invalidError(json, schema))
 
+  private def tooDeep: Either[AvroException, AvroValue] =
+    Left(
+      new InvalidJson(
+        s"Invalid depth ${MaxDepth + 1}, exceeding the limit of $MaxDepth nested records, arrays and maps"
+      )
+    )
+
   private def invalidError(json: Json, schema: String): AvroException =
     new InvalidJson(s"Invalid ${describe(json)} for $schema")
 

tests/shared/src/test/scala/vulcan/BinaryValueSpec.scala

@@ -258,7 +258,7 @@ object BinaryValueSpec extends BaseSuite {
     )
   }
 
-  pureTest("BinaryEncoder and BinaryDecoder should limit the number of nested values of recursive schemas") {
+  pureTest("BinaryEncoder and BinaryDecoder should limit the depth of nested records, arrays and maps") {
     val limit = BinaryDecoder.MaxDepth
 
     def nodes(count: Int): ByteVector = {
@@ -272,12 +272,22 @@ object BinaryValueSpec extends BaseSuite {
       writer.toByteVector
     }
 
-    val error =
-      s"Invalid depth ${limit + 1} at position ${2 * (limit + 1)}, exceeding the limit of $limit nested values of recursive schemas"
+    // Records whose next record is within 300 nested arrays, encoded as a union index and block counts
+    val arrays = (1 to 300).foldLeft[Schema](Schema.Ref(name("Deep")))((items, _) => Schema.Array(items))
+    val deep =
+      Schema.Record(name("Deep"), Vector(Schema.Field("next", Schema.Union(Vector(Schema.Null, arrays)))))
 
-    matches(decode(node, nodes(limit + 1))) { case Right(_) => success } &&
-    expect.eql(Left(error), decode(node, nodes(limit + 2))) &&
-    expect.eql(Left(error), decode(node, nodes(100000)))
+    def deepRecords(count: Int): ByteVector =
+      ByteVector.fill(301L * (count - 1))(2) ++ ByteVector(0) ++ ByteVector.fill(300L * (count - 1))(0)
+
+    def error(position: Int): String =
+      s"Invalid depth ${limit + 1} at position $position, exceeding the limit of $limit nested records, arrays and maps"
+
+    matches(decode(node, nodes(limit))) { case Right(_) => success } &&
+    expect.eql(Left(error(2 * limit)), decode(node, nodes(limit + 1))) &&
+    expect.eql(Left(error(2 * limit)), decode(node, nodes(100000))) &&
+    matches(decode(deep, deepRecords(2))) { case Right(_) => success } &&
+    expect.eql(Left(error(limit)), decode(deep, deepRecords(50)))
   }
 
   pureTest("BinaryEncoder and BinaryDecoder should not decode records which always contain themselves") {

tests/shared/src/test/scala/vulcan/JsonValueSpec.scala

@@ -187,22 +187,35 @@ object JsonValueSpec extends BaseSuite {
     expect.eql(Left("Invalid null for union"), decode(Schema.Union(Vector(Schema.Int)), "null"))
   }
 
-  pureTest("JsonEncoder and JsonDecoder should limit the number of nested values of recursive schemas") {
+  pureTest("JsonEncoder and JsonDecoder should limit the depth of nested records, arrays and maps") {
     val limit = JsonDecoder.MaxDepth
 
     def nodes(count: Int): Json =
       (1 until count).foldLeft(Json.obj("value" -> Json.fromInt(0), "next" -> Json.Null)) { (next, _) =>
         Json.obj("value" -> Json.fromInt(0), "next" -> Json.obj("Node" -> next))
       }
 
-    val decoder = JsonDecoder(node)
+    val arrays = (1 to 300).foldLeft[Schema](Schema.Ref(name("Deep")))((items, _) => Schema.Array(items))
+    val deep =
+      Schema.Record(name("Deep"), Vector(Schema.Field("next", Schema.Union(Vector(Schema.Null, arrays)))))
+
+    def deepRecords(count: Int): Json =
+      (1 until count).foldLeft(Json.obj("next" -> Json.Null)) { (next, _) =>
+        val nested = (1 to 300).foldLeft(next)((items, _) => Json.arr(items))
+        Json.obj("next" -> Json.obj("array" -> nested))
+      }
+
+    def decodeJson(schema: Schema, json: Json): Either[String, AvroValue] =
+      JsonDecoder(schema).flatMap(_.decode(json)).left.map(_.message)
 
     val error =
-      s"Invalid depth ${limit + 1}, exceeding the limit of $limit nested values of recursive schemas"
+      s"Invalid depth ${limit + 1}, exceeding the limit of $limit nested records, arrays and maps"
 
-    matches(decoder.flatMap(_.decode(nodes(limit + 1)))) { case Right(_) => success } &&
-    expect.eql(Left(error), decoder.flatMap(_.decode(nodes(limit + 2))).left.map(_.message)) &&
-    expect.eql(Left(error), decoder.flatMap(_.decode(nodes(100000))).left.map(_.message))
+    matches(decodeJson(node, nodes(limit))) { case Right(_) => success } &&
+    expect.eql(Left(error), decodeJson(node, nodes(limit + 1))) &&
+    expect.eql(Left(error), decodeJson(node, nodes(100000))) &&
+    matches(decodeJson(deep, deepRecords(2))) { case Right(_) => success } &&
+    expect.eql(Left(error), decodeJson(deep, deepRecords(50)))
   }
 
   pureTest("JsonEncoder and JsonDecoder should not compile invalid schemas") {