diff --git a/build.sbt b/build.sbt index 60ea079a..fd5e069f 100644 --- a/build.sbt +++ b/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, diff --git a/modules/core/src/main/scala/vulcan/BinaryDecoder.scala b/modules/core/src/main/scala/vulcan/BinaryDecoder.scala index c69a56b5..667e9eae 100644 --- a/modules/core/src/main/scala/vulcan/BinaryDecoder.scala +++ b/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 } } diff --git a/modules/core/src/main/scala/vulcan/Codec.scala b/modules/core/src/main/scala/vulcan/Codec.scala index 0bc76faf..e998731f 100644 --- a/modules/core/src/main/scala/vulcan/Codec.scala +++ b/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 => diff --git a/modules/core/src/main/scala/vulcan/JsonDecoder.scala b/modules/core/src/main/scala/vulcan/JsonDecoder.scala index b1054022..807f3482 100644 --- a/modules/core/src/main/scala/vulcan/JsonDecoder.scala +++ b/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,10 +129,7 @@ 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 @@ -140,13 +137,7 @@ object JsonDecoder { 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") diff --git a/tests/shared/src/test/scala/vulcan/BinaryValueSpec.scala b/tests/shared/src/test/scala/vulcan/BinaryValueSpec.scala index d7d0dd8a..114b6783 100644 --- a/tests/shared/src/test/scala/vulcan/BinaryValueSpec.scala +++ b/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") { diff --git a/tests/shared/src/test/scala/vulcan/JsonValueSpec.scala b/tests/shared/src/test/scala/vulcan/JsonValueSpec.scala index ec94a5d3..b4fb1f24 100644 --- a/tests/shared/src/test/scala/vulcan/JsonValueSpec.scala +++ b/tests/shared/src/test/scala/vulcan/JsonValueSpec.scala @@ -187,7 +187,7 @@ 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 = @@ -195,14 +195,27 @@ object JsonValueSpec extends BaseSuite { 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") {