From 3433ff31dc8c8e9dfc95b766bc4f61c6c4e0c486 Mon Sep 17 00:00:00 2001 From: Arturo Bernal Date: Sun, 12 Jul 2026 14:50:02 +0200 Subject: [PATCH] Hardened WebSocket transport and permessage-deflate for RFC 6455/7692 conformance Fixed the client permessage-deflate encoder, which stripped the flush trailer on every fragment and could truncate a SYNC_FLUSH, so a compressed message split across frames now decodes, and an empty final fragment is encoded as a single 0x00 octet; the frame writer derives the masking key from SecureRandom. The client no longer offers client_max_window_bits, since the JDK Deflater cannot honour a reduced window, and rejects a handshake response that echoes it. The session engine fails the connection on RSV1 set on a control frame, no longer delivers a truncated message after a 1009 close, and surfaces a selection key cancelled during shutdown as an I/O error on write so the engine takes its close path instead of retaining the frame as back-pressure. Extension negotiation on the server is stricter. Offers are parsed from every Sec-WebSocket-Extensions field and combined as RFC 6455 requires, so a client's fallback offer in a later field is still examined; a parameter repeated within an offer drops that offer; and the permessage-deflate factory accepts only recognised parameters with well-formed values, the no-context-takeover parameters carrying no value and the window-bits parameters being exactly 15, the only size the JDK Deflater supports. A given extension name is accepted only once, and an unacceptable offer is declined rather than failing the handshake. On the HTTP/2 server the handler advertises a bounded inbound flow-control window and bounds the outbound queue by a byte budget, splitting large writes into chunks and signalling the reactor after each so a write larger than the budget cannot deadlock; the frame reader rejects a 64-bit length with the most significant bit set, the frame writer enforces the 125-byte control frame limit and truncates the CLOSE reason, and the processor fails a data frame received in the middle of a fragmented message. The HTTP/2 client accepts any 2xx handshake response. Both permessage-deflate implementations release their Deflater and Inflater when the session ends. The extension chains close every codec they hold even if one of them fails, the negotiation closes each extension exactly once, and the session engine coordinates the encoder and decoder close with try/finally so a failure in one still releases the other. --- .../testing/websocket/H2WebSocketEchoIT.java | 1 - .../websocket/WebSocketClientTest.java | 1 - .../websocket/performance/WsPerfHarness.java | 1 - .../websocket/api/WebSocketClientConfig.java | 37 +- .../impl/protocol/Http1UpgradeProtocol.java | 88 ++-- .../Http2ExtendedConnectProtocol.java | 13 +- .../transport/DataStreamChannelTransport.java | 17 +- .../transport/WebSocketSessionEngine.java | 35 +- .../websocket/PerMessageDeflateExtension.java | 27 +- .../PerMessageDeflateExtensionFactory.java | 98 +++-- .../core5/websocket/WebSocketExtension.java | 9 + .../WebSocketExtensionNegotiation.java | 33 +- .../websocket/WebSocketExtensionRegistry.java | 40 +- .../core5/websocket/WebSocketExtensions.java | 38 +- .../core5/websocket/WebSocketFrameReader.java | 3 +- .../core5/websocket/WebSocketFrameWriter.java | 14 +- .../websocket/extension/ExtensionChain.java | 48 ++ .../extension/PerMessageDeflate.java | 62 ++- .../extension/WebSocketExtensionChain.java | 16 + .../websocket/frame/WebSocketFrameWriter.java | 7 +- .../WebSocketH2ServerExchangeHandler.java | 315 ++++++++++--- .../server/WebSocketServerProcessor.java | 7 + .../server/WebSocketServerRequestHandler.java | 85 ++-- .../Http1UpgradeProtocolExtensionTest.java | 97 ++++- .../example/WebSocketEchoClient.java | 2 - .../DataStreamChannelTransportTest.java | 93 ++++ .../transport/WebSocketInboundRfcGapTest.java | 127 ++++++ ...PerMessageDeflateExtensionFactoryTest.java | 81 ++++ .../PerMessageDeflateExtensionTest.java | 67 +++ .../WebSocketExtensionNegotiationTest.java | 24 + .../WebSocketExtensionRegistryTest.java | 114 +++++ .../websocket/WebSocketExtensionsTest.java | 36 ++ .../websocket/WebSocketFrameWriterTest.java | 27 ++ .../extension/ExtensionChainTest.java | 73 ++++ .../extension/MessageDeflateTest.java | 63 +++ ...MessageDeflateFragmentedRoundTripTest.java | 122 ++++++ .../WebSocketH2ServerExchangeHandlerTest.java | 412 ++++++++++++++++++ .../server/WebSocketServerProcessorTest.java | 33 ++ .../WebSocketServerRequestHandlerTest.java | 216 +++++++++ 39 files changed, 2270 insertions(+), 312 deletions(-) create mode 100644 httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransportTest.java create mode 100644 httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/transport/WebSocketInboundRfcGapTest.java create mode 100644 httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/PerMessageDeflateFragmentedRoundTripTest.java diff --git a/httpclient5-testing/src/test/java/org/apache/hc/client5/testing/websocket/H2WebSocketEchoIT.java b/httpclient5-testing/src/test/java/org/apache/hc/client5/testing/websocket/H2WebSocketEchoIT.java index c9e7ccb57b..aaa6c2d7dd 100644 --- a/httpclient5-testing/src/test/java/org/apache/hc/client5/testing/websocket/H2WebSocketEchoIT.java +++ b/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(); diff --git a/httpclient5-testing/src/test/java/org/apache/hc/client5/testing/websocket/WebSocketClientTest.java b/httpclient5-testing/src/test/java/org/apache/hc/client5/testing/websocket/WebSocketClientTest.java index 34fcdc697b..0fd08a3455 100644 --- a/httpclient5-testing/src/test/java/org/apache/hc/client5/testing/websocket/WebSocketClientTest.java +++ b/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()) { diff --git a/httpclient5-testing/src/test/java/org/apache/hc/client5/testing/websocket/performance/WsPerfHarness.java b/httpclient5-testing/src/test/java/org/apache/hc/client5/testing/websocket/performance/WsPerfHarness.java index cc063e16ef..b24f35a923 100644 --- a/httpclient5-testing/src/test/java/org/apache/hc/client5/testing/websocket/performance/WsPerfHarness.java +++ b/httpclient5-testing/src/test/java/org/apache/hc/client5/testing/websocket/performance/WsPerfHarness.java @@ -166,7 +166,6 @@ private static void runClient( b.enablePerMessageDeflate(true) .offerClientNoContextTakeover(false) .offerServerNoContextTakeover(false) - .offerClientMaxWindowBits(null) .offerServerMaxWindowBits(null); } final WebSocketClientConfig cfg = b.build(); diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/api/WebSocketClientConfig.java b/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/api/WebSocketClientConfig.java index a25acb10c5..693e5d3913 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/api/WebSocketClientConfig.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/api/WebSocketClientConfig.java @@ -55,7 +55,6 @@ public final class WebSocketClientConfig { private final boolean perMessageDeflateEnabled; private final boolean offerServerNoContextTakeover; private final boolean offerClientNoContextTakeover; - private final Integer offerClientMaxWindowBits; private final Integer offerServerMaxWindowBits; // Framing / flow @@ -82,7 +81,6 @@ private WebSocketClientConfig( final boolean perMessageDeflateEnabled, final boolean offerServerNoContextTakeover, final boolean offerClientNoContextTakeover, - final Integer offerClientMaxWindowBits, final Integer offerServerMaxWindowBits, final int maxFrameSize, final int outgoingChunkSize, @@ -102,7 +100,6 @@ private WebSocketClientConfig( this.perMessageDeflateEnabled = perMessageDeflateEnabled; this.offerServerNoContextTakeover = offerServerNoContextTakeover; this.offerClientNoContextTakeover = offerClientNoContextTakeover; - this.offerClientMaxWindowBits = offerClientMaxWindowBits; this.offerServerMaxWindowBits = offerServerMaxWindowBits; this.maxFrameSize = maxFrameSize; this.outgoingChunkSize = outgoingChunkSize; @@ -169,19 +166,6 @@ public boolean isOfferClientNoContextTakeover() { return offerClientNoContextTakeover; } - /** - * Optional value for {@code client_max_window_bits} in the PMCE offer. - * - *

Valid values are in range 8..15 when non-null. The client encoder - * currently supports only {@code 15} due to JDK Deflater limitations.

- * - * @return offered {@code client_max_window_bits}, or {@code null} if not offered - * @since 5.7 - */ - public Integer getOfferClientMaxWindowBits() { - return offerClientMaxWindowBits; - } - /** * Optional value for {@code server_max_window_bits} in the PMCE offer. * @@ -332,7 +316,6 @@ public static final class Builder { private boolean perMessageDeflateEnabled = true; private boolean offerServerNoContextTakeover = true; private boolean offerClientNoContextTakeover = true; - private Integer offerClientMaxWindowBits = 15; private Integer offerServerMaxWindowBits = null; private int maxFrameSize = 64 * 1024; @@ -409,21 +392,6 @@ public Builder offerClientNoContextTakeover(final boolean v) { return this; } - /** - * Offers {@code client_max_window_bits} in the PMCE offer. - * - *

Valid values are in range 8..15 when non-null. The client encoder - * currently supports only {@code 15} due to JDK Deflater limitations.

- * - * @param v window bits, or {@code null} to omit the parameter - * @return this builder - * @since 5.7 - */ - public Builder offerClientMaxWindowBits(final Integer v) { - this.offerClientMaxWindowBits = v; - return this; - } - /** * Offers {@code server_max_window_bits} in the PMCE offer. * @@ -570,9 +538,6 @@ 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 (offerServerMaxWindowBits != null && (offerServerMaxWindowBits < 8 || offerServerMaxWindowBits > 15)) { throw new IllegalArgumentException("offerServerMaxWindowBits must be in range [8..15]"); } @@ -600,7 +565,7 @@ public WebSocketClientConfig build() { return new WebSocketClientConfig( connectTimeout, subprotocols, perMessageDeflateEnabled, offerServerNoContextTakeover, offerClientNoContextTakeover, - offerClientMaxWindowBits, offerServerMaxWindowBits, + offerServerMaxWindowBits, maxFrameSize, outgoingChunkSize, maxFramesPerTick, directBuffers, autoPong, closeWaitTimeout, maxMessageSize, diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http1UpgradeProtocol.java b/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http1UpgradeProtocol.java index 186b22680e..8bf1391f10 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http1UpgradeProtocol.java +++ b/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) { @@ -393,10 +414,7 @@ static ExtensionChain buildExtensionChain(final WebSocketClientConfig cfg, final return chain; } 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(); + Integer serverBits = null; final Integer offerServerBits = cfg.getOfferServerMaxWindowBits(); final String[] tokens = ext.split(","); @@ -414,19 +432,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 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); @@ -438,24 +460,13 @@ static ExtensionChain buildExtensionChain(final WebSocketClientConfig cfg, final v = v.substring(1, v.length() - 1); // strip quotes if any } if ("client_max_window_bits".equalsIgnoreCase(k)) { - if (offerClientBits == null) { - throw new IllegalStateException("Server selected client_max_window_bits not offered"); - } - try { - if (v.isEmpty()) { - throw new IllegalStateException("client_max_window_bits must have a value"); - } - clientBits = Integer.parseInt(v); - if (clientBits < 8 || clientBits > 15) { - throw new IllegalStateException("client_max_window_bits out of range: " + clientBits); - } - } catch (final NumberFormatException nfe) { - throw new IllegalStateException("Invalid client_max_window_bits: " + v, nfe); - } + // The client never offers client_max_window_bits (the JDK Deflater cannot honor + // a window smaller than 15, RFC 7692 section 7.2.1), so a server that selects it + // has echoed a parameter that was never offered. + throw new IllegalStateException("Server selected client_max_window_bits not offered"); } 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"); @@ -478,21 +489,12 @@ static ExtensionChain buildExtensionChain(final WebSocketClientConfig cfg, final if (!cfg.isPerMessageDeflateEnabled()) { throw new IllegalStateException("Server negotiated PMCE but client disabled it"); } - if (clientBits != null) { - if (offerClientBits == null) { - throw new IllegalStateException("Server selected client_max_window_bits not offered"); - } - if (!clientBits.equals(offerClientBits)) { - throw new IllegalStateException("Unsupported client_max_window_bits: " + clientBits - + " (offered " + offerClientBits + ")"); - } - } if (serverBits != null) { if (offerServerBits != null && serverBits > offerServerBits) { throw new IllegalStateException("server_max_window_bits exceeds offer: " + serverBits); } } - chain.add(new PerMessageDeflate(true, serverNoCtx, clientNoCtx, clientBits, serverBits)); + chain.add(new PerMessageDeflate(true, serverNoCtx, clientNoCtx, null, serverBits)); } return chain; } diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http2ExtendedConnectProtocol.java b/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http2ExtendedConnectProtocol.java index b3b50c5b41..c94f13dfef 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http2ExtendedConnectProtocol.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http2ExtendedConnectProtocol.java @@ -160,9 +160,8 @@ public CompletableFuture 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) { diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransport.java b/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransport.java index d002d32e2f..2a8c980139 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransport.java +++ b/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,26 @@ 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 ex) { + // A cancelled selection key is a terminal state, not transient back-pressure. Surface it + // as an I/O error so the engine takes its close path instead of retaining the frame and + // waiting for an output callback that will never arrive. + throw new IOException("WebSocket transport channel is closed", ex); + } } @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. + } } } diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/WebSocketSessionEngine.java b/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/WebSocketSessionEngine.java index f2e9b3a27c..4ab599a024 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/WebSocketSessionEngine.java +++ b/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,7 +700,22 @@ private void drainAndRelease() { dataQueuedBytes.addAndGet(-f.size); } } - cancelCloseTimeout(); + // Close the encoder and decoder chains coordinated with try/finally so a failure closing + // one still closes the other, then cancel the close timeout unconditionally. Each chain's + // close() is itself exception-safe and releases every codec it holds. + try { + if (encChain != null) { + encChain.close(); + } + } finally { + try { + if (decChain != null) { + decChain.close(); + } + } finally { + cancelCloseTimeout(); + } + } } // ================================================================== diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtension.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtension.java index 6c1068f6bf..957a7f600c 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtension.java +++ b/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; } diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactory.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactory.java index 6166efe842..d486156d04 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactory.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactory.java @@ -26,14 +26,18 @@ */ package org.apache.hc.core5.websocket; +import java.util.Map; + /** * Factory for {@code permessage-deflate} extensions (RFC 7692). * - *

Note: the JDK {@link java.util.zip.Deflater} / {@link java.util.zip.Inflater} - * only support a 15-bit window size. This factory therefore accepts - * {@code client_max_window_bits} / {@code server_max_window_bits} only when - * they are either absent or explicitly set to {@code 15}. Other values are - * rejected during negotiation.

+ *

An offer is accepted only when every parameter is recognised and well-formed. The + * {@code *_no_context_takeover} parameters must appear without a value, and the window-bits + * parameters must be exactly {@code 15} because the JDK {@link java.util.zip.Deflater} / + * {@link java.util.zip.Inflater} only support a 15-bit window. {@code client_max_window_bits} may + * also be offered without a value, in which case the client advertises support and the server + * selects the window. Any unknown parameter, any value on a valueless parameter, and any window + * size other than {@code 15} cause the offer to be declined (RFC 7692 section 7.1).

*/ public final class PerMessageDeflateExtensionFactory implements WebSocketExtensionFactory { @@ -44,33 +48,46 @@ public String getName() { @Override public WebSocketExtension create(final WebSocketExtensionData request, final boolean server) { - if (request == null) { - return null; - } - final String name = request.getName(); - if (!"permessage-deflate".equals(name)) { + if (request == null || !"permessage-deflate".equals(request.getName())) { return null; } - final boolean serverNoContextTakeover = request.getParameters().containsKey("server_no_context_takeover"); - final boolean clientNoContextTakeover = request.getParameters().containsKey("client_no_context_takeover"); - final boolean clientMaxBitsPresent = request.getParameters().containsKey("client_max_window_bits"); - final boolean serverMaxBitsPresent = request.getParameters().containsKey("server_max_window_bits"); - Integer clientMaxWindowBits = parseWindowBits(request.getParameters().get("client_max_window_bits")); - Integer serverMaxWindowBits = parseWindowBits(request.getParameters().get("server_max_window_bits")); - if (clientMaxBitsPresent && clientMaxWindowBits == null) { - clientMaxWindowBits = 15; - } - if (serverMaxBitsPresent && serverMaxWindowBits == null) { - serverMaxWindowBits = 15; - } - - if (!PerMessageDeflateExtension.isValidWindowBits(clientMaxWindowBits) - || !PerMessageDeflateExtension.isValidWindowBits(serverMaxWindowBits)) { - return null; - } - // JDK Deflater/Inflater only supports 15-bit window size. - if (!isSupportedWindowBits(clientMaxWindowBits) || !isSupportedWindowBits(serverMaxWindowBits)) { - return null; + boolean serverNoContextTakeover = false; + boolean clientNoContextTakeover = false; + Integer clientMaxWindowBits = null; + Integer serverMaxWindowBits = null; + for (final Map.Entry param : request.getParameters().entrySet()) { + final String value = param.getValue(); + switch (param.getKey()) { + case "server_no_context_takeover": + if (value != null) { + return null; + } + serverNoContextTakeover = true; + break; + case "client_no_context_takeover": + if (value != null) { + return null; + } + clientNoContextTakeover = true; + break; + case "client_max_window_bits": + // A valueless parameter advertises support; the server selects the window (15). + if (value != null && !isWindowBits15(value)) { + return null; + } + clientMaxWindowBits = 15; + break; + case "server_max_window_bits": + // Must carry a value, and only 15 is supported. + if (!isWindowBits15(value)) { + return null; + } + serverMaxWindowBits = 15; + break; + default: + // Unknown parameters make the offer invalid. + return null; + } } return new PerMessageDeflateExtension( serverNoContextTakeover, @@ -79,18 +96,13 @@ public WebSocketExtension create(final WebSocketExtensionData request, final boo serverMaxWindowBits); } - private static Integer parseWindowBits(final String value) { - if (value == null || value.isEmpty()) { - return null; - } - try { - return Integer.parseInt(value); - } catch (final NumberFormatException ignore) { - return null; - } - } - - private static boolean isSupportedWindowBits(final Integer bits) { - return bits == null || bits == 15; + /** + * The only window size the JDK Deflater/Inflater supports. A valid value is exactly the two + * decimal digits {@code "15"}; this rejects a leading zero ({@code "015"}), a sign + * ({@code "+15"}) and any missing or non-numeric value, which {@link Integer#parseInt} would + * otherwise accept. + */ + private static boolean isWindowBits15(final String value) { + return "15".equals(value); } } diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtension.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtension.java index 4364a5008f..98d213c857 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtension.java +++ b/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() { + } } diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiation.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiation.java index 4acf9779ce..768c6aa4c5 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiation.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiation.java @@ -29,12 +29,13 @@ import java.util.Collections; import java.util.List; import java.util.StringJoiner; +import java.util.concurrent.atomic.AtomicBoolean; - -public final class WebSocketExtensionNegotiation { +public final class WebSocketExtensionNegotiation implements AutoCloseable { private final List extensions; private final List responseData; + private final AtomicBoolean closed = new AtomicBoolean(false); WebSocketExtensionNegotiation( final List extensions, @@ -58,4 +59,32 @@ public String formatResponseHeader() { } return joiner.length() > 0 ? joiner.toString() : null; } + + /** + * Releases the native resources held by the negotiated extensions. Idempotent; only the first + * invocation closes the extensions and each extension is closed exactly once. A failure to + * close one extension does not prevent the remaining extensions from being closed. + * + * @since 5.7 + */ + @Override + public void close() { + if (closed.compareAndSet(false, true)) { + RuntimeException failure = null; + for (final WebSocketExtension extension : extensions) { + try { + extension.close(); + } catch (final RuntimeException ex) { + if (failure == null) { + failure = ex; + } else { + failure.addSuppressed(ex); + } + } + } + if (failure != null) { + throw failure; + } + } + } } diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistry.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistry.java index 12fa259f3f..95203a4b84 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistry.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistry.java @@ -27,9 +27,11 @@ package org.apache.hc.core5.websocket; import java.util.ArrayList; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Set; import org.apache.hc.core5.util.Args; @@ -52,17 +54,39 @@ public WebSocketExtensionNegotiation negotiate( final boolean server) throws WebSocketException { final List extensions = new ArrayList<>(); final List responseData = new ArrayList<>(); - if (requested != null) { - for (final WebSocketExtensionData request : requested) { - final WebSocketExtensionFactory factory = factories.get(request.getName()); - if (factory != null) { - final WebSocketExtension extension = factory.create(request, server); - if (extension != null) { - extensions.add(extension); - responseData.add(extension.getResponseData()); + final Set accepted = new HashSet<>(); + try { + if (requested != null) { + for (final WebSocketExtensionData request : requested) { + // At most one offer per extension name is accepted; a second accepted offer of the + // same extension would claim the same RSV bit and break both the response header and + // the RSV bookkeeping in the frame reader. + if (accepted.contains(request.getName())) { + continue; } + final WebSocketExtensionFactory factory = factories.get(request.getName()); + if (factory != null) { + final WebSocketExtension extension = factory.create(request, server); + if (extension != null) { + extensions.add(extension); + responseData.add(extension.getResponseData()); + accepted.add(request.getName()); + } + } + } + } + } catch (final WebSocketException | RuntimeException ex) { + // A later factory failed after earlier extensions were already created; close them so + // their native resources are not leaked, preserving the original failure and recording + // any close failure as a suppressed exception. + for (final WebSocketExtension extension : extensions) { + try { + extension.close(); + } catch (final RuntimeException closeEx) { + ex.addSuppressed(closeEx); } } + throw ex; } return new WebSocketExtensionNegotiation(extensions, responseData); } diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensions.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensions.java index 6c853b50fb..38ee16dddc 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensions.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensions.java @@ -27,6 +27,7 @@ package org.apache.hc.core5.websocket; import java.util.ArrayList; +import java.util.Iterator; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -44,8 +45,30 @@ private WebSocketExtensions() { public static List parse(final Header header) { final List extensions = new ArrayList<>(); + addOffers(header, extensions); + return extensions; + } + + /** + * Parses and combines every {@code Sec-WebSocket-Extensions} header field. RFC 6455 permits the + * header to be split across multiple fields and requires them to be treated as a single combined + * value, so a client's fallback offers carried in a later field are examined as well. + * + * @since 5.7 + */ + public static List parse(final Iterator
headers) { + final List extensions = new ArrayList<>(); + if (headers != null) { + while (headers.hasNext()) { + addOffers(headers.next(), extensions); + } + } + return extensions; + } + + private static void addOffers(final Header header, final List extensions) { if (header == null) { - return extensions; + return; } for (final HeaderElement element : MessageSupport.parseElements(header)) { final String name = element.getName(); @@ -53,11 +76,22 @@ public static List parse(final Header header) { continue; } final Map params = new LinkedHashMap<>(); + boolean duplicateParameter = false; for (final NameValuePair param : element.getParameters()) { + // RFC 7692 section 7.1: an extension parameter MUST NOT appear more than once in a + // negotiation offer. A repeated parameter makes the whole offer invalid, so it is + // dropped here and the extension is left un-negotiated rather than silently collapsed + // into a single map entry. + if (params.containsKey(param.getName())) { + duplicateParameter = true; + break; + } params.put(param.getName(), param.getValue()); } + if (duplicateParameter) { + continue; + } extensions.add(new WebSocketExtensionData(name, params)); } - return extensions; } } diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameReader.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameReader.java index fb14e941b7..2bbdf8cf40 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameReader.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameReader.java @@ -98,7 +98,8 @@ WebSocketFrame readFrame() throws IOException { len = (len << 8) | (readByte() & 0xFF); } if (len < 0) { - throw new WebSocketException("Negative frame payload length"); + // RFC 6455 section 5.2: the most significant bit of a 64-bit length MUST be 0. + throw new WebSocketException("64-bit frame length must have the most significant bit set to 0"); } } if (len > Integer.MAX_VALUE) { diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameWriter.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameWriter.java index 69ef31e5b8..abc4a2c9c6 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameWriter.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameWriter.java @@ -34,6 +34,7 @@ import java.util.List; import org.apache.hc.core5.util.Args; +import org.apache.hc.core5.websocket.message.CloseCodec; class WebSocketFrameWriter { @@ -63,13 +64,9 @@ void writePong(final ByteBuffer payload) throws IOException { } void writeClose(final int statusCode, final String reason) throws IOException { - final byte[] reasonBytes = reason != null ? reason.getBytes(StandardCharsets.UTF_8) : new byte[0]; - final int len = 2 + reasonBytes.length; - final ByteBuffer buffer = ByteBuffer.allocate(len); - buffer.put((byte) ((statusCode >> 8) & 0xFF)); - buffer.put((byte) (statusCode & 0xFF)); - buffer.put(reasonBytes); - buffer.flip(); + // CloseCodec truncates the reason to 123 UTF-8 bytes so the payload never + // exceeds the 125-byte control frame limit (RFC 6455 section 5.5). + final ByteBuffer buffer = ByteBuffer.wrap(CloseCodec.encode(statusCode, reason)); writeFrame(WebSocketFrameType.CLOSE, buffer, false, false, false); } @@ -102,6 +99,9 @@ private void writeFrame( Args.notNull(type, "Frame type"); final ByteBuffer buffer = payload != null ? payload.asReadOnlyBuffer() : ByteBuffer.allocate(0); final int payloadLen = buffer.remaining(); + if (type.isControl() && payloadLen > 125) { + throw new IllegalArgumentException("Control frame payload > 125 bytes: " + payloadLen); + } int firstByte = 0x80 | (type.getOpcode() & 0x0F); if (rsv1) { firstByte |= 0x40; diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/ExtensionChain.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/ExtensionChain.java index 3b73baf01b..5fcc1fef5b 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/ExtensionChain.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/ExtensionChain.java @@ -112,6 +112,30 @@ public WebSocketExtensionChain.Encoded encode(final byte[] data, final boolean f } return new WebSocketExtensionChain.Encoded(out, setRsv1); } + + /** + * Releases native resources held by the encoders in this chain. Every encoder is closed even + * if a previous one fails; the first failure is rethrown with the rest attached as suppressed. + * + * @since 5.7 + */ + public void close() { + RuntimeException failure = null; + for (final WebSocketExtensionChain.Encoder e : encs) { + try { + e.close(); + } catch (final RuntimeException ex) { + if (failure == null) { + failure = ex; + } else { + failure.addSuppressed(ex); + } + } + } + if (failure != null) { + throw failure; + } + } } public static final class DecodeChain { @@ -143,5 +167,29 @@ public byte[] decode(final byte[] data, final long maxDecodedSize) throws Except } return out; } + + /** + * Releases native resources held by the decoders in this chain. Every decoder is closed even + * if a previous one fails; the first failure is rethrown with the rest attached as suppressed. + * + * @since 5.7 + */ + public void close() { + RuntimeException failure = null; + for (final WebSocketExtensionChain.Decoder d : decs) { + try { + d.close(); + } catch (final RuntimeException ex) { + if (failure == null) { + failure = ex; + } else { + failure.addSuppressed(ex); + } + } + } + if (failure != null) { + throw failure; + } + } } } \ No newline at end of file diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/PerMessageDeflate.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/PerMessageDeflate.java index 002b9d486e..61b64c480a 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/PerMessageDeflate.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/PerMessageDeflate.java @@ -87,44 +87,57 @@ public Encoded encode(final byte[] data, final boolean first, final boolean fin) return new Encoded(out, first); } + @Override + public void close() { + def.end(); + } + private byte[] compressMessage(final byte[] data) { return doDeflate(data, true, true, clientNoContextTakeover); } private byte[] compressFragment(final byte[] data, final boolean fin) { - return doDeflate(data, fin, true, fin && clientNoContextTakeover); + // 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). + return doDeflate(data, fin, fin, fin && clientNoContextTakeover); } private byte[] doDeflate(final byte[] data, final boolean fin, final boolean stripTail, final boolean maybeReset) { - if (data == null || data.length == 0) { - if (fin && maybeReset) { - def.reset(); - } - return new byte[0]; - } - def.setInput(data); - final ByteArrayOutputStream out = new ByteArrayOutputStream(Math.max(128, data.length / 2)); - final byte[] buf = new byte[Math.min(16384, Math.max(1024, data.length))]; - while (!def.needsInput()) { - final int n = def.deflate(buf, 0, buf.length, Deflater.SYNC_FLUSH); + // Empty input is deflated through SYNC_FLUSH as well: an empty message or empty + // final fragment must compress to the single octet 0x00 once the 00 00 FF FF flush + // trailer is stripped (RFC 7692 section 7.2.3.6), not to an empty payload. + final byte[] input = data != null ? data : new byte[0]; + def.setInput(input); + final ByteArrayOutputStream out = new ByteArrayOutputStream(Math.max(128, input.length / 2)); + final byte[] buf = new byte[Math.min(16384, Math.max(1024, input.length))]; + // Drain until a SYNC_FLUSH leaves the output buffer partly filled. Testing + // needsInput() would stop as soon as the input is consumed, which can drop the + // trailing 00 00 FF FF flush bytes when the buffer fills exactly and then have + // stripTail cut into real data. + int n; + do { + n = def.deflate(buf, 0, buf.length, Deflater.SYNC_FLUSH); if (n > 0) { out.write(buf, 0, n); - } else { - break; } - } + } while (n == buf.length); byte[] all = out.toByteArray(); - if (stripTail && all.length >= 4) { - final int newLen = all.length - 4; // strip 00 00 FF FF - if (newLen <= 0) { - all = new byte[0]; - } else { - final byte[] trimmed = new byte[newLen]; - System.arraycopy(all, 0, trimmed, 0, newLen); + if (stripTail) { + // Strip the trailing 00 00 FF FF SYNC_FLUSH marker. When nothing remains the + // message (or final fragment) is empty; RFC 7692 section 7.2.3.6 represents an + // empty payload as the single octet 0x00, not as an empty payload. A fresh + // Deflater already yields 0x00 here, but one holding context (context takeover) + // emits only the 4-octet marker, so the 0x00 must be supplied explicitly. + if (all.length > 4) { + final byte[] trimmed = new byte[all.length - 4]; + System.arraycopy(all, 0, trimmed, 0, trimmed.length); all = trimmed; + } else { + all = new byte[]{0x00}; } } if (fin && maybeReset) { @@ -182,6 +195,11 @@ public byte[] decode(final byte[] compressedMessage, final long maxDecodedSize) } return out.toByteArray(); } + + @Override + public void close() { + inf.end(); + } }; } diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/WebSocketExtensionChain.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/WebSocketExtensionChain.java index 92ad719015..ad082c44a0 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/WebSocketExtensionChain.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/WebSocketExtensionChain.java @@ -69,6 +69,14 @@ interface Encoder { * Encode one fragment; return transformed payload and whether to set RSV on FIRST frame. */ Encoded encode(byte[] data, boolean first, boolean fin); + + /** + * Releases any native resources held by this encoder (e.g. a {@code Deflater}). + * + * @since 5.7 + */ + default void close() { + } } interface Decoder { @@ -88,5 +96,13 @@ interface Decoder { default byte[] decode(final byte[] payload, final long maxDecodedSize) throws Exception { return decode(payload); } + + /** + * Releases any native resources held by this decoder (e.g. an {@code Inflater}). + * + * @since 5.7 + */ + default void close() { + } } } diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/frame/WebSocketFrameWriter.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/frame/WebSocketFrameWriter.java index 5839f0caca..0126a513c4 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/frame/WebSocketFrameWriter.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/frame/WebSocketFrameWriter.java @@ -35,7 +35,7 @@ import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.nio.charset.StandardCharsets; -import java.util.concurrent.ThreadLocalRandom; +import java.security.SecureRandom; import org.apache.hc.core5.annotation.Internal; import org.apache.hc.core5.websocket.message.CloseCodec; @@ -48,6 +48,9 @@ @Internal public final class WebSocketFrameWriter { + // RFC 6455 section 5.3 requires the masking key to be derived from a strong source of entropy. + private static final SecureRandom MASK_RNG = new SecureRandom(); + // -- Text/Binary ----------------------------------------------------------- public ByteBuffer text(final CharSequence data, final boolean fin) { @@ -164,7 +167,7 @@ public ByteBuffer frameIntoWithRSV(final int opcode, final ByteBuffer payload, f int maskInt = 0; if (mask) { - maskInt = ThreadLocalRandom.current().nextInt(); + maskInt = MASK_RNG.nextInt(); out.put((byte) (maskInt >>> 24)).put((byte) (maskInt >>> 16)) .put((byte) (maskInt >>> 8)).put((byte) maskInt); } diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandler.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandler.java index 64cb417143..7d424faf81 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandler.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandler.java @@ -35,6 +35,8 @@ import java.util.concurrent.Executor; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; import org.apache.hc.core5.concurrent.DefaultThreadFactory; @@ -55,6 +57,7 @@ import org.apache.hc.core5.websocket.WebSocketCloseStatus; import org.apache.hc.core5.websocket.WebSocketConfig; import org.apache.hc.core5.websocket.WebSocketConstants; +import org.apache.hc.core5.websocket.WebSocketException; import org.apache.hc.core5.websocket.WebSocketExtensionNegotiation; import org.apache.hc.core5.websocket.WebSocketExtensionRegistry; import org.apache.hc.core5.websocket.WebSocketExtensions; @@ -68,6 +71,15 @@ final class WebSocketH2ServerExchangeHandler implements AsyncServerExchangeHandl private static final byte[] END_INBOUND = new byte[0]; private static final ByteBuffer END_OUTBOUND = ByteBuffer.allocate(0); + /** Bounded in-flight inbound byte budget advertised via HTTP/2 flow control. */ + private static final int INBOUND_WINDOW = 256 * 1024; + + /** Default budget for buffered outbound bytes before the application writer is back-pressured. */ + private static final int DEFAULT_OUTBOUND_BUFFER_SIZE = 1024 * 1024; + + /** Default maximum size of a single buffered outbound chunk (an implementation detail). */ + private static final int DEFAULT_OUTBOUND_CHUNK_SIZE = 64 * 1024; + /** * Default execution strategy (no explicit thread creation in the handler). * Note: tasks are typically long-lived (one per WS session). The bootstrap should ideally inject an executor. @@ -81,15 +93,26 @@ final class WebSocketH2ServerExchangeHandler implements AsyncServerExchangeHandl private final Executor executor; private final BlockingQueue inbound = new LinkedBlockingQueue<>(); + // Entry count is unbounded; outbound memory is bounded by a byte budget (see reserveOutboundBudget). private final BlockingQueue outbound = new LinkedBlockingQueue<>(); private final ReentrantLock outLock = new ReentrantLock(); private ByteBuffer currentOutbound; + // Byte-budget back-pressure for the outbound queue: the application writer blocks once the + // in-flight outbound bytes would exceed the budget and resumes as produce() drains chunks. + private final ReentrantLock outboundBudgetLock = new ReentrantLock(); + private final Condition outboundBudgetAvailable = outboundBudgetLock.newCondition(); + private final int outboundBudgetBytes; + private final int outboundChunkSize; + private long pendingOutboundBytes; + private volatile boolean responseSent; private volatile boolean outboundEnd; private volatile boolean shutdown; private volatile DataStreamChannel dataChannel; + private volatile CapacityChannel capacityChannel; + private final AtomicBoolean initialCreditGranted = new AtomicBoolean(false); WebSocketH2ServerExchangeHandler( final WebSocketHandler handler, @@ -103,10 +126,25 @@ final class WebSocketH2ServerExchangeHandler implements AsyncServerExchangeHandl final WebSocketConfig config, final WebSocketExtensionRegistry extensionRegistry, final Executor executor) { + this(handler, config, extensionRegistry, executor, + DEFAULT_OUTBOUND_BUFFER_SIZE, DEFAULT_OUTBOUND_CHUNK_SIZE); + } + + WebSocketH2ServerExchangeHandler( + final WebSocketHandler handler, + final WebSocketConfig config, + final WebSocketExtensionRegistry extensionRegistry, + final Executor executor, + final int outboundBufferSize, + final int outboundChunkSize) { this.handler = Args.notNull(handler, "WebSocket handler"); this.config = config != null ? config : WebSocketConfig.DEFAULT; this.extensionRegistry = extensionRegistry != null ? extensionRegistry : WebSocketExtensionRegistry.createDefault(); this.executor = executor != null ? executor : DEFAULT_EXECUTOR; + this.outboundBudgetBytes = Args.positive(outboundBufferSize, "Outbound buffer size"); + // A single chunk must never exceed the budget, or a large write could block forever waiting + // for capacity that can never be freed; cap the chunk size at the budget. + this.outboundChunkSize = Math.min(Args.positive(outboundChunkSize, "Outbound chunk size"), this.outboundBudgetBytes); this.responseSent = false; this.outboundEnd = false; this.shutdown = false; @@ -133,64 +171,109 @@ public void handleRequest( } final WebSocketExtensionNegotiation negotiation = extensionRegistry.negotiate( - WebSocketExtensions.parse(request.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER)), + WebSocketExtensions.parse(request.headerIterator(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER)), true); - final BasicHttpResponse response = new BasicHttpResponse(HttpStatus.SC_OK); - final String extensionsHeader = negotiation.formatResponseHeader(); - if (extensionsHeader != null) { - response.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER, extensionsHeader); - } + // The worker thread takes ownership of the negotiated extensions once the task is accepted + // and releases them in its finally after the stream has been torn down. If the hand-off + // never happens (a failed sendResponse or a rejected executor) this method releases them, + // attaching any close failure as a suppressed exception rather than masking the original. + try { + final BasicHttpResponse response = new BasicHttpResponse(HttpStatus.SC_OK); + final String extensionsHeader = negotiation.formatResponseHeader(); + if (extensionsHeader != null) { + response.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER, extensionsHeader); + } - final List offeredProtocols = WebSocketHandshake.parseSubprotocols( - request.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL_LOWER)); - final String protocolResponse = handler.selectSubprotocol(offeredProtocols); - if (protocolResponse != null) { - response.addHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL_LOWER, protocolResponse); - } + final List offeredProtocols = WebSocketHandshake.parseSubprotocols( + request.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL_LOWER)); + final String protocolResponse = handler.selectSubprotocol(offeredProtocols); + if (protocolResponse != null) { + response.addHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL_LOWER, protocolResponse); + } - responseChannel.sendResponse(response, new BasicEntityDetails(-1, null), context); - responseSent = true; + responseChannel.sendResponse(response, new BasicEntityDetails(-1, null), context); + responseSent = true; - final InputStream inputStream = new QueueInputStream(inbound); - final OutputStream outputStream = new QueueOutputStream(outbound); - final WebSocketSession session = new WebSocketSession( - config, inputStream, outputStream, null, null, negotiation.getExtensions()); + final InputStream inputStream = new QueueInputStream(inbound); + final OutputStream outputStream = new QueueOutputStream(); + final WebSocketSession session = new WebSocketSession( + config, inputStream, outputStream, null, null, negotiation.getExtensions()); - executor.execute(() -> { - try { - handler.onOpen(session); - new WebSocketServerProcessor(session, handler, config.getMaxMessageSize()).process(); - } catch (final WebSocketProtocolException ex) { - handler.onError(session, ex); + executor.execute(() -> { try { - session.close(ex.closeCode, ex.getMessage()); - } catch (final IOException ignore) { - // ignore - } - } catch (final Exception ex) { - handler.onError(session, ex); - try { - session.close(WebSocketCloseStatus.INTERNAL_ERROR.getCode(), "WebSocket error"); - } catch (final IOException ignore) { - // ignore - } - } finally { - shutdown = true; - outbound.offer(END_OUTBOUND); - inbound.offer(END_INBOUND); - - final DataStreamChannel channel = dataChannel; - if (channel != null) { - channel.requestOutput(); + handler.onOpen(session); + new WebSocketServerProcessor(session, handler, config.getMaxMessageSize()).process(); + } catch (final WebSocketProtocolException ex) { + handler.onError(session, ex); + try { + session.close(ex.closeCode, ex.getMessage()); + } catch (final IOException ignore) { + // ignore + } + } catch (final WebSocketException ex) { + handler.onError(session, ex); + try { + session.close(WebSocketCloseStatus.PROTOCOL_ERROR.getCode(), ex.getMessage()); + } catch (final IOException ignore) { + // ignore + } + } catch (final Exception ex) { + handler.onError(session, ex); + try { + session.close(WebSocketCloseStatus.INTERNAL_ERROR.getCode(), "WebSocket error"); + } catch (final IOException ignore) { + // ignore + } + } finally { + // Tear the stream down first so a failing extension close cannot leave the + // HTTP/2 stream without an END_STREAM; release the extensions last, always. + try { + shutdown = true; + abortOutboundBudget(); + enqueueSentinel(END_OUTBOUND); + inbound.offer(END_INBOUND); + + final DataStreamChannel channel = dataChannel; + if (channel != null) { + channel.requestOutput(); + } + } finally { + negotiation.close(); + } } + }); + } catch (final Exception primary) { + // The task was never handed off to the worker (sendResponse failed or the executor + // rejected it); release the negotiated extensions here, preserving the original failure. + try { + negotiation.close(); + } catch (final RuntimeException closeEx) { + primary.addSuppressed(closeEx); } - }); + throw primary; + } } @Override public void updateCapacity(final CapacityChannel capacityChannel) throws IOException { - capacityChannel.update(Integer.MAX_VALUE); + this.capacityChannel = capacityChannel; + // Advertise a bounded window once; further credit is replenished as the worker thread + // drains buffered bytes, so inbound memory stays bounded even if the worker stalls. + if (initialCreditGranted.compareAndSet(false, true)) { + capacityChannel.update(INBOUND_WINDOW); + } + } + + private void replenishInbound(final int n) { + final CapacityChannel channel = capacityChannel; + if (channel != null && n > 0) { + try { + channel.update(n); + } catch (final IOException ignore) { + // channel already gone; nothing to replenish + } + } } @Override @@ -271,6 +354,7 @@ public void produce(final DataStreamChannel channel) throws IOException { } finally { outLock.unlock(); } + releaseOutboundBudget(buf.capacity()); continue; } @@ -290,12 +374,19 @@ public void produce(final DataStreamChannel channel) throws IOException { } finally { outLock.unlock(); } + // The chunk has been fully written; release its bytes so a back-pressured writer resumes. + releaseOutboundBudget(buf.capacity()); } } @Override public void failed(final Exception cause) { shutdown = true; + // Release any writer blocked on the outbound byte budget so it observes the shutdown, then + // clear the queues and post the sentinels so the worker's blocking reads/writes unwind. + abortOutboundBudget(); + outbound.clear(); + inbound.clear(); outbound.offer(END_OUTBOUND); inbound.offer(END_INBOUND); @@ -308,8 +399,10 @@ public void failed(final Exception cause) { @Override public void releaseResources() { shutdown = true; + abortOutboundBudget(); outbound.clear(); inbound.clear(); + inbound.offer(END_INBOUND); outLock.lock(); try { currentOutbound = null; @@ -318,7 +411,97 @@ public void releaseResources() { } } - private static final class QueueInputStream extends InputStream { + private void enqueueOutbound(final byte[] data, final int off, final int len) throws IOException { + // Once the exchange has failed or been released nothing drains the queue any more, so a + // blocking reservation would wedge the worker thread; fail the write instead. + if (shutdown) { + throw new IOException("WebSocket stream already terminated"); + } + // Split the write into budget-sized chunks. Each chunk reserves its bytes against the + // outbound budget and blocks the worker until the reactor has drained enough previously + // queued bytes, so a single large write cannot buffer an unbounded amount of memory. + int position = off; + int remaining = len; + while (remaining > 0) { + final int n = Math.min(remaining, outboundChunkSize); + reserveOutboundBudget(n); + final byte[] chunk = new byte[n]; + System.arraycopy(data, position, chunk, 0, n); + outbound.add(ByteBuffer.wrap(chunk)); + // Signal after every chunk, before the next reservation can block: a write larger than + // the budget must let the reactor drain and free capacity, otherwise the worker would + // park with the reactor never told there is output to drain (deadlock). + requestOutput(); + position += n; + remaining -= n; + } + } + + /** + * Enqueues a zero-length control sentinel (END_OUTBOUND). Sentinels carry no payload and never + * block on the byte budget, so the HTTP/2 stream can always be terminated. + */ + private void enqueueSentinel(final ByteBuffer sentinel) { + outbound.add(sentinel); + requestOutput(); + } + + private void requestOutput() { + final DataStreamChannel channel = dataChannel; + if (responseSent && channel != null) { + channel.requestOutput(); + } + } + + private void reserveOutboundBudget(final int n) throws IOException { + outboundBudgetLock.lock(); + try { + // Wait until this chunk fits within the budget. A single chunk is always admitted when + // the queue is empty (pendingOutboundBytes == 0) so an oversized chunk cannot deadlock. + while (!shutdown && pendingOutboundBytes > 0 && pendingOutboundBytes + n > outboundBudgetBytes) { + try { + outboundBudgetAvailable.await(); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while waiting for outbound capacity", ex); + } + } + if (shutdown) { + throw new IOException("WebSocket stream already terminated"); + } + pendingOutboundBytes += n; + } finally { + outboundBudgetLock.unlock(); + } + } + + private void releaseOutboundBudget(final int n) { + if (n <= 0) { + return; + } + outboundBudgetLock.lock(); + try { + pendingOutboundBytes -= n; + if (pendingOutboundBytes < 0) { + pendingOutboundBytes = 0; + } + outboundBudgetAvailable.signalAll(); + } finally { + outboundBudgetLock.unlock(); + } + } + + private void abortOutboundBudget() { + outboundBudgetLock.lock(); + try { + pendingOutboundBytes = 0; + outboundBudgetAvailable.signalAll(); + } finally { + outboundBudgetLock.unlock(); + } + } + + private final class QueueInputStream extends InputStream { private final BlockingQueue queue; private byte[] current; @@ -331,6 +514,11 @@ private static final class QueueInputStream extends InputStream { @Override public int read() throws IOException { if (current == null || pos >= current.length) { + // The previous chunk is fully consumed: replenish inbound flow-control credit for + // its bytes so the peer may send more, keeping buffered memory bounded. + if (current != null && current != END_INBOUND && current.length > 0) { + replenishInbound(current.length); + } try { current = queue.take(); } catch (final InterruptedException ex) { @@ -348,40 +536,31 @@ public int read() throws IOException { private final class QueueOutputStream extends OutputStream { - private final BlockingQueue queue; - - QueueOutputStream(final BlockingQueue queue) { - this.queue = queue; - } - @Override public void write(final int b) throws IOException { - queue.offer(ByteBuffer.wrap(new byte[]{(byte) b})); - requestOutput(); + // enqueueOutbound signals the reactor after the chunk is queued. + enqueueOutbound(new byte[]{(byte) b}, 0, 1); } @Override public void write(final byte[] b, final int off, final int len) throws IOException { + // Validate the range before reserving budget: an invalid range must fail cleanly rather + // than reserve bytes and then throw in arraycopy, which would leak the reservation and + // wedge later writes. + if ((off | len) < 0 || len > b.length - off) { + throw new IndexOutOfBoundsException(); + } if (len == 0) { return; } - final byte[] copy = new byte[len]; - System.arraycopy(b, off, copy, 0, len); - queue.offer(ByteBuffer.wrap(copy)); - requestOutput(); + enqueueOutbound(b, off, len); } @Override - public void close() { - queue.offer(END_OUTBOUND); - requestOutput(); - } - - private void requestOutput() { - final DataStreamChannel channel = dataChannel; - if (responseSent && channel != null) { - channel.requestOutput(); - } + public void close() throws IOException { + // The END_OUTBOUND sentinel must always terminate the HTTP/2 stream, so it bypasses the + // byte budget even after a shutdown has cleared the queue. + enqueueSentinel(END_OUTBOUND); } } } diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessor.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessor.java index 97e1be1bf9..066de09319 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessor.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessor.java @@ -53,6 +53,10 @@ class WebSocketServerProcessor { } void process() throws IOException { + readLoop(); + } + + private void readLoop() throws IOException { ByteBuffer continuationBuf = null; WebSocketFrameType continuationType = null; while (true) { @@ -85,6 +89,9 @@ void process() throws IOException { return; case TEXT: case BINARY: + if (continuationBuf != null) { + throw new WebSocketProtocolException(1002, "Data frame during fragmented message"); + } if (frame.isFin()) { dispatchMessage(type, frame.getPayload()); } else { diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandler.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandler.java index 4a7adb556f..62edae6040 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandler.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandler.java @@ -81,7 +81,13 @@ public void handle( return; } if (!WebSocketHandshake.isWebSocketUpgrade(request)) { - trigger.submitResponse(new BasicClassicHttpResponse(HttpStatus.SC_UPGRADE_REQUIRED)); + final ClassicHttpResponse rejection = new BasicClassicHttpResponse(HttpStatus.SC_UPGRADE_REQUIRED); + // RFC 9110 section 15.5.22 requires a 426 response to carry an Upgrade header field, and + // section 7.8 the matching Connection: Upgrade option; RFC 6455 section 4.4 adds the version. + rejection.addHeader(HttpHeaders.CONNECTION, HeaderElements.UPGRADE); + rejection.addHeader(HttpHeaders.UPGRADE, "websocket"); + rejection.addHeader(WebSocketConstants.SEC_WEBSOCKET_VERSION, "13"); + trigger.submitResponse(rejection); return; } final WebSocketHandler handler = supplier.get(); @@ -96,42 +102,47 @@ public void handle( response.addHeader(HttpHeaders.CONNECTION, HeaderElements.UPGRADE); response.addHeader(HttpHeaders.UPGRADE, "websocket"); response.addHeader(WebSocketConstants.SEC_WEBSOCKET_ACCEPT, accept); - final WebSocketExtensionNegotiation negotiation = extensionRegistry.negotiate( - WebSocketExtensions.parse(request.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS)), - true); - final String extensionsHeader = negotiation.formatResponseHeader(); - if (extensionsHeader != null) { - response.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, extensionsHeader); - } - final List offeredProtocols = WebSocketHandshake.parseSubprotocols( - request.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL)); - final String protocol = handler.selectSubprotocol(offeredProtocols); - if (protocol != null) { - response.addHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL, protocol); - } - trigger.submitResponse(response); - final InputStream inputStream = connection.getSocketInputStream(); - final OutputStream outputStream = connection.getSocketOutputStream(); - if (inputStream == null || outputStream == null) { - connection.close(); - return; - } - final WebSocketSession session = new WebSocketSession(config, inputStream, outputStream, - connection.getRemoteAddress(), connection.getLocalAddress(), negotiation.getExtensions()); - try { - handler.onOpen(session); - new WebSocketServerProcessor(session, handler, config.getMaxMessageSize()).process(); - } catch (final WebSocketProtocolException ex) { - handler.onError(session, ex); - session.close(ex.closeCode, ex.getMessage()); - } catch (final WebSocketException ex) { - handler.onError(session, ex); - session.close(WebSocketCloseStatus.PROTOCOL_ERROR.getCode(), ex.getMessage()); - } catch (final Exception ex) { - handler.onError(session, ex); - session.close(WebSocketCloseStatus.INTERNAL_ERROR.getCode(), "WebSocket error"); - } finally { - connection.close(); + // Try-with-resources transfers ownership of the negotiated extensions to this scope: they + // are released on every exit path (a failed submitResponse, unavailable streams, a throwing + // onOpen and normal or exceptional processor termination), and a close failure is attached + // as a suppressed exception rather than masking the primary failure. + try (final WebSocketExtensionNegotiation negotiation = extensionRegistry.negotiate( + WebSocketExtensions.parse(request.headerIterator(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS)), + true)) { + final String extensionsHeader = negotiation.formatResponseHeader(); + if (extensionsHeader != null) { + response.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, extensionsHeader); + } + final List offeredProtocols = WebSocketHandshake.parseSubprotocols( + request.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL)); + final String protocol = handler.selectSubprotocol(offeredProtocols); + if (protocol != null) { + response.addHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL, protocol); + } + trigger.submitResponse(response); + final InputStream inputStream = connection.getSocketInputStream(); + final OutputStream outputStream = connection.getSocketOutputStream(); + if (inputStream == null || outputStream == null) { + connection.close(); + return; + } + final WebSocketSession session = new WebSocketSession(config, inputStream, outputStream, + connection.getRemoteAddress(), connection.getLocalAddress(), negotiation.getExtensions()); + try { + handler.onOpen(session); + new WebSocketServerProcessor(session, handler, config.getMaxMessageSize()).process(); + } catch (final WebSocketProtocolException ex) { + handler.onError(session, ex); + session.close(ex.closeCode, ex.getMessage()); + } catch (final WebSocketException ex) { + handler.onError(session, ex); + session.close(WebSocketCloseStatus.PROTOCOL_ERROR.getCode(), ex.getMessage()); + } catch (final Exception ex) { + handler.onError(session, ex); + session.close(WebSocketCloseStatus.INTERNAL_ERROR.getCode(), "WebSocket error"); + } finally { + connection.close(); + } } } diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http1UpgradeProtocolExtensionTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http1UpgradeProtocolExtensionTest.java index 153a632cdd..aa5359eca1 100644 --- a/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http1UpgradeProtocolExtensionTest.java +++ b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http1UpgradeProtocolExtensionTest.java @@ -29,8 +29,10 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import org.apache.hc.client5.http.websocket.api.WebSocketClientConfig; +import org.apache.hc.core5.http.message.BasicHttpResponse; import org.apache.hc.core5.websocket.extension.ExtensionChain; import org.apache.hc.core5.websocket.frame.FrameHeaderBits; import org.junit.jupiter.api.Test; @@ -47,36 +49,102 @@ void pmce_rejectedWhenDisabled() { } @Test - void pmce_rejectedWhenParametersNotOffered() { + void pmce_rejectedWhenClientWindowBitsNotOffered() { final WebSocketClientConfig cfg = WebSocketClientConfig.custom() .offerServerNoContextTakeover(false) .offerClientNoContextTakeover(false) - .offerClientMaxWindowBits(null) .offerServerMaxWindowBits(null) .build(); - assertThrows(IllegalStateException.class, () -> - Http1UpgradeProtocol.buildExtensionChain(cfg, "permessage-deflate; server_no_context_takeover")); + // The client never offers client_max_window_bits, so a response that echoes it must be rejected. assertThrows(IllegalStateException.class, () -> Http1UpgradeProtocol.buildExtensionChain(cfg, "permessage-deflate; client_max_window_bits=15")); } @Test - void pmce_rejectedOnUnknownOrDuplicate() { + void pmce_acceptsUninvitedNoContextTakeover() { + // RFC 7692 sections 7.1.1.1 and 7.1.1.2: the server may include either + // no-context-takeover parameter in the response even when the offer did not. + final WebSocketClientConfig cfg = WebSocketClientConfig.custom() + .offerServerNoContextTakeover(false) + .offerClientNoContextTakeover(false) + .build(); + final ExtensionChain chain = Http1UpgradeProtocol.buildExtensionChain(cfg, + "permessage-deflate; server_no_context_takeover; client_no_context_takeover"); + assertFalse(chain.isEmpty()); + } + + @Test + void pmce_acceptsUninvitedServerMaxWindowBits() { + // RFC 7692 section 7.1.2.1: the server may constrain its own window uninvited. + final WebSocketClientConfig cfg = WebSocketClientConfig.custom() + .offerServerMaxWindowBits(null) + .build(); + final ExtensionChain chain = Http1UpgradeProtocol.buildExtensionChain(cfg, + "permessage-deflate; server_max_window_bits=12"); + assertFalse(chain.isEmpty()); + } + + @Test + void headerValue_combinesRepeatedHeaders() { + // RFC 6455 section 9.1: Sec-WebSocket-Extensions may be split across multiple + // header fields; all instances must be taken into account. + final BasicHttpResponse response = new BasicHttpResponse(101); + response.addHeader("Sec-WebSocket-Extensions", "permessage-deflate"); + response.addHeader("Sec-WebSocket-Extensions", "x-unknown"); + assertEquals("permessage-deflate, x-unknown", + Http1UpgradeProtocol.headerValue(response, "Sec-WebSocket-Extensions")); + // The combined value must flow into validation: the unknown extension is rejected. final WebSocketClientConfig cfg = WebSocketClientConfig.custom().build(); assertThrows(IllegalStateException.class, () -> - Http1UpgradeProtocol.buildExtensionChain(cfg, "permessage-deflate; unknown=1")); + Http1UpgradeProtocol.buildExtensionChain(cfg, + Http1UpgradeProtocol.headerValue(response, "Sec-WebSocket-Extensions"))); + } + + @Test + void pmce_rejectsDuplicateServerNoContextTakeover() { + // RFC 7692 section 7.1: a parameter name must not appear more than once in a response. + final WebSocketClientConfig cfg = WebSocketClientConfig.custom().build(); + final IllegalStateException ex = assertThrows(IllegalStateException.class, () -> + Http1UpgradeProtocol.buildExtensionChain(cfg, + "permessage-deflate; server_no_context_takeover; server_no_context_takeover")); + assertTrue(ex.getMessage().startsWith("Duplicate permessage-deflate parameter"), ex.getMessage()); + } + + @Test + void pmce_rejectsDuplicateClientNoContextTakeover() { + final WebSocketClientConfig cfg = WebSocketClientConfig.custom().build(); + final IllegalStateException ex = assertThrows(IllegalStateException.class, () -> + Http1UpgradeProtocol.buildExtensionChain(cfg, + "permessage-deflate; client_no_context_takeover; client_no_context_takeover")); + assertTrue(ex.getMessage().startsWith("Duplicate permessage-deflate parameter"), ex.getMessage()); + } + + @Test + void pmce_rejectsDuplicateServerMaxWindowBits() { + final WebSocketClientConfig cfg = WebSocketClientConfig.custom().build(); + final IllegalStateException ex = assertThrows(IllegalStateException.class, () -> + Http1UpgradeProtocol.buildExtensionChain(cfg, + "permessage-deflate; server_max_window_bits=12; server_max_window_bits=10")); + assertTrue(ex.getMessage().startsWith("Duplicate permessage-deflate parameter"), ex.getMessage()); + } + + @Test + void pmce_rejectsDuplicateClientMaxWindowBits() { + // client_max_window_bits can no longer be offered, so the first occurrence is already + // rejected as "not offered"; a syntactically invalid duplicated response must still fail. + final WebSocketClientConfig cfg = WebSocketClientConfig.custom().build(); assertThrows(IllegalStateException.class, () -> - Http1UpgradeProtocol.buildExtensionChain(cfg, "permessage-deflate, permessage-deflate")); + Http1UpgradeProtocol.buildExtensionChain(cfg, + "permessage-deflate; client_max_window_bits=15; client_max_window_bits=12")); } @Test - void pmce_rejectedOnUnsupportedClientWindowBits() { - final WebSocketClientConfig cfg = WebSocketClientConfig.custom() - .offerClientMaxWindowBits(15) - .offerServerMaxWindowBits(15) - .build(); + void pmce_rejectedOnUnknownOrDuplicate() { + final WebSocketClientConfig cfg = WebSocketClientConfig.custom().build(); assertThrows(IllegalStateException.class, () -> - Http1UpgradeProtocol.buildExtensionChain(cfg, "permessage-deflate; client_max_window_bits=12")); + Http1UpgradeProtocol.buildExtensionChain(cfg, "permessage-deflate; unknown=1")); + assertThrows(IllegalStateException.class, () -> + Http1UpgradeProtocol.buildExtensionChain(cfg, "permessage-deflate, permessage-deflate")); } @Test @@ -93,13 +161,12 @@ void pmce_acceptsServerWindowBitsBelow15() { @Test void pmce_validNegotiation_buildsChain() { final WebSocketClientConfig cfg = WebSocketClientConfig.custom() - .offerClientMaxWindowBits(15) .offerServerMaxWindowBits(15) .offerClientNoContextTakeover(true) .offerServerNoContextTakeover(true) .build(); final ExtensionChain chain = Http1UpgradeProtocol.buildExtensionChain(cfg, - "permessage-deflate; client_no_context_takeover; server_no_context_takeover; client_max_window_bits=15"); + "permessage-deflate; client_no_context_takeover; server_no_context_takeover; server_max_window_bits=15"); assertFalse(chain.isEmpty()); assertEquals(FrameHeaderBits.RSV1, chain.rsvMask()); } diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/example/WebSocketEchoClient.java b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/example/WebSocketEchoClient.java index e21526690f..1f131eb846 100644 --- a/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/example/WebSocketEchoClient.java +++ b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/example/WebSocketEchoClient.java @@ -54,7 +54,6 @@ *
    *
  • Per-message deflate enabled
    • *
  • Server and client context takeover disabled (stateless compression)
    • - *
  • Client maximum window bits set to 15
    • *
* *

The lifecycle follows these steps:

@@ -89,7 +88,6 @@ public static void main(final String[] args) throws Exception { .enablePerMessageDeflate(true) .offerServerNoContextTakeover(true) .offerClientNoContextTakeover(true) - .offerClientMaxWindowBits(15) .setCloseWaitTimeout(Timeout.ofSeconds(2)) .build(); diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransportTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransportTest.java new file mode 100644 index 0000000000..e65826f459 --- /dev/null +++ b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransportTest.java @@ -0,0 +1,93 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.client5.http.websocket.transport; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.CancelledKeyException; +import java.util.List; + +import org.apache.hc.core5.http.Header; +import org.apache.hc.core5.http.nio.DataStreamChannel; +import org.junit.jupiter.api.Test; + +/** + * Verifies the transport's behaviour when the channel's selection key has been cancelled by a + * concurrent shutdown: {@code requestOutput()} tolerates it as a best-effort nudge, while + * {@code write(ByteBuffer)} surfaces it as an {@link IOException} so the engine takes its close + * path instead of misreading it as transient back-pressure. + */ +class DataStreamChannelTransportTest { + + private static final class CancelledChannel implements DataStreamChannel { + + @Override + public int write(final ByteBuffer src) { + throw new CancelledKeyException(); + } + + @Override + public void requestOutput() { + throw new CancelledKeyException(); + } + + @Override + public void endStream() { + } + + @Override + public void endStream(final List trailers) { + } + } + + @Test + void requestOutputSwallowsCancelledKeyDuringShutdown() { + final DataStreamChannelTransport transport = new DataStreamChannelTransport(); + transport.setChannel(new CancelledChannel()); + assertDoesNotThrow(transport::requestOutput); + } + + @Test + void writeThrowsOnCancelledChannel() { + final DataStreamChannelTransport transport = new DataStreamChannelTransport(); + transport.setChannel(new CancelledChannel()); + // A cancelled selection key is terminal; write() must surface an IOException rather than + // report written == 0, which the engine would misread as transient back-pressure. + final IOException ex = assertThrows(IOException.class, + () -> transport.write(ByteBuffer.allocate(4))); + assertTrue(ex.getCause() instanceof CancelledKeyException); + } + + @Test + void requestOutputIsNoOpWithoutChannel() { + assertDoesNotThrow(new DataStreamChannelTransport()::requestOutput); + } +} diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/transport/WebSocketInboundRfcGapTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/transport/WebSocketInboundRfcGapTest.java new file mode 100644 index 0000000000..2d9f5427d6 --- /dev/null +++ b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/transport/WebSocketInboundRfcGapTest.java @@ -0,0 +1,127 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.client5.http.websocket.transport; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; + +import java.nio.ByteBuffer; +import java.nio.CharBuffer; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.hc.client5.http.websocket.api.WebSocketClientConfig; +import org.apache.hc.client5.http.websocket.api.WebSocketListener; +import org.apache.hc.core5.websocket.extension.ExtensionChain; +import org.apache.hc.core5.websocket.extension.PerMessageDeflate; +import org.apache.hc.core5.websocket.frame.FrameOpcode; +import org.junit.jupiter.api.Test; + +/** + * Reproduces two inbound RFC 6455 conformance gaps in the live NIO client engine that + * {@link WebSocketInboundRfcTest} does not cover: a control frame carrying RSV1 while an extension + * is negotiated (RFC 6455 ยง5.2), and delivery of a truncated over-limit message to the listener + * after the connection has already been failed with 1009 (RFC 6455 ยง5.4 / ยง7.1.7). + */ +class WebSocketInboundRfcGapTest { + + @Test + void rsv1OnControlFrameWithNegotiatedExtension_closesWith1002() { + final CaptureListener listener = new CaptureListener(); + // permessage-deflate negotiated: the engine owns RSV1 for DATA frames, but a control frame + // is never compressed, so RSV1 on a PING has no defined meaning and MUST fail the connection. + final ExtensionChain chain = new ExtensionChain(); + chain.add(new PerMessageDeflate(true, false, false, null, null)); + final WebSocketSessionEngine engine = newEngine(listener, 0L, chain); + + engine.onData(controlFrameWithRsv1(FrameOpcode.PING)); + + assertEquals(1002, listener.closeCode.get(), + "RSV1 on a control frame must fail the connection with 1002"); + } + + @Test + void overLimitFinalFragment_doesNotDeliverTruncatedMessageAfter1009() { + final CaptureListener listener = new CaptureListener(); + final WebSocketSessionEngine engine = newEngine(listener, 4L, null); + + engine.onData(unmaskedFrame(FrameOpcode.BINARY, false, false, new byte[]{1, 2, 3})); + engine.onData(unmaskedFrame(FrameOpcode.CONT, true, false, new byte[]{4, 5})); + + assertEquals(1009, listener.closeCode.get(), "over-limit message must fail with 1009"); + assertFalse(listener.messageDelivered.get(), + "a message must not be delivered to the listener after the connection is failed"); + } + + private static WebSocketSessionEngine newEngine(final CaptureListener listener, + final long maxMessageSize, + final ExtensionChain chain) { + final WebSocketClientConfig.Builder cfg = WebSocketClientConfig.custom() + .enablePerMessageDeflate(false); + if (maxMessageSize > 0) { + cfg.setMaxMessageSize(maxMessageSize); + } + return new WebSocketSessionEngine(new StubTransport(), listener, cfg.build(), chain, null); + } + + private static ByteBuffer unmaskedFrame(final int opcode, final boolean fin, final boolean rsv1, + final byte[] payload) { + final int len = payload != null ? payload.length : 0; + final ByteBuffer buf = ByteBuffer.allocate(2 + len); + buf.put((byte) ((fin ? 0x80 : 0x00) | (rsv1 ? 0x40 : 0x00) | (opcode & 0x0F))); + buf.put((byte) len); + if (len > 0) { + buf.put(payload); + } + buf.flip(); + return buf; + } + + private static ByteBuffer controlFrameWithRsv1(final int opcode) { + return unmaskedFrame(opcode, true, true, new byte[0]); + } + + private static final class CaptureListener implements WebSocketListener { + private final AtomicInteger closeCode = new AtomicInteger(-1); + private final AtomicBoolean messageDelivered = new AtomicBoolean(false); + + @Override + public void onText(final CharBuffer data, final boolean last) { + messageDelivered.set(true); + } + + @Override + public void onBinary(final ByteBuffer data, final boolean last) { + messageDelivered.set(true); + } + + @Override + public void onClose(final int statusCode, final String reason) { + closeCode.compareAndSet(-1, statusCode); + } + } +} diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactoryTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactoryTest.java index 964ed201ce..d034ba2473 100644 --- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactoryTest.java +++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactoryTest.java @@ -32,8 +32,10 @@ import java.util.Collections; import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; +import org.apache.hc.core5.http.message.BasicHeader; import org.junit.jupiter.api.Test; class PerMessageDeflateExtensionFactoryTest { @@ -69,4 +71,83 @@ void echoesNegotiatedWindowBitsWhenRequested() { assertEquals("15", response.getParameters().get("client_max_window_bits")); assertEquals("15", response.getParameters().get("server_max_window_bits")); } + + @Test + void rejectsServerMaxWindowBitsWithoutValue() { + assertNull(negotiateServer("permessage-deflate; server_max_window_bits"), + "server_max_window_bits must carry a value"); + } + + @Test + void rejectsServerMaxWindowBitsNonNumeric() { + final Map params = new LinkedHashMap<>(); + params.put("server_max_window_bits", "abc"); + assertNull(create(params), "a non-numeric server_max_window_bits must be rejected"); + } + + @Test + void rejectsServerMaxWindowBitsBelowRange() { + final Map params = new LinkedHashMap<>(); + params.put("server_max_window_bits", "7"); + assertNull(create(params), "server_max_window_bits below 8 must be rejected"); + } + + @Test + void rejectsClientMaxWindowBitsNonNumeric() { + final Map params = new LinkedHashMap<>(); + params.put("client_max_window_bits", "abc"); + assertNull(create(params), "a present but non-numeric client_max_window_bits must be rejected"); + } + + @Test + void acceptsValuelessClientMaxWindowBits() { + assertNotNull(negotiateServer("permessage-deflate; client_max_window_bits"), + "a valueless client_max_window_bits merely advertises support and is accepted"); + } + + @Test + void rejectsUnknownParameter() { + final Map params = new LinkedHashMap<>(); + params.put("unknown", "1"); + assertNull(create(params), "an unknown parameter must be rejected"); + } + + @Test + void rejectsValuedNoContextTakeover() { + final Map params = new LinkedHashMap<>(); + params.put("server_no_context_takeover", "1"); + assertNull(create(params), "server_no_context_takeover must not carry a value"); + } + + @Test + void acceptsValuelessNoContextTakeover() { + assertNotNull(negotiateServer("permessage-deflate; server_no_context_takeover; client_no_context_takeover"), + "valueless no-context-takeover parameters are accepted"); + } + + @Test + void rejectsLeadingZeroWindowBits() { + final Map params = new LinkedHashMap<>(); + params.put("server_max_window_bits", "015"); + assertNull(create(params), "a leading-zero window-bits value must be rejected"); + } + + @Test + void rejectsPlusSignedWindowBits() { + final Map params = new LinkedHashMap<>(); + params.put("server_max_window_bits", "+15"); + assertNull(create(params), "a signed window-bits value must be rejected"); + } + + private static WebSocketExtension create(final Map params) { + return new PerMessageDeflateExtensionFactory().create( + new WebSocketExtensionData("permessage-deflate", params), true); + } + + private static WebSocketExtension negotiateServer(final String headerValue) { + final List offers = WebSocketExtensions.parse( + new BasicHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, headerValue)); + assertEquals(1, offers.size(), "the header must parse to a single offer"); + return new PerMessageDeflateExtensionFactory().create(offers.get(0), true); + } } diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionTest.java index 20fd8b019b..b561fbd700 100644 --- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionTest.java +++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionTest.java @@ -96,6 +96,73 @@ void decodeZeroLimitMeansUnlimited() throws Exception { assertArrayEquals(plain, toBytes(out)); } + @Test + void fragmentedEncodeRoundTripsThroughDecode() throws Exception { + // Encode a message as three frames (fin=false, false, true), then decode them back. + // Non-final fragments must retain the 00 00 FF FF flush trailer so the reassembled + // DEFLATE stream stays valid (RFC 7692 section 7.2.1). + final PerMessageDeflateExtension enc = new PerMessageDeflateExtension(); + final byte[] p1 = "The quick brown fox ".getBytes(StandardCharsets.UTF_8); + final byte[] p2 = "jumps over the lazy ".getBytes(StandardCharsets.UTF_8); + final byte[] p3 = "dog, and again the fox.".getBytes(StandardCharsets.UTF_8); + + final ByteBuffer f1 = enc.encode(WebSocketFrameType.TEXT, false, ByteBuffer.wrap(p1)); + final ByteBuffer f2 = enc.encode(WebSocketFrameType.CONTINUATION, false, ByteBuffer.wrap(p2)); + final ByteBuffer f3 = enc.encode(WebSocketFrameType.CONTINUATION, true, ByteBuffer.wrap(p3)); + + final PerMessageDeflateExtension dec = new PerMessageDeflateExtension(); + final ByteArrayOutputStream joined = new ByteArrayOutputStream(); + joined.write(toBytes(dec.decode(WebSocketFrameType.TEXT, false, f1))); + joined.write(toBytes(dec.decode(WebSocketFrameType.CONTINUATION, false, f2))); + joined.write(toBytes(dec.decode(WebSocketFrameType.CONTINUATION, true, f3))); + + final byte[] expected = new byte[p1.length + p2.length + p3.length]; + System.arraycopy(p1, 0, expected, 0, p1.length); + System.arraycopy(p2, 0, expected, p1.length, p2.length); + System.arraycopy(p3, 0, expected, p1.length + p2.length, p3.length); + assertArrayEquals(expected, joined.toByteArray()); + } + + @Test + void encodesEmptyMessageAsSingleZeroOctet() throws Exception { + final PerMessageDeflateExtension enc = new PerMessageDeflateExtension(); + final ByteBuffer encoded = enc.encode(WebSocketFrameType.BINARY, true, ByteBuffer.allocate(0)); + + // RFC 7692 section 7.2.3.6: an empty compressed message is the single octet 0x00. + assertArrayEquals(new byte[]{0x00}, toBytes(encoded)); + + final PerMessageDeflateExtension dec = new PerMessageDeflateExtension(); + assertArrayEquals(new byte[0], + toBytes(dec.decode(WebSocketFrameType.BINARY, true, ByteBuffer.wrap(new byte[]{0x00})))); + } + + @Test + void encodesEmptyMessageAsZeroOctetWithContextTakeover() throws Exception { + // The default extension keeps deflate/inflate context across messages. + final PerMessageDeflateExtension enc = new PerMessageDeflateExtension(); + final PerMessageDeflateExtension dec = new PerMessageDeflateExtension(); + + final byte[] hello = "hello".getBytes(StandardCharsets.UTF_8); + assertArrayEquals(hello, toBytes(dec.decode(WebSocketFrameType.TEXT, true, + enc.encode(WebSocketFrameType.TEXT, true, ByteBuffer.wrap(hello))))); + + // The deflater now holds context, so an empty message flushes to only 00 00 FF FF; it must + // still encode to the single octet 0x00 (RFC 7692 section 7.2.3.6), not to an empty payload. + final ByteBuffer emptyEncoded = enc.encode(WebSocketFrameType.TEXT, true, ByteBuffer.allocate(0)); + assertArrayEquals(new byte[]{0x00}, toBytes(emptyEncoded)); + assertArrayEquals(new byte[0], toBytes(dec.decode(WebSocketFrameType.TEXT, true, emptyEncoded))); + } + + @Test + void closeReleasesNativeCodecs() throws Exception { + final PerMessageDeflateExtension ext = new PerMessageDeflateExtension(); + ext.encode(WebSocketFrameType.TEXT, true, ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8))); + ext.close(); + // After close() the Deflater has been ended; reusing the extension must fail. + assertThrows(Exception.class, + () -> ext.encode(WebSocketFrameType.TEXT, true, ByteBuffer.wrap("again".getBytes(StandardCharsets.UTF_8)))); + } + private static byte[] deflateWithSyncFlush(final byte[] input) { final Deflater deflater = new Deflater(Deflater.DEFAULT_COMPRESSION, true); deflater.setInput(input); diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiationTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiationTest.java index c1c5a415d2..6697567310 100644 --- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiationTest.java +++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiationTest.java @@ -30,6 +30,7 @@ import static org.junit.jupiter.api.Assertions.assertNull; import java.util.Collections; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; @@ -51,4 +52,27 @@ void formatsEmptyHeaderAsNull() { Collections.emptyList(), Collections.emptyList()); assertNull(negotiation.formatResponseHeader()); } + + @Test + void closeIsIdempotentAndClosesEachExtensionOnce() { + final AtomicInteger closed = new AtomicInteger(); + final WebSocketExtension extension = new WebSocketExtension() { + @Override + public String getName() { + return "x-test"; + } + + @Override + public void close() { + closed.incrementAndGet(); + } + }; + final WebSocketExtensionNegotiation negotiation = new WebSocketExtensionNegotiation( + Collections.singletonList(extension), Collections.emptyList()); + + negotiation.close(); + negotiation.close(); + + assertEquals(1, closed.get(), "each extension must be closed exactly once"); + } } diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistryTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistryTest.java index f3b6b7418b..16b21512ea 100644 --- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistryTest.java +++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistryTest.java @@ -28,9 +28,15 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; @@ -102,4 +108,112 @@ void defaultRegistryContainsPerMessageDeflate() throws Exception { true); assertEquals(1, negotiation.getExtensions().size()); } + + @Test + void duplicateOffersAreDeduplicated() throws Exception { + final WebSocketExtensionRegistry registry = WebSocketExtensionRegistry.createDefault(); + // A client may offer the same extension more than once; accepting both would claim RSV1 + // twice and later break the frame reader with an IllegalStateException. + final WebSocketExtensionNegotiation negotiation = registry.negotiate( + Arrays.asList( + new WebSocketExtensionData("permessage-deflate", Collections.emptyMap()), + new WebSocketExtensionData("permessage-deflate", Collections.emptyMap())), + true); + assertEquals(1, negotiation.getExtensions().size(), "duplicate offers must collapse to one extension"); + assertEquals(1, negotiation.getResponseData().size(), "response must list the extension once"); + } + + @Test + void fallbackOfferIsAcceptedWhenTheFirstIsDeclined() throws Exception { + final WebSocketExtensionRegistry registry = WebSocketExtensionRegistry.createDefault(); + final Map unsupported = new LinkedHashMap<>(); + unsupported.put("server_max_window_bits", "12"); + // The first offer asks for a 12-bit window the JDK Deflater cannot honour and is declined; + // the registry must move on and accept the client's plain fallback offer. + final WebSocketExtensionNegotiation negotiation = registry.negotiate( + Arrays.asList( + new WebSocketExtensionData("permessage-deflate", unsupported), + new WebSocketExtensionData("permessage-deflate", Collections.emptyMap())), + true); + assertEquals(1, negotiation.getExtensions().size(), "the fallback offer must be negotiated"); + } + + @Test + void negotiateClosesCreatedExtensionsWhenLaterFactoryThrows() { + final AtomicInteger closed = new AtomicInteger(); + final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry() + .register(closingFactory("ext-a", closed, null)) + .register(throwingFactory("ext-b", new WebSocketException("boom"))); + + final WebSocketException ex = assertThrows(WebSocketException.class, () -> + registry.negotiate(Arrays.asList( + new WebSocketExtensionData("ext-a", Collections.emptyMap()), + new WebSocketExtensionData("ext-b", Collections.emptyMap())), + true)); + + assertEquals("boom", ex.getMessage()); + assertEquals(1, closed.get(), "the already-created extension must be closed exactly once"); + } + + @Test + void negotiateSuppressesCloseFailureOnOriginalException() { + final WebSocketException primary = new WebSocketException("primary"); + final IllegalStateException closeFailure = new IllegalStateException("close failed"); + final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry() + .register(closingFactory("ext-a", new AtomicInteger(), closeFailure)) + .register(throwingFactory("ext-b", primary)); + + final WebSocketException ex = assertThrows(WebSocketException.class, () -> + registry.negotiate(Arrays.asList( + new WebSocketExtensionData("ext-a", Collections.emptyMap()), + new WebSocketExtensionData("ext-b", Collections.emptyMap())), + true)); + + assertSame(primary, ex, "the original negotiation failure must be preserved"); + assertEquals(1, ex.getSuppressed().length, "the close failure must be attached as suppressed"); + assertSame(closeFailure, ex.getSuppressed()[0]); + } + + private static WebSocketExtensionFactory closingFactory( + final String name, final AtomicInteger closed, final RuntimeException closeFailure) { + return new WebSocketExtensionFactory() { + @Override + public String getName() { + return name; + } + + @Override + public WebSocketExtension create(final WebSocketExtensionData request, final boolean server) { + return new WebSocketExtension() { + @Override + public String getName() { + return name; + } + + @Override + public void close() { + closed.incrementAndGet(); + if (closeFailure != null) { + throw closeFailure; + } + } + }; + } + }; + } + + private static WebSocketExtensionFactory throwingFactory(final String name, final WebSocketException failure) { + return new WebSocketExtensionFactory() { + @Override + public String getName() { + return name; + } + + @Override + public WebSocketExtension create(final WebSocketExtensionData request, final boolean server) + throws WebSocketException { + throw failure; + } + }; + } } diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionsTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionsTest.java index 5b48cae8f8..03da54f692 100644 --- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionsTest.java +++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionsTest.java @@ -27,10 +27,13 @@ package org.apache.hc.core5.websocket; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import java.util.Arrays; import java.util.List; import java.util.Map; +import org.apache.hc.core5.http.Header; import org.apache.hc.core5.http.message.BasicHeader; import org.junit.jupiter.api.Test; @@ -49,4 +52,37 @@ void parsesExtensionsHeader() { final Map params = data.get(1).getParameters(); assertEquals("1", params.get("bar")); } + + @Test + void dropsOfferWithDuplicateParameter() { + final BasicHeader header = new BasicHeader( + WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, + "permessage-deflate; client_no_context_takeover; client_no_context_takeover"); + assertTrue(WebSocketExtensions.parse(header).isEmpty(), + "a repeated parameter makes the offer invalid and it must be dropped, not collapsed"); + } + + @Test + void dropsOnlyTheOfferWithDuplicateParameter() { + final BasicHeader header = new BasicHeader( + WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, + "permessage-deflate; server_max_window_bits=10; server_max_window_bits=12, foo; bar=1"); + final List data = WebSocketExtensions.parse(header); + assertEquals(1, data.size(), "the malformed permessage-deflate offer is dropped, the valid one kept"); + assertEquals("foo", data.get(0).getName()); + } + + @Test + void combinesMultipleExtensionHeaders() { + // RFC 6455: multiple Sec-WebSocket-Extensions fields are a single combined value, so a + // client's fallback offer in a later field must still be examined. + final List data = WebSocketExtensions.parse(Arrays.
asList( + new BasicHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, + "permessage-deflate; server_max_window_bits=12"), + new BasicHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, + "permessage-deflate")).iterator()); + assertEquals(2, data.size(), "offers from every Sec-WebSocket-Extensions field must be combined"); + assertEquals("12", data.get(0).getParameters().get("server_max_window_bits")); + assertTrue(data.get(1).getParameters().isEmpty(), "the fallback offer must be preserved"); + } } diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketFrameWriterTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketFrameWriterTest.java index 0e80917d70..1f1fbf79cd 100644 --- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketFrameWriterTest.java +++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketFrameWriterTest.java @@ -27,6 +27,7 @@ package org.apache.hc.core5.websocket; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.ByteArrayOutputStream; @@ -89,6 +90,32 @@ void setsRsv1WhenExtensionUsesIt() throws Exception { assertTrue((out[0] & 0x40) != 0, "RSV1 must be set"); } + @Test + void rejectsOversizedControlFrame() { + final ByteArrayOutputStream baos = new ByteArrayOutputStream(); + final WebSocketFrameWriter writer = new WebSocketFrameWriter(baos, Collections.emptyList()); + assertThrows(IllegalArgumentException.class, + () -> writer.writePing(ByteBuffer.wrap(new byte[126]))); + } + + @Test + void closeReasonTruncatedToControlFrameLimit() throws Exception { + final ByteArrayOutputStream baos = new ByteArrayOutputStream(); + final WebSocketFrameWriter writer = new WebSocketFrameWriter(baos, Collections.emptyList()); + + final StringBuilder reason = new StringBuilder(); + for (int i = 0; i < 200; i++) { + reason.append('x'); + } + writer.writeClose(1000, reason.toString()); + + final byte[] out = baos.toByteArray(); + assertEquals((byte) 0x88, out[0]); // FIN + CLOSE + final int payloadLen = out[1] & 0xFF; // no MASK bit on server frames + assertTrue(payloadLen <= 125, "CLOSE payload must fit the 125-byte control frame limit, was " + payloadLen); + assertEquals(out.length - 2, payloadLen, "single-byte length header, no extended length"); + } + private static byte[] writeBinary(final byte[] payload, final List extensions) throws Exception { final ByteArrayOutputStream baos = new ByteArrayOutputStream(); final WebSocketFrameWriter writer = new WebSocketFrameWriter(baos, extensions); diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/ExtensionChainTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/ExtensionChainTest.java index a79abeb83d..8ef3ff3929 100644 --- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/ExtensionChainTest.java +++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/ExtensionChainTest.java @@ -27,8 +27,13 @@ package org.apache.hc.core5.websocket.extension; import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; @@ -47,4 +52,72 @@ void addAndUsePmce_decodeRoundTrip() throws Exception { assertArrayEquals(data, back); } + + @Test + void encodeChainClosesAllEncodersEvenWhenOneThrows() { + final AtomicInteger closed = new AtomicInteger(); + final RuntimeException boom = new IllegalStateException("boom"); + final WebSocketExtensionChain.Encoder throwing = new WebSocketExtensionChain.Encoder() { + @Override + public WebSocketExtensionChain.Encoded encode(final byte[] data, final boolean first, final boolean fin) { + return new WebSocketExtensionChain.Encoded(data, first); + } + + @Override + public void close() { + throw boom; + } + }; + final WebSocketExtensionChain.Encoder counting = new WebSocketExtensionChain.Encoder() { + @Override + public WebSocketExtensionChain.Encoded encode(final byte[] data, final boolean first, final boolean fin) { + return new WebSocketExtensionChain.Encoded(data, first); + } + + @Override + public void close() { + closed.incrementAndGet(); + } + }; + final ExtensionChain.EncodeChain chain = new ExtensionChain.EncodeChain(Arrays.asList(throwing, counting)); + + final RuntimeException ex = assertThrows(RuntimeException.class, chain::close); + + assertSame(boom, ex); + assertEquals(1, closed.get(), "the second encoder must still be closed"); + } + + @Test + void decodeChainClosesAllDecodersEvenWhenOneThrows() { + final AtomicInteger closed = new AtomicInteger(); + final RuntimeException boom = new IllegalStateException("boom"); + final WebSocketExtensionChain.Decoder throwing = new WebSocketExtensionChain.Decoder() { + @Override + public byte[] decode(final byte[] payload) { + return payload; + } + + @Override + public void close() { + throw boom; + } + }; + final WebSocketExtensionChain.Decoder counting = new WebSocketExtensionChain.Decoder() { + @Override + public byte[] decode(final byte[] payload) { + return payload; + } + + @Override + public void close() { + closed.incrementAndGet(); + } + }; + final ExtensionChain.DecodeChain chain = new ExtensionChain.DecodeChain(Arrays.asList(throwing, counting)); + + final RuntimeException ex = assertThrows(RuntimeException.class, chain::close); + + assertSame(boom, ex); + assertEquals(1, closed.get(), "the second decoder must still be closed"); + } } diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/MessageDeflateTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/MessageDeflateTest.java index 284566dac4..b1235cf05d 100644 --- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/MessageDeflateTest.java +++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/MessageDeflateTest.java @@ -84,6 +84,47 @@ void roundTrip_message() throws Exception { assertArrayEquals(plain, roundTrip); } + @Test + void encode_emptyMessage_isSingleZeroOctet() throws Exception { + final PerMessageDeflate pmce = new PerMessageDeflate(true, false, false, null, null); + final WebSocketExtensionChain.Encoder enc = pmce.newEncoder(); + + final WebSocketExtensionChain.Encoded encoded = enc.encode(new byte[0], true, true); + + // RFC 7692 section 7.2.3.6: an empty compressed message is the single octet 0x00. + assertArrayEquals(new byte[]{0x00}, encoded.payload); + assertTrue(encoded.setRsvOnFirst, "RSV1 on the first (and only) frame"); + assertArrayEquals(new byte[0], pmce.newDecoder().decode(encoded.payload), + "an empty compressed message must round-trip to empty"); + } + + @Test + void encode_emptyMessage_withContextTakeover_roundTrips() throws Exception { + // clientNoContextTakeover = false: the deflater keeps its context across messages. + final PerMessageDeflate pmce = new PerMessageDeflate(true, false, false, null, null); + final WebSocketExtensionChain.Encoder enc = pmce.newEncoder(); + final WebSocketExtensionChain.Decoder dec = pmce.newDecoder(); + + final byte[] hello = "hello".getBytes(StandardCharsets.UTF_8); + assertArrayEquals(hello, dec.decode(enc.encode(hello, true, true).payload)); + + final byte[] empty = enc.encode(new byte[0], true, true).payload; + assertArrayEquals(new byte[]{0x00}, empty, "empty message compresses to 0x00 with context takeover"); + assertArrayEquals(new byte[0], dec.decode(empty)); + } + + @Test + void encode_emptyMessage_withNoContextTakeover_roundTrips() throws Exception { + // clientNoContextTakeover = true: the deflater is reset after each message. + final PerMessageDeflate pmce = new PerMessageDeflate(true, false, true, null, null); + final WebSocketExtensionChain.Encoder enc = pmce.newEncoder(); + final WebSocketExtensionChain.Decoder dec = pmce.newDecoder(); + + final byte[] empty = enc.encode(new byte[0], true, true).payload; + assertArrayEquals(new byte[]{0x00}, empty, "empty message compresses to 0x00 with no-context-takeover"); + assertArrayEquals(new byte[0], dec.decode(empty)); + } + @Test void decode_withinLimit_succeeds() throws Exception { final PerMessageDeflate pmce = new PerMessageDeflate(true, true, true, null, null); @@ -134,6 +175,28 @@ void decode_zeroLimitMeansUnlimited() throws Exception { assertArrayEquals(plain, roundTrip); } + @Test + void encoderCloseReleasesDeflater() { + final PerMessageDeflate pmce = new PerMessageDeflate(true, false, false, null, null); + final WebSocketExtensionChain.Encoder enc = pmce.newEncoder(); + enc.encode("hello".getBytes(StandardCharsets.UTF_8), true, true); + enc.close(); + // After close() the Deflater has been ended; reusing the encoder must fail. + assertThrows(Exception.class, + () -> enc.encode("again".getBytes(StandardCharsets.UTF_8), true, true)); + } + + @Test + void decoderCloseReleasesInflater() throws Exception { + final PerMessageDeflate pmce = new PerMessageDeflate(true, false, false, null, null); + final WebSocketExtensionChain.Encoder enc = pmce.newEncoder(); + final WebSocketExtensionChain.Decoder dec = pmce.newDecoder(); + final byte[] wire = enc.encode("hello".getBytes(StandardCharsets.UTF_8), true, true).payload; + dec.decode(wire); + dec.close(); + assertThrows(Exception.class, () -> dec.decode(wire)); + } + private static boolean endsWithTail(final byte[] b) { if (b.length < 4) { return false; diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/PerMessageDeflateFragmentedRoundTripTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/PerMessageDeflateFragmentedRoundTripTest.java new file mode 100644 index 0000000000..cb7d69e33b --- /dev/null +++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/PerMessageDeflateFragmentedRoundTripTest.java @@ -0,0 +1,122 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.core5.websocket.extension; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; + +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; + +import org.apache.hc.core5.websocket.extension.WebSocketExtensionChain.Decoder; +import org.apache.hc.core5.websocket.extension.WebSocketExtensionChain.Encoder; +import org.junit.jupiter.api.Test; + +/** + * Proves that a permessage-deflate message split across several WebSocket data frames survives a + * compress/decompress round trip (RFC 7692 ยง7.2). The existing {@link MessageDeflateTest} only + * exercises the single-frame path ({@code encode(data, true, true)}); the fragmented path + * ({@code compressFragment}) is never decoded back. + */ +final class PerMessageDeflateFragmentedRoundTripTest { + + private static final String TEXT = + "The quick brown fox jumps over the lazy dog. " + + "Pack my box with five dozen liquor jugs. " + + "How vexingly quick daft zebras jump! " + + "The five boxing wizards jump quickly."; + + private static byte[] encodeFragmented(final Encoder enc, final byte[] plain, final int fragments) { + final ByteArrayOutputStream wire = new ByteArrayOutputStream(); + final int chunk = (plain.length + fragments - 1) / fragments; + for (int i = 0; i < fragments; i++) { + final int off = i * chunk; + final int len = Math.min(chunk, plain.length - off); + final byte[] part = new byte[len]; + System.arraycopy(plain, off, part, 0, len); + final boolean first = i == 0; + final boolean fin = i == fragments - 1; + final byte[] out = enc.encode(part, first, fin).payload; + wire.write(out, 0, out.length); + } + return wire.toByteArray(); + } + + @Test + void fragmentedMessageRoundTripsAcrossThreeFrames() throws Exception { + final PerMessageDeflate pmce = new PerMessageDeflate(true, false, false, null, null); + final Encoder enc = pmce.newEncoder(); + final Decoder dec = pmce.newDecoder(); + + final byte[] plain = TEXT.getBytes(StandardCharsets.UTF_8); + + final byte[] wire = encodeFragmented(enc, plain, 3); + final byte[] roundTrip = dec.decode(wire); + + assertArrayEquals(plain, roundTrip, + "A fragmented compressed message must reassemble to the original bytes"); + } + + @Test + void fragmentedMessageWithEmptyFinalFragmentRoundTrips() throws Exception { + final PerMessageDeflate pmce = new PerMessageDeflate(true, false, false, null, null); + final Encoder enc = pmce.newEncoder(); + final Decoder dec = pmce.newDecoder(); + + final byte[] head = "hello".getBytes(StandardCharsets.UTF_8); + final ByteArrayOutputStream wire = new ByteArrayOutputStream(); + final byte[] firstFragment = enc.encode(head, true, false).payload; + wire.write(firstFragment, 0, firstFragment.length); + final byte[] finalFragment = enc.encode(new byte[0], false, true).payload; + wire.write(finalFragment, 0, finalFragment.length); + + // RFC 7692 section 7.2.3.6: an empty final fragment is the single octet 0x00. + assertArrayEquals(new byte[]{0x00}, finalFragment, + "an empty final fragment must be encoded as the single octet 0x00"); + assertArrayEquals(head, dec.decode(wire.toByteArray()), + "a fragmented message whose final fragment is empty must round-trip"); + } + + @Test + void fragmentedMessageMatchesSingleFrameEncoding() throws Exception { + final byte[] plain = TEXT.getBytes(StandardCharsets.UTF_8); + + final byte[] fragmentedWire = encodeFragmented( + new PerMessageDeflate(true, false, false, null, null).newEncoder(), plain, 4); + final byte[] singleWire = + new PerMessageDeflate(true, false, false, null, null).newEncoder() + .encode(plain, true, true).payload; + + // The wire bytes may differ (block boundaries), but both must decode to the same plaintext. + final byte[] fromFragmented = + new PerMessageDeflate(true, false, false, null, null).newDecoder().decode(fragmentedWire); + final byte[] fromSingle = + new PerMessageDeflate(true, false, false, null, null).newDecoder().decode(singleWire); + + assertArrayEquals(plain, fromSingle, "single-frame control must round-trip"); + assertArrayEquals(plain, fromFragmented, "fragmented message must round-trip to the same plaintext"); + } +} diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandlerTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandlerTest.java index ae88106701..e54b6aa334 100644 --- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandlerTest.java +++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandlerTest.java @@ -26,22 +26,49 @@ */ package org.apache.hc.core5.websocket.server; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.apache.hc.core5.http.EntityDetails; +import org.apache.hc.core5.http.Header; import org.apache.hc.core5.http.HttpRequest; import org.apache.hc.core5.http.HttpResponse; import org.apache.hc.core5.http.HttpStatus; import org.apache.hc.core5.http.Method; import org.apache.hc.core5.http.message.BasicHttpRequest; import org.apache.hc.core5.http.nio.AsyncPushProducer; +import org.apache.hc.core5.http.nio.CapacityChannel; +import org.apache.hc.core5.http.nio.DataStreamChannel; import org.apache.hc.core5.http.nio.ResponseChannel; import org.apache.hc.core5.http.protocol.HttpContext; import org.apache.hc.core5.http.protocol.HttpCoreContext; import org.apache.hc.core5.websocket.WebSocketConstants; +import org.apache.hc.core5.websocket.WebSocketExtension; +import org.apache.hc.core5.websocket.WebSocketExtensionData; +import org.apache.hc.core5.websocket.WebSocketExtensionFactory; import org.apache.hc.core5.websocket.WebSocketExtensionRegistry; import org.apache.hc.core5.websocket.WebSocketHandler; +import org.apache.hc.core5.websocket.WebSocketSession; import org.junit.jupiter.api.Test; class WebSocketH2ServerExchangeHandlerTest { @@ -69,6 +96,35 @@ HttpResponse getResponse() { } } + private static final class CountingExtensionFactory implements WebSocketExtensionFactory { + + private final AtomicInteger closed; + + CountingExtensionFactory(final AtomicInteger closed) { + this.closed = closed; + } + + @Override + public String getName() { + return "x-test"; + } + + @Override + public WebSocketExtension create(final WebSocketExtensionData request, final boolean server) { + return new WebSocketExtension() { + @Override + public String getName() { + return "x-test"; + } + + @Override + public void close() { + closed.incrementAndGet(); + } + }; + } + } + @Test void rejectsNonConnectMethod() throws Exception { final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler( @@ -115,4 +171,360 @@ void rejectsUnknownProtocol() throws Exception { assertNotNull(channel.getResponse()); assertEquals(HttpStatus.SC_BAD_REQUEST, channel.getResponse().getCode()); } + + @Test + void releasesExtensionsWhenOnOpenThrows() throws Exception { + final AtomicInteger closed = new AtomicInteger(); + final AtomicReference worker = new AtomicReference<>(); + final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry() + .register(new CountingExtensionFactory(closed)); + final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler( + new WebSocketHandler() { + @Override + public void onOpen(final WebSocketSession session) { + throw new IllegalStateException("onOpen failed"); + } + }, null, registry, worker::set); + + final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo"); + request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket"); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER, "x-test"); + handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create()); + + worker.get().run(); + + assertEquals(1, closed.get(), "onOpen failure must release negotiated extensions exactly once"); + } + + @Test + void releasesExtensionsWhenExecutorRejects() throws Exception { + final AtomicInteger closed = new AtomicInteger(); + final Executor rejecting = command -> { + throw new RejectedExecutionException("rejected"); + }; + final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry() + .register(new CountingExtensionFactory(closed)); + final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler( + new WebSocketHandler() { + }, null, registry, rejecting); + + final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo"); + request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket"); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER, "x-test"); + + assertThrows(RejectedExecutionException.class, () -> + handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create())); + + assertEquals(1, closed.get(), "a rejected executor must release negotiated extensions exactly once"); + } + + @Test + void releasesExtensionsOnNormalCompletion() throws Exception { + final AtomicInteger closed = new AtomicInteger(); + final AtomicReference worker = new AtomicReference<>(); + final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry() + .register(new CountingExtensionFactory(closed)); + final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler( + new WebSocketHandler() { + }, null, registry, worker::set); + + final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo"); + request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket"); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER, "x-test"); + handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create()); + + // A clean end-of-stream lets the processor read loop terminate normally. + handler.streamEnd(null); + worker.get().run(); + + assertEquals(1, closed.get(), "normal processor completion must release negotiated extensions exactly once"); + } + + @Test + void executorRejectionIsNotMaskedByExtensionCloseFailure() throws Exception { + final RuntimeException closeFailure = new IllegalStateException("close failed"); + final Executor rejecting = command -> { + throw new RejectedExecutionException("rejected"); + }; + final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry() + .register(throwingCloseFactory(closeFailure)); + final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler( + new WebSocketHandler() { + }, null, registry, rejecting); + + final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo"); + request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket"); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER, "x-test"); + + final RejectedExecutionException ex = assertThrows(RejectedExecutionException.class, () -> + handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create())); + assertEquals("rejected", ex.getMessage(), "the original rejection must be preserved"); + assertEquals(1, ex.getSuppressed().length, "the extension close failure must be suppressed, not masking"); + assertSame(closeFailure, ex.getSuppressed()[0]); + } + + @Test + void streamIsTornDownEvenWhenExtensionCloseThrows() throws Exception { + final RuntimeException closeFailure = new IllegalStateException("close failed"); + final AtomicReference worker = new AtomicReference<>(); + final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry() + .register(throwingCloseFactory(closeFailure)); + final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler( + new WebSocketHandler() { + }, null, registry, worker::set); + + final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo"); + request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket"); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER, "x-test"); + handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create()); + + handler.streamEnd(null); + // The worker's extension close() throws, but the stream teardown must still have run first. + assertThrows(RuntimeException.class, () -> worker.get().run()); + + final CollectingDataStreamChannel channel = new CollectingDataStreamChannel(); + while (handler.available() > 0) { + handler.produce(channel); + } + assertTrue(channel.endStreamCalled(), + "the HTTP/2 stream must be terminated with END_STREAM despite the extension close failure"); + } + + @Test + void protocolViolationClosesWith1002() throws Exception { + final AtomicReference worker = new AtomicReference<>(); + final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler( + new WebSocketHandler() { + }, null, WebSocketExtensionRegistry.createDefault(), worker::set); + + final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo"); + request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket"); + handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create()); + + // A fragmented (FIN=0) masked PING violates RFC 6455 section 5.5 and raises a + // checked WebSocketException, which must map to close code 1002, not 1011. + handler.consume(ByteBuffer.wrap(new byte[]{0x09, (byte) 0x80, 1, 2, 3, 4})); + worker.get().run(); + + final CollectingDataStreamChannel channel = new CollectingDataStreamChannel(); + while (handler.available() > 0) { + handler.produce(channel); + } + final byte[] out = channel.bytes(); + assertEquals((byte) 0x88, out[0], "expected a CLOSE frame"); + final int code = ((out[2] & 0xFF) << 8) | (out[3] & 0xFF); + assertEquals(1002, code, "protocol violation must close with 1002"); + } + + private static WebSocketExtensionFactory throwingCloseFactory(final RuntimeException closeFailure) { + return new WebSocketExtensionFactory() { + @Override + public String getName() { + return "x-test"; + } + + @Override + public WebSocketExtension create(final WebSocketExtensionData request, final boolean server) { + return new WebSocketExtension() { + @Override + public String getName() { + return "x-test"; + } + + @Override + public void close() { + throw closeFailure; + } + }; + } + }; + } + + private static final class CollectingDataStreamChannel implements DataStreamChannel { + private final ByteArrayOutputStream collected = new ByteArrayOutputStream(); + private boolean endStreamCalled; + + @Override + public void requestOutput() { + } + + @Override + public int write(final ByteBuffer src) { + final int n = src.remaining(); + final byte[] chunk = new byte[n]; + src.get(chunk); + collected.write(chunk, 0, n); + return n; + } + + @Override + public void endStream() { + endStreamCalled = true; + } + + @Override + public void endStream(final List trailers) { + endStreamCalled = true; + } + + byte[] bytes() { + return collected.toByteArray(); + } + + boolean endStreamCalled() { + return endStreamCalled; + } + } + + @Test + void writeAfterStreamFailureFailsInsteadOfBlocking() throws Exception { + final AtomicReference worker = new AtomicReference<>(); + final AtomicReference error = new AtomicReference<>(); + final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler( + new WebSocketHandler() { + @Override + public void onOpen(final WebSocketSession session) { + try { + session.sendText("hello"); + } catch (final Exception ex) { + throw new RuntimeException(ex); + } + } + + @Override + public void onError(final WebSocketSession session, final Exception cause) { + error.set(cause); + } + }, null, WebSocketExtensionRegistry.createDefault(), worker::set); + + final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo"); + request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket"); + handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create()); + + // The stream fails before the worker gets to run; a write must then surface an + // error through the session instead of blocking on the dead outbound queue. + handler.failed(new IOException("stream reset")); + worker.get().run(); + + assertNotNull(error.get(), "write after stream failure must fail, not block"); + } + + @Test + void largeOutboundWriteCompletesViaRequestOutputSignals() throws Exception { + // A tiny budget (8 bytes, 4-byte chunks) cannot hold the whole 66-byte frame, so the writer + // blocks mid-write. Draining is driven ONLY by requestOutput() signals here, never by + // spontaneous polling, so the write can only complete if requestOutput() fires per chunk. + final StringBuilder sb = new StringBuilder(); + for (int i = 0; i < 64; i++) { + sb.append((char) ('a' + (i % 26))); + } + final String text = sb.toString(); + + final CountDownLatch writeDone = new CountDownLatch(1); + final AtomicReference writeError = new AtomicReference<>(); + final AtomicReference worker = new AtomicReference<>(); + final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler( + new WebSocketHandler() { + @Override + public void onOpen(final WebSocketSession session) { + try { + session.sendText(text); + } catch (final Exception ex) { + writeError.set(ex); + } finally { + writeDone.countDown(); + } + } + }, null, WebSocketExtensionRegistry.createDefault(), worker::set, 8, 4); + + final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo"); + request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket"); + handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create()); + + final Semaphore outputReady = new Semaphore(0); + final AtomicInteger requestOutputCalls = new AtomicInteger(); + final AtomicBoolean ended = new AtomicBoolean(); + final ByteArrayOutputStream collected = new ByteArrayOutputStream(); + final DataStreamChannel channel = new DataStreamChannel() { + @Override + public void requestOutput() { + requestOutputCalls.incrementAndGet(); + outputReady.release(); + } + + @Override + public int write(final ByteBuffer src) { + final int n = src.remaining(); + final byte[] c = new byte[n]; + src.get(c); + collected.write(c, 0, n); + return n; + } + + @Override + public void endStream() { + ended.set(true); + outputReady.release(); + } + + @Override + public void endStream(final List trailers) { + endStream(); + } + }; + + final Thread reactor = new Thread(() -> { + try { + handler.produce(channel); // HttpCore drives the first produce() after the response + while (!ended.get() && outputReady.tryAcquire(5, TimeUnit.SECONDS)) { + handler.produce(channel); + } + } catch (final IOException | InterruptedException ex) { + throw new RuntimeException(ex); + } + }, "ws-reactor"); + reactor.setDaemon(true); + final Thread writerThread = new Thread(worker.get(), "ws-writer"); + writerThread.setDaemon(true); + + reactor.start(); + writerThread.start(); + + // Without a per-chunk requestOutput() the writer parks and the reactor is never re-signalled, + // so this await would time out; with the signal it completes. + assertTrue(writeDone.await(10, TimeUnit.SECONDS), + "a write larger than the budget must complete via requestOutput() signals, not deadlock"); + assertNull(writeError.get(), "the write must succeed"); + + handler.streamEnd(null); // end the read loop so the worker enqueues END_OUTBOUND + writerThread.join(TimeUnit.SECONDS.toMillis(5)); + reactor.join(TimeUnit.SECONDS.toMillis(5)); + + assertTrue(requestOutputCalls.get() > 0, "draining must be driven by requestOutput() signals"); + final byte[] out = collected.toByteArray(); + assertEquals(66, out.length, "the 64-byte text frame must be reassembled losslessly from 4-byte chunks"); + assertEquals((byte) 0x81, out[0], "FIN + text opcode"); + assertArrayEquals(text.getBytes(StandardCharsets.UTF_8), Arrays.copyOfRange(out, 2, out.length)); + } + + @Test + void advertisesBoundedInboundCapacity() throws Exception { + final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler( + new WebSocketHandler() { + }, null, WebSocketExtensionRegistry.createDefault()); + + final AtomicInteger granted = new AtomicInteger(); + final CapacityChannel channel = new CapacityChannel() { + @Override + public void update(final int increment) { + granted.addAndGet(increment); + } + }; + + handler.updateCapacity(channel); + handler.updateCapacity(channel); // a repeated query must not re-grant the initial window + + assertEquals(256 * 1024, granted.get(), + "inbound credit must be a bounded window, not Integer.MAX_VALUE"); + } } diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessorTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessorTest.java index 31ee63b14a..e921a21b29 100644 --- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessorTest.java +++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessorTest.java @@ -27,6 +27,7 @@ package org.apache.hc.core5.websocket.server; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.ByteArrayInputStream; @@ -39,6 +40,7 @@ import org.apache.hc.core5.websocket.WebSocketConfig; import org.apache.hc.core5.websocket.WebSocketHandler; import org.apache.hc.core5.websocket.WebSocketSession; +import org.apache.hc.core5.websocket.exceptions.WebSocketProtocolException; import org.junit.jupiter.api.Test; class WebSocketServerProcessorTest { @@ -106,6 +108,37 @@ void processesTextAndCloseFrames() throws Exception { assertTrue(out.size() > 0, "server should send close response"); } + @Test + void dataFrameDuringFragmentedMessage_isRejected() throws Exception { + final ByteArrayOutputStream frames = new ByteArrayOutputStream(); + // TEXT, not final -> starts a fragmented message + frames.write(maskedFrame(0x1, false, "a".getBytes(StandardCharsets.UTF_8))); + // a new TEXT frame while the fragmented message is still in progress (RFC 6455 section 5.4) + frames.write(maskedFrame(0x1, true, "b".getBytes(StandardCharsets.UTF_8))); + + final WebSocketSession session = new WebSocketSession( + WebSocketConfig.DEFAULT, + new ByteArrayInputStream(frames.toByteArray()), + new ByteArrayOutputStream(), + null, + null, + Collections.emptyList()); + final WebSocketServerProcessor processor = + new WebSocketServerProcessor(session, new TrackingHandler(), 1024); + + final WebSocketProtocolException ex = assertThrows(WebSocketProtocolException.class, processor::process); + assertEquals(1002, ex.closeCode); + assertTrue(ex.getMessage().contains("Data frame during fragmented message"), ex.getMessage()); + } + + private static byte[] maskedFrame(final int opcode, final boolean fin, final byte[] payload) { + final byte[] framed = maskedFrame(opcode, payload); + if (!fin) { + framed[0] &= 0x7F; // clear the FIN bit + } + return framed; + } + private static final class TrackingHandler implements WebSocketHandler { private String text; private int closeCode; diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandlerTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandlerTest.java index 817f093ad5..d1ceb6811f 100644 --- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandlerTest.java +++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandlerTest.java @@ -28,6 +28,8 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.IOException; @@ -36,9 +38,11 @@ import java.nio.ByteBuffer; import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Supplier; import org.apache.hc.core5.http.ClassicHttpResponse; +import org.apache.hc.core5.http.HeaderElements; import org.apache.hc.core5.http.HttpHeaders; import org.apache.hc.core5.http.HttpRequestMapper; import org.apache.hc.core5.http.HttpStatus; @@ -48,6 +52,9 @@ import org.apache.hc.core5.http.protocol.HttpContext; import org.apache.hc.core5.websocket.WebSocketConfig; import org.apache.hc.core5.websocket.WebSocketConstants; +import org.apache.hc.core5.websocket.WebSocketExtension; +import org.apache.hc.core5.websocket.WebSocketExtensionData; +import org.apache.hc.core5.websocket.WebSocketExtensionFactory; import org.apache.hc.core5.websocket.WebSocketExtensionRegistry; import org.apache.hc.core5.websocket.WebSocketHandler; import org.apache.hc.core5.websocket.WebSocketSession; @@ -120,6 +127,152 @@ public String selectSubprotocol(final List protocols) { connection.close(); } + @Test + void releasesExtensionsWhenOnOpenThrows() throws Exception { + final AtomicInteger closed = new AtomicInteger(); + final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry() + .register(new CountingExtensionFactory(closed)); + final Supplier supplier = () -> new WebSocketHandler() { + @Override + public void onOpen(final WebSocketSession session) { + throw new IllegalStateException("onOpen failed"); + } + }; + final HttpRequestMapper> mapper = (request, context) -> supplier; + final WebSocketServerRequestHandler handler = new WebSocketServerRequestHandler( + mapper, WebSocketConfig.DEFAULT, registry); + + final BasicClassicHttpRequest request = new BasicClassicHttpRequest("GET", "/ws"); + request.addHeader(HttpHeaders.CONNECTION, "Upgrade"); + request.addHeader(HttpHeaders.UPGRADE, "websocket"); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_VERSION, "13"); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_KEY, "dGhlIHNhbXBsZSBub25jZQ=="); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, "x-test"); + + final RecordingTrigger trigger = new RecordingTrigger(); + final HttpContext context = HttpCoreContext.create(); + final WebSocketServerConnection connection = createConnection(); + context.setAttribute(WebSocketContextKeys.CONNECTION, connection); + try { + handler.handle(request, trigger, context); + } catch (final IOException ignore) { + // A failed handshake may surface an I/O error while writing the close frame; this + // is independent of whether the negotiated extensions were released. + } + + assertEquals(1, closed.get(), "onOpen failure must release negotiated extensions exactly once"); + connection.close(); + } + + @Test + void releasesExtensionsWhenSubmitResponseFails() throws Exception { + final AtomicInteger closed = new AtomicInteger(); + final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry() + .register(new CountingExtensionFactory(closed)); + final HttpRequestMapper> mapper = + (request, context) -> () -> new WebSocketHandler() { + }; + final WebSocketServerRequestHandler handler = new WebSocketServerRequestHandler( + mapper, WebSocketConfig.DEFAULT, registry); + + final BasicClassicHttpRequest request = new BasicClassicHttpRequest("GET", "/ws"); + request.addHeader(HttpHeaders.CONNECTION, "Upgrade"); + request.addHeader(HttpHeaders.UPGRADE, "websocket"); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_VERSION, "13"); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_KEY, "dGhlIHNhbXBsZSBub25jZQ=="); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, "x-test"); + + final ResponseTrigger throwingTrigger = new ResponseTrigger() { + @Override + public void sendInformation(final ClassicHttpResponse response) { + } + + @Override + public void submitResponse(final ClassicHttpResponse response) throws IOException { + throw new IOException("submit failed"); + } + }; + final HttpContext context = HttpCoreContext.create(); + final WebSocketServerConnection connection = createConnection(); + context.setAttribute(WebSocketContextKeys.CONNECTION, connection); + + assertThrows(IOException.class, () -> handler.handle(request, throwingTrigger, context)); + + assertEquals(1, closed.get(), "a failed submitResponse must release negotiated extensions exactly once"); + connection.close(); + } + + @Test + void releasesExtensionsOnNormalCompletion() throws Exception { + final AtomicInteger closed = new AtomicInteger(); + final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry() + .register(new CountingExtensionFactory(closed)); + final HttpRequestMapper> mapper = + (request, context) -> () -> new WebSocketHandler() { + }; + final WebSocketServerRequestHandler handler = new WebSocketServerRequestHandler( + mapper, WebSocketConfig.DEFAULT, registry); + + final BasicClassicHttpRequest request = new BasicClassicHttpRequest("GET", "/ws"); + request.addHeader(HttpHeaders.CONNECTION, "Upgrade"); + request.addHeader(HttpHeaders.UPGRADE, "websocket"); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_VERSION, "13"); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_KEY, "dGhlIHNhbXBsZSBub25jZQ=="); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, "x-test"); + + final RecordingTrigger trigger = new RecordingTrigger(); + final HttpContext context = HttpCoreContext.create(); + // The peer end of the socket is already closed, so the processor reads EOF and the + // read loop terminates normally. + final WebSocketServerConnection connection = createConnection(); + context.setAttribute(WebSocketContextKeys.CONNECTION, connection); + + handler.handle(request, trigger, context); + + assertEquals(1, closed.get(), "normal processor completion must release negotiated extensions exactly once"); + connection.close(); + } + + @Test + void submitResponseFailureIsNotMaskedByExtensionCloseFailure() throws Exception { + final RuntimeException closeFailure = new IllegalStateException("close failed"); + final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry() + .register(throwingCloseFactory(closeFailure)); + final HttpRequestMapper> mapper = + (request, context) -> () -> new WebSocketHandler() { + }; + final WebSocketServerRequestHandler handler = new WebSocketServerRequestHandler( + mapper, WebSocketConfig.DEFAULT, registry); + + final BasicClassicHttpRequest request = new BasicClassicHttpRequest("GET", "/ws"); + request.addHeader(HttpHeaders.CONNECTION, "Upgrade"); + request.addHeader(HttpHeaders.UPGRADE, "websocket"); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_VERSION, "13"); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_KEY, "dGhlIHNhbXBsZSBub25jZQ=="); + request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, "x-test"); + + final ResponseTrigger throwingTrigger = new ResponseTrigger() { + @Override + public void sendInformation(final ClassicHttpResponse response) { + } + + @Override + public void submitResponse(final ClassicHttpResponse response) throws IOException { + throw new IOException("submit failed"); + } + }; + final HttpContext context = HttpCoreContext.create(); + final WebSocketServerConnection connection = createConnection(); + context.setAttribute(WebSocketContextKeys.CONNECTION, connection); + + final IOException ex = assertThrows(IOException.class, + () -> handler.handle(request, throwingTrigger, context)); + assertEquals("submit failed", ex.getMessage(), "the original failure must be preserved"); + assertEquals(1, ex.getSuppressed().length, "the extension close failure must be suppressed, not masking"); + assertSame(closeFailure, ex.getSuppressed()[0]); + connection.close(); + } + @Test void returnsUpgradeRequiredForNonUpgradeRequest() throws Exception { final WebSocketServerRequestHandler handler = new WebSocketServerRequestHandler( @@ -132,6 +285,16 @@ void returnsUpgradeRequiredForNonUpgradeRequest() throws Exception { handler.handle(request, trigger, HttpCoreContext.create()); assertEquals(HttpStatus.SC_UPGRADE_REQUIRED, trigger.response.getCode()); + // RFC 9110 section 15.5.22: a 426 response MUST carry an Upgrade header field, with the + // matching Connection: Upgrade option (section 7.8). + assertNotNull(trigger.response.getFirstHeader(HttpHeaders.UPGRADE)); + assertEquals("websocket", trigger.response.getFirstHeader(HttpHeaders.UPGRADE).getValue()); + assertNotNull(trigger.response.getFirstHeader(HttpHeaders.CONNECTION)); + assertTrue(trigger.response.getFirstHeader(HttpHeaders.CONNECTION).getValue() + .equalsIgnoreCase(HeaderElements.UPGRADE)); + // RFC 6455 section 4.4: the rejection must advertise the supported version. + assertNotNull(trigger.response.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_VERSION)); + assertEquals("13", trigger.response.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_VERSION).getValue()); } @Test @@ -162,6 +325,59 @@ private static WebSocketServerConnection createConnection() throws IOException { return factory.createConnection(socket); } + private static final class CountingExtensionFactory implements WebSocketExtensionFactory { + + private final AtomicInteger closed; + + CountingExtensionFactory(final AtomicInteger closed) { + this.closed = closed; + } + + @Override + public String getName() { + return "x-test"; + } + + @Override + public WebSocketExtension create(final WebSocketExtensionData request, final boolean server) { + return new WebSocketExtension() { + @Override + public String getName() { + return "x-test"; + } + + @Override + public void close() { + closed.incrementAndGet(); + } + }; + } + } + + private static WebSocketExtensionFactory throwingCloseFactory(final RuntimeException closeFailure) { + return new WebSocketExtensionFactory() { + @Override + public String getName() { + return "x-test"; + } + + @Override + public WebSocketExtension create(final WebSocketExtensionData request, final boolean server) { + return new WebSocketExtension() { + @Override + public String getName() { + return "x-test"; + } + + @Override + public void close() { + throw closeFailure; + } + }; + } + }; + } + private static final class RecordingTrigger implements ResponseTrigger { ClassicHttpResponse response;