Correctly handle fragmentation in JdkZlibDecoder (#10948)

Motivation:

We had multiple bugs in JdkZlibDecoder which could lead to decoding errors when the data was received in a fragmentated manner.

Modifications:

- Correctly handle skipping of comments
- Correctly handle footer / header decoding
- Add unit test that verifies the correct handling of fragmentation

Result:

Fixes https://github.com/netty/netty/issues/10875

codec/src/main/java/io/netty/handler/codec/compression/JdkZlibDecoder.java

@@ -201,29 +201,32 @@ public class JdkZlibDecoder extends ZlibDecoder {
         }
 
         if (crc != null) {
-            switch (gzipState) {
+            if (gzipState != GzipState.HEADER_END) {
-                case FOOTER_START:
+                if (gzipState == GzipState.FOOTER_START) {
-                    if (readGZIPFooter(in)) {
+                    if (!handleGzipFooter(in)) {
-                        finished = true;
+                        return;
                     }
+                } else {
+                    if (!readGZIPHeader(in)) {
+                        return;
+                    }
+                }
+                // Some bytes may have been consumed, and so we must re-set the number of readable bytes.
+                readableBytes = in.readableBytes();
+                if (readableBytes == 0) {
                     return;
-                default:
+                }
-                    if (gzipState != GzipState.HEADER_END) {
-                        if (!readGZIPHeader(in)) {
-                            return;
-                        }
-                    }
             }
-            // Some bytes may have been consumed, and so we must re-set the number of readable bytes.
-            readableBytes = in.readableBytes();
         }
 
-        if (in.hasArray()) {
+        if (inflater.needsInput()) {
-            inflater.setInput(in.array(), in.arrayOffset() + in.readerIndex(), readableBytes);
+            if (in.hasArray()) {
-        } else {
+                inflater.setInput(in.array(), in.arrayOffset() + in.readerIndex(), readableBytes);
-            byte[] array = new byte[readableBytes];
+            } else {
-            in.getBytes(in.readerIndex(), array);
+                byte[] array = new byte[readableBytes];
-            inflater.setInput(array);
+                in.getBytes(in.readerIndex(), array);
+                inflater.setInput(array);
+            }
         }
 
         ByteBuf decompressed = prepareDecompressBuffer(ctx, null, inflater.getRemaining() << 1);
@@ -233,20 +236,19 @@ public class JdkZlibDecoder extends ZlibDecoder {
                 byte[] outArray = decompressed.array();
                 int writerIndex = decompressed.writerIndex();
                 int outIndex = decompressed.arrayOffset() + writerIndex;
-                int outputLength = inflater.inflate(outArray, outIndex, decompressed.writableBytes());
+                int writable = decompressed.writableBytes();
+                int outputLength = inflater.inflate(outArray, outIndex, writable);
                 if (outputLength > 0) {
                     decompressed.writerIndex(writerIndex + outputLength);
                     if (crc != null) {
                         crc.update(outArray, outIndex, outputLength);
                     }
-                } else {
+                } else  if (inflater.needsDictionary()) {
-                    if (inflater.needsDictionary()) {
+                    if (dictionary == null) {
-                        if (dictionary == null) {
+                        throw new DecompressionException(
-                            throw new DecompressionException(
+                                "decompression failure, unable to set dictionary as non was specified");
-                                    "decompression failure, unable to set dictionary as non was specified");
-                        }
-                        inflater.setDictionary(dictionary);
                     }
+                    inflater.setDictionary(dictionary);
                 }
 
                 if (inflater.finished()) {
@@ -265,20 +267,11 @@ public class JdkZlibDecoder extends ZlibDecoder {
 
             if (readFooter) {
                 gzipState = GzipState.FOOTER_START;
-                if (readGZIPFooter(in)) {
+                handleGzipFooter(in);
-                    finished = !decompressConcatenated;
-
-                    if (!finished) {
-                        inflater.reset();
-                        crc.reset();
-                        gzipState = GzipState.HEADER_START;
-                    }
-                }
             }
         } catch (DataFormatException e) {
             throw new DecompressionException("decompression failure", e);
         } finally {
-
             if (decompressed.isReadable()) {
                 ctx.fireChannelRead(decompressed);
             } else {
@@ -287,6 +280,20 @@ public class JdkZlibDecoder extends ZlibDecoder {
         }
     }
 
+    private boolean handleGzipFooter(ByteBuf in) {
+        if (readGZIPFooter(in)) {
+            finished = !decompressConcatenated;
+
+            if (!finished) {
+                inflater.reset();
+                crc.reset();
+                gzipState = GzipState.HEADER_START;
+                return true;
+            }
+        }
+        return false;
+    }
+
     @Override
     protected void decompressionBufferExhausted(ByteBuf buffer) {
         finished = true;
@@ -365,41 +372,22 @@ public class JdkZlibDecoder extends ZlibDecoder {
                 gzipState = GzipState.SKIP_FNAME;
                 // fall through
             case SKIP_FNAME:
-                if ((flags & FNAME) != 0) {
+                if (!skipIfNeeded(in, FNAME)) {
-                    if (!in.isReadable()) {
+                    return false;
-                        return false;
-                    }
-                    do {
-                        int b = in.readUnsignedByte();
-                        crc.update(b);
-                        if (b == 0x00) {
-                            break;
-                        }
-                    } while (in.isReadable());
                 }
                 gzipState = GzipState.SKIP_COMMENT;
                 // fall through
             case SKIP_COMMENT:
-                if ((flags & FCOMMENT) != 0) {
+                if (!skipIfNeeded(in, FCOMMENT)) {
-                    if (!in.isReadable()) {
+                    return false;
-                        return false;
-                    }
-                    do {
-                        int b = in.readUnsignedByte();
-                        crc.update(b);
-                        if (b == 0x00) {
-                            break;
-                        }
-                    } while (in.isReadable());
                 }
                 gzipState = GzipState.PROCESS_FHCRC;
                 // fall through
             case PROCESS_FHCRC:
                 if ((flags & FHCRC) != 0) {
-                    if (in.readableBytes() < 4) {
+                    if (!verifyCrc(in)) {
                         return false;
                     }
-                    verifyCrc(in);
                 }
                 crc.reset();
                 gzipState = GzipState.HEADER_END;
@@ -411,17 +399,50 @@ public class JdkZlibDecoder extends ZlibDecoder {
         }
     }
 
-    private boolean readGZIPFooter(ByteBuf buf) {
+    /**
-        if (buf.readableBytes() < 8) {
+     * Skip bytes in the input if needed until we find the end marker {@code 0x00}.
+     * @param   in the input
+     * @param   flagMask the mask that should be present in the {@code flags} when we need to skip bytes.
+     * @return  {@code true} if the operation is complete and we can move to the next state, {@code false} if we need
+     *          the retry again once we have more readable bytes.
+     */
+    private boolean skipIfNeeded(ByteBuf in, int flagMask) {
+        if ((flags & flagMask) != 0) {
+            for (;;) {
+                if (!in.isReadable()) {
+                    // We didnt find the end yet, need to retry again once more data is readable
+                    return false;
+                }
+                int b = in.readUnsignedByte();
+                crc.update(b);
+                if (b == 0x00) {
+                    break;
+                }
+            }
+        }
+        // Skip is handled, we can move to the next processing state.
+        return true;
+    }
+
+    /**
+     * Read the GZIP footer.
+     *
+     * @param   in the input.
+     * @return  {@code true} if the footer could be read, {@code false} if the read could not be performed as
+     *          the input {@link ByteBuf} doesn't have enough readable bytes (8 bytes).
+     */
+    private boolean readGZIPFooter(ByteBuf in) {
+        if (in.readableBytes() < 8) {
             return false;
         }
 
-        verifyCrc(buf);
+        boolean enoughData = verifyCrc(in);
+        assert enoughData;
 
         // read ISIZE and verify
         int dataLength = 0;
         for (int i = 0; i < 4; ++i) {
-            dataLength |= buf.readUnsignedByte() << i * 8;
+            dataLength |= in.readUnsignedByte() << i * 8;
         }
         int readLength = inflater.getTotalOut();
         if (dataLength != readLength) {
@@ -431,7 +452,17 @@ public class JdkZlibDecoder extends ZlibDecoder {
         return true;
     }
 
-    private void verifyCrc(ByteBuf in) {
+    /**
+     * Verifies CRC.
+     *
+     * @param   in the input.
+     * @return  {@code true} if verification could be performed, {@code false} if verification could not be performed as
+     *          the input {@link ByteBuf} doesn't have enough readable bytes (4 bytes).
+     */
+    private boolean verifyCrc(ByteBuf in) {
+        if (in.readableBytes() < 4) {
+            return false;
+        }
         long crcValue = 0;
         for (int i = 0; i < 4; ++i) {
             crcValue |= (long) in.readUnsignedByte() << i * 8;
@@ -441,6 +472,7 @@ public class JdkZlibDecoder extends ZlibDecoder {
             throw new DecompressionException(
                     "CRC value mismatch. Expected: " + crcValue + ", Got: " + readCrc);
         }
+        return true;
     }
 
     /*

codec/src/test/java/io/netty/handler/codec/compression/JdkZlibTest.java

@@ -90,4 +90,34 @@ public class JdkZlibTest extends ZlibTest {
             chDecoderGZip.close();
         }
     }
+
+    @Test
+    public void testConcatenatedStreamsReadFullyWhenFragmented() throws IOException {
+        EmbeddedChannel chDecoderGZip = new EmbeddedChannel(new JdkZlibDecoder(true));
+
+        try {
+            byte[] bytes = IOUtils.toByteArray(getClass().getResourceAsStream("/multiple.gz"));
+
+            // Let's feed the input byte by byte to simulate fragmentation.
+            ByteBuf buf = Unpooled.copiedBuffer(bytes);
+            boolean written = false;
+            while (buf.isReadable()) {
+                written |= chDecoderGZip.writeInbound(buf.readRetainedSlice(1));
+            }
+            buf.release();
+
+            assertTrue(written);
+            Queue<Object> messages = chDecoderGZip.inboundMessages();
+            assertEquals(2, messages.size());
+
+            for (String s : Arrays.asList("a", "b")) {
+                ByteBuf msg = (ByteBuf) messages.poll();
+                assertEquals(s, msg.toString(CharsetUtil.UTF_8));
+                ReferenceCountUtil.release(msg);
+            }
+        } finally {
+            assertFalse(chDecoderGZip.finish());
+            chDecoderGZip.close();
+        }
+    }
 }

codec/src/test/java/io/netty/handler/codec/compression/ZlibTest.java

@@ -105,9 +105,20 @@ public abstract class ZlibTest {
 
         EmbeddedChannel chDecoderGZip = new EmbeddedChannel(createDecoder(ZlibWrapper.GZIP));
         try {
-            chDecoderGZip.writeInbound(deflatedData);
+            while (deflatedData.isReadable()) {
+                chDecoderGZip.writeInbound(deflatedData.readRetainedSlice(1));
+            }
+            deflatedData.release();
             assertTrue(chDecoderGZip.finish());
-            ByteBuf buf = chDecoderGZip.readInbound();
+            ByteBuf buf = Unpooled.buffer();
+            for (;;) {
+                ByteBuf b = chDecoderGZip.readInbound();
+                if (b == null) {
+                    break;
+                }
+                buf.writeBytes(b);
+                b.release();
+            }
             assertEquals(buf, data);
             assertNull(chDecoderGZip.readInbound());
             data.release();