This is an automated email from the ASF dual-hosted git repository. iemejia pushed a commit to branch AVRO-4295-csharp-available-bytes in repository https://gitbox.apache.org/repos/asf/avro.git
commit 1ed2417604a362e7102f68d43f69247d97fe4191 Author: Ismaël Mejía <[email protected]> AuthorDate: Fri Aug 7 18:34:14 2026 +0200 AVRO-4295: [csharp] Apply collection allocation caps to all readers Commit ed310ad6b hardened only DefaultReader (used by GenericReader<T>). The other public readers -- GenericDatumReader<T>/SpecificDatumReader<T> (via PreresolvingDatumReader<T>), SpecificDefaultReader, and ReflectDefaultReader -- read array/map blocks without any bound, so a tiny payload could still decode a huge collection (e.g. 100,000,000 zero-byte elements from a few bytes) or preallocate an oversized array. Extract the shared guards (min-bytes-per-element, structural/item caps, and EnsureCollectionAvailable) into an internal CollectionBounds helper so the caps cannot drift between the reader implementations, and wire every reader's ReadArray/ReadMap/skip paths through it. The block count is read as a long and validated before the int cast, and PreresolvingDatumReader grows its backing array in bounded, geometric chunks so a huge count on a non-seekable stream cannot preallocate the whole block up front. The per-datum zero-byte-element budget is tracked in a thread-static nesting scope (mirroring the Java fix) rather than on the reader instance, preserving PreresolvingDatumReader's documented thread-sharing contract; nested scopes accumulate into the enclosing datum and only the outermost resets. Adds regression tests for the Generic, Specific, and Reflect reader paths, including a concurrency test for a shared reader. --- .../src/apache/main/Generic/CollectionBounds.cs | 259 +++++++++++++++++++++ .../src/apache/main/Generic/GenericReader.cs | 205 ++-------------- .../apache/main/Generic/PreresolvingDatumReader.cs | 83 ++++++- .../apache/main/Reflect/ReflectDefaultReader.cs | 17 +- .../src/apache/main/Specific/SpecificReader.cs | 16 +- lang/csharp/src/apache/test/IO/BinaryCodecTests.cs | 193 +++++++++++++++ lang/csharp/src/apache/test/Reflect/TestArray.cs | 14 ++ .../src/apache/test/Specific/SpecificTests.cs | 25 ++ 8 files changed, 610 insertions(+), 202 deletions(-) diff --git a/lang/csharp/src/apache/main/Generic/CollectionBounds.cs b/lang/csharp/src/apache/main/Generic/CollectionBounds.cs new file mode 100644 index 0000000000..d9566816fa --- /dev/null +++ b/lang/csharp/src/apache/main/Generic/CollectionBounds.cs @@ -0,0 +1,259 @@ +/* + * 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 + * + * https://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. + */ +using System; +using Avro.IO; + +namespace Avro.Generic +{ + /// <summary> + /// Shared allocation guards for decoding Avro collections (arrays and maps), + /// used by every generic/specific reader. Avro encodes a collection as one or + /// more blocks, each prefixed with an element count; a malicious or truncated + /// input can declare far more elements than the stream could ever hold, driving + /// an unbounded allocation from a tiny payload. These helpers reject such counts + /// before anything is allocated. The same logic backs both reader + /// implementations (<see cref="DefaultReader"/> and + /// <see cref="PreresolvingDatumReader{T}"/>) so the caps cannot drift apart. + /// </summary> + internal static class CollectionBounds + { + // Collection allocation limits, guarding against a block-count DoS. Both + // default to the same values as the other Avro SDKs and can be overridden + // (to a single value capping both) via the AVRO_MAX_COLLECTION_ITEMS + // environment variable. + internal static readonly long MaxCollectionItems = ReadCollectionLimit(10_000_000L); + + // The largest array the runtime can allocate. Mirrors + // BinaryDecoder.MaxDotNetArrayLength: the readers size .NET arrays from the + // (cumulative) block count, which throws (OutOfMemoryException/ + // OverflowException) above this length rather than a deterministic + // AvroException. +#if NETSTANDARD2_0 + private const int MaxDotNetArrayLength = 0x3FFFFFFF; +#else + private const int MaxDotNetArrayLength = 0x7FFFFFC7; +#endif + + // The structural cap is additionally clamped to the runtime's maximum + // array length: the callers cast the (cumulative) block count to int to + // size .NET collections, and a limit above the max array length (e.g. from + // a large env override, or int.MaxValue itself) would let a collection + // that passes EnsureCollectionAvailable still fault inside Array.Resize + // instead of failing deterministically. + internal static readonly long MaxCollectionStructural = + Math.Min(ReadCollectionLimit(2147483639L), MaxDotNetArrayLength); + + // Upper bound on how many elements the backing array is grown by in a + // single step while decoding. The array still grows to hold every element + // actually read; this only avoids resizing to the full (possibly + // attacker-declared) block count up front, before any element is read. + // That matters most for non-seekable streams, where the bytes-available + // check cannot bound the declared count, so a single resize to the block + // count could allocate a huge array before the truncated stream is + // detected. + internal const int MaxCollectionPrealloc = 1024; + + private static long ReadCollectionLimit(long defaultValue) + { + string env = Environment.GetEnvironmentVariable("AVRO_MAX_COLLECTION_ITEMS"); + if (!string.IsNullOrEmpty(env) && long.TryParse(env, out long value) && value >= 0) + { + return value; + } + + return defaultValue; + } + + // Per-thread, per-datum cumulative count of zero-byte-encoded collection + // elements (e.g. an array of nulls). Such elements consume no input, so + // the bytes-remaining check cannot bound them, and a per-collection cap is + // not enough either: a record's schema can declare many zero-byte + // collection fields, each block under the limit but jointly unbounded. The + // budget is therefore cumulative across a whole datum. It is thread-static, + // not reader instance state, because a resolved reader may be reused or + // shared among threads (see PreresolvingDatumReader), which such an + // instance field would make unsafe. + [ThreadStatic] private static long zeroByteItemsRead; + + // Nesting depth of the active decode scope on this thread. A delegated + // reader or a skipped writer field decodes within the enclosing datum's + // scope and accumulates into its budget; only the outermost scope resets + // the running total. + [ThreadStatic] private static int scopeDepth; + + /// <summary> + /// Opens a decode scope bounding the cumulative zero-byte-element + /// allocation for the current datum. Scopes nest: a nested scope (a + /// delegated reader or a skipped field) accumulates into the enclosing + /// datum's budget, and only the outermost scope resets the running total, + /// so the cap applies across the whole datum rather than per collection. + /// Dispose the returned scope (via <c>using</c>) once the datum is decoded; + /// the budget is thread-static, so the scope must be closed on the same + /// thread, and it is always closed so state cannot leak into later decodes. + /// </summary> + internal static Scope EnterScope() + { + if (scopeDepth == 0) + { + zeroByteItemsRead = 0; + } + + scopeDepth++; + return default; + } + + /// <summary> + /// The disposable returned by <see cref="EnterScope"/>. A stateless struct + /// so <c>using</c> incurs no allocation; closing the outermost scope resets + /// the per-datum budget. + /// </summary> + internal readonly struct Scope : IDisposable + { + /// <inheritdoc/> + public void Dispose() + { + if (--scopeDepth == 0) + { + zeroByteItemsRead = 0; + } + } + } + + /// <summary> + /// Minimum number of bytes a single value of the given schema can occupy + /// on the wire. Used to reject an array/map block count that could not be + /// backed by the bytes remaining. A type that encodes to zero bytes + /// returns 0 (not only <c>null</c>, but also composites that encode to + /// nothing, e.g. a record whose fields are all zero-byte), which disables + /// the bytes-remaining check for it (so an array of such elements is not + /// falsely rejected; they are instead bounded by the zero-byte item cap). + /// A depth limit breaks self-referencing schemas. + /// </summary> + internal static int MinBytesPerElement(Schema schema, int depth = 0) + { + if (schema == null) + { + return 0; + } + + switch (schema.Tag) + { + case Schema.Type.Null: + return 0; + case Schema.Type.Float: + return 4; + case Schema.Type.Double: + return 8; + case Schema.Type.Fixed: + return ((FixedSchema)schema).Size; + case Schema.Type.Record: + case Schema.Type.Error: + if (depth > 64) + { + // A cyclic or pathologically deep record. Return 1 (not + // 0) so the collection check stays enabled; a valid + // recursive value always encodes to >= 1 byte. The depth + // guard is applied only here, so zero-byte leaf types + // such as null still return 0 regardless of depth. + return 1; + } + + // Accumulate in a long and clamp so a deeply nested schema + // cannot overflow int into a value <= 0, which would disable + // the collection check. + long total = 0; + foreach (Field f in (RecordSchema)schema) + { + total += MinBytesPerElement(f.Schema, depth + 1); + if (total >= int.MaxValue) + { + return int.MaxValue; + } + } + + return (int)total; + default: + // boolean, int, long, bytes, string, enum, union, array, map: + // all encode to at least one byte. + return 1; + } + } + + /// <summary> + /// Rejects a collection (array or map) block that could drive an unbounded + /// allocation, before allocating for it. A block whose declared element + /// count could not be backed by the bytes actually remaining is rejected; + /// zero-byte element blocks (where the bytes-remaining check does not + /// apply) are bounded by a cumulative item cap; and every collection is + /// bounded by a structural cap. Returns the running total across blocks. + /// </summary> + /// <param name="d">Decoder the collection is being read from.</param> + /// <param name="total">Running element total across the blocks decoded so far for this collection.</param> + /// <param name="count">Element count declared by the current block.</param> + /// <param name="minBytesPerElement">Minimum on-wire size of one element (see <see cref="MinBytesPerElement"/>).</param> + internal static long EnsureCollectionAvailable(Decoder d, long total, long count, long minBytesPerElement) + { + // A negative count is corrupt/malicious data (it can also arise from + // long.MinValue overflow when negating a negative block count), and + // the callers cast the block count to int; reject it explicitly. + if (count < 0) + { + throw new AvroException($"Invalid negative collection block count: {count}"); + } + + // Reject before adding so an oversized block count cannot overflow + // `total` (wrapping it negative and bypassing the caps below). The + // running total is always <= MaxCollectionStructural on entry (the + // invariant this method maintains) and count >= 0, so the subtraction + // cannot underflow or overflow. + if (count > MaxCollectionStructural - total) + { + throw new AvroException( + $"Collection size {total} + {count} exceeds the maximum allowed size of {MaxCollectionStructural}"); + } + + total += count; + + if (minBytesPerElement <= 0) + { + // Zero-byte elements (e.g. null) consume no input, so the + // bytes-remaining check cannot bound them. Cap the cumulative + // count across the whole datum, not just this collection: a + // record's schema can declare many zero-byte collection fields, + // each block under the limit but jointly unbounded. + zeroByteItemsRead += count; + if (zeroByteItemsRead > MaxCollectionItems) + { + throw new AvroException( + $"Collection of zero-byte elements ({zeroByteItemsRead}) exceeds the maximum allowed size of {MaxCollectionItems}"); + } + } + else if (d is BinaryDecoder bd) + { + long remaining = bd.RemainingBytes(); + if (remaining >= 0 && count > remaining / minBytesPerElement) + { + throw new AvroException( + $"Collection claims {count} elements with at least {minBytesPerElement} bytes each, but only {remaining} bytes are available"); + } + } + + return total; + } + } +} diff --git a/lang/csharp/src/apache/main/Generic/GenericReader.cs b/lang/csharp/src/apache/main/Generic/GenericReader.cs index 503ea82e82..8163c43075 100644 --- a/lang/csharp/src/apache/main/Generic/GenericReader.cs +++ b/lang/csharp/src/apache/main/Generic/GenericReader.cs @@ -110,15 +110,6 @@ namespace Avro.Generic /// </summary> public Schema WriterSchema { get; private set; } - // Cumulative number of zero-byte-encoded collection elements (e.g. an - // array of nulls) seen while decoding the current datum. Reset per - // top-level Read. Such elements consume no input, so the bytes-remaining - // check cannot bound them, and a per-collection cap is not enough either: - // a record's schema can declare many collection fields, each block under - // the limit but jointly unbounded. The cap is therefore applied across - // the whole datum. See EnsureCollectionAvailable. - private long zeroByteItemsRead; - /// <summary> /// Constructs the default reader for the given schemas using the DefaultReader. If the @@ -146,11 +137,13 @@ namespace Avro.Generic /// <returns>Object read from the decoder.</returns> public T Read<T>(T reuse, Decoder decoder) { - // Start a fresh zero-byte-element budget for this datum. The cap is + // Open a fresh zero-byte-element budget for this datum. The cap is // cumulative across every collection decoded in this datum (see - // EnsureCollectionAvailable), not per collection. - zeroByteItemsRead = 0; - return (T)Read(reuse, WriterSchema, ReaderSchema, decoder); + // CollectionBounds.EnsureCollectionAvailable), not per collection. + using (CollectionBounds.EnterScope()) + { + return (T)Read(reuse, WriterSchema, ReaderSchema, decoder); + } } /// <summary> @@ -417,7 +410,7 @@ namespace Avro.Generic ArraySchema rs = (ArraySchema)readerSchema; object result = CreateArray(reuse, rs); int i = 0; - long minBytes = MinBytesPerElement(writerSchema.ItemSchema); + long minBytes = CollectionBounds.MinBytesPerElement(writerSchema.ItemSchema); long total = 0; for (long nl = d.ReadArrayStart(); nl != 0; nl = d.ReadArrayNext()) { @@ -425,7 +418,7 @@ namespace Avro.Generic // bytes remaining (or, for zero-byte elements, that exceeds the // item cap) before allocating for it. Checked on the raw long, // which also avoids the int cast below overflowing. - total = EnsureCollectionAvailable(d, total, nl, minBytes); + total = CollectionBounds.EnsureCollectionAvailable(d, total, nl, minBytes); int n = (int)nl; // Preallocate only a bounded amount up front, then grow on demand // below. On a non-seekable stream EnsureCollectionAvailable cannot @@ -435,7 +428,7 @@ namespace Avro.Generic // the cap keep the original single-resize fast path. Compute in // long and clamp so a large i near the structural cap cannot // overflow the int sum. - long preallocLong = Math.Min((long)i + Math.Min(n, MaxCollectionPrealloc), MaxCollectionStructural); + long preallocLong = Math.Min((long)i + Math.Min(n, CollectionBounds.MaxCollectionPrealloc), CollectionBounds.MaxCollectionStructural); int prealloc = (int)preallocLong; if (GetArraySize(result) < prealloc) ResizeArray(ref result, prealloc); for (int j = 0; j < n; j++, i++) @@ -455,9 +448,9 @@ namespace Avro.Generic grown = i + 1; } - if (grown > MaxCollectionStructural) + if (grown > CollectionBounds.MaxCollectionStructural) { - grown = MaxCollectionStructural; + grown = CollectionBounds.MaxCollectionStructural; } ResizeArray(ref result, (int)grown); @@ -545,11 +538,11 @@ namespace Avro.Generic MapSchema rs = (MapSchema)readerSchema; object result = CreateMap(reuse, rs); // Map keys are strings (>= 1 byte length prefix) plus the value. - long minBytes = 1L + MinBytesPerElement(writerSchema.ValueSchema); + long minBytes = 1L + CollectionBounds.MinBytesPerElement(writerSchema.ValueSchema); long total = 0; for (long nl = d.ReadMapStart(); nl != 0; nl = d.ReadMapNext()) { - total = EnsureCollectionAvailable(d, total, nl, minBytes); + total = CollectionBounds.EnsureCollectionAvailable(d, total, nl, minBytes); int n = (int)nl; for (int j = 0; j < n; j++) { @@ -560,170 +553,6 @@ namespace Avro.Generic return result; } - /// <summary> - /// Minimum number of bytes a single value of the given schema can occupy - /// on the wire. Used to reject an array/map block count that could not be - /// backed by the bytes remaining. A type that encodes to zero bytes - /// returns 0 (not only <c>null</c>, but also composites that encode to - /// nothing, e.g. a record whose fields are all zero-byte), which disables - /// the bytes-remaining check for it (so an array of such elements is not - /// falsely rejected; they are instead bounded by the zero-byte item cap). - /// A depth limit breaks self-referencing schemas. - /// </summary> - private static int MinBytesPerElement(Schema schema, int depth = 0) - { - if (schema == null) - { - return 0; - } - - switch (schema.Tag) - { - case Schema.Type.Null: - return 0; - case Schema.Type.Float: - return 4; - case Schema.Type.Double: - return 8; - case Schema.Type.Fixed: - return ((FixedSchema)schema).Size; - case Schema.Type.Record: - case Schema.Type.Error: - if (depth > 64) - { - // A cyclic or pathologically deep record. Return 1 (not - // 0) so the collection check stays enabled; a valid - // recursive value always encodes to >= 1 byte. The depth - // guard is applied only here, so zero-byte leaf types - // such as null still return 0 regardless of depth. - return 1; - } - - // Accumulate in a long and clamp so a deeply nested schema - // cannot overflow int into a value <= 0, which would disable - // the collection check. - long total = 0; - foreach (Field f in (RecordSchema)schema) - { - total += MinBytesPerElement(f.Schema, depth + 1); - if (total >= int.MaxValue) - { - return int.MaxValue; - } - } - - return (int)total; - default: - // boolean, int, long, bytes, string, enum, union, array, map: - // all encode to at least one byte. - return 1; - } - } - - // Collection allocation limits, guarding against a block-count DoS. Both - // default to the same values as the other Avro SDKs and can be overridden - // (to a single value capping both) via the AVRO_MAX_COLLECTION_ITEMS - // environment variable. - private static readonly long MaxCollectionItems = ReadCollectionLimit(10_000_000L); - - // The largest array the runtime can allocate. Mirrors - // BinaryDecoder.MaxDotNetArrayLength: the default reader grows its backing - // array via Array.Resize, which throws (OutOfMemoryException/OverflowException) - // above this length rather than a deterministic AvroException. -#if NETSTANDARD2_0 - private const int MaxDotNetArrayLength = 0x3FFFFFFF; -#else - private const int MaxDotNetArrayLength = 0x7FFFFFC7; -#endif - - // The structural cap is additionally clamped to the runtime's maximum - // array length: the callers cast the (cumulative) block count to int to - // size .NET collections, and a limit above the max array length (e.g. from - // a large env override, or int.MaxValue itself) would let a collection - // that passes EnsureCollectionAvailable still fault inside Array.Resize - // instead of failing deterministically. - private static readonly long MaxCollectionStructural = - Math.Min(ReadCollectionLimit(2147483639L), MaxDotNetArrayLength); - - // Upper bound on how many elements the backing array is grown by in a - // single step while decoding. The array still grows to hold every element - // actually read; this only avoids resizing to the full (possibly - // attacker-declared) block count up front, before any element is read. - // That matters most for non-seekable streams, where the bytes-available - // check cannot bound the declared count, so a single Array.Resize to the - // block count could allocate a huge array before the truncated stream is - // detected. - private const int MaxCollectionPrealloc = 1024; - - private static long ReadCollectionLimit(long defaultValue) - { - string env = Environment.GetEnvironmentVariable("AVRO_MAX_COLLECTION_ITEMS"); - if (!string.IsNullOrEmpty(env) && long.TryParse(env, out long value) && value >= 0) - { - return value; - } - - return defaultValue; - } - - /// <summary> - /// Rejects a collection (array or map) block that could drive an unbounded - /// allocation, before allocating for it. A block whose declared element - /// count could not be backed by the bytes actually remaining is rejected; - /// zero-byte element blocks (where the bytes-remaining check does not - /// apply) are bounded by a cumulative item cap; and every collection is - /// bounded by a structural cap. Returns the running total across blocks. - /// </summary> - private long EnsureCollectionAvailable(Decoder d, long total, long count, long minBytesPerElement) - { - // A negative count is corrupt/malicious data (it can also arise from - // long.MinValue overflow when negating a negative block count), and - // the callers cast the block count to int; reject it explicitly. - if (count < 0) - { - throw new AvroException($"Invalid negative collection block count: {count}"); - } - - // Reject before adding so an oversized block count cannot overflow - // `total` (wrapping it negative and bypassing the caps below). The - // running total is always <= MaxCollectionStructural on entry (the - // invariant this method maintains) and count >= 0, so the subtraction - // cannot underflow or overflow. - if (count > MaxCollectionStructural - total) - { - throw new AvroException( - $"Collection size {total} + {count} exceeds the maximum allowed size of {MaxCollectionStructural}"); - } - - total += count; - - if (minBytesPerElement <= 0) - { - // Zero-byte elements (e.g. null) consume no input, so the - // bytes-remaining check cannot bound them. Cap the cumulative - // count across the whole datum, not just this collection: a - // record's schema can declare many zero-byte collection fields, - // each block under the limit but jointly unbounded. - zeroByteItemsRead += count; - if (zeroByteItemsRead > MaxCollectionItems) - { - throw new AvroException( - $"Collection of zero-byte elements ({zeroByteItemsRead}) exceeds the maximum allowed size of {MaxCollectionItems}"); - } - } - else if (d is BinaryDecoder bd) - { - long remaining = bd.RemainingBytes(); - if (remaining >= 0 && count > remaining / minBytesPerElement) - { - throw new AvroException( - $"Collection claims {count} elements with at least {minBytesPerElement} bytes each, but only {remaining} bytes are available"); - } - } - - return total; - } - /// <summary> /// Used by the default implementation of ReadMap() to create a fresh map object. The default /// implementation of this method returns a IDictionary<string, map>. @@ -884,11 +713,11 @@ namespace Avro.Generic case Schema.Type.Array: { Schema s = (writerSchema as ArraySchema).ItemSchema; - long minBytes = MinBytesPerElement(s); + long minBytes = CollectionBounds.MinBytesPerElement(s); long total = 0; for (long n = d.ReadArrayStart(); n != 0; n = d.ReadArrayNext()) { - total = EnsureCollectionAvailable(d, total, n, minBytes); + total = CollectionBounds.EnsureCollectionAvailable(d, total, n, minBytes); for (long i = 0; i < n; i++) Skip(s, d); } } @@ -896,11 +725,11 @@ namespace Avro.Generic case Schema.Type.Map: { Schema s = (writerSchema as MapSchema).ValueSchema; - long minBytes = 1L + MinBytesPerElement(s); + long minBytes = 1L + CollectionBounds.MinBytesPerElement(s); long total = 0; for (long n = d.ReadMapStart(); n != 0; n = d.ReadMapNext()) { - total = EnsureCollectionAvailable(d, total, n, minBytes); + total = CollectionBounds.EnsureCollectionAvailable(d, total, n, minBytes); for (long i = 0; i < n; i++) { d.SkipString(); Skip(s, d); } } } diff --git a/lang/csharp/src/apache/main/Generic/PreresolvingDatumReader.cs b/lang/csharp/src/apache/main/Generic/PreresolvingDatumReader.cs index 53270faecd..73133f382f 100644 --- a/lang/csharp/src/apache/main/Generic/PreresolvingDatumReader.cs +++ b/lang/csharp/src/apache/main/Generic/PreresolvingDatumReader.cs @@ -15,6 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ +using System; using System.Collections.Generic; using System.IO; using Avro.IO; @@ -69,7 +70,15 @@ namespace Avro.Generic /// <inheritdoc/> public T Read(T reuse, Decoder decoder) { - return (T)_reader(reuse, decoder); + // Open a fresh zero-byte-element budget for this datum. The cap is + // cumulative across every collection decoded in this datum (see + // CollectionBounds.EnsureCollectionAvailable), not per collection. The + // budget is thread-static, so this reader stays safe to share among + // threads as documented. + using (CollectionBounds.EnterScope()) + { + return (T)_reader(reuse, decoder); + } } /// <summary> @@ -364,15 +373,24 @@ namespace Avro.Generic var reader = ResolveReader(ws, rs); var mapAccess = GetMapAccess(readerSchema); - return (r,d) => ReadMap(r, d, mapAccess, reader); + // Map keys are strings (>= 1 byte length prefix) plus the value. + long valueMinBytes = 1L + CollectionBounds.MinBytesPerElement(ws); + return (r,d) => ReadMap(r, d, mapAccess, reader, valueMinBytes); } - private object ReadMap(object reuse, Decoder decoder, MapAccess mapAccess, ReadItem valueReader) + private object ReadMap(object reuse, Decoder decoder, MapAccess mapAccess, ReadItem valueReader, long valueMinBytes) { object map = mapAccess.Create(reuse); - for (int n = (int)decoder.ReadMapStart(); n != 0; n = (int)decoder.ReadMapNext()) + long total = 0; + for (long nl = decoder.ReadMapStart(); nl != 0; nl = decoder.ReadMapNext()) { + // Reject a block whose element count could not be backed by the + // bytes remaining (or, for zero-byte elements, that exceeds the + // item cap) before allocating for it. Checked on the raw long, + // which also avoids the int cast below overflowing. + total = CollectionBounds.EnsureCollectionAvailable(decoder, total, nl, valueMinBytes); + int n = (int)nl; mapAccess.AddElements(map, n, valueReader, decoder, false); } return map; @@ -383,18 +401,57 @@ namespace Avro.Generic var itemReader = ResolveReader(writerSchema.ItemSchema, readerSchema.ItemSchema); var arrayAccess = GetArrayAccess(readerSchema); - return (r, d) => ReadArray(r, d, arrayAccess, itemReader, IsReusable(readerSchema.ItemSchema.Tag)); + long itemMinBytes = CollectionBounds.MinBytesPerElement(writerSchema.ItemSchema); + return (r, d) => ReadArray(r, d, arrayAccess, itemReader, IsReusable(readerSchema.ItemSchema.Tag), itemMinBytes); } - private object ReadArray(object reuse, Decoder decoder, ArrayAccess arrayAccess, ReadItem itemReader, bool itemReusable) + private object ReadArray(object reuse, Decoder decoder, ArrayAccess arrayAccess, ReadItem itemReader, bool itemReusable, long itemMinBytes) { object array = arrayAccess.Create(reuse); int i = 0; - for (int n = (int)decoder.ReadArrayStart(); n != 0; n = (int)decoder.ReadArrayNext()) + // Capacity we have requested from arrayAccess.EnsureSize so far. The + // block is read in bounded chunks that grow this geometrically, so a + // huge declared count on a non-seekable stream (where the + // bytes-remaining check cannot bound it) does not preallocate the + // whole block before any element is read; a truncated stream instead + // faults after a bounded growth. + int capacity = 0; + long total = 0; + for (long nl = decoder.ReadArrayStart(); nl != 0; nl = decoder.ReadArrayNext()) { - arrayAccess.EnsureSize(ref array, i + n); - arrayAccess.AddElements(array, n, i, itemReader, decoder, itemReusable); - i += n; + total = CollectionBounds.EnsureCollectionAvailable(decoder, total, nl, itemMinBytes); + int n = (int)nl; + int remaining = n; + while (remaining > 0) + { + int chunk = Math.Min(remaining, CollectionBounds.MaxCollectionPrealloc); + int needed = i + chunk; + if (capacity < needed) + { + // Grow ~1.5x (amortized O(n), so a legitimate large array + // is not resized on every chunk) plus one chunk, then + // clamp to the structural cap (which is <= the runtime's + // max array length). The validated element count never + // exceeds that cap. + long grown = (long)capacity + (capacity >> 1) + CollectionBounds.MaxCollectionPrealloc; + if (grown < needed) + { + grown = needed; + } + + if (grown > CollectionBounds.MaxCollectionStructural) + { + grown = CollectionBounds.MaxCollectionStructural; + } + + capacity = (int)grown; + arrayAccess.EnsureSize(ref array, capacity); + } + + arrayAccess.AddElements(array, chunk, i, itemReader, decoder, itemReusable); + i += chunk; + remaining -= chunk; + } } arrayAccess.Resize(ref array, i); return array; @@ -486,20 +543,26 @@ namespace Avro.Generic return d => d.SkipFixed(size); case Schema.Type.Array: var itemSkip = GetSkip(((ArraySchema)writerSchema).ItemSchema); + var arrayItemMinBytes = CollectionBounds.MinBytesPerElement(((ArraySchema)writerSchema).ItemSchema); return d => { + long total = 0; for (long n = d.ReadArrayStart(); n != 0; n = d.ReadArrayNext()) { + total = CollectionBounds.EnsureCollectionAvailable(d, total, n, arrayItemMinBytes); for (long i = 0; i < n; i++) itemSkip(d); } }; case Schema.Type.Map: { var valueSkip = GetSkip(((MapSchema)writerSchema).ValueSchema); + var mapValueMinBytes = 1L + CollectionBounds.MinBytesPerElement(((MapSchema)writerSchema).ValueSchema); return d => { + long total = 0; for (long n = d.ReadMapStart(); n != 0; n = d.ReadMapNext()) { + total = CollectionBounds.EnsureCollectionAvailable(d, total, n, mapValueMinBytes); for (long i = 0; i < n; i++) { d.SkipString(); valueSkip(d); } } }; diff --git a/lang/csharp/src/apache/main/Reflect/ReflectDefaultReader.cs b/lang/csharp/src/apache/main/Reflect/ReflectDefaultReader.cs index 034cb89f88..2a4a400665 100644 --- a/lang/csharp/src/apache/main/Reflect/ReflectDefaultReader.cs +++ b/lang/csharp/src/apache/main/Reflect/ReflectDefaultReader.cs @@ -20,6 +20,7 @@ using System; using System.Collections; using System.Collections.Generic; using System.Globalization; +using Avro.Generic; using Avro.IO; using Avro.Specific; using Newtonsoft.Json.Linq; @@ -496,8 +497,16 @@ namespace Avro.Reflect } int i = 0; - for (int n = (int)dec.ReadArrayStart(); n != 0; n = (int)dec.ReadArrayNext()) + long minBytes = CollectionBounds.MinBytesPerElement(writerSchema.ItemSchema); + long total = 0; + for (long nl = dec.ReadArrayStart(); nl != 0; nl = dec.ReadArrayNext()) { + // Reject a block whose element count could not be backed by the + // bytes remaining (or, for zero-byte elements, that exceeds the + // cumulative item cap) before allocating for it. Checked on the + // raw long, which also avoids the int cast below overflowing. + total = CollectionBounds.EnsureCollectionAvailable(dec, total, nl, minBytes); + int n = (int)nl; for (int j = 0; j < n; j++, i++) { arrayHelper.Add(Read(null, writerSchema.ItemSchema, rs.ItemSchema, dec)); @@ -532,8 +541,12 @@ namespace Avro.Reflect map = (System.Collections.IDictionary)Activator.CreateInstance(GetTypeFromSchema(rs, false)); } - for (int n = (int)d.ReadMapStart(); n != 0; n = (int)d.ReadMapNext()) + long minBytes = 1L + CollectionBounds.MinBytesPerElement(writerSchema.ValueSchema); + long total = 0; + for (long nl = d.ReadMapStart(); nl != 0; nl = d.ReadMapNext()) { + total = CollectionBounds.EnsureCollectionAvailable(d, total, nl, minBytes); + int n = (int)nl; for (int j = 0; j < n; j++) { string k = d.ReadString(); diff --git a/lang/csharp/src/apache/main/Specific/SpecificReader.cs b/lang/csharp/src/apache/main/Specific/SpecificReader.cs index 1019fa36ce..d6223f00a2 100644 --- a/lang/csharp/src/apache/main/Specific/SpecificReader.cs +++ b/lang/csharp/src/apache/main/Specific/SpecificReader.cs @@ -213,8 +213,16 @@ namespace Avro.Specific array = ObjectCreator.Instance.New(getTargetType(readerSchema), Schema.Type.Array) as System.Collections.IList; int i = 0; - for (int n = (int)dec.ReadArrayStart(); n != 0; n = (int)dec.ReadArrayNext()) + long minBytes = CollectionBounds.MinBytesPerElement(writerSchema.ItemSchema); + long total = 0; + for (long nl = dec.ReadArrayStart(); nl != 0; nl = dec.ReadArrayNext()) { + // Reject a block whose element count could not be backed by the + // bytes remaining (or, for zero-byte elements, that exceeds the + // cumulative item cap) before allocating for it. Checked on the + // raw long, which also avoids the int cast below overflowing. + total = CollectionBounds.EnsureCollectionAvailable(dec, total, nl, minBytes); + int n = (int)nl; for (int j = 0; j < n; j++, i++) array.Add(Read(null, writerSchema.ItemSchema, rs.ItemSchema, dec)); } @@ -245,8 +253,12 @@ namespace Avro.Specific else map = ObjectCreator.Instance.New(getTargetType(readerSchema), Schema.Type.Map) as System.Collections.IDictionary; - for (int n = (int)d.ReadMapStart(); n != 0; n = (int)d.ReadMapNext()) + long minBytes = 1L + CollectionBounds.MinBytesPerElement(writerSchema.ValueSchema); + long total = 0; + for (long nl = d.ReadMapStart(); nl != 0; nl = d.ReadMapNext()) { + total = CollectionBounds.EnsureCollectionAvailable(d, total, nl, minBytes); + int n = (int)nl; for (int j = 0; j < n; j++) { string k = d.ReadString(); diff --git a/lang/csharp/src/apache/test/IO/BinaryCodecTests.cs b/lang/csharp/src/apache/test/IO/BinaryCodecTests.cs index e8e86929ab..4c5e0acb30 100644 --- a/lang/csharp/src/apache/test/IO/BinaryCodecTests.cs +++ b/lang/csharp/src/apache/test/IO/BinaryCodecTests.cs @@ -784,6 +784,199 @@ namespace Avro.Test } } + // C# has a second reader implementation, GenericDatumReader<T> (based on + // PreresolvingDatumReader<T>), which is public and used directly by + // callers. It must enforce the same collection-allocation caps as the + // DefaultReader used above; the following tests exercise it independently. + + // A zero-byte element type (null) consumes no input, so the + // bytes-remaining check cannot bound its block count: a tiny payload + // declaring a huge count must be rejected by the item cap before building + // the array, rather than decoding e.g. 100,000,000 elements from 5 bytes. + [Test] + public void TestDatumReaderReadArrayOfNullRejectsHugeCount() + { + var schema = Avro.Schema.Parse("{\"type\":\"array\",\"items\":\"null\"}"); + var ms = new MemoryStream(); + new BinaryEncoder(ms).WriteLong(100_000_000); // well over the zero-byte item cap + ms.Position = 0; + var reader = new GenericDatumReader<object>(schema, schema); + Assert.Throws<AvroException>(() => reader.Read(null, new BinaryDecoder(ms))); + } + + // A non-zero-byte element block whose declared count could not be backed + // by the bytes remaining must be rejected before allocating. + [Test] + public void TestDatumReaderReadArrayRejectsCountBeyondStream() + { + var schema = Avro.Schema.Parse("{\"type\":\"array\",\"items\":\"long\"}"); + var ms = new MemoryStream(); + new BinaryEncoder(ms).WriteLong(1000000); // 1,000,000 longs, no data + ms.Position = 0; + var reader = new GenericDatumReader<object>(schema, schema); + Assert.Throws<AvroException>(() => reader.Read(null, new BinaryDecoder(ms))); + } + + // The complement: a legitimate array of nulls under the cap still decodes. + [Test] + public void TestDatumReaderReadArrayOfNullNotFalselyRejected() + { + var schema = Avro.Schema.Parse("{\"type\":\"array\",\"items\":\"null\"}"); + var ms = new MemoryStream(); + var enc = new BinaryEncoder(ms); + enc.WriteLong(100000); // one block of 100,000 nulls (zero bytes each) + enc.WriteLong(0); // end-of-array marker + ms.Position = 0; + var reader = new GenericDatumReader<object>(schema, schema); + var result = (object[])reader.Read(null, new BinaryDecoder(ms)); + Assert.AreEqual(100000, result.Length); + } + + [Test] + public void TestDatumReaderReadMapRejectsCountBeyondStream() + { + var schema = Avro.Schema.Parse("{\"type\":\"map\",\"values\":\"long\"}"); + var ms = new MemoryStream(); + new BinaryEncoder(ms).WriteLong(1000000); + ms.Position = 0; + var reader = new GenericDatumReader<object>(schema, schema); + Assert.Throws<AvroException>(() => reader.Read(null, new BinaryDecoder(ms))); + } + + // The zero-byte item cap is cumulative across a decoded datum, not per + // collection: a record declaring many array<null> fields, each block under + // the limit but jointly over it, must be rejected before allocating the + // offending field. + [Test] + public void TestDatumReaderRecordOfNullArrayFieldsRejectedCumulatively() + { + var schema = Avro.Schema.Parse( + "{\"type\":\"record\",\"name\":\"R\",\"fields\":[" + + "{\"name\":\"a\",\"type\":{\"type\":\"array\",\"items\":\"null\"}}," + + "{\"name\":\"b\",\"type\":{\"type\":\"array\",\"items\":\"null\"}}]}"); + var ms = new MemoryStream(); + var enc = new BinaryEncoder(ms); + enc.WriteLong(1); // field a: one null + enc.WriteLong(0); // end of a + enc.WriteLong(10_000_000); // field b: cumulative 10,000,001 > cap; rejected before allocating + enc.WriteLong(0); // end of b (not reached) + ms.Position = 0; + var reader = new GenericDatumReader<object>(schema, schema); + Assert.Throws<AvroException>(() => reader.Read(null, new BinaryDecoder(ms))); + } + + // The complement: jointly-under-the-cap fields decode, and the per-datum + // budget resets between datums so reusing the reader does not accumulate. + [Test] + public void TestDatumReaderRecordOfNullArrayFieldsWithinLimitReads() + { + var schema = Avro.Schema.Parse( + "{\"type\":\"record\",\"name\":\"R\",\"fields\":[" + + "{\"name\":\"a\",\"type\":{\"type\":\"array\",\"items\":\"null\"}}," + + "{\"name\":\"b\",\"type\":{\"type\":\"array\",\"items\":\"null\"}}]}"); + var ms = new MemoryStream(); + var enc = new BinaryEncoder(ms); + enc.WriteLong(3); enc.WriteLong(0); // field a: 3 nulls + enc.WriteLong(3); enc.WriteLong(0); // field b: 3 nulls + var reader = new GenericDatumReader<object>(schema, schema); + for (int i = 0; i < 2; i++) + { + ms.Position = 0; + var rec = (GenericRecord)reader.Read(null, new BinaryDecoder(ms)); + Assert.AreEqual(3, ((object[])rec["a"]).Length); + Assert.AreEqual(3, ((object[])rec["b"]).Length); + } + } + + // The skip path (a writer field absent from the reader schema) must be + // bounded too, so skipping a huge zero-byte block cannot loop endlessly. + [Test] + public void TestDatumReaderSkipArrayOfNullRejectsHugeCount() + { + var writer = Avro.Schema.Parse( + "{\"type\":\"record\",\"name\":\"Foo\",\"fields\":[" + + "{\"name\":\"arr\",\"type\":{\"type\":\"array\",\"items\":\"null\"}}," + + "{\"name\":\"val\",\"type\":\"int\"}]}"); + var reader = Avro.Schema.Parse( + "{\"type\":\"record\",\"name\":\"Foo\",\"fields\":[" + + "{\"name\":\"val\",\"type\":\"int\"}]}"); + var ms = new MemoryStream(); + new BinaryEncoder(ms).WriteLong(100_000_000); // well over the zero-byte item cap + ms.Position = 0; + var r = new GenericDatumReader<object>(writer, reader); + Assert.Throws<AvroException>(() => r.Read(null, new BinaryDecoder(ms))); + } + + // A block count larger than int.MaxValue must be rejected before the int + // cast, even for a null-element array where the byte check is skipped. + [Test] + public void TestDatumReaderReadArrayRejectsCountAboveIntMax() + { + var schema = Avro.Schema.Parse("{\"type\":\"array\",\"items\":\"null\"}"); + var ms = new MemoryStream(); + new BinaryEncoder(ms).WriteLong((long)int.MaxValue + 1); + ms.Position = 0; + var reader = new GenericDatumReader<object>(schema, schema); + Assert.Throws<AvroException>(() => reader.Read(null, new BinaryDecoder(ms))); + } + + // A huge non-zero-byte count on a non-seekable stream (where the + // bytes-remaining check cannot bound it) must not preallocate the whole + // block: the backing store grows on demand, so a truncated stream fails + // with a bounded AvroException instead of a multi-gigabyte allocation. + [Test] + public void TestDatumReaderReadArrayHugeCountOnStreamClampsPreallocation() + { + var schema = Avro.Schema.Parse("{\"type\":\"array\",\"items\":\"long\"}"); + var backing = new MemoryStream(); + new BinaryEncoder(backing).WriteLong(200_000_000); // block count; no element data + byte[] encoded = backing.ToArray(); + using (var ns = new NonSeekableStream(new MemoryStream(encoded))) + { + var reader = new GenericDatumReader<object>(schema, schema); + Assert.Throws<AvroException>(() => reader.Read(null, new BinaryDecoder(ns))); + } + } + + // A resolved PreresolvingDatumReader is documented as safe to share among + // threads. The per-datum zero-byte-element budget must therefore not be + // reader instance state (a shared counter would let concurrent decodes + // reset and increment each other); it is thread-static. Many threads + // decoding within-cap data through one shared reader must all succeed. + [Test] + public void TestDatumReaderSharedAcrossThreadsIsThreadSafe() + { + var schema = Avro.Schema.Parse( + "{\"type\":\"record\",\"name\":\"R\",\"fields\":[" + + "{\"name\":\"a\",\"type\":{\"type\":\"array\",\"items\":\"null\"}}," + + "{\"name\":\"b\",\"type\":{\"type\":\"array\",\"items\":\"null\"}}]}"); + var ms = new MemoryStream(); + var enc = new BinaryEncoder(ms); + enc.WriteLong(3); enc.WriteLong(0); // field a: 3 nulls + enc.WriteLong(3); enc.WriteLong(0); // field b: 3 nulls + byte[] encoded = ms.ToArray(); + + var reader = new GenericDatumReader<object>(schema, schema); + var errors = new System.Collections.Concurrent.ConcurrentQueue<Exception>(); + System.Threading.Tasks.Parallel.For(0, 32, _ => + { + try + { + for (int i = 0; i < 500; i++) + { + var rec = (GenericRecord)reader.Read(null, new BinaryDecoder(new MemoryStream(encoded))); + Assert.AreEqual(3, ((object[])rec["a"]).Length); + Assert.AreEqual(3, ((object[])rec["b"]).Length); + } + } + catch (Exception ex) + { + errors.Enqueue(ex); + } + }); + Assert.IsEmpty(errors); + } + // Minimal read-only, forward-only stream wrapper reporting CanSeek=false. private sealed class NonSeekableStream : Stream { diff --git a/lang/csharp/src/apache/test/Reflect/TestArray.cs b/lang/csharp/src/apache/test/Reflect/TestArray.cs index ede5af3359..21dd2da953 100644 --- a/lang/csharp/src/apache/test/Reflect/TestArray.cs +++ b/lang/csharp/src/apache/test/Reflect/TestArray.cs @@ -82,6 +82,20 @@ namespace Avro.Test } } + // A malicious array block declaring far more elements than the stream + // could hold must be rejected before allocating, matching the Generic and + // Specific readers. Exercises ReflectDefaultReader.ReadArray. + [TestCase] + public void ListRejectsCountBeyondStream() + { + var schema = Schema.Parse(_simpleList); // array<string> + var ms = new MemoryStream(); + new BinaryEncoder(ms).WriteLong(1000000); // 1,000,000 strings, no data + ms.Seek(0, SeekOrigin.Begin); + var reader = new ReflectReader<List<string>>(schema, schema); + Assert.Throws<AvroException>(() => reader.Read(new BinaryDecoder(ms))); + } + [TestCase] public void ListRecTest() { diff --git a/lang/csharp/src/apache/test/Specific/SpecificTests.cs b/lang/csharp/src/apache/test/Specific/SpecificTests.cs index 1aa3c3a03a..1fecddf300 100644 --- a/lang/csharp/src/apache/test/Specific/SpecificTests.cs +++ b/lang/csharp/src/apache/test/Specific/SpecificTests.cs @@ -511,6 +511,31 @@ namespace Avro.Test } + // A malicious array block declaring far more elements than the stream + // could hold must be rejected before allocating, matching the Generic + // reader. Exercises SpecificDefaultReader.ReadArray (via SpecificReader) + // and SpecificDatumReader (via PreresolvingDatumReader). + [TestCase] + public void TestSpecificReaderRejectsArrayCountBeyondStream() + { + var schema = EmbeddedGenericsRecord._SCHEMA; + + byte[] Malicious() + { + var ms = new MemoryStream(); + var enc = new BinaryEncoder(ms); + enc.WriteUnionIndex(0); // OptionalInt: union branch 0 (null) + enc.WriteLong(1000000); // OptionalIntList: 1,000,000 items, no data + return ms.ToArray(); + } + + var r1 = new SpecificReader<EmbeddedGenericsRecord>(schema, schema); + Assert.Throws<AvroException>(() => r1.Read(null, new BinaryDecoder(new MemoryStream(Malicious())))); + + var r2 = new SpecificDatumReader<EmbeddedGenericsRecord>(schema, schema); + Assert.Throws<AvroException>(() => r2.Read(null, new BinaryDecoder(new MemoryStream(Malicious())))); + } + private static S deserialize<S>(Stream ms, Schema ws, Schema rs) where S : class, ISpecificRecord { long initialPos = ms.Position;
