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).
@@ -819,12 +845,67 @@ public static void skip(Schema schema, Decoder in) throws IOException {
   private static void skipInternal(Schema schema, Decoder in) throws IOException {
     switch (schema.getType()) {
     case RECORD:
-      for (Field field : schema.getFields())
-        skipInternal(field.schema(), in);
+    case ARRAY:
+    case MAP:
+    case UNION:
+      // Skipping descends into nested structural values with the same recursive
+      // call chain as reading, so bound the nesting depth here too. Otherwise a
+      // recursive schema whose deeply nested value is skipped (a writer-only field
+      // during resolution, the fast reader's skip steps, or BinaryData.compare)
+      // would still overflow the stack.
+      SystemLimitException.incrementDecodeDepth();
+      try {
+        skipStructural(schema, in);
+      } finally {
+        SystemLimitException.decrementDecodeDepth();
+      }
       break;
     case ENUM:
       in.readEnum();
       break;
+    case FIXED:
+      in.skipFixed(schema.getFixedSize());
+      break;
+    case STRING:
+      in.skipString();
+      break;
+    case BYTES:
+      in.skipBytes();
+      break;
+    case INT:
+      in.readInt();
+      break;
+    case LONG:
+      in.readLong();
+      break;
+    case FLOAT:
+      in.readFloat();
+      break;
+    case DOUBLE:
+      in.readDouble();
+      break;
+    case BOOLEAN:
+      in.readBoolean();
+      break;
+    case NULL:
+      in.readNull();
+      break;
+    default:
+      throw new RuntimeException("Unknown type: " + schema);
+    }
+  }
+
+  /**
+   * Skips the structural (nesting) value types. Split out of
+   * {@link #skipInternal} so the decode-depth guard wraps exactly the types that
+   * grow the recursive call stack, mirroring {@link #readStructural}.
+   */
+  private static void skipStructural(Schema schema, Decoder in) throws IOException {
+    switch (schema.getType()) {
+    case RECORD:
+      for (Field field : schema.getFields())
+        skipInternal(field.schema(), in);
+      break;
     case ARRAY:
       Schema elementType = schema.getElementType();
       // Bound the cumulative element count: skipping a huge block of elements
@@ -865,35 +946,8 @@ private static void skipInternal(Schema schema, Decoder in) throws IOException {
     case UNION:
       skipInternal(schema.getTypes().get(in.readIndex()), in);
       break;
-    case FIXED:
-      in.skipFixed(schema.getFixedSize());
-      break;
-    case STRING:
-      in.skipString();
-      break;
-    case BYTES:
-      in.skipBytes();
-      break;
-    case INT:
-      in.readInt();
-      break;
-    case LONG:
-      in.readLong();
-      break;
-    case FLOAT:
-      in.readFloat();
-      break;
-    case DOUBLE:
-      in.readDouble();
-      break;
-    case BOOLEAN:
-      in.readBoolean();
-      break;
-    case NULL:
-      in.readNull();
-      break;
     default:
-      throw new RuntimeException("Unknown type: " + schema);
+      throw new RuntimeException("Not a structural type: " + schema);
     }
   }
 

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

@@ -23,6 +23,7 @@
 import org.apache.avro.Schema;
 import org.apache.avro.Schema.Field;
 import org.apache.avro.AvroRuntimeException;
+import org.apache.avro.SystemLimitException;
 import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.util.internal.ThreadLocalWithInitial;
 
@@ -70,11 +71,15 @@ public static int compare(byte[] b1, int s1, byte[] b2, int s2, Schema schema) {
   public static int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2, Schema schema) {
     Decoders decoders = DECODERS.get();
     decoders.set(b1, s1, l1, b2, s2, l2);
+    // Delimit a decode scope so the recursion-depth counter starts (and is reset)
+    // at this top-level comparison, mirroring GenericDatumReader.
+    SystemLimitException.beginCollectionAllocationScope();
     try {
       return compare(decoders, schema);
     } catch (IOException e) {
       throw new AvroRuntimeException(e);
     } finally {
+      SystemLimitException.endCollectionAllocationScope();
       decoders.clear();
     }
   }
@@ -84,6 +89,25 @@ public static int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2,
    * less than, return LT.
    */
   private static int compare(Decoders d, Schema schema) throws IOException {
+    switch (schema.getType()) {
+    case RECORD:
+    case ARRAY:
+    case UNION:
+      // These descend recursively into nested values, so bound the nesting depth
+      // to prevent a recursive schema comparing deeply nested data from
+      // overflowing the stack (mirrors GenericDatumReader's decode-depth guard).
+      SystemLimitException.incrementDecodeDepth();
+      try {
+        return compareStructural(d, schema);
+      } finally {
+        SystemLimitException.decrementDecodeDepth();
+      }
+    default:
+      return compareScalar(d, schema);
+    }
+  }
+
+  private static int compareStructural(Decoders d, Schema schema) throws IOException {
     Decoder d1 = d.d1;
     Decoder d2 = d.d2;
     switch (schema.getType()) {
@@ -101,17 +125,6 @@ private static int compare(Decoders d, Schema schema) throws IOException {
       }
       return 0;
     }
-    case ENUM:
-    case INT:
-      return Integer.compare(d1.readInt(), d2.readInt());
-    case LONG:
-      return Long.compare(d1.readLong(), d2.readLong());
-    case FLOAT:
-      return Float.compare(d1.readFloat(), d2.readFloat());
-    case DOUBLE:
-      return Double.compare(d1.readDouble(), d2.readDouble());
-    case BOOLEAN:
-      return Boolean.compare(d1.readBoolean(), d2.readBoolean());
     case ARRAY: {
       long i = 0; // position in array
       long r1 = 0, r2 = 0; // remaining in current block
@@ -146,14 +159,34 @@ private static int compare(Decoders d, Schema schema) throws IOException {
         }
       }
     }
-    case MAP:
-      throw new AvroRuntimeException("Can't compare maps!");
     case UNION: {
       int i1 = d1.readInt();
       int i2 = d2.readInt();
       int c = Integer.compare(i1, i2);
       return c == 0 ? compare(d, schema.getTypes().get(i1)) : c;
     }
+    default:
+      throw new AvroRuntimeException("Not a structural type to compare: " + schema);
+    }
+  }
+
+  private static int compareScalar(Decoders d, Schema schema) throws IOException {
+    Decoder d1 = d.d1;
+    Decoder d2 = d.d2;
+    switch (schema.getType()) {
+    case ENUM:
+    case INT:
+      return Integer.compare(d1.readInt(), d2.readInt());
+    case LONG:
+      return Long.compare(d1.readLong(), d2.readLong());
+    case FLOAT:
+      return Float.compare(d1.readFloat(), d2.readFloat());
+    case DOUBLE:
+      return Double.compare(d1.readDouble(), d2.readDouble());
+    case BOOLEAN:
+      return Boolean.compare(d1.readBoolean(), d2.readBoolean());
+    case MAP:
+      throw new AvroRuntimeException("Can't compare maps!");
     case FIXED: {
       int size = schema.getFixedSize();
       int c = compareBytes(d.d1.getBuf(), d.d1.getPos(), size, d.d2.getBuf(), d.d2.getPos(), size);