handle BFINAL deflate blocks in permessage-deflate decoders

httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtension.java

@@ -106,7 +106,7 @@ public ByteBuffer decode(final WebSocketFrameType type,
         final byte[] buffer = new byte[Math.min(16384, Math.max(1024, input.length * 2))];
         long produced = 0L;
         try {
-            while (!inflater.needsInput()) {
+            while (!inflater.needsInput() && !inflater.finished()) {
                 final int count = inflater.inflate(buffer);
                 if (count == 0 && inflater.needsInput()) {
                     break;
@@ -124,6 +124,11 @@ public ByteBuffer decode(final WebSocketFrameType type,
         } catch (final Exception ex) {
             throw new WebSocketException("Unable to inflate payload", ex);
         }
+        if (inflater.finished()) {
+            // A DEFLATE block with BFINAL set ends the stream (RFC 7692 section 7.2.3.4);
+            // the sender starts a new one for the data that follows.
+            inflater.reset();
+        }
         if (fin) {
             decodingMessage = false;
             if (clientNoContextTakeover) {

httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/PerMessageDeflate.java

@@ -190,7 +190,9 @@ public byte[] decode(final byte[] compressedMessage, final long maxDecodedSize)
                         break;
                     }
                 }
-                if (serverNoContextTakeover) {
+                // A DEFLATE block with BFINAL set ends the stream (RFC 7692 section 7.2.3.4);
+                // the next message then starts a new one.
+                if (serverNoContextTakeover || inf.finished()) {
                     inf.reset();
                 }
                 return out.toByteArray();

httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionTest.java

@@ -29,10 +29,12 @@
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
 
 import java.io.ByteArrayOutputStream;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.zip.Deflater;
 
@@ -96,6 +98,19 @@ void decodeZeroLimitMeansUnlimited() throws Exception {
         assertArrayEquals(plain, toBytes(out));
     }
 
+    @Test
+    void decodesMessagesEndingWithBfinalBlock() {
+        // RFC 7692 section 7.2.3.4: "Hello" compressed using a DEFLATE block with BFINAL set to 1.
+        final byte[] compressed = {(byte) 0xf3, 0x48, (byte) 0xcd, (byte) 0xc9, (byte) 0xc9, 0x07, 0x00, 0x00};
+        final byte[] plain = "Hello".getBytes(StandardCharsets.UTF_8);
+
+        final PerMessageDeflateExtension ext = new PerMessageDeflateExtension();
+        assertTimeoutPreemptively(Duration.ofSeconds(5), () -> {
+            assertArrayEquals(plain, toBytes(ext.decode(WebSocketFrameType.TEXT, true, ByteBuffer.wrap(compressed))));
+            assertArrayEquals(plain, toBytes(ext.decode(WebSocketFrameType.TEXT, true, ByteBuffer.wrap(compressed))));
+        });
+    }
+
     @Test
     void fragmentedEncodeRoundTripsThroughDecode() throws Exception {
         // Encode a message as three frames (fin=false, false, true), then decode them back.

httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/MessageDeflateTest.java

@@ -125,6 +125,19 @@ void encode_emptyMessage_withNoContextTakeover_roundTrips() throws Exception {
         assertArrayEquals(new byte[0], dec.decode(empty));
     }
 
+    @Test
+    void decode_messagesEndingWithBfinalBlock_withContextTakeover() throws Exception {
+        // RFC 7692 section 7.2.3.4: "Hello" compressed using a DEFLATE block with BFINAL set to 1.
+        final byte[] wire = {(byte) 0xf3, 0x48, (byte) 0xcd, (byte) 0xc9, (byte) 0xc9, 0x07, 0x00, 0x00};
+        final byte[] plain = "Hello".getBytes(StandardCharsets.UTF_8);
+
+        final PerMessageDeflate pmce = new PerMessageDeflate(true, false, false, null, null);
+        final WebSocketExtensionChain.Decoder dec = pmce.newDecoder();
+
+        assertArrayEquals(plain, dec.decode(wire));
+        assertArrayEquals(plain, dec.decode(wire), "message after a BFINAL block must not decode to empty");
+    }
+
     @Test
     void decode_withinLimit_succeeds() throws Exception {
         final PerMessageDeflate pmce = new PerMessageDeflate(true, true, true, null, null);