Fix stream buffer double release, header recursion, and eager chunk allocation

src/main/java/org/xerial/snappy/Snappy.java

@@ -428,6 +428,10 @@ public static int maxCompressedLength(int byteSize)
 
     /**
      * Zero-copy compress using memory addresses.
+     * <p/>
+     * <b>Unsafe:</b> this method cannot check memory bounds. The caller must ensure that destAddr points to at
+     * least {@link #maxCompressedLength(int)} bytes of writable memory; otherwise memory after it is overwritten.
+     * Use {@link #compress(ByteBuffer, ByteBuffer)} for bounds-checked compression of off-heap data.
      *
      * @param inputAddr input memory address
      * @param inputSize input byte size
@@ -443,6 +447,11 @@ public static long rawCompress(long inputAddr, long inputSize, long destAddr)
 
     /**
      * Zero-copy decompress using memory addresses.
+     * <p/>
+     * <b>Unsafe:</b> this method cannot check memory bounds, and the output size is read from the compressed
+     * data. The caller must check {@link #uncompressedLength(long, long)} against the capacity at destAddr before
+     * calling this method; otherwise untrusted input can overwrite memory after it. Use
+     * {@link #uncompress(ByteBuffer, ByteBuffer)} for bounds-checked decompression of off-heap data.
      *
      * @param inputAddr input memory address
      * @param inputSize input byte size

src/main/java/org/xerial/snappy/SnappyFramedInputStream.java

@@ -242,16 +242,21 @@ public SnappyFramedInputStream(ReadableByteChannel in,
      */
     private void allocateBuffersBasedOnSize(int size)
     {
+        // Clear each released buffer so that close() does not release it to the pool again if an allocation below
+        // fails, which would let two streams share the same buffer
         if (input != null) {
             bufferPool.releaseDirect(input);
+            input = null;
         }
 
         if (uncompressedDirect != null) {
             bufferPool.releaseDirect(uncompressedDirect);
+            uncompressedDirect = null;
         }
 
         if (buffer != null) {
             bufferPool.releaseArray(buffer);
+            buffer = null;
         }
 
         input = bufferPool.allocateDirect(size);
@@ -561,6 +566,10 @@ private boolean ensureBuffer()
             if (uncompressedLength > uncompressedDirect.capacity()) {
                 bufferPool.releaseDirect(uncompressedDirect);
                 bufferPool.releaseArray(buffer);
+                // Clear the released buffers first so that close() does not release them to the pool again if the
+                // allocation below fails, which would let two streams share the same buffer
+                uncompressedDirect = null;
+                buffer = null;
                 uncompressedDirect = bufferPool.allocateDirect(uncompressedLength);
                 buffer = bufferPool.allocateArray(uncompressedLength);
             }

src/main/java/org/xerial/snappy/SnappyInputStream.java

@@ -27,6 +27,7 @@
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Arrays;
 
 /**
  * A stream filter for reading data compressed by {@link SnappyOutputStream}.
@@ -37,6 +38,7 @@ public class SnappyInputStream
         extends InputStream
 {
     public static final int MAX_CHUNK_SIZE = 512 * 1024 * 1024; // 512 MiB
+    private static final int INITIAL_CHUNK_BUFFER_SIZE = 64 * 1024;
     // Maximum size of input without a SnappyOutputStream header, which is read entirely into a single byte array
     private static final int MAX_RAW_INPUT_SIZE = Integer.MAX_VALUE - 8;
 
@@ -428,24 +430,28 @@ protected boolean hasNextChunk()
         uncompressedCursor = 0;
         uncompressedLimit = 0;
 
-        int readBytes = readNext(header, 0, 4);
-        if (readBytes < 4) {
-            return false;
-        }
+        int readBytes;
+        int chunkSize;
+        // Skip the headers of concatenated streams in a loop rather than recursively, since a long run of 16-byte
+        // headers would otherwise exhaust the stack
+        while (true) {
+            readBytes = readNext(header, 0, 4);
+            if (readBytes < 4) {
+                return false;
+            }
 
-        int chunkSize = SnappyOutputStream.readInt(header, 0);
-        if (chunkSize == SnappyCodec.MAGIC_HEADER_HEAD) {
+            chunkSize = SnappyOutputStream.readInt(header, 0);
+            if (chunkSize != SnappyCodec.MAGIC_HEADER_HEAD) {
+                break;
+            }
             // Concatenated data
             int remainingHeaderSize = SnappyCodec.headerSize() - 4;
             readBytes = readNext(header, 4, remainingHeaderSize);
             if(readBytes < remainingHeaderSize) {
                 throw new SnappyIOException(SnappyErrorCode.FAILED_TO_UNCOMPRESS, String.format("Insufficient header size in a concatenated block"));
             }
 
-            if (isValidHeader(header)) {
-                return hasNextChunk();
-            }
-            else {
+            if (!isValidHeader(header)) {
                 return false;
             }
         }
@@ -460,14 +466,18 @@ protected boolean hasNextChunk()
             throw new SnappyError(SnappyErrorCode.FAILED_TO_UNCOMPRESS, String.format("Received chunkSize %,d is greater than max configured chunk size %,d", chunkSize, maxChunkSize));
         }
 
-        // extend the compressed data buffer size
-        if (compressed == null || chunkSize > compressed.length) {
-            // chunkSize exceeds limit
-            compressed = new byte[chunkSize];
+        if (compressed == null) {
+            compressed = new byte[Math.min(chunkSize, INITIAL_CHUNK_BUFFER_SIZE)];
         }
         readBytes = 0;
         while (readBytes < chunkSize) {
-            int ret = in.read(compressed, readBytes, chunkSize - readBytes);
+            if (readBytes == compressed.length) {
+                // Extend the compressed data buffer as the chunk data arrives, so that a chunk header alone cannot
+                // force a large allocation
+                compressed = Arrays.copyOf(compressed,
+                        (int) Math.min(chunkSize, Math.max(compressed.length * 2L, INITIAL_CHUNK_BUFFER_SIZE)));
+            }
+            int ret = in.read(compressed, readBytes, Math.min(chunkSize, compressed.length) - readBytes);
             if (ret == -1) {
                 break;
             }

src/test/java/org/xerial/snappy/SnappyBoundsCheckTest.java

@@ -498,4 +498,117 @@ public void write(byte[] b, int off, int len)
         }
         assertTrue(failAfterHeader.closed);
     }
+
+    // GHSA-6gp7-6wmv-gxqw: a long run of concatenated stream headers must not exhaust the stack
+    @Test
+    public void inputStreamWithManyConcatenatedHeaders()
+            throws Exception
+    {
+        ByteArrayOutputStream single = new ByteArrayOutputStream();
+        SnappyOutputStream out = new SnappyOutputStream(single);
+        out.write("hello".getBytes("UTF-8"));
+        out.close();
+        byte[] stream = single.toByteArray();
+        byte[] header = Arrays.copyOf(stream, SnappyCodec.headerSize());
+
+        ByteArrayOutputStream data = new ByteArrayOutputStream();
+        data.write(stream);
+        for (int i = 0; i < 100000; i++) {
+            data.write(header);
+        }
+        data.write(stream, header.length, stream.length - header.length);
+
+        byte[] result = readAll(new SnappyInputStream(new ByteArrayInputStream(data.toByteArray())));
+        assertEquals("hellohello", new String(result, "UTF-8"));
+    }
+
+    // GHSA-6gp7-6wmv-gxqw: a chunk header alone must not allocate the declared chunk size
+    @Test
+    public void inputStreamWithTruncatedLargeChunk()
+            throws Exception
+    {
+        ByteArrayOutputStream data = new ByteArrayOutputStream();
+        ByteArrayOutputStream single = new ByteArrayOutputStream();
+        new SnappyOutputStream(single).close();
+        data.write(Arrays.copyOf(single.toByteArray(), SnappyCodec.headerSize()));
+        // declare a 500 MiB chunk followed by only a few bytes
+        int chunkSize = 500 * 1024 * 1024;
+        data.write(new byte[] {(byte) (chunkSize >>> 24), (byte) (chunkSize >>> 16), (byte) (chunkSize >>> 8),
+                (byte) chunkSize, 1, 2, 3});
+
+        long before = usedHeap();
+        try {
+            readAll(new SnappyInputStream(new ByteArrayInputStream(data.toByteArray())));
+            fail("expected IOException");
+        }
+        catch (IOException e) {
+            // expected: the chunk is truncated
+        }
+        assertTrue("allocated too much memory for a truncated chunk", usedHeap() - before < 100L * 1024 * 1024);
+    }
+
+    private static long usedHeap()
+    {
+        Runtime rt = Runtime.getRuntime();
+        return rt.totalMemory() - rt.freeMemory();
+    }
+
+    // GHSA-c73m-r934-8qvg: a failed buffer allocation must not lead to releasing the same buffer twice
+    @Test
+    public void framedStreamDoesNotReleaseBuffersTwiceWhenAllocationFails()
+            throws Exception
+    {
+        final java.util.IdentityHashMap<Object, Boolean> released = new java.util.IdentityHashMap<Object, Boolean>();
+        org.xerial.snappy.pool.BufferPool pool = new org.xerial.snappy.pool.BufferPool()
+        {
+            @Override
+            public byte[] allocateArray(int size)
+            {
+                return new byte[size];
+            }
+
+            @Override
+            public void releaseArray(byte[] buffer)
+            {
+                assertNull("array released twice", released.put(buffer, Boolean.TRUE));
+            }
+
+            @Override
+            public ByteBuffer allocateDirect(int size)
+            {
+                if (size > 1024 * 1024) {
+                    throw new OutOfMemoryError("simulated");
+                }
+                return ByteBuffer.allocateDirect(size);
+            }
+
+            @Override
+            public void releaseDirect(ByteBuffer buffer)
+            {
+                assertNull("direct buffer released twice", released.put(buffer, Boolean.TRUE));
+            }
+        };
+
+        // a valid frame whose uncompressed size (2 MiB) exceeds the initial buffers
+        byte[] compressed = Snappy.compress(new byte[2 * 1024 * 1024]);
+        ByteArrayOutputStream body = new ByteArrayOutputStream();
+        body.write(new byte[4]);
+        body.write(compressed);
+        ByteArrayOutputStream stream = new ByteArrayOutputStream();
+        stream.write(SnappyFramed.HEADER_BYTES);
+        writeChunk(stream, 0x00, body.toByteArray());
+
+        SnappyFramedInputStream in = new SnappyFramedInputStream(
+                java.nio.channels.Channels.newChannel(new ByteArrayInputStream(stream.toByteArray())), false, pool);
+        try {
+            in.read();
+            fail("expected OutOfMemoryError");
+        }
+        catch (OutOfMemoryError e) {
+            // expected
+        }
+        finally {
+            in.close();
+        }
+    }
 }