AVRO-4329: [Java] Bound bytes/string allocation on non-seekable streams

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

@@ -297,6 +297,16 @@ public double readDouble() throws IOException {
   @Override
   public Utf8 readString(Utf8 old) throws IOException {
     int length = SystemLimitException.checkMaxStringLength(readLong());
+    // Only take the bounded growing-buffer path when we would otherwise have to
+    // allocate a new backing array. When a reusable buffer of sufficient capacity
+    // is supplied there is no large up-front allocation to guard against, so honor
+    // the reuse contract and read straight into it.
+    if (length != 0 && requiresBoundedRead(length) && (old == null || length > old.getBytes().length)) {
+      // Large declared length on a non-seekable stream: read via a growing buffer
+      // so a truncated/hostile stream fails after a bounded allocation rather than
+      // allocating the full declared length up front. See requiresBoundedRead.
+      return new Utf8(readBoundedByteArray(length));
+    }
     ensureAvailableBytes(length);
     Utf8 result = (old != null ? old : new Utf8());
     result.setByteLength(length);
@@ -321,6 +331,18 @@ public void skipString() throws IOException {
   @Override
   public ByteBuffer readBytes(ByteBuffer old) throws IOException {
     int length = SystemLimitException.checkMaxBytesLength(readLong());
+    // Only take the bounded growing-buffer path when we would otherwise have to
+    // allocate a new buffer. A reusable buffer of sufficient capacity carries no
+    // large up-front allocation to guard against, so honor the reuse contract
+    // (see Decoder#readBytes) and read straight into it.
+    if (length != 0 && requiresBoundedRead(length) && (old == null || length > old.capacity())) {
+      // Large declared length on a non-seekable stream: read via a growing buffer
+      // so a truncated/hostile stream fails after a bounded allocation rather than
+      // allocating the full declared length up front. See requiresBoundedRead.
+      ByteBuffer result = ByteBuffer.wrap(readBoundedByteArray(length));
+      result.limit(length);
+      return result;
+    }
     ensureAvailableBytes(length);
     final ByteBuffer result;
     if (old != null && length <= old.capacity()) {
@@ -592,6 +614,68 @@ private void ensureAvailableBytes(int length) throws EOFException {
     }
   }
 
+  /**
+   * Largest buffer allocated up front for a length-prefixed {@code bytes} or
+   * {@code string} value whose backing data cannot be shown to be present (see
+   * {@link #requiresBoundedRead(int)}). Above this size the value is read via a
+   * buffer that grows in bounded steps so the cost of a truncated or hostile
+   * declared length is proportional to the bytes actually delivered, not to the
+   * (attacker-chosen) declared length.
+   */
+  static final int MAX_UNVERIFIED_ALLOCATION = 16 * 1024;
+
+  /**
+   * Whether a length-prefixed value of the given declared {@code length} must be
+   * read via the bounded growing-buffer path rather than a single up-front
+   * allocation.
+   * <p>
+   * The up-front allocation is safe when the source can report that at least
+   * {@code length} bytes remain (a memory-backed or seekable source):
+   * {@link #ensureAvailableBytes(int)} rejects an over-long declaration before
+   * any allocation. On a non-seekable stream (socket, pipe, decompression stream)
+   * the remaining byte count is unknown, so a huge declared length would
+   * otherwise drive a single large allocation before a single payload byte is
+   * read. In that case, for lengths above {@link #MAX_UNVERIFIED_ALLOCATION},
+   * read incrementally instead.
+   *
+   * @param length the declared length of the value to read
+   * @return {@code true} if the value should be read incrementally
+   */
+  private boolean requiresBoundedRead(int length) {
+    return length > MAX_UNVERIFIED_ALLOCATION && (source == null || source.remainingBytes() < 0);
+  }
+
+  /**
+   * Reads exactly {@code length} bytes into a freshly allocated array, growing
+   * the backing buffer in bounded steps (starting at
+   * {@link #MAX_UNVERIFIED_ALLOCATION} and doubling, capped at {@code length}).
+   * <p>
+   * This is used when the number of bytes remaining is unknown so that a
+   * truncated or hostile stream declaring a huge length fails with an
+   * {@link EOFException} after a bounded allocation, instead of allocating the
+   * full declared length up front. The returned array is exactly {@code length}
+   * bytes long, so callers may take ownership of it without copying.
+   *
+   * @param length the number of bytes to read; must be positive
+   * @return a newly allocated array of exactly {@code length} bytes
+   * @throws EOFException if the source is exhausted before {@code length} bytes
+   *                      are read
+   */
+  byte[] readBoundedByteArray(int length) throws IOException {
+    byte[] data = new byte[Math.min(length, MAX_UNVERIFIED_ALLOCATION)];
+    int read = 0;
+    while (read < length) {
+      if (read == data.length) {
+        int next = (int) Math.min((long) length, (long) data.length * 2);
+        data = Arrays.copyOf(data, next);
+      }
+      int chunk = data.length - read;
+      doReadBytes(data, read, chunk);
+      read += chunk;
+    }
+    return data;
+  }
+
   /**
    * Returns an {@link java.io.InputStream} that is aware of any buffering that
    * may occur in this BinaryDecoder. Readers that need to interleave decoding

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

@@ -155,8 +155,21 @@ public double readDouble() throws IOException {
 
   @Override
   public ByteBuffer readBytes(ByteBuffer old) throws IOException {
-    long length = readLong();
-    return byteReader.read(old, SystemLimitException.checkMaxBytesLength(length));
+    int length = SystemLimitException.checkMaxBytesLength(readLong());
+    // A DirectBinaryDecoder reads straight from a (typically non-seekable) stream,
+    // so the number of bytes remaining is unknown. For a large declared length
+    // that cannot be satisfied from a reusable buffer, read via a bounded growing
+    // buffer so a truncated/hostile stream fails after a bounded allocation rather
+    // than allocating the full declared length up front. The in-memory
+    // ByteBufferInputStream path (ReuseByteReader) is left untouched: it slices
+    // from buffers already held, so it does not over-allocate from the length.
+    if (length > MAX_UNVERIFIED_ALLOCATION && (old == null || old.capacity() < length)
+        && !(byteReader instanceof ReuseByteReader)) {
+      ByteBuffer result = ByteBuffer.wrap(readBoundedByteArray(length));
+      result.limit(length);
+      return result;
+    }
+    return byteReader.read(old, length);
   }
 
   @Override

lang/java/avro/src/test/java/org/apache/avro/io/TestBinaryDecoderBoundedRead.java

@@ -0,0 +1,198 @@
+/*
+ * 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.
+ */
+package org.apache.avro.io;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.EOFException;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+/**
+ * Regression tests for AVRO-4303: a {@code bytes}/{@code string} length prefix
+ * read from a non-seekable stream must not drive a single up-front allocation
+ * of the (attacker-declared) length. A truncated or hostile stream declaring a
+ * huge length must fail with a bounded {@link EOFException} after only a
+ * bounded allocation, rather than an {@link OutOfMemoryError} or a huge
+ * allocation.
+ */
+public class TestBinaryDecoderBoundedRead {
+
+  /**
+   * Wraps the bytes in a stream whose class is neither
+   * {@link ByteArrayInputStream} nor {@code ByteBufferInputStream}, so the
+   * decoder reports an unknown number of remaining bytes (the non-seekable case).
+   */
+  private static InputStream nonSeekable(byte[] data) {
+    return new FilterInputStream(new ByteArrayInputStream(data)) {
+    };
+  }
+
+  /**
+   * Encodes a bytes/string length prefix followed by {@code payload} raw bytes.
+   */
+  private static byte[] lengthPrefixed(long declaredLength, byte[] payload) throws IOException {
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    BinaryEncoder e = EncoderFactory.get().directBinaryEncoder(out, null);
+    e.writeLong(declaredLength);
+    e.flush();
+    out.write(payload);
+    return out.toByteArray();
+  }
+
+  /**
+   * A near-2GB declared length that a single {@code new byte[len]} could not
+   * satisfy on a normal test heap, so reaching an {@link EOFException} proves the
+   * decoder never attempted the full up-front allocation.
+   */
+  private static final long HUGE_LENGTH = Integer.MAX_VALUE - 8L;
+
+  @Test
+  @Timeout(value = 30, unit = TimeUnit.SECONDS)
+  void bufferedDecoderRejectsHugeBytesLengthOnStreamWithoutHugeAllocation() throws IOException {
+    byte[] data = lengthPrefixed(HUGE_LENGTH, new byte[] { 1, 2, 3, 4, 5 });
+    BinaryDecoder d = DecoderFactory.get().binaryDecoder(nonSeekable(data), null);
+    assertThrows(EOFException.class, () -> d.readBytes(null));
+  }
+
+  @Test
+  @Timeout(value = 30, unit = TimeUnit.SECONDS)
+  void bufferedDecoderRejectsHugeStringLengthOnStreamWithoutHugeAllocation() throws IOException {
+    byte[] data = lengthPrefixed(HUGE_LENGTH, new byte[] { 'a', 'b', 'c' });
+    BinaryDecoder d = DecoderFactory.get().binaryDecoder(nonSeekable(data), null);
+    assertThrows(EOFException.class, () -> d.readString(null));
+  }
+
+  @Test
+  @Timeout(value = 30, unit = TimeUnit.SECONDS)
+  void directDecoderRejectsHugeBytesLengthOnStreamWithoutHugeAllocation() throws IOException {
+    byte[] data = lengthPrefixed(HUGE_LENGTH, new byte[] { 1, 2, 3, 4, 5 });
+    BinaryDecoder d = DecoderFactory.get().directBinaryDecoder(nonSeekable(data), null);
+    assertThrows(EOFException.class, () -> d.readBytes(null));
+  }
+
+  @Test
+  @Timeout(value = 30, unit = TimeUnit.SECONDS)
+  void directDecoderRejectsHugeStringLengthOnStreamWithoutHugeAllocation() throws IOException {
+    byte[] data = lengthPrefixed(HUGE_LENGTH, new byte[] { 'a', 'b', 'c' });
+    BinaryDecoder d = DecoderFactory.get().directBinaryDecoder(nonSeekable(data), null);
+    assertThrows(EOFException.class, () -> d.readString(null));
+  }
+
+  @Test
+  void legitimateLargeBytesStillRoundTripsOnNonSeekableStream() throws IOException {
+    // Larger than the bounded-read threshold so it exercises the growing-buffer
+    // path, but a genuinely present payload must still decode intact.
+    byte[] payload = new byte[1_000_000];
+    for (int i = 0; i < payload.length; i++) {
+      payload[i] = (byte) i;
+    }
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    BinaryEncoder e = EncoderFactory.get().directBinaryEncoder(out, null);
+    e.writeBytes(payload);
+    e.flush();
+
+    BinaryDecoder d = DecoderFactory.get().binaryDecoder(nonSeekable(out.toByteArray()), null);
+    ByteBuffer result = d.readBytes(null);
+    byte[] decoded = new byte[result.remaining()];
+    result.get(decoded);
+    assertArrayEquals(payload, decoded);
+  }
+
+  @Test
+  void reusableBufferWithSufficientCapacityIsHonoredOnNonSeekableStream() throws IOException {
+    // A large value on a non-seekable stream, supplied with a reusable buffer that
+    // is already big enough, must reuse it (Decoder#readBytes reuse contract)
+    // rather than take the bounded-read path and allocate a new array.
+    byte[] payload = new byte[1_000_000];
+    for (int i = 0; i < payload.length; i++) {
+      payload[i] = (byte) i;
+    }
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    BinaryEncoder e = EncoderFactory.get().directBinaryEncoder(out, null);
+    e.writeBytes(payload);
+    e.flush();
+
+    ByteBuffer reusable = ByteBuffer.allocate(payload.length + 100);
+    BinaryDecoder d = DecoderFactory.get().binaryDecoder(nonSeekable(out.toByteArray()), null);
+    ByteBuffer result = d.readBytes(reusable);
+    assertSame(reusable.array(), result.array(), "supplied buffer with sufficient capacity should be reused");
+    byte[] decoded = new byte[result.remaining()];
+    result.get(decoded);
+    assertArrayEquals(payload, decoded);
+  }
+
+  @Test
+  void legitimateLargeStringStillRoundTripsOnNonSeekableStream() throws IOException {
+    StringBuilder sb = new StringBuilder();
+    while (sb.length() < 100_000) {
+      sb.append("avro-4303-");
+    }
+    String expected = sb.toString();
+    byte[] utf8 = expected.getBytes(StandardCharsets.UTF_8);
+
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    BinaryEncoder e = EncoderFactory.get().directBinaryEncoder(out, null);
+    e.writeString(expected);
+    e.flush();
+
+    BinaryDecoder d = DecoderFactory.get().directBinaryDecoder(nonSeekable(out.toByteArray()), null);
+    String decoded = d.readString();
+    assertEquals(expected, decoded);
+    // Sanity: the bounded path produced exactly the encoded bytes.
+    assertArrayEquals(utf8, decoded.getBytes(StandardCharsets.UTF_8));
+  }
+
+  @Test
+  void smallValuesAndSeekableSourcesKeepDirectAllocationPath() throws IOException {
+    // Small value: below the bounded-read threshold, read directly.
+    byte[] small = new byte[100];
+    Arrays.fill(small, (byte) 7);
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    BinaryEncoder e = EncoderFactory.get().directBinaryEncoder(out, null);
+    e.writeBytes(small);
+    e.flush();
+    byte[] encoded = out.toByteArray();
+
+    // Seekable (byte-array) source: existing available-bytes guard rejects an
+    // over-long declared length up front, unchanged by AVRO-4303.
+    byte[] truncated = lengthPrefixed(HUGE_LENGTH, new byte[] { 1, 2, 3 });
+    BinaryDecoder seekable = DecoderFactory.get().binaryDecoder(truncated, null);
+    assertThrows(EOFException.class, () -> seekable.readBytes(null));
+
+    // Small value still decodes on a non-seekable stream via the direct path.
+    BinaryDecoder d = DecoderFactory.get().binaryDecoder(nonSeekable(encoded), null);
+    ByteBuffer result = d.readBytes(null);
+    byte[] decoded = new byte[result.remaining()];
+    result.get(decoded);
+    assertArrayEquals(small, decoded);
+  }
+}