GH-XXXX: Defer byte[] allocation until after bounds check

Also, document that when enable_unsafe_memory_access is enabled,
all bets are off.

Reported by n0mi1k.

Author: lidavidm

memory/memory-core/src/main/java/org/apache/arrow/memory/ArrowBuf.java

@@ -731,6 +731,24 @@ public void getBytes(long index, byte[] dst, int dstIndex, int length) {
     }
   }
 
+  /**
+   * Copy data from this ArrowBuf into a newly allocated array.
+   *
+   * <p>This method is more resilient to invalid data inadvertently causing large allocations,
+   * as the byte[] will not be allocated until we check the length.
+   *
+   * @param index index (0 based relative to the portion of memory this ArrowBuf has access to)
+   * @param length length of data to copy from this ArrowBuf
+   */
+  public byte[] getBytesAsArray(long index, int length) {
+    checkIndex(index, length);
+    byte[] dst = new byte[length];
+    if (length != 0) {
+      MemoryUtil.copyFromMemory(addr(index), dst, 0, length);
+    }
+    return dst;
+  }
+
   /**
    * Copy data from a given byte array into this ArrowBuf starting at a given index.
    *
@@ -1008,8 +1026,7 @@ public int setBytes(long index, InputStream in, int length) throws IOException {
   /**
    * Copy a certain length of bytes from this ArrowBuf at a given index into the given OutputStream.
    *
-   * @param index index index (0 based relative to the portion of memory this ArrowBuf has access
-   *     to)
+   * @param index index (0 based relative to the portion of memory this ArrowBuf has access to)
    * @param out dst stream to copy data into
    * @param length length of data to copy
    * @throws IOException on failing to write to stream

memory/memory-core/src/main/java/org/apache/arrow/memory/BoundsChecking.java

@@ -24,6 +24,10 @@
  * "arrow.enable_unsafe_memory_access" or "drill.enable_unsafe_memory_access". The latter is
  * deprecated. The environmental variable is named "ARROW_ENABLE_UNSAFE_MEMORY_ACCESS". When both
  * the system property and the environmental variable are set, the system property takes precedence.
+ *
+ * <p>WARNING: disabling bounds checking means that out-of-bounds memory access is possible! This can
+ * lead to security vulnerabilities. You should not read or write untrusted data when bounds checking
+ * is disabled.
  */
 public class BoundsChecking {
 

vector/src/main/java/org/apache/arrow/vector/BaseVariableWidthViewVector.java

@@ -1556,7 +1556,6 @@ public int hashCode(int index, ArrowBufHasher hasher) {
    */
   protected byte[] getData(int index) {
     final int dataLength = getValueLength(index);
-    byte[] result = new byte[dataLength];
     if (dataLength > INLINE_SIZE) {
       // data is in the data buffer
       // get buffer index
@@ -1566,12 +1565,11 @@ protected byte[] getData(int index) {
       final int dataOffset =
           viewBuffer.getInt(
               ((long) index * ELEMENT_SIZE) + LENGTH_WIDTH + PREFIX_WIDTH + BUF_INDEX_WIDTH);
-      dataBuffers.get(bufferIndex).getBytes(dataOffset, result, 0, dataLength);
-    } else {
-      // data is in the view buffer
-      viewBuffer.getBytes((long) index * ELEMENT_SIZE + BUF_INDEX_WIDTH, result, 0, dataLength);
+      ArrowBuf dataBuffer = dataBuffers.get(bufferIndex);
+      return dataBuffer.getBytesAsArray(dataOffset, dataLength);
     }
-    return result;
+    // data is in the view buffer
+    return viewBuffer.getBytesAsArray((long) index * ELEMENT_SIZE + BUF_INDEX_WIDTH, dataLength);
   }
 
   protected void getData(int index, ReusableBuffer<?> buffer) {

vector/src/test/java/org/apache/arrow/vector/TestVariableWidthViewVector.java

@@ -2921,4 +2921,20 @@ public void testValidate() {
       assertTrue(e.getMessage().contains("Not enough capacity for data buffer"));
     }
   }
+
+  @Test
+  public void testValidateInvalidOffsets() {
+    try (final ViewVarCharVector vector = new ViewVarCharVector("v", allocator)) {
+      vector.allocateNew(8, 1);
+      vector.allocateOrGetLastDataBuffer(8);
+      var offsets = vector.getDataBuffer();
+      offsets.setInt(0, 64);
+      offsets.setInt(1, 0);
+      offsets.setInt(2, 0);
+      offsets.setInt(3, 1024);
+      vector.setValueCount(1);
+      vector.setIndexDefined(0);
+      assertThrows(IndexOutOfBoundsException.class, vector::validateFull);
+    }
+  }
 }