From 35ac770f3dbd5307695d564d1b3272f54b0f0e6d Mon Sep 17 00:00:00 2001 From: Norman Maurer Date: Mon, 18 Jan 2021 14:02:37 +0100 Subject: [PATCH] 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/compression/JdkZlibDecoder.java | 162 +++++++++++------- .../codec/compression/JdkZlibTest.java | 30 ++++ .../handler/codec/compression/ZlibTest.java | 15 +- 3 files changed, 140 insertions(+), 67 deletions(-) diff --git a/codec/src/main/java/io/netty/handler/codec/compression/JdkZlibDecoder.java b/codec/src/main/java/io/netty/handler/codec/compression/JdkZlibDecoder.java index 83d7191f84..3c714139ae 100644 --- a/codec/src/main/java/io/netty/handler/codec/compression/JdkZlibDecoder.java +++ b/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) { - case FOOTER_START: - if (readGZIPFooter(in)) { - finished = true; + if (gzipState != GzipState.HEADER_END) { + if (gzipState == GzipState.FOOTER_START) { + if (!handleGzipFooter(in)) { + 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()) { - inflater.setInput(in.array(), in.arrayOffset() + in.readerIndex(), readableBytes); - } else { - byte[] array = new byte[readableBytes]; - in.getBytes(in.readerIndex(), array); - inflater.setInput(array); + if (inflater.needsInput()) { + if (in.hasArray()) { + inflater.setInput(in.array(), in.arrayOffset() + in.readerIndex(), readableBytes); + } else { + byte[] array = new byte[readableBytes]; + 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 { - if (inflater.needsDictionary()) { - if (dictionary == null) { - throw new DecompressionException( - "decompression failure, unable to set dictionary as non was specified"); - } - inflater.setDictionary(dictionary); + } else if (inflater.needsDictionary()) { + if (dictionary == null) { + throw new DecompressionException( + "decompression failure, unable to set dictionary as non was specified"); } + inflater.setDictionary(dictionary); } if (inflater.finished()) { @@ -265,20 +267,11 @@ public class JdkZlibDecoder extends ZlibDecoder { if (readFooter) { gzipState = GzipState.FOOTER_START; - if (readGZIPFooter(in)) { - finished = !decompressConcatenated; - - if (!finished) { - inflater.reset(); - crc.reset(); - gzipState = GzipState.HEADER_START; - } - } + handleGzipFooter(in); } } 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 (!in.isReadable()) { - return false; - } - do { - int b = in.readUnsignedByte(); - crc.update(b); - if (b == 0x00) { - break; - } - } while (in.isReadable()); + if (!skipIfNeeded(in, FNAME)) { + return false; } gzipState = GzipState.SKIP_COMMENT; // fall through case SKIP_COMMENT: - if ((flags & FCOMMENT) != 0) { - if (!in.isReadable()) { - return false; - } - do { - int b = in.readUnsignedByte(); - crc.update(b); - if (b == 0x00) { - break; - } - } while (in.isReadable()); + if (!skipIfNeeded(in, FCOMMENT)) { + return false; } 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; } /* diff --git a/codec/src/test/java/io/netty/handler/codec/compression/JdkZlibTest.java b/codec/src/test/java/io/netty/handler/codec/compression/JdkZlibTest.java index e9515a9a35..fcd9dbca6c 100644 --- a/codec/src/test/java/io/netty/handler/codec/compression/JdkZlibTest.java +++ b/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 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(); + } + } } diff --git a/codec/src/test/java/io/netty/handler/codec/compression/ZlibTest.java b/codec/src/test/java/io/netty/handler/codec/compression/ZlibTest.java index 3302a74712..dbd091488a 100644 --- a/codec/src/test/java/io/netty/handler/codec/compression/ZlibTest.java +++ b/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();