AVRO-4327: [Java] Bound decode recursion depth

lang/java/avro/src/main/java/org/apache/avro/SystemLimitException.java

@@ -42,6 +42,13 @@
  * 0 minimum) that may be allocated at once. Unlike other element types, these
  * cannot be bounded by the number of bytes remaining in the stream, so the
  * limit defaults to a fraction of the maximum heap.</li>
+ * <li><tt>org.apache.avro.limits.decode.maxDepth</tt> limits how deeply nested
+ * a value may be while decoding. A recursive schema (e.g. a linked list or a
+ * tree) lets a small, hostile payload drive arbitrarily deep nesting,
+ * exhausting the call stack ({@link StackOverflowError}) before any allocation
+ * limit is reached. The limit is enforced by counting structural descents (into
+ * a record, array, map or union) and rejecting input that nests deeper than the
+ * configured maximum.</li>
  * </ul>
  *
  * The default is to permit sizes up to {@link #MAX_ARRAY_VM_LIMIT}.
@@ -62,9 +69,24 @@ public class SystemLimitException extends AvroRuntimeException {
   public static final String MAX_COLLECTION_LENGTH_PROPERTY = "org.apache.avro.limits.collectionItems.maxLength";
   public static final String MAX_STRING_LENGTH_PROPERTY = "org.apache.avro.limits.string.maxLength";
 
+  /**
+   * System property bounding how deeply nested a value may be while decoding:
+   * {@value}. See {@link #incrementDecodeDepth()}.
+   */
+  public static final String MAX_DECODE_DEPTH_PROPERTY = "org.apache.avro.limits.decode.maxDepth";
+
+  /**
+   * Default maximum decode nesting depth. Comfortably exceeds any realistic
+   * schema nesting while remaining far below where a recursive decode would
+   * exhaust the call stack. Aligns with the well-known default used by Protocol
+   * Buffers.
+   */
+  static final int DEFAULT_MAX_DECODE_DEPTH = 100;
+
   private static int maxBytesLength = MAX_ARRAY_VM_LIMIT;
   private static int maxCollectionLength = MAX_ARRAY_VM_LIMIT;
   private static int maxStringLength = MAX_ARRAY_VM_LIMIT;
+  private static int maxDecodeDepth = DEFAULT_MAX_DECODE_DEPTH;
 
   private static final Logger LOG = LoggerFactory.getLogger(SystemLimitException.class);
 
@@ -132,6 +154,14 @@ public class SystemLimitException extends AvroRuntimeException {
   private static final class CollectionAllocationScope {
     private int depth;
     private long allocated;
+    /**
+     * Current decode nesting depth (structural descents into records, arrays, maps
+     * and unions). Tracked per thread rather than per reader instance so a reader
+     * reused concurrently cannot corrupt another thread's counter, and so the depth
+     * is threaded implicitly through the recursive decode without changing the
+     * reader method signatures. See {@link #incrementDecodeDepth()}.
+     */
+    private int decodeDepth;
   }
 
   private static final ThreadLocal<CollectionAllocationScope> COLLECTION_ALLOCATION_SCOPE = ThreadLocal
@@ -364,6 +394,11 @@ public static void beginCollectionAllocationScope() {
     CollectionAllocationScope scope = COLLECTION_ALLOCATION_SCOPE.get();
     if (scope.depth == 0) {
       scope.allocated = 0;
+      // Defensively clear the decode depth at the outermost datum boundary. The
+      // counter is already kept balanced by the try/finally around every
+      // increment, but resetting here guarantees a stale value from an abnormally
+      // terminated earlier decode on this thread cannot leak into this one.
+      scope.decodeDepth = 0;
     }
     scope.depth++;
   }
@@ -414,6 +449,47 @@ public static long checkMaxCollectionAllocation(long items) {
     return total;
   }
 
+  /**
+   * Record a structural descent (into a record, array, map or union) while
+   * decoding and verify the nesting has not grown past
+   * {@link #MAX_DECODE_DEPTH_PROPERTY the configured maximum}.
+   * <p>
+   * Avro's binary decoders decode nested values with a recursive call chain, so
+   * the call stack grows in lockstep with the nesting of the data. A recursive
+   * schema (e.g. a linked list or tree) lets a tiny, hostile payload declare
+   * arbitrarily deep nesting, overflowing the stack ({@link StackOverflowError})
+   * long before any allocation limit is reached. Bounding the depth turns such
+   * input into a clean, catchable failure instead of a crash.
+   * <p>
+   * Every call that succeeds must be paired with a matching
+   * {@link #decrementDecodeDepth()} in a {@code finally} block. When the limit
+   * would be exceeded this method throws <em>without</em> incrementing, so the
+   * counter stays balanced as the exception unwinds the enclosing
+   * (already-incremented) frames.
+   *
+   * @throws SystemLimitException if the decode nesting would exceed the maximum.
+   */
+  public static void incrementDecodeDepth() {
+    CollectionAllocationScope scope = COLLECTION_ALLOCATION_SCOPE.get();
+    if (scope.decodeDepth >= maxDecodeDepth) {
+      throw new SystemLimitException("Decode nesting depth exceeds the maximum allowed of " + maxDecodeDepth
+          + " (configure with the system property " + MAX_DECODE_DEPTH_PROPERTY + ")");
+    }
+    scope.decodeDepth++;
+  }
+
+  /**
+   * Record leaving a structural value opened by {@link #incrementDecodeDepth()}.
+   * Must be called from a {@code finally} block so the depth is restored even
+   * when decoding the nested value fails.
+   */
+  public static void decrementDecodeDepth() {
+    CollectionAllocationScope scope = COLLECTION_ALLOCATION_SCOPE.get();
+    if (scope.decodeDepth > 0) {
+      scope.decodeDepth--;
+    }
+  }
+
   /**
    * Check to ensure that reading the string size is within the specified limits.
    *
@@ -468,5 +544,6 @@ static void resetLimits() {
     // 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);
+    maxDecodeDepth = getLimitFromProperty(MAX_DECODE_DEPTH_PROPERTY, DEFAULT_MAX_DECODE_DEPTH);
   }
 }

lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader.java

@@ -214,15 +214,21 @@ protected Object readWithConversion(Object old, Schema expected, LogicalType log
   protected Object readWithoutConversion(Object old, Schema expected, ResolvingDecoder in) throws IOException {
     switch (expected.getType()) {
     case RECORD:
-      return readRecord(old, expected, in);
-    case ENUM:
-      return readEnum(expected, in);
     case ARRAY:
-      return readArray(old, expected, in);
     case MAP:
-      return readMap(old, expected, in);
     case UNION:
-      return read(old, expected.getTypes().get(in.readIndex()), in);
+      // Descending into a structural value grows the decode call stack. Bound the
+      // nesting depth so a recursive schema fed a deeply nested payload fails with
+      // a SystemLimitException instead of a StackOverflowError. The counter is
+      // decremented on exit via the finally so it stays balanced even on error.
+      SystemLimitException.incrementDecodeDepth();
+      try {
+        return readStructural(old, expected, in);
+      } finally {
+        SystemLimitException.decrementDecodeDepth();
+      }
+    case ENUM:
+      return readEnum(expected, in);
     case FIXED:
       return readFixed(old, expected, in);
     case STRING:
@@ -247,6 +253,26 @@ protected Object readWithoutConversion(Object old, Schema expected, ResolvingDec
     }
   }
 
+  /**
+   * Dispatches the structural (nesting) value types. Split out of
+   * {@link #readWithoutConversion} so the decode-depth guard wraps exactly the
+   * types that grow the recursive call stack.
+   */
+  private Object readStructural(Object old, Schema expected, ResolvingDecoder in) throws IOException {
+    switch (expected.getType()) {
+    case RECORD:
+      return readRecord(old, expected, in);
+    case ARRAY:
+      return readArray(old, expected, in);
+    case MAP:
+      return readMap(old, expected, in);
+    case UNION:
+      return read(old, expected.getTypes().get(in.readIndex()), in);
+    default:
+      throw new AvroRuntimeException("Not a structural type: " + expected);
+    }
+  }
+
   /**
    * Convert an underlying representation of a logical type (such as a ByteBuffer)
    * to a higher level object (such as a BigDecimal).

lang/java/avro/src/main/java/org/apache/avro/io/FastReaderBuilder.java

@@ -419,12 +419,17 @@ 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);
+      SystemLimitException.incrementDecodeDepth();
+      try {
+        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);
+      } finally {
+        SystemLimitException.decrementDecodeDepth();
       }
-      return unionReaders[selection].read(null, decoder);
     });
 
   }
@@ -478,48 +483,57 @@ private FieldReader createArrayReader(Schema readerSchema, Container action) thr
     boolean zeroByteElements = GenericDatumReader.isZeroByteSchema(elementType);
 
     return reusingReader((reuse, decoder) -> {
-      // Open a decode scope so the zero-byte element allocation cap is cumulative
-      // across every block of this array even when the fast reader is used
-      // standalone (i.e. without GenericDatumReader.read opening the outer datum
-      // scope); otherwise a huge array split into many small blocks would bypass
-      // the cap. The scope nests: when a datum scope is already open this simply
-      // accumulates into it, and only the outermost scope resets the running
-      // total (see SystemLimitException). The try/finally guarantees the scope is
-      // always closed so ThreadLocal state cannot leak into later decodes on the
-      // same thread.
-      SystemLimitException.beginCollectionAllocationScope();
+      // Descending into an array grows the decode call stack; bound the nesting
+      // depth first so a recursive schema cannot overflow the stack. Kept outside
+      // the collection-allocation scope below so that when the depth check throws
+      // (before incrementing) no unbalanced decrement occurs.
+      SystemLimitException.incrementDecodeDepth();
       try {
-        if (reuse instanceof GenericArray) {
-          GenericArray<Object> reuseArray = (GenericArray<Object>) reuse;
-          long l = decoder.readArrayStart();
-          checkArrayBlock(decoder, elementType, zeroByteElements, l);
-          reuseArray.clear();
-
-          while (l > 0) {
-            for (long i = 0; i < l; i++) {
-              reuseArray.add(elementReader.read(reuseArray.peek(), decoder));
-            }
-            l = decoder.arrayNext();
+        // Open a decode scope so the zero-byte element allocation cap is cumulative
+        // across every block of this array even when the fast reader is used
+        // standalone (i.e. without GenericDatumReader.read opening the outer datum
+        // scope); otherwise a huge array split into many small blocks would bypass
+        // the cap. The scope nests: when a datum scope is already open this simply
+        // accumulates into it, and only the outermost scope resets the running
+        // total (see SystemLimitException). The try/finally guarantees the scope is
+        // always closed so ThreadLocal state cannot leak into later decodes on the
+        // same thread.
+        SystemLimitException.beginCollectionAllocationScope();
+        try {
+          if (reuse instanceof GenericArray) {
+            GenericArray<Object> reuseArray = (GenericArray<Object>) reuse;
+            long l = decoder.readArrayStart();
             checkArrayBlock(decoder, elementType, zeroByteElements, l);
-          }
-          return reuseArray;
-        } else {
-          long l = decoder.readArrayStart();
-          checkArrayBlock(decoder, elementType, zeroByteElements, l);
-          List<Object> array = (reuse instanceof List) ? (List<Object>) reuse
-              : 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));
+            reuseArray.clear();
+
+            while (l > 0) {
+              for (long i = 0; i < l; i++) {
+                reuseArray.add(elementReader.read(reuseArray.peek(), decoder));
+              }
+              l = decoder.arrayNext();
+              checkArrayBlock(decoder, elementType, zeroByteElements, l);
             }
-            l = decoder.arrayNext();
+            return reuseArray;
+          } else {
+            long l = decoder.readArrayStart();
             checkArrayBlock(decoder, elementType, zeroByteElements, l);
+            List<Object> array = (reuse instanceof List) ? (List<Object>) reuse
+                : 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));
+              }
+              l = decoder.arrayNext();
+              checkArrayBlock(decoder, elementType, zeroByteElements, l);
+            }
+            return array;
           }
-          return array;
+        } finally {
+          SystemLimitException.endCollectionAllocationScope();
         }
       } finally {
-        SystemLimitException.endCollectionAllocationScope();
+        SystemLimitException.decrementDecodeDepth();
       }
     });
   }
@@ -637,11 +651,18 @@ public boolean canReuse() {
 
     @Override
     public Object read(Object reuse, Decoder decoder) throws IOException {
-      Object object = supplier.newInstance(reuse, schema);
-      for (ExecutionStep thisStep : readSteps) {
-        thisStep.execute(object, decoder);
+      // Bound decode nesting depth: a recursive schema fed deeply nested data
+      // would otherwise overflow the stack via this recursive descent.
+      SystemLimitException.incrementDecodeDepth();
+      try {
+        Object object = supplier.newInstance(reuse, schema);
+        for (ExecutionStep thisStep : readSteps) {
+          thisStep.execute(object, decoder);
+        }
+        return object;
+      } finally {
+        SystemLimitException.decrementDecodeDepth();
       }
-      return object;
     }
   }
 
@@ -657,19 +678,24 @@ public MapReader(FieldReader keyReader, FieldReader valueReader) {
 
     @Override
     public Object read(Object reuse, Decoder decoder) throws IOException {
-      long l = decoder.readMapStart();
-      Map<Object, Object> targetMap = new HashMap<>();
-
-      while (l > 0) {
-        for (int i = 0; i < l; i++) {
-          Object key = keyReader.read(null, decoder);
-          Object value = valueReader.read(null, decoder);
-          targetMap.put(key, value);
+      SystemLimitException.incrementDecodeDepth();
+      try {
+        long l = decoder.readMapStart();
+        Map<Object, Object> targetMap = new HashMap<>();
+
+        while (l > 0) {
+          for (int i = 0; i < l; i++) {
+            Object key = keyReader.read(null, decoder);
+            Object value = valueReader.read(null, decoder);
+            targetMap.put(key, value);
+          }
+          l = decoder.mapNext();
         }
-        l = decoder.mapNext();
-      }
 
-      return targetMap;
+        return targetMap;
+      } finally {
+        SystemLimitException.decrementDecodeDepth();
+      }
     }
   }