Harden WebSocket transport and permessage-deflate for RFC 6455/7692 conformance

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/websocket/H2WebSocketEchoIT.java

@@ -169,7 +169,6 @@ void echoesOverHttp2ExtendedConnectWithPmce() throws Exception {
         final WebSocketClientConfig cfg = WebSocketClientConfig.custom()
                 .enableHttp2(true)
                 .enablePerMessageDeflate(true)
-                .offerClientMaxWindowBits(15)
                 .setCloseWaitTimeout(Timeout.ofSeconds(2))
                 .build();
 

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/websocket/WebSocketClientTest.java

@@ -186,7 +186,6 @@ void echo_compressed_pmce() throws Exception {
                 .enablePerMessageDeflate(true)
                 .offerServerNoContextTakeover(true)
                 .offerClientNoContextTakeover(true)
-                .offerClientMaxWindowBits(15)
                 .build();
 
         try (final CloseableWebSocketClient client = WebSocketClients.createDefault()) {

httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/api/WebSocketClientConfig.java

@@ -170,12 +170,13 @@ public boolean isOfferClientNoContextTakeover() {
     }
 
     /**
-     * Optional value for {@code client_max_window_bits} in the PMCE offer.
+     * Offered value for {@code client_max_window_bits}.
      *
-     * <p>Valid values are in range 8..15 when non-null. The client encoder
-     * currently supports only {@code 15} due to JDK Deflater limitations.</p>
+     * <p>The client never offers {@code client_max_window_bits}: the JDK Deflater cannot honor a
+     * server-selected window smaller than 15, which RFC 7692 section 7.2.1 would require. This
+     * always returns {@code null}; configuring a value is rejected when the configuration is built.</p>
      *
-     * @return offered {@code client_max_window_bits}, or {@code null} if not offered
+     * @return always {@code null}
      * @since 5.7
      */
     public Integer getOfferClientMaxWindowBits() {
@@ -332,7 +333,7 @@ public static final class Builder {
         private boolean perMessageDeflateEnabled = true;
         private boolean offerServerNoContextTakeover = true;
         private boolean offerClientNoContextTakeover = true;
-        private Integer offerClientMaxWindowBits = 15;
+        private Integer offerClientMaxWindowBits = null;
         private Integer offerServerMaxWindowBits = null;
 
         private int maxFrameSize = 64 * 1024;
@@ -410,12 +411,12 @@ public Builder offerClientNoContextTakeover(final boolean v) {
         }
 
         /**
-         * Offers {@code client_max_window_bits} in the PMCE offer.
+         * Configures {@code client_max_window_bits}. This parameter is not supported: the JDK
+         * Deflater cannot honor a server-selected window smaller than 15 (RFC 7692 section 7.2.1),
+         * so the client never offers it. Only {@code null} is accepted; any non-null value is
+         * rejected when {@link #build()} is called.
          *
-         * <p>Valid values are in range 8..15 when non-null. The client encoder
-         * currently supports only {@code 15} due to JDK Deflater limitations.</p>
-         *
-         * @param v window bits, or {@code null} to omit the parameter
+         * @param v must be {@code null}
          * @return this builder
          * @since 5.7
          */
@@ -570,8 +571,10 @@ public Builder enableHttp2(final boolean enabled) {
          * @since 5.7
          */
         public WebSocketClientConfig build() {
-            if (offerClientMaxWindowBits != null && (offerClientMaxWindowBits < 8 || offerClientMaxWindowBits > 15)) {
-                throw new IllegalArgumentException("offerClientMaxWindowBits must be in range [8..15]");
+            if (offerClientMaxWindowBits != null) {
+                throw new IllegalArgumentException(
+                        "client_max_window_bits cannot be offered: the JDK Deflater cannot honor a "
+                        + "server-selected window smaller than 15 (RFC 7692 section 7.2.1)");
             }
             if (offerServerMaxWindowBits != null && (offerServerMaxWindowBits < 8 || offerServerMaxWindowBits > 15)) {
                 throw new IllegalArgumentException("offerServerMaxWindowBits must be in range [8..15]");

httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http1UpgradeProtocol.java

@@ -33,7 +33,10 @@
 import java.security.MessageDigest;
 import java.security.SecureRandom;
 import java.util.Base64;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
+import java.util.Set;
 import java.util.StringJoiner;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -157,9 +160,8 @@ public void completed(final WebSocketEndpointConnector.ProtoEndpoint endpoint) {
                                 if (cfg.isOfferClientNoContextTakeover()) {
                                     ext.append("; client_no_context_takeover");
                                 }
-                                if (cfg.getOfferClientMaxWindowBits() != null) {
-                                    ext.append("; client_max_window_bits=").append(cfg.getOfferClientMaxWindowBits());
-                                }
+                                // client_max_window_bits is never offered: the JDK Deflater cannot
+                                // honor a server-selected window smaller than 15 (RFC 7692 7.2.1).
                                 if (cfg.getOfferServerMaxWindowBits() != null) {
                                     ext.append("; server_max_window_bits=").append(cfg.getOfferServerMaxWindowBits());
                                 }
@@ -359,9 +361,28 @@ public void cancelled() {
         }
     }
 
-    private static String headerValue(final HttpResponse r, final String name) {
-        final Header h = r.getFirstHeader(name);
-        return h != null ? h.getValue() : null;
+    static String headerValue(final HttpResponse r, final String name) {
+        // RFC 6455 section 9.1 permits Sec-WebSocket-Extensions and Sec-WebSocket-Protocol
+        // to be split across multiple header fields; combine them as a comma-separated list.
+        final Header[] headers = r.getHeaders(name);
+        if (headers.length == 0) {
+            return null;
+        }
+        if (headers.length == 1) {
+            return headers[0].getValue();
+        }
+        final StringBuilder buf = new StringBuilder();
+        for (final Header h : headers) {
+            final String value = h.getValue();
+            if (value == null || value.isEmpty()) {
+                continue;
+            }
+            if (buf.length() > 0) {
+                buf.append(", ");
+            }
+            buf.append(value);
+        }
+        return buf.length() > 0 ? buf.toString() : null;
     }
 
     private static boolean containsToken(final HttpResponse r, final String header, final String token) {
@@ -394,8 +415,6 @@ static ExtensionChain buildExtensionChain(final WebSocketClientConfig cfg, final
         }
         boolean pmceSeen = false, serverNoCtx = false, clientNoCtx = false;
         Integer clientBits = null, serverBits = null;
-        final boolean offerServerNoCtx = cfg.isOfferServerNoContextTakeover();
-        final boolean offerClientNoCtx = cfg.isOfferClientNoContextTakeover();
         final Integer offerClientBits = cfg.getOfferClientMaxWindowBits();
         final Integer offerServerBits = cfg.getOfferServerMaxWindowBits();
 
@@ -414,19 +433,23 @@ static ExtensionChain buildExtensionChain(final WebSocketClientConfig cfg, final
             }
             pmceSeen = true;
 
+            // RFC 7692 section 7.1: a parameter name must not appear more than once within an
+            // accepted extension; a repeated name makes the response invalid.
+            final Set<String> parameterNames = new HashSet<>();
             for (int i = 1; i < parts.length; i++) {
                 final String p = parts[i].trim();
                 final int eq = p.indexOf('=');
+                final String parameterName = (eq >= 0 ? p.substring(0, eq) : p).trim().toLowerCase(Locale.ROOT);
+                if (!parameterNames.add(parameterName)) {
+                    throw new IllegalStateException("Duplicate permessage-deflate parameter: " + parameterName);
+                }
                 if (eq < 0) {
+                    // RFC 7692 sections 7.1.1.1 and 7.1.1.2 permit the server to include either
+                    // no-context-takeover parameter in the response even when the offer did not;
+                    // both are always safe to honour.
                     if ("server_no_context_takeover".equalsIgnoreCase(p)) {
-                        if (!offerServerNoCtx) {
-                            throw new IllegalStateException("Server selected server_no_context_takeover not offered");
-                        }
                         serverNoCtx = true;
                     } else if ("client_no_context_takeover".equalsIgnoreCase(p)) {
-                        if (!offerClientNoCtx) {
-                            throw new IllegalStateException("Server selected client_no_context_takeover not offered");
-                        }
                         clientNoCtx = true;
                     } else {
                         throw new IllegalStateException("Unsupported permessage-deflate parameter: " + p);
@@ -453,9 +476,8 @@ static ExtensionChain buildExtensionChain(final WebSocketClientConfig cfg, final
                             throw new IllegalStateException("Invalid client_max_window_bits: " + v, nfe);
                         }
                     } else if ("server_max_window_bits".equalsIgnoreCase(k)) {
-                        if (offerServerBits == null) {
-                            throw new IllegalStateException("Server selected server_max_window_bits not offered");
-                        }
+                        // RFC 7692 section 7.1.2.1 permits the server to include this parameter
+                        // uninvited; a server constraining its own window is always acceptable.
                         try {
                             if (v.isEmpty()) {
                                 throw new IllegalStateException("server_max_window_bits must have a value");

httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http2ExtendedConnectProtocol.java

@@ -160,9 +160,8 @@ public CompletableFuture<WebSocket> connect(
             if (cfg.isOfferClientNoContextTakeover()) {
                 ext.append("; client_no_context_takeover");
             }
-            if (cfg.getOfferClientMaxWindowBits() != null && cfg.getOfferClientMaxWindowBits() == 15) {
-                ext.append("; client_max_window_bits=15");
-            }
+            // client_max_window_bits is never offered: the JDK Deflater cannot honor a
+            // server-selected window smaller than 15 (RFC 7692 7.2.1).
             if (cfg.getOfferServerMaxWindowBits() != null) {
                 ext.append("; server_max_window_bits=").append(cfg.getOfferServerMaxWindowBits());
             }
@@ -211,8 +210,9 @@ public void produceRequest(final RequestChannel channel, final HttpContext conte
         public void consumeResponse(final HttpResponse response, final EntityDetails entityDetails,
                                     final HttpContext context) throws HttpException, IOException {
 
-            if (response.getCode() != HttpStatus.SC_OK) {
-                failFuture(new IllegalStateException("Unexpected status: " + response.getCode()));
+            final int code = response.getCode();
+            if (code < HttpStatus.SC_OK || code >= HttpStatus.SC_REDIRECTION) {
+                failFuture(new IllegalStateException("Unexpected status: " + code));
                 return;
             }
 
@@ -321,8 +321,7 @@ public void cancel() {
         }
 
         private static String headerValue(final HttpResponse r, final String name) {
-            final Header h = r.getFirstHeader(name);
-            return h != null ? h.getValue() : null;
+            return Http1UpgradeProtocol.headerValue(r, name);
         }
 
         private void failFuture(final Exception ex) {

httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransport.java

@@ -28,6 +28,7 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.channels.CancelledKeyException;
 
 import org.apache.hc.core5.annotation.Internal;
 import org.apache.hc.core5.http.nio.DataStreamChannel;
@@ -51,14 +52,24 @@ public int write(final ByteBuffer src) throws IOException {
         if (ch == null) {
             return 0;
         }
-        return ch.write(src);
+        try {
+            return ch.write(src);
+        } catch (final CancelledKeyException ignore) {
+            // The selection key was cancelled by a concurrent shutdown; the channel is gone.
+            return 0;
+        }
     }
 
     @Override
     public void requestOutput() {
         final DataStreamChannel ch = channel;
         if (ch != null) {
-            ch.requestOutput();
+            try {
+                ch.requestOutput();
+            } catch (final CancelledKeyException ignore) {
+                // requestOutput is a best-effort nudge; if the channel's selection key was
+                // cancelled by a concurrent shutdown there is nothing left to flush.
+            }
         }
     }
 

httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/WebSocketSessionEngine.java

@@ -109,6 +109,7 @@ public final class WebSocketSessionEngine {
     final AtomicBoolean open = new AtomicBoolean(true);
     final AtomicBoolean closeSent = new AtomicBoolean(false);
     private final AtomicBoolean closeReceived = new AtomicBoolean(false);
+    private final AtomicBoolean released = new AtomicBoolean(false);
     volatile boolean closeAfterFlush;
     private volatile ScheduledFuture<?> closeTimeoutFuture;
 
@@ -332,6 +333,12 @@ private void handleFrame() {
             return;
         }
         if (FrameOpcode.isControl(op)) {
+            if (r1) {
+                // Control frames are never compressed, so an extension never defines RSV1 for them.
+                initiateClose(1002, "RSV1 set on control frame");
+                inbuf.clear();
+                return;
+            }
             if (!fin) {
                 initiateClose(1002, "fragmented control frame");
                 inbuf.clear();
@@ -379,8 +386,7 @@ private void handleFrame() {
                     inbuf.clear();
                     return;
                 }
-                appendToMessage(payload);
-                if (fin) {
+                if (appendToMessage(payload) && fin) {
                     deliverAssembledMessage();
                 }
                 break;
@@ -484,15 +490,16 @@ private void startMessage(final int opcode, final ByteBuffer payload, final bool
         appendToMessage(payload);
     }
 
-    private void appendToMessage(final ByteBuffer payload) {
+    private boolean appendToMessage(final ByteBuffer payload) {
         final int n = payload.remaining();
         assemblingSize += n;
         if (cfg.getMaxMessageSize() > 0 && assemblingSize > cfg.getMaxMessageSize()) {
             initiateClose(1009, "Message too big");
-            return;
+            return false;
         }
         assemblingBuf = WebSocketBufferOps.ensureCapacity(assemblingBuf, n);
         assemblingBuf.put(payload.asReadOnlyBuffer());
+        return true;
     }
 
     private void deliverAssembledMessage() {
@@ -673,6 +680,9 @@ boolean enqueueData(final OutFrame frame) {
     }
 
     private void drainAndRelease() {
+        if (!released.compareAndSet(false, true)) {
+            return;
+        }
         if (activeWrite != null) {
             if (activeWrite.dataFrame) {
                 dataQueuedBytes.addAndGet(-activeWrite.size);
@@ -690,6 +700,12 @@ private void drainAndRelease() {
                 dataQueuedBytes.addAndGet(-f.size);
             }
         }
+        if (encChain != null) {
+            encChain.close();
+        }
+        if (decChain != null) {
+            decChain.close();
+        }
         cancelCloseTimeout();
     }
 

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

@@ -144,18 +144,25 @@ public ByteBuffer encode(final WebSocketFrameType type, final boolean fin, final
         deflater.setInput(input);
         final ByteArrayOutputStream out = new ByteArrayOutputStream(Math.max(128, input.length / 2));
         final byte[] buffer = new byte[Math.min(16384, Math.max(1024, input.length))];
-        while (!deflater.needsInput()) {
-            final int count = deflater.deflate(buffer, 0, buffer.length, Deflater.SYNC_FLUSH);
+        // Drain until a SYNC_FLUSH leaves the output buffer partly filled; testing needsInput()
+        // would stop once the input is consumed and could drop the trailing 00 00 FF FF flush bytes.
+        int count;
+        do {
+            count = deflater.deflate(buffer, 0, buffer.length, Deflater.SYNC_FLUSH);
             if (count > 0) {
                 out.write(buffer, 0, count);
-            } else {
-                break;
             }
-        }
+        } while (count == buffer.length);
         final byte[] data = out.toByteArray();
         final ByteBuffer encoded;
-        if (data.length >= 4) {
-            encoded = ByteBuffer.wrap(data, 0, data.length - 4);
+        // Strip the 00 00 FF FF flush trailer only on the final fragment; non-final fragments
+        // must keep it so each intermediate empty stored block stays valid and the reassembled
+        // DEFLATE stream decodes (RFC 7692 section 7.2.1). When stripping leaves nothing the final
+        // fragment is empty, which RFC 7692 section 7.2.3.6 represents as the single octet 0x00.
+        if (fin) {
+            encoded = data.length > 4
+                    ? ByteBuffer.wrap(data, 0, data.length - 4)
+                    : ByteBuffer.wrap(new byte[]{0x00});
         } else {
             encoded = ByteBuffer.wrap(data);
         }
@@ -183,6 +190,12 @@ public WebSocketExtensionData getResponseData() {
         return new WebSocketExtensionData(getName(), params);
     }
 
+    @Override
+    public void close() {
+        deflater.end();
+        inflater.end();
+    }
+
     private static boolean isDataFrame(final WebSocketFrameType type) {
         return type == WebSocketFrameType.TEXT || type == WebSocketFrameType.BINARY;
     }

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

@@ -77,4 +77,13 @@ default ByteBuffer encode(
     default WebSocketExtensionData getResponseData() {
         return new WebSocketExtensionData(getName(), null);
     }
+
+    /**
+     * Releases any native resources held by this extension (e.g. a {@code Deflater} or
+     * {@code Inflater}). Called once when the owning session terminates.
+     *
+     * @since 5.7
+     */
+    default void close() {
+    }
 }