diff --git a/lang/java/avro/src/main/java/org/apache/avro/SystemLimitException.java b/lang/java/avro/src/main/java/org/apache/avro/SystemLimitException.java index 886a939735c..5ce0b25ba6f 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/SystemLimitException.java +++ b/lang/java/avro/src/main/java/org/apache/avro/SystemLimitException.java @@ -28,13 +28,20 @@ * The following system properties can be set to limit the size of bytes, * strings and collection types to be allocated: * * * The default is to permit sizes up to {@link #MAX_ARRAY_VM_LIMIT}. @@ -83,6 +90,35 @@ public class SystemLimitException extends AvroRuntimeException { public static final long MAX_DECOMPRESS_LENGTH = getLongLimitFromProperty(MAX_DECOMPRESS_LENGTH_PROPERTY, defaultMaxDecompressLength()); + /** + * System property declaring the maximum number of array elements whose minimum + * encoded size is zero (e.g. {@code null}, a zero-length fixed, a record whose + * fields are all zero-byte, or a recursive schema conservatively treated as a 0 + * minimum) to allocate at once: {@value}. + */ + public static final String MAX_COLLECTION_ALLOCATION_PROPERTY = "org.apache.avro.limits.collectionItems.maxAllocation"; + + /** + * Fraction of the maximum heap a single decoded collection of elements whose + * minimum encoded size is zero may occupy by default. Keeps the backing + * allocation below the heap so a small payload declaring a huge block count + * cannot exhaust the JVM. + */ + private static final long DEFAULT_MAX_COLLECTION_ALLOCATION_HEAP_FRACTION = 4; + + /** + * Estimated bytes retained per pre-allocated collection slot (a single object + * reference), used to translate the heap budget into an element count. + */ + private static final long BYTES_PER_COLLECTION_SLOT = 8; + + /** + * Maximum number of array elements whose minimum encoded size is zero to + * allocate at once. Recomputed from the system property (or the heap) by + * {@link #resetLimits()}. + */ + private static long maxCollectionAllocation = defaultMaxCollectionAllocation(); + static { resetLimits(); } @@ -153,6 +189,20 @@ private static long defaultMaxDecompressLength() { Math.max(1L, Runtime.getRuntime().maxMemory() / DEFAULT_MAX_DECOMPRESS_HEAP_FRACTION)); } + /** + * Calculate the default maximum number of array elements whose minimum encoded + * size is zero to allocate at once, as a fraction of the maximum heap. Such + * elements consume no guaranteed input bytes, so the usual "bytes remaining" + * bound does not apply and the allocation must instead be capped relative to + * the available memory. + * + * @return the calculated default max element count. + */ + private static long defaultMaxCollectionAllocation() { + long heapBudget = Math.max(1L, Runtime.getRuntime().maxMemory() / DEFAULT_MAX_COLLECTION_ALLOCATION_HEAP_FRACTION); + return Math.max(1L, heapBudget / BYTES_PER_COLLECTION_SLOT); + } + /** * Check to ensure that reading the bytes is within the specified limits. * @@ -246,6 +296,44 @@ public static int checkMaxCollectionLength(long items) { return (int) items; } + /** + * Check to ensure that allocating storage for the specified number of array + * elements whose minimum encoded size is zero remains within the heap-aware + * limit. + *

+ * Elements whose minimum encoded size is zero (e.g. {@code null}, a zero-length + * fixed, a record whose fields are all zero-byte, or a recursive schema whose + * cycle is conservatively broken with a 0 minimum) consume no guaranteed input + * bytes, so the number that may be declared is not bounded by the bytes + * remaining in the stream. Without a cap, a tiny payload can declare an + * enormous block count and drive an unbounded backing-array allocation. This + * limit is derived from the maximum heap (see + * {@link #MAX_COLLECTION_ALLOCATION_PROPERTY}). + * + * @param existing The number of elements already allocated for the collection. + * @param items The next number of elements to allocate. + * @return The cumulative element count if and only if it is within the limit. + * @throws SystemLimitException if the cumulative allocation would exceed the + * limit. + * @throws AvroRuntimeException if either argument is negative. + */ + public static long checkMaxCollectionAllocation(long existing, long items) { + if (existing < 0) { + throw new AvroRuntimeException("Malformed data. Length is negative: " + existing); + } + if (items < 0) { + throw new AvroRuntimeException("Malformed data. Length is negative: " + items); + } + long total = existing + items; + if (total < existing || total > maxCollectionAllocation) { + throw new SystemLimitException("Cannot allocate " + (total < existing ? "more than Long.MAX_VALUE" : total) + + " collection elements whose minimum encoded size is zero: exceeds the maximum allowed of " + + maxCollectionAllocation + " (configure with the system property " + MAX_COLLECTION_ALLOCATION_PROPERTY + + ")"); + } + return total; + } + /** * Check to ensure that reading the string size is within the specified limits. * @@ -294,5 +382,11 @@ static void resetLimits() { maxBytesLength = getLimitFromProperty(MAX_BYTES_LENGTH_PROPERTY, MAX_ARRAY_VM_LIMIT); maxCollectionLength = getLimitFromProperty(MAX_COLLECTION_LENGTH_PROPERTY, MAX_ARRAY_VM_LIMIT); maxStringLength = getLimitFromProperty(MAX_STRING_LENGTH_PROPERTY, MAX_ARRAY_VM_LIMIT); + maxCollectionAllocation = getLongLimitFromProperty(MAX_COLLECTION_ALLOCATION_PROPERTY, + defaultMaxCollectionAllocation()); + // A collection cannot hold more than MAX_ARRAY_VM_LIMIT elements, so keep the + // zero-byte allocation cap consistent with the other collection limits even + // when it is configured (or derived from a very large heap) above that. + maxCollectionAllocation = Math.min(maxCollectionAllocation, MAX_ARRAY_VM_LIMIT); } } diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader.java b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader.java index 178ca50ccad..cc72297a2eb 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader.java +++ b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader.java @@ -37,6 +37,7 @@ import org.apache.avro.LogicalType; import org.apache.avro.Schema; import org.apache.avro.Schema.Field; +import org.apache.avro.SystemLimitException; import org.apache.avro.io.DatumReader; import org.apache.avro.io.Decoder; import org.apache.avro.io.DecoderFactory; @@ -114,6 +115,26 @@ public void setExpected(Schema reader) { private static final ThreadLocal>> RESOLVER_CACHE = ThreadLocalWithInitial .of(WeakIdentityHashMap::new); + /** + * Upper bound on the initial capacity eagerly allocated for a collection from + * its declared block count. The backing array/map grows on demand as elements + * are read, so this is only a starting hint: it prevents a large declared count + * from driving a huge up-front allocation before any element is decoded. This + * matters most for stream sources, where the decoder cannot know how many bytes + * remain and so cannot otherwise bound the declared count against the input. + */ + private static final int MAX_COLLECTION_PREALLOC = 1024; + + /** + * Clamp a declared collection block count to a safe initial allocation size. + * + * @param count the declared (already limit-checked) block count + * @return {@code count} capped at {@link #MAX_COLLECTION_PREALLOC} + */ + public static int initialCollectionCapacity(long count) { + return (int) Math.min(count, MAX_COLLECTION_PREALLOC); + } + /** * Gets a resolving decoder for use by this GenericDatumReader. Unstable API. * Currently uses a thread local cache to prevent constructing the resolvers too @@ -296,9 +317,20 @@ protected Object readArray(Object old, Schema expected, ResolvingDecoder in) thr long base = 0; if (l > 0) { ensureAvailableCollectionBytes(in, l, expectedType); + // Elements whose minimum encoded size is zero (null, a zero-length fixed, a + // record whose fields are all zero-byte, or a recursive schema where the + // cycle is broken with a 0 minimum) consume no guaranteed input, so + // ensureAvailableCollectionBytes cannot bound their count from the bytes + // remaining. Cap such collections against a heap-aware limit so a tiny + // payload cannot declare a huge block count and drive an unbounded + // backing-array allocation. + boolean zeroByteElements = minBytesPerElement(expectedType) == 0; + if (zeroByteElements) { + SystemLimitException.checkMaxCollectionAllocation(base, l); + } LogicalType logicalType = expectedType.getLogicalType(); Conversion conversion = getData().getConversionFor(logicalType); - Object array = newArray(old, (int) l, expected); + Object array = newArray(old, initialCollectionCapacity(l), expected); do { if (logicalType != null && conversion != null) { for (long i = 0; i < l; i++) { @@ -311,7 +343,11 @@ protected Object readArray(Object old, Schema expected, ResolvingDecoder in) thr } } base += l; - } while ((l = arrayNext(in, expectedType)) > 0); + l = arrayNext(in, expectedType); + if (zeroByteElements && l > 0) { + SystemLimitException.checkMaxCollectionAllocation(base, l); + } + } while (l > 0); return pruneArray(array); } else { return pruneArray(newArray(old, 0, expected)); @@ -364,7 +400,7 @@ protected Object readMap(Object old, Schema expected, ResolvingDecoder in) throw LogicalType logicalType = eValue.getLogicalType(); Conversion conversion = getData().getConversionFor(logicalType); ensureAvailableMapBytes(in, l, eValue); - Object map = newMap(old, (int) l); + Object map = newMap(old, initialCollectionCapacity(l)); if (l > 0) { do { if (logicalType != null && conversion != null) { @@ -441,6 +477,23 @@ static int minBytesPerElement(Schema schema) { return minBytesPerElement(schema, Collections.newSetFromMap(new IdentityHashMap<>())); } + /** + * Whether the minimum encoded size of the given schema is zero, i.e. + * {@link #minBytesPerElement(Schema)} is {@code 0}. This is true for values + * that always encode to zero bytes (e.g. {@code null}, a zero-length + * {@code fixed}, or a record whose fields are all zero-byte), and + * conservatively for recursive schemas, where the cycle is broken by returning + * a 0 minimum. Such elements cannot be bounded by the number of bytes remaining + * in the stream, so a collection of them must be bounded by a heap-aware + * allocation limit instead. + * + * @param schema the element (or map value) schema + * @return {@code true} if the schema's minimum encoded size is zero + */ + public static boolean isZeroByteSchema(Schema schema) { + return minBytesPerElement(schema) == 0; + } + private static int minBytesPerElement(Schema schema, Set visited) { switch (schema.getType()) { case NULL: @@ -481,9 +534,11 @@ private static int minBytesPerElement(Schema schema, Set visited) { * reports fewer remaining bytes than required. *

* This check prevents out-of-memory errors from pre-allocating huge backing - * arrays when the source data is truncated or malicious. + * arrays when the source data is truncated or malicious. It is exposed so the + * fast reader ({@code FastReaderBuilder}) can apply the same guard as this + * classic reader. */ - private static void ensureAvailableCollectionBytes(Decoder decoder, long count, Schema elementSchema) + public static void ensureAvailableCollectionBytes(Decoder decoder, long count, Schema elementSchema) throws EOFException { if (count <= 0) { return; @@ -747,7 +802,23 @@ public static void skip(Schema schema, Decoder in) throws IOException { break; case ARRAY: Schema elementType = schema.getElementType(); + // Bound the cumulative element count: skipping a huge block of elements + // whose minimum encoded size is zero (e.g. null) would otherwise loop + // unboundedly (a CPU exhaustion) even though it reads nothing. Such + // elements use the heap-aware allocation cap; others the structural + // collection cap. + boolean zeroByteElements = isZeroByteSchema(elementType); + long arrayTotal = 0; for (long l = in.skipArray(); l > 0; l = in.skipArray()) { + // Always enforce the cumulative structural cap, then additionally the + // heap-aware allocation cap for zero-byte elements (which the structural + // cap alone does not bound tightly), so a huge count split across blocks + // cannot drive an unbounded skip loop. + SystemLimitException.checkMaxCollectionLength(arrayTotal, l); + if (zeroByteElements) { + SystemLimitException.checkMaxCollectionAllocation(arrayTotal, l); + } + arrayTotal += l; for (long i = 0; i < l; i++) { skip(elementType, in); } @@ -755,7 +826,11 @@ public static void skip(Schema schema, Decoder in) throws IOException { break; case MAP: Schema value = schema.getValueType(); + // Map entries always carry a >= 1 byte key, so the structural cap applies. + long mapTotal = 0; for (long l = in.skipMap(); l > 0; l = in.skipMap()) { + SystemLimitException.checkMaxCollectionLength(mapTotal, l); + mapTotal += l; for (long i = 0; i < l; i++) { in.skipString(); skip(value, in); diff --git a/lang/java/avro/src/main/java/org/apache/avro/io/BinaryDecoder.java b/lang/java/avro/src/main/java/org/apache/avro/io/BinaryDecoder.java index 77fc8490764..c2e2ab8351c 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/io/BinaryDecoder.java +++ b/lang/java/avro/src/main/java/org/apache/avro/io/BinaryDecoder.java @@ -408,8 +408,21 @@ protected void doReadBytes(byte[] bytes, int start, int length) throws IOExcepti protected long doReadItemCount() throws IOException { long result = readLong(); if (result < 0L) { - // Consume byte-count if present - readLong(); + // A negative block count is followed by a block byte-size; consume it. + final long bytecount = readLong(); + if (result == Long.MIN_VALUE) { + // Long.MIN_VALUE cannot be negated (-Long.MIN_VALUE overflows back to + // Long.MIN_VALUE), so it is not a valid block count. Reject it rather + // than letting it fall through as a negative "count" that would later be + // truncated to 0 and silently terminate the collection without consuming + // the end marker, desynchronizing decoding of subsequent fields. + throw new AvroRuntimeException("Malformed data. Block count is invalid: " + result); + } + if (bytecount < 0L) { + // The block byte-size is a byte count and must be non-negative, matching + // doSkipItems(). + throw new AvroRuntimeException("Malformed data. Block byte-size is negative: " + bytecount); + } result = -result; } return result; @@ -436,7 +449,17 @@ protected long doReadItemCount() throws IOException { private long doSkipItems() throws IOException { long result = readLong(); while (result < 0L) { + if (result == Long.MIN_VALUE) { + // Consistent with doReadItemCount: Long.MIN_VALUE is not a valid block + // count (it cannot be negated), so reject it rather than treating it as + // a byte-sized block and continuing to skip. + readLong(); + throw new AvroRuntimeException("Malformed data. Block count is invalid: " + result); + } final long bytecount = readLong(); + if (bytecount < 0L) { + throw new AvroRuntimeException("Malformed data. Block byte-size is negative: " + bytecount); + } doSkipBytes(bytecount); result = readLong(); } @@ -458,7 +481,7 @@ public long arrayNext() throws IOException { @Override public long skipArray() throws IOException { - return doSkipItems(); + return SystemLimitException.checkMaxCollectionLength(doSkipItems()); } @Override @@ -476,7 +499,7 @@ public long mapNext() throws IOException { @Override public long skipMap() throws IOException { - return doSkipItems(); + return SystemLimitException.checkMaxCollectionLength(doSkipItems()); } @Override diff --git a/lang/java/avro/src/main/java/org/apache/avro/io/FastReaderBuilder.java b/lang/java/avro/src/main/java/org/apache/avro/io/FastReaderBuilder.java index 512c9ebf34f..96767ad9934 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/io/FastReaderBuilder.java +++ b/lang/java/avro/src/main/java/org/apache/avro/io/FastReaderBuilder.java @@ -40,6 +40,7 @@ import org.apache.avro.Resolver.WriterUnion; import org.apache.avro.Schema; import org.apache.avro.Schema.Field; +import org.apache.avro.SystemLimitException; import org.apache.avro.generic.GenericArray; import org.apache.avro.generic.GenericData; import org.apache.avro.generic.GenericData.InstanceSupplier; @@ -419,6 +420,10 @@ private FieldReader createUnionReader(WriterUnion action) throws IOException { private FieldReader createUnionReader(FieldReader[] unionReaders) { return reusingReader((reuse, decoder) -> { final int selection = decoder.readIndex(); + if (selection < 0 || selection >= unionReaders.length) { + throw new AvroTypeException( + "Union branch index out of range: must be in [0, " + unionReaders.length + "), but received " + selection); + } return unionReaders[selection].read(null, decoder); }); @@ -469,39 +474,82 @@ private Optional> findClass(String clazz) { @SuppressWarnings("unchecked") private FieldReader createArrayReader(Schema readerSchema, Container action) throws IOException { FieldReader elementReader = getReaderFor(action.elementAction, null); + Schema elementType = readerSchema.getElementType(); + + // Elements whose minimum encoded size is zero (null, a record with only + // zero-byte fields, or a recursive schema whose cycle is broken with a 0 + // minimum) consume no guaranteed input, so the block count cannot be bounded + // by the bytes remaining in the stream. Cap such collections against a + // heap-aware limit so a tiny payload cannot declare a huge block count and + // drive an unbounded backing-array allocation (AVRO-4300). + boolean zeroByteElements = GenericDatumReader.isZeroByteSchema(elementType); return reusingReader((reuse, decoder) -> { if (reuse instanceof GenericArray) { GenericArray reuseArray = (GenericArray) reuse; long l = decoder.readArrayStart(); + long total = 0; + checkArrayBlock(decoder, elementType, zeroByteElements, total, l); reuseArray.clear(); while (l > 0) { for (long i = 0; i < l; i++) { reuseArray.add(elementReader.read(reuseArray.peek(), decoder)); } + total += l; l = decoder.arrayNext(); + checkArrayBlock(decoder, elementType, zeroByteElements, total, l); } return reuseArray; } else { long l = decoder.readArrayStart(); + long total = 0; + checkArrayBlock(decoder, elementType, zeroByteElements, total, l); List array = (reuse instanceof List) ? (List) reuse - : new GenericData.Array<>((int) l, readerSchema); + : new GenericData.Array<>(GenericDatumReader.initialCollectionCapacity(l), readerSchema); array.clear(); while (l > 0) { for (long i = 0; i < l; i++) { array.add(elementReader.read(null, decoder)); } + total += l; l = decoder.arrayNext(); + checkArrayBlock(decoder, elementType, zeroByteElements, total, l); } return array; } }); } + /** + * Validates an array block count before its elements are allocated, applying + * the same guards as the classic {@code GenericDatumReader}: the + * bytes-remaining check for elements with a positive minimum size, and the + * heap-aware allocation cap for zero-byte elements (which the bytes check + * cannot bound). + */ + private static void checkArrayBlock(Decoder decoder, Schema elementType, boolean zeroByteElements, long total, + long count) throws IOException { + if (count <= 0) { + return; + } + if (zeroByteElements) { + // The bytes-remaining check cannot bound zero-byte elements (minBytes is + // 0, so ensureAvailableCollectionBytes would no-op after recomputing it); + // apply the heap-aware allocation cap instead. + SystemLimitException.checkMaxCollectionAllocation(total, count); + } else { + GenericDatumReader.ensureAvailableCollectionBytes(decoder, count, elementType); + } + } + private FieldReader createEnumReader(EnumAdjust action) { return reusingReader((reuse, decoder) -> { int index = decoder.readEnum(); + if (index < 0 || index >= action.values.length) { + throw new AvroTypeException( + "Enumeration out of range: must be in [0, " + action.values.length + "), but received " + index); + } Object resultObject = action.values[index]; if (resultObject == null) { throw new AvroTypeException("No match for " + action.writer.getEnumSymbols().get(index)); diff --git a/lang/java/avro/src/main/java/org/apache/avro/io/ResolvingDecoder.java b/lang/java/avro/src/main/java/org/apache/avro/io/ResolvingDecoder.java index 6bdb16a332c..0b124579f71 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/io/ResolvingDecoder.java +++ b/lang/java/avro/src/main/java/org/apache/avro/io/ResolvingDecoder.java @@ -260,8 +260,19 @@ public int readEnum() throws IOException { Symbol.EnumAdjustAction top = (Symbol.EnumAdjustAction) parser.popSymbol(); int n = in.readEnum(); if (top.noAdjustments) { + // n is used directly as an index into the reader enum's symbols, so it + // must fall within the reader symbol count. + if (n < 0 || n >= top.size) { + throw new AvroTypeException("Enumeration out of range: must be in [0, " + top.size + "), but received " + n); + } return n; } + // Otherwise n indexes the writer-to-reader adjustment table; reject an index + // outside it rather than letting the array access throw. + if (n < 0 || n >= top.adjustments.length) { + throw new AvroTypeException( + "Enumeration out of range: must be in [0, " + top.adjustments.length + "), but received " + n); + } Object o = top.adjustments[n]; if (o instanceof Integer) { return (Integer) o; diff --git a/lang/java/avro/src/main/java/org/apache/avro/io/ValidatingDecoder.java b/lang/java/avro/src/main/java/org/apache/avro/io/ValidatingDecoder.java index 26f79a16ff2..3c5e0842189 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/io/ValidatingDecoder.java +++ b/lang/java/avro/src/main/java/org/apache/avro/io/ValidatingDecoder.java @@ -164,7 +164,7 @@ public int readEnum() throws IOException { Symbol.IntCheckAction top = (Symbol.IntCheckAction) parser.popSymbol(); int result = in.readEnum(); if (result < 0 || result >= top.size) { - throw new AvroTypeException("Enumeration out of range: max is " + top.size + " but received " + result); + throw new AvroTypeException("Enumeration out of range: must be in [0, " + top.size + "), but received " + result); } return result; } diff --git a/lang/java/avro/src/main/java/org/apache/avro/io/parsing/Symbol.java b/lang/java/avro/src/main/java/org/apache/avro/io/parsing/Symbol.java index b5dcbeb68f0..039b1517fd6 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/io/parsing/Symbol.java +++ b/lang/java/avro/src/main/java/org/apache/avro/io/parsing/Symbol.java @@ -26,6 +26,7 @@ import java.util.NoSuchElementException; import java.util.Set; +import org.apache.avro.AvroTypeException; import org.apache.avro.Schema; /** @@ -457,6 +458,10 @@ private Alternative(Symbol[] symbols, String[] labels) { } public Symbol getSymbol(int index) { + if (index < 0 || index >= symbols.length) { + throw new AvroTypeException( + "Union branch index out of range: must be in [0, " + symbols.length + "), but received " + index); + } return symbols[index]; } diff --git a/lang/java/avro/src/test/java/org/apache/avro/TestSystemLimitException.java b/lang/java/avro/src/test/java/org/apache/avro/TestSystemLimitException.java index 0da39179506..a0b59f55e27 100644 --- a/lang/java/avro/src/test/java/org/apache/avro/TestSystemLimitException.java +++ b/lang/java/avro/src/test/java/org/apache/avro/TestSystemLimitException.java @@ -46,6 +46,7 @@ void reset() { System.clearProperty(MAX_BYTES_LENGTH_PROPERTY); System.clearProperty(MAX_COLLECTION_LENGTH_PROPERTY); System.clearProperty(MAX_STRING_LENGTH_PROPERTY); + System.clearProperty(MAX_COLLECTION_ALLOCATION_PROPERTY); resetLimits(); } @@ -110,6 +111,49 @@ void testCheckMaxStringLength() { "String length 1024 exceeds maximum allowed"); } + @Test + void testCheckMaxCollectionAllocation() { + // With a small custom limit, cumulative allocations beyond it are rejected. + System.setProperty(MAX_COLLECTION_ALLOCATION_PROPERTY, "1000"); + resetLimits(); + + // Values within the limit pass through and return the running total. + assertEquals(0L, checkMaxCollectionAllocation(0L, 0L)); + assertEquals(1000L, checkMaxCollectionAllocation(0L, 1000L)); + assertEquals(1000L, checkMaxCollectionAllocation(400L, 600L)); + + // A single block over the limit is rejected. + SystemLimitException ex = assertThrows(SystemLimitException.class, () -> checkMaxCollectionAllocation(0L, 1001L)); + assertTrue(ex.getMessage().contains("exceeds the maximum allowed of 1000"), ex.getMessage()); + + // Cumulative blocks that cross the limit are rejected. + ex = assertThrows(SystemLimitException.class, () -> checkMaxCollectionAllocation(600L, 401L)); + assertTrue(ex.getMessage().contains("exceeds the maximum allowed of 1000"), ex.getMessage()); + + // Negative arguments are rejected as malformed. + Exception nex = assertThrows(AvroRuntimeException.class, () -> checkMaxCollectionAllocation(-1L, 10L)); + assertEquals(ERROR_NEGATIVE, nex.getMessage()); + nex = assertThrows(AvroRuntimeException.class, () -> checkMaxCollectionAllocation(10L, -1L)); + assertEquals(ERROR_NEGATIVE, nex.getMessage()); + + // Additive overflow is rejected rather than wrapping to a small value. + assertThrows(SystemLimitException.class, () -> checkMaxCollectionAllocation(Long.MAX_VALUE, 10L)); + } + + @Test + void testCheckMaxCollectionAllocationDefaultsToHeapFraction() { + // With no property set, the default is derived from the heap and is well + // below Integer.MAX_VALUE on a normally sized JVM, yet generous enough for + // legitimate small collections. + resetLimits(); + assertEquals(1024L, checkMaxCollectionAllocation(0L, 1024L)); + // A pathologically large zero-byte collection is rejected without allocating. + // Use MAX_ARRAY_VM_LIMIT + 1 so it exceeds the cap regardless of heap size + // (the default is derived from the heap and then clamped to MAX_ARRAY_VM_LIMIT, + // so Integer.MAX_VALUE - 8 alone would not exceed it on a very large heap). + assertThrows(SystemLimitException.class, () -> checkMaxCollectionAllocation(0L, (long) MAX_ARRAY_VM_LIMIT + 1L)); + } + @Test void testCheckMaxCollectionLengthFromNonZero() { // Correct values pass through diff --git a/lang/java/avro/src/test/java/org/apache/avro/generic/TestGenericDatumReader.java b/lang/java/avro/src/test/java/org/apache/avro/generic/TestGenericDatumReader.java index 5586b828999..211692c4d7b 100644 --- a/lang/java/avro/src/test/java/org/apache/avro/generic/TestGenericDatumReader.java +++ b/lang/java/avro/src/test/java/org/apache/avro/generic/TestGenericDatumReader.java @@ -19,19 +19,26 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import java.io.BufferedInputStream; +import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.EOFException; import java.io.IOException; +import java.io.InputStream; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Random; import java.util.stream.Collectors; import java.util.stream.IntStream; +import org.apache.avro.AvroRuntimeException; import org.apache.avro.Schema; +import org.apache.avro.SystemLimitException; import org.apache.avro.io.BinaryDecoder; import org.apache.avro.io.BinaryEncoder; import org.apache.avro.io.DecoderFactory; @@ -216,6 +223,49 @@ void arrayOfIntsRejectsHugeCount() throws Exception { assertThrows(EOFException.class, () -> reader.read(null, decoder)); } + /** + * On a stream source the decoder cannot know how many bytes remain, so the + * bytes-available guard is skipped and a large declared array count reaches the + * allocation path directly. The initial backing-array capacity must be clamped + * so a hostile count cannot drive a huge up-front allocation; a truncated + * stream therefore fails with {@link EOFException} (once the declared elements + * cannot be read) rather than attempting to preallocate hundreds of millions of + * slots. + */ + @Test + void arrayHugeCountOnStreamClampsPreallocation() throws Exception { + Schema schema = Schema.createArray(Schema.create(Schema.Type.LONG)); + // Record the largest capacity ever requested from newArray so we can assert + // the preallocation is clamped rather than sized to the declared block count. + // Without this, a regression that reintroduces new Object[(int) count] could + // allocate ~200M slots and still pass (with an EOFException) on a large-heap + // JVM, hiding the regression. + final int[] maxRequestedCapacity = { 0 }; + GenericDatumReader reader = new GenericDatumReader(schema) { + @Override + protected Object newArray(Object old, int size, Schema schema) { + maxRequestedCapacity[0] = Math.max(maxRequestedCapacity[0], size); + return super.newArray(old, size, schema); + } + }; + + // A huge (non-zero-byte) block count followed by no element data. Wrapping in + // a BufferedInputStream keeps the source from being a ByteArrayInputStream, so + // it cannot report its remaining byte count (remainingBytes() == -1) and the + // bytes-available guard is disabled -- exercising the preallocation clamp. + byte[] data = encodeVarints(200_000_000L); + InputStream stream = new BufferedInputStream(new ByteArrayInputStream(data)); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(stream, null); + // Confirm the guard-disabling precondition: the decoder cannot report how + // many bytes remain, so this really is the unknown-remaining-bytes path. + assertEquals(-1, decoder.remainingBytes()); + assertThrows(EOFException.class, () -> reader.read(null, decoder)); + // The declared count is 200,000,000 but the initial allocation must stay + // clamped to GenericDatumReader.initialCollectionCapacity (<= 1024). + assertTrue(maxRequestedCapacity[0] <= GenericDatumReader.initialCollectionCapacity(200_000_000L), + "preallocation was not clamped: requested capacity " + maxRequestedCapacity[0]); + } + /** * Verify that reading an array of nulls with a large count SUCCEEDS because * null elements are 0 bytes each, so the byte check is correctly skipped. @@ -303,4 +353,337 @@ void mapOfRecordsRejectsHugeCountUsingFullRecordSize() throws Exception { BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); assertThrows(EOFException.class, () -> reader.read(null, decoder)); } + + // --- Zero-byte element collection allocation limit (AVRO-4300) --- + + /** + * An array of {@code null} elements encodes each element as 0 bytes, so the + * bytes-remaining check cannot bound the block count. A huge count must be + * rejected by the heap-aware allocation limit rather than driving an unbounded + * backing-array allocation. + */ + @Test + void arrayOfNullsRejectsCountAboveAllocationLimit() throws Exception { + System.setProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY, "1000"); + org.apache.avro.TestSystemLimitException.resetLimits(); + try { + Schema schema = Schema.createArray(Schema.create(Schema.Type.NULL)); + GenericDatumReader reader = new GenericDatumReader<>(schema); + + // A single block declaring 200,000 null elements: only ~4 payload bytes, + // but would allocate a 200,000-slot backing array. Exceeds the limit of + // 1000, so it must be rejected before allocating. + byte[] data = encodeVarints(200_000L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + assertThrows(SystemLimitException.class, () -> reader.read(null, decoder)); + + // Cumulative growth across multiple blocks is also rejected: two blocks of + // 600 nulls each (1200 > 1000) must throw on the second block. + byte[] cumulative = encodeVarints(600L, 600L, 0L); + BinaryDecoder decoder2 = DecoderFactory.get().binaryDecoder(cumulative, null); + assertThrows(SystemLimitException.class, () -> reader.read(null, decoder2)); + } finally { + System.clearProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY); + org.apache.avro.TestSystemLimitException.resetLimits(); + } + } + + /** + * A legitimate array of {@code null} elements within the allocation limit still + * decodes correctly, so the guard does not reject valid data. + */ + @Test + void arrayOfNullsWithinAllocationLimitStillDecodes() throws Exception { + System.setProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY, "1000"); + org.apache.avro.TestSystemLimitException.resetLimits(); + try { + Schema schema = Schema.createArray(Schema.create(Schema.Type.NULL)); + GenericDatumReader reader = new GenericDatumReader<>(schema); + + byte[] data = encodeVarints(1000L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + GenericData.Array result = (GenericData.Array) reader.read(null, decoder); + assertEquals(1000, result.size()); + } finally { + System.clearProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY); + org.apache.avro.TestSystemLimitException.resetLimits(); + } + } + + /** + * An array whose element is a record with only {@code null} fields also encodes + * to 0 bytes per element and must be bounded by the allocation limit. + */ + @Test + void arrayOfAllNullRecordsRejectsCountAboveAllocationLimit() throws Exception { + System.setProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY, "1000"); + org.apache.avro.TestSystemLimitException.resetLimits(); + try { + Schema recWithNull = Schema.createRecord("AllNull", null, "test", false); + recWithNull.setFields(Collections.singletonList(new Schema.Field("n", Schema.create(Schema.Type.NULL)))); + Schema schema = Schema.createArray(recWithNull); + GenericDatumReader reader = new GenericDatumReader<>(schema); + + byte[] data = encodeVarints(200_000L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + assertThrows(SystemLimitException.class, () -> reader.read(null, decoder)); + } finally { + System.clearProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY); + org.apache.avro.TestSystemLimitException.resetLimits(); + } + } + + private static GenericDatumReader arrayReader(Schema elementType, boolean fastReader) { + return readerFor(Schema.createArray(elementType), fastReader); + } + + private static GenericDatumReader readerFor(Schema schema, boolean fastReader) { + GenericData data = new GenericData(); + data.setFastReaderEnabled(fastReader); + return new GenericDatumReader<>(schema, schema, data); + } + + /** + * Full matrix: every collection kind must be rejected (never OOM) with a huge + * declared block count and effectively no element data (only a trailing {@code + * 0L} varint, used as a single element value or a 0-length map key), on both + * the fast (default) and the classic reader. Zero-byte-element arrays are + * bounded by the heap-aware allocation cap (SystemLimitException); every other + * kind is bounded by the bytes-remaining check (EOFException). Maps always + * carry a >=1-byte key so they fall in the latter group regardless of the value + * type. + */ + @Test + void hugeCollectionsRejectedOnBothReaderPaths() throws Exception { + // element/value type -> expected exception when the count is huge and no data + Schema nullType = Schema.create(Schema.Type.NULL); + Schema longType = Schema.create(Schema.Type.LONG); + Schema intType = Schema.create(Schema.Type.INT); + + // (schema, expected exception). array is the only zero-byte case here. + Object[][] cases = { { Schema.createArray(nullType), SystemLimitException.class }, + { Schema.createArray(longType), EOFException.class }, { Schema.createArray(intType), EOFException.class }, + { Schema.createMap(nullType), EOFException.class }, { Schema.createMap(longType), EOFException.class }, }; + + // Small allocation limit so the zero-byte array case is rejected + // deterministically regardless of the test JVM heap size. + System.setProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY, "1000"); + org.apache.avro.TestSystemLimitException.resetLimits(); + try { + for (Object[] c : cases) { + Schema schema = (Schema) c[0]; + @SuppressWarnings("unchecked") + Class expected = (Class) c[1]; + for (boolean fast : new boolean[] { true, false }) { + GenericDatumReader reader = readerFor(schema, fast); + byte[] data = encodeVarints(200_000_000L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + assertThrows(expected, () -> reader.read(null, decoder), () -> schema + " fast=" + fast); + } + } + } finally { + System.clearProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY); + org.apache.avro.TestSystemLimitException.resetLimits(); + } + } + + /** + * The fast reader (the default decode path) must apply the same guards as the + * classic reader. A non-zero-byte element array with a huge count and no data + * must be rejected by the bytes-remaining check rather than pre-allocating a + * huge backing array. Also verifies the cumulative allocation cap across + * multiple blocks for the zero-byte case, which the single-block matrix does + * not exercise. + */ + @Test + void fastAndClassicReaderRejectCumulativeNullBlocks() throws Exception { + System.setProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY, "1000"); + org.apache.avro.TestSystemLimitException.resetLimits(); + try { + for (boolean fast : new boolean[] { true, false }) { + GenericDatumReader reader = arrayReader(Schema.create(Schema.Type.NULL), fast); + // Two blocks of 600 nulls each (1200 > 1000) must throw on the second. + byte[] data = encodeVarints(600L, 600L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + assertThrows(SystemLimitException.class, () -> reader.read(null, decoder), "fastReader=" + fast); + } + } finally { + System.clearProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY); + org.apache.avro.TestSystemLimitException.resetLimits(); + } + } + + /** + * A negative block count encodes {@code abs(count)} zero-byte elements preceded + * by a block byte-size; the decoder normalizes it to a positive count, which + * must still be bounded by the allocation cap on both reader paths (matching + * the negative-block-count coverage of the C and Python SDKs). + */ + @Test + void fastAndClassicReaderRejectNegativeNullBlockCount() throws Exception { + System.setProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY, "1000"); + org.apache.avro.TestSystemLimitException.resetLimits(); + try { + for (boolean fast : new boolean[] { true, false }) { + GenericDatumReader reader = arrayReader(Schema.create(Schema.Type.NULL), fast); + // -200000 items (zigzag negative), followed by a block byte-size of 0, + // then the end-of-array terminator. Normalized to 200000 > 1000. + byte[] data = encodeVarints(-200_000L, 0L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + assertThrows(SystemLimitException.class, () -> reader.read(null, decoder), "fastReader=" + fast); + } + } finally { + System.clearProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY); + org.apache.avro.TestSystemLimitException.resetLimits(); + } + } + + /** + * {@code Long.MIN_VALUE} as a block count is the pathological overflow case: + * negating it overflows back to a negative value. Rather than letting it be + * truncated to {@code 0} (which would silently end the collection without + * consuming the end marker and desynchronize decoding of following fields), the + * decoder rejects it outright as malformed, matching the C SDK. + */ + @Test + void fastAndClassicReaderRejectMinValueBlockCount() throws Exception { + for (boolean fast : new boolean[] { true, false }) { + GenericDatumReader reader = arrayReader(Schema.create(Schema.Type.NULL), fast); + // Long.MIN_VALUE items (zigzag), a block byte-size of 0, then the + // end-of-array terminator. Negating Long.MIN_VALUE overflows, so it must be + // rejected as malformed instead of decoded as an empty array. + byte[] data = encodeVarints(Long.MIN_VALUE, 0L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + assertThrows(AvroRuntimeException.class, () -> reader.read(null, decoder), "fastReader=" + fast); + } + } + + /** + * A union branch index outside {@code [0, branch count)} is malformed and must + * be rejected on both reader paths rather than throwing a generic + * {@code IndexOutOfBoundsException}. + */ + @Test + void fastAndClassicReaderRejectOutOfRangeUnionIndex() throws Exception { + Schema schema = Schema.createUnion(Schema.create(Schema.Type.NULL), Schema.create(Schema.Type.LONG)); + for (boolean fast : new boolean[] { true, false }) { + for (long index : new long[] { 5L, -1L }) { + GenericDatumReader reader = readerFor(schema, fast); + byte[] data = encodeVarints(index); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + assertThrows(org.apache.avro.AvroTypeException.class, () -> reader.read(null, decoder), + "fastReader=" + fast + " index=" + index); + } + } + } + + /** + * An enum symbol index outside {@code [0, symbol count)} is malformed and must + * be rejected on both reader paths. + */ + @Test + void fastAndClassicReaderRejectOutOfRangeEnumIndex() throws Exception { + Schema schema = Schema.createEnum("E", null, null, Arrays.asList("A", "B")); + for (boolean fast : new boolean[] { true, false }) { + for (int index : new int[] { 9, -1 }) { + GenericDatumReader reader = readerFor(schema, fast); + byte[] data = encodeVarints(index); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + assertThrows(org.apache.avro.AvroTypeException.class, () -> reader.read(null, decoder), + "fastReader=" + fast + " index=" + index); + } + } + } + + /** + * A legitimate array of nulls within the limit still decodes on both reader + * paths. + */ + @Test + void fastAndClassicReaderDecodeSmallNullArray() throws Exception { + for (boolean fast : new boolean[] { true, false }) { + GenericDatumReader reader = arrayReader(Schema.create(Schema.Type.NULL), fast); + byte[] data = encodeVarints(1000L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + Collection result = (Collection) reader.read(null, decoder); + assertEquals(1000, result.size(), "fastReader=" + fast); + } + } + + /** + * Skipping a huge zero-byte array (e.g. during schema projection) is bounded by + * the heap-aware allocation cap, so it cannot loop unboundedly even though it + * reads nothing. + */ + @Test + void skipArrayOfNullRejectsHugeCount() throws Exception { + System.setProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY, "1000"); + org.apache.avro.TestSystemLimitException.resetLimits(); + try { + Schema schema = Schema.createArray(Schema.create(Schema.Type.NULL)); + byte[] data = encodeVarints(200_000L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + assertThrows(SystemLimitException.class, () -> GenericDatumReader.skip(schema, decoder)); + } finally { + System.clearProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY); + org.apache.avro.TestSystemLimitException.resetLimits(); + } + } + + /** + * A legitimate small zero-byte array is skipped without error. + */ + @Test + void skipSmallNullArraySucceeds() throws Exception { + Schema schema = Schema.createArray(Schema.create(Schema.Type.NULL)); + // 1000 nulls then the end marker, followed by a trailing long we can read to + // confirm the skip advanced correctly. + byte[] data = encodeVarints(1000L, 0L, 42L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + GenericDatumReader.skip(schema, decoder); + assertEquals(42L, decoder.readLong()); + } + + /** + * Skipping a huge map is bounded by the structural collection cap. + */ + @Test + void skipMapRejectsHugeCount() throws Exception { + Schema schema = Schema.createMap(Schema.create(Schema.Type.NULL)); + // A single block declaring more than Integer.MAX_VALUE - 8 entries. + byte[] data = encodeVarints((long) Integer.MAX_VALUE, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + // Integer.MAX_VALUE exceeds MAX_ARRAY_VM_LIMIT, so this hits the VM + // structural-limit path (an UnsupportedOperationException). + assertThrows(UnsupportedOperationException.class, () -> GenericDatumReader.skip(schema, decoder)); + } + + /** + * A writer array field that the reader schema omits is skipped during + * resolution through the decoder's skipArray; a huge count must be bounded on + * both the fast and classic reader paths. + */ + @Test + void resolvingSkipOfHugeNullArrayFieldIsBounded() throws Exception { + System.setProperty(SystemLimitException.MAX_COLLECTION_LENGTH_PROPERTY, "1000"); + org.apache.avro.TestSystemLimitException.resetLimits(); + try { + Schema writer = new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"R\",\"fields\":[" + + "{\"name\":\"arr\",\"type\":{\"type\":\"array\",\"items\":\"null\"}}," + + "{\"name\":\"a\",\"type\":\"long\"}]}"); + Schema reader = new Schema.Parser() + .parse("{\"type\":\"record\",\"name\":\"R\",\"fields\":[{\"name\":\"a\",\"type\":\"long\"}]}"); + byte[] data = encodeVarints(2000L, 0L, 42L); + for (boolean fast : new boolean[] { true, false }) { + GenericData data2 = new GenericData(); + data2.setFastReaderEnabled(fast); + GenericDatumReader r = new GenericDatumReader<>(writer, reader, data2); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + assertThrows(SystemLimitException.class, () -> r.read(null, decoder), "fastReader=" + fast); + } + } finally { + System.clearProperty(SystemLimitException.MAX_COLLECTION_LENGTH_PROPERTY); + org.apache.avro.TestSystemLimitException.resetLimits(); + } + } }