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-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..1522970515 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 @@ -170,12 +170,13 @@ public boolean isOfferClientNoContextTakeover() { } /** - * Optional value for {@code client_max_window_bits} in the PMCE offer. + * Offered value for {@code client_max_window_bits}. * - *

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

+ *

The client never offers {@code client_max_window_bits}: the JDK Deflater cannot honor a + * server-selected window smaller than 15, which RFC 7692 section 7.2.1 would require. This + * always returns {@code null}; configuring a value is rejected when the configuration is built.

* - * @return offered {@code client_max_window_bits}, or {@code null} if not offered + * @return always {@code null} * @since 5.7 */ public Integer getOfferClientMaxWindowBits() { @@ -332,7 +333,7 @@ public static final class Builder { private boolean perMessageDeflateEnabled = true; private boolean offerServerNoContextTakeover = true; private boolean offerClientNoContextTakeover = true; - private Integer offerClientMaxWindowBits = 15; + private Integer offerClientMaxWindowBits = null; private Integer offerServerMaxWindowBits = null; private int maxFrameSize = 64 * 1024; @@ -410,12 +411,12 @@ public Builder offerClientNoContextTakeover(final boolean v) { } /** - * Offers {@code client_max_window_bits} in the PMCE offer. + * Configures {@code client_max_window_bits}. This parameter is not supported: the JDK + * Deflater cannot honor a server-selected window smaller than 15 (RFC 7692 section 7.2.1), + * so the client never offers it. Only {@code null} is accepted; any non-null value is + * rejected when {@link #build()} is called. * - *

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 + * @param v must be {@code null} * @return this builder * @since 5.7 */ @@ -570,8 +571,10 @@ public Builder enableHttp2(final boolean enabled) { * @since 5.7 */ public WebSocketClientConfig build() { - if (offerClientMaxWindowBits != null && (offerClientMaxWindowBits < 8 || offerClientMaxWindowBits > 15)) { - throw new IllegalArgumentException("offerClientMaxWindowBits must be in range [8..15]"); + if (offerClientMaxWindowBits != null) { + throw new IllegalArgumentException( + "client_max_window_bits cannot be offered: the JDK Deflater cannot honor a " + + "server-selected window smaller than 15 (RFC 7692 section 7.2.1)"); } if (offerServerMaxWindowBits != null && (offerServerMaxWindowBits < 8 || offerServerMaxWindowBits > 15)) { throw new IllegalArgumentException("offerServerMaxWindowBits must be in range [8..15]"); 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..c0c33e3812 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) { @@ -394,8 +415,6 @@ static ExtensionChain buildExtensionChain(final WebSocketClientConfig cfg, final } boolean pmceSeen = false, serverNoCtx = false, clientNoCtx = false; Integer clientBits = null, serverBits = null; - final boolean offerServerNoCtx = cfg.isOfferServerNoContextTakeover(); - final boolean offerClientNoCtx = cfg.isOfferClientNoContextTakeover(); final Integer offerClientBits = cfg.getOfferClientMaxWindowBits(); final Integer offerServerBits = cfg.getOfferServerMaxWindowBits(); @@ -414,19 +433,23 @@ static ExtensionChain buildExtensionChain(final WebSocketClientConfig cfg, final } pmceSeen = true; + // RFC 7692 section 7.1: a parameter name must not appear more than once within an + // accepted extension; a repeated name makes the response invalid. + final Set parameterNames = new HashSet<>(); for (int i = 1; i < parts.length; i++) { final String p = parts[i].trim(); final int eq = p.indexOf('='); + final String parameterName = (eq >= 0 ? p.substring(0, eq) : p).trim().toLowerCase(Locale.ROOT); + if (!parameterNames.add(parameterName)) { + throw new IllegalStateException("Duplicate permessage-deflate parameter: " + parameterName); + } if (eq < 0) { + // RFC 7692 sections 7.1.1.1 and 7.1.1.2 permit the server to include either + // no-context-takeover parameter in the response even when the offer did not; + // both are always safe to honour. if ("server_no_context_takeover".equalsIgnoreCase(p)) { - if (!offerServerNoCtx) { - throw new IllegalStateException("Server selected server_no_context_takeover not offered"); - } serverNoCtx = true; } else if ("client_no_context_takeover".equalsIgnoreCase(p)) { - if (!offerClientNoCtx) { - throw new IllegalStateException("Server selected client_no_context_takeover not offered"); - } clientNoCtx = true; } else { throw new IllegalStateException("Unsupported permessage-deflate parameter: " + p); @@ -453,9 +476,8 @@ static ExtensionChain buildExtensionChain(final WebSocketClientConfig cfg, final throw new IllegalStateException("Invalid client_max_window_bits: " + v, nfe); } } else if ("server_max_window_bits".equalsIgnoreCase(k)) { - if (offerServerBits == null) { - throw new IllegalStateException("Server selected server_max_window_bits not offered"); - } + // RFC 7692 section 7.1.2.1 permits the server to include this parameter + // uninvited; a server constraining its own window is always acceptable. try { if (v.isEmpty()) { throw new IllegalStateException("server_max_window_bits must have a value"); 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..6243173bae 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,24 @@ public int write(final ByteBuffer src) throws IOException { if (ch == null) { return 0; } - return ch.write(src); + try { + return ch.write(src); + } catch (final CancelledKeyException ignore) { + // The selection key was cancelled by a concurrent shutdown; the channel is gone. + return 0; + } } @Override public void requestOutput() { final DataStreamChannel ch = channel; if (ch != null) { - ch.requestOutput(); + try { + ch.requestOutput(); + } catch (final CancelledKeyException ignore) { + // requestOutput is a best-effort nudge; if the channel's selection key was + // cancelled by a concurrent shutdown there is nothing left to flush. + } } } 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..66d15596da 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,6 +700,12 @@ private void drainAndRelease() { dataQueuedBytes.addAndGet(-f.size); } } + if (encChain != null) { + encChain.close(); + } + if (decChain != null) { + decChain.close(); + } 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/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..78240e45e0 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,14 @@ 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 +60,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/WebSocketFrameReader.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameReader.java index fb14e941b7..34101aafae 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 @@ -101,6 +101,9 @@ WebSocketFrame readFrame() throws IOException { throw new WebSocketException("Negative frame payload length"); } } + if (len < 0) { + throw new WebSocketException("64-bit frame length must have the most significant bit set to 0"); + } if (len > Integer.MAX_VALUE) { throw new WebSocketException("Frame payload too large: " + len); } 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..b869b6ccea 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,17 @@ 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. + * + * @since 5.7 + */ + public void close() { + for (final WebSocketExtensionChain.Encoder e : encs) { + e.close(); + } + } } public static final class DecodeChain { @@ -143,5 +154,16 @@ public byte[] decode(final byte[] data, final long maxDecodedSize) throws Except } return out; } + + /** + * Releases native resources held by the decoders in this chain. + * + * @since 5.7 + */ + public void close() { + for (final WebSocketExtensionChain.Decoder d : decs) { + d.close(); + } + } } } \ 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..bb77a0804c 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,7 @@ 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.ReentrantLock; import org.apache.hc.core5.concurrent.DefaultThreadFactory; @@ -55,6 +56,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 +70,12 @@ 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; + + /** Maximum number of buffered outbound frames before the application writer is back-pressured. */ + private static final int OUTBOUND_QUEUE_CAPACITY = 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,7 +89,7 @@ final class WebSocketH2ServerExchangeHandler implements AsyncServerExchangeHandl private final Executor executor; private final BlockingQueue inbound = new LinkedBlockingQueue<>(); - private final BlockingQueue outbound = new LinkedBlockingQueue<>(); + private final BlockingQueue outbound = new LinkedBlockingQueue<>(OUTBOUND_QUEUE_CAPACITY); private final ReentrantLock outLock = new ReentrantLock(); private ByteBuffer currentOutbound; @@ -90,6 +98,8 @@ final class WebSocketH2ServerExchangeHandler implements AsyncServerExchangeHandl 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, @@ -136,61 +146,105 @@ public void handleRequest( WebSocketExtensions.parse(request.getFirstHeader(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); - try { - session.close(ex.closeCode, ex.getMessage()); - } catch (final IOException ignore) { - // ignore - } - } catch (final Exception ex) { - handler.onError(session, ex); + executor.execute(() -> { 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; + enqueueOutboundQuietly(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 @@ -296,6 +350,10 @@ public void produce(final DataStreamChannel channel) throws IOException { @Override public void failed(final Exception cause) { shutdown = true; + // Clear first so a worker blocked on a full outbound queue is released, then post the + // sentinels so the worker's blocking reads/writes unwind. + outbound.clear(); + inbound.clear(); outbound.offer(END_OUTBOUND); inbound.offer(END_INBOUND); @@ -310,6 +368,7 @@ public void releaseResources() { shutdown = true; outbound.clear(); inbound.clear(); + inbound.offer(END_INBOUND); outLock.lock(); try { currentOutbound = null; @@ -318,7 +377,31 @@ public void releaseResources() { } } - private static final class QueueInputStream extends InputStream { + private void enqueueOutbound(final ByteBuffer buf) throws IOException { + // Once the exchange has failed or been released nothing drains the queue any more, + // so a blocking put would wedge the worker thread; fail the write instead. + if (shutdown) { + throw new IOException("WebSocket stream already terminated"); + } + try { + // Bounded blocking put: applies backpressure to the application writer (worker thread) + // when the reactor has not yet drained the outbound queue. + outbound.put(buf); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while queuing outbound data", ex); + } + } + + private void enqueueOutboundQuietly(final ByteBuffer buf) { + try { + outbound.put(buf); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + } + } + + private final class QueueInputStream extends InputStream { private final BlockingQueue queue; private byte[] current; @@ -331,6 +414,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,15 +436,9 @@ 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})); + enqueueOutbound(ByteBuffer.wrap(new byte[]{(byte) b})); requestOutput(); } @@ -367,13 +449,13 @@ public void write(final byte[] b, final int off, final int len) throws IOExcepti } final byte[] copy = new byte[len]; System.arraycopy(b, off, copy, 0, len); - queue.offer(ByteBuffer.wrap(copy)); + enqueueOutbound(ByteBuffer.wrap(copy)); requestOutput(); } @Override - public void close() { - queue.offer(END_OUTBOUND); + public void close() throws IOException { + enqueueOutbound(END_OUTBOUND); requestOutput(); } 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..4a18e52c24 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( + // 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.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(); + 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/api/WebSocketClientConfigTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/api/WebSocketClientConfigTest.java index d4d94b2ff0..9e3f406eb1 100644 --- a/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/api/WebSocketClientConfigTest.java +++ b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/api/WebSocketClientConfigTest.java @@ -28,6 +28,8 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import org.apache.hc.core5.util.Timeout; @@ -54,4 +56,19 @@ void builderDefaultsAndCustom() { assertEquals(2048, cfg.getMaxMessageSize()); assertEquals(Timeout.ofSeconds(3), cfg.getConnectTimeout()); } + + @Test + void defaultConfigDoesNotOfferClientMaxWindowBits() { + // RFC 7692 section 7.2.1: the JDK Deflater cannot honor a reduced window, so the client + // never offers client_max_window_bits. + assertNull(WebSocketClientConfig.custom().build().getOfferClientMaxWindowBits()); + } + + @Test + void configuringClientMaxWindowBitsIsRejected() { + assertThrows(IllegalArgumentException.class, () -> + WebSocketClientConfig.custom().offerClientMaxWindowBits(15).build()); + assertThrows(IllegalArgumentException.class, () -> + WebSocketClientConfig.custom().offerClientMaxWindowBits(10).build()); + } } 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..81e0af81b4 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,104 @@ 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")); + // RFC 7692 section 7.1.2.2: client_max_window_bits may appear in a response only + // when the offer included it. 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 +163,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..3a7ba17db8 --- /dev/null +++ b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransportTest.java @@ -0,0 +1,85 @@ +/* + * ==================================================================== + * 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.assertEquals; + +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 that the transport tolerates a channel whose selection key has been cancelled by a + * concurrent shutdown, so a late CLOSE frame does not surface a {@link CancelledKeyException}. + */ +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 writeTreatsCancelledChannelAsGone() throws Exception { + final DataStreamChannelTransport transport = new DataStreamChannelTransport(); + transport.setChannel(new CancelledChannel()); + assertEquals(0, transport.write(ByteBuffer.allocate(4))); + } + + @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/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..e3c8ee60c0 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,13 @@ 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.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; @@ -102,4 +106,97 @@ 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 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/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/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..26248c50fe 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 @@ -28,20 +28,39 @@ 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.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; +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 +88,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 +163,262 @@ 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 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;