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..13c4f1bad4 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 @@ -158,7 +158,9 @@ public void completed(final WebSocketEndpointConnector.ProtoEndpoint endpoint) { ext.append("; client_no_context_takeover"); } if (cfg.getOfferClientMaxWindowBits() != null) { - ext.append("; client_max_window_bits=").append(cfg.getOfferClientMaxWindowBits()); + // The JDK Deflater always compresses with a 15-bit window, so a + // smaller configured value must not be promised to the server. + ext.append("; client_max_window_bits=15"); } 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,9 +415,10 @@ 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(); + // The offer always advertises client_max_window_bits=15 because the JDK Deflater + // cannot compress with a smaller window, so the response is validated against 15 + // regardless of the configured value. + final Integer offerClientBits = cfg.getOfferClientMaxWindowBits() != null ? Integer.valueOf(15) : null; final Integer offerServerBits = cfg.getOfferServerMaxWindowBits(); final String[] tokens = ext.split(","); @@ -418,15 +440,12 @@ static ExtensionChain buildExtensionChain(final WebSocketClientConfig cfg, final final String p = parts[i].trim(); final int eq = p.indexOf('='); 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 +472,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..efb7fca68e 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 @@ -211,8 +211,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 +322,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..5d33452306 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,17 +144,21 @@ 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) { + // 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). + if (fin && data.length >= 4) { encoded = ByteBuffer.wrap(data, 0, data.length - 4); } else { encoded = ByteBuffer.wrap(data); @@ -183,6 +187,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/WebSocketExtensionRegistry.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistry.java index 12fa259f3f..9d1ab04f6f 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,14 +54,22 @@ public WebSocketExtensionNegotiation negotiate( final boolean server) throws WebSocketException { final List extensions = new ArrayList<>(); final List responseData = new ArrayList<>(); + final Set accepted = new HashSet<>(); 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()); } } } 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 702cd74cce..4c2ff9c364 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameReader.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameReader.java @@ -98,6 +98,9 @@ WebSocketFrame readFrame() throws IOException { len = (len << 8) | (readByte() & 0xFF); } } + 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/WebSocketSession.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketSession.java index 4fe9e78f60..d783c6de11 100644 --- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketSession.java +++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketSession.java @@ -52,6 +52,7 @@ public final class WebSocketSession { private final SocketAddress localAddress; private final WebSocketFrameReader reader; private final WebSocketFrameWriter writer; + private final List extensions; private final ReentrantLock writeLock = new ReentrantLock(); private volatile boolean closeSent; @@ -68,11 +69,24 @@ public WebSocketSession( this.remoteAddress = remoteAddress; this.localAddress = localAddress; final List negotiated = extensions != null ? extensions : Collections.emptyList(); + this.extensions = negotiated; this.reader = new WebSocketFrameReader(this.config, this.inputStream, negotiated); this.writer = new WebSocketFrameWriter(this.outputStream, negotiated); this.closeSent = false; } + /** + * Releases native resources held by the negotiated extensions. Invoked once when the + * session processing loop terminates. + * + * @since 5.7 + */ + public void releaseExtensions() { + for (final WebSocketExtension extension : extensions) { + extension.close(); + } + } + public SocketAddress getRemoteAddress() { return remoteAddress; } 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..831e9b1ad7 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,12 +87,20 @@ 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, @@ -108,14 +116,17 @@ private byte[] doDeflate(final byte[] data, 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); + // 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 @@ -182,6 +193,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..0bcf533907 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, @@ -153,7 +163,7 @@ public void handleRequest( responseSent = true; final InputStream inputStream = new QueueInputStream(inbound); - final OutputStream outputStream = new QueueOutputStream(outbound); + final OutputStream outputStream = new QueueOutputStream(); final WebSocketSession session = new WebSocketSession( config, inputStream, outputStream, null, null, negotiation.getExtensions()); @@ -168,6 +178,13 @@ public void handleRequest( } 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 { @@ -177,7 +194,7 @@ public void handleRequest( } } finally { shutdown = true; - outbound.offer(END_OUTBOUND); + enqueueOutboundQuietly(END_OUTBOUND); inbound.offer(END_INBOUND); final DataStreamChannel channel = dataChannel; @@ -190,7 +207,23 @@ public void handleRequest( @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 +329,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 +347,7 @@ public void releaseResources() { shutdown = true; outbound.clear(); inbound.clear(); + inbound.offer(END_INBOUND); outLock.lock(); try { currentOutbound = null; @@ -318,7 +356,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 +393,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 +415,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 +428,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..f0c5d7cdfc 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,14 @@ class WebSocketServerProcessor { } void process() throws IOException { + try { + readLoop(); + } finally { + session.releaseExtensions(); + } + } + + private void readLoop() throws IOException { ByteBuffer continuationBuf = null; WebSocketFrameType continuationType = null; while (true) { @@ -85,6 +93,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..c278150af0 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,10 @@ 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 6455 section 4.4 requires the rejection to advertise the supported version. + rejection.addHeader(WebSocketConstants.SEC_WEBSOCKET_VERSION, "13"); + trigger.submitResponse(rejection); return; } final WebSocketHandler handler = supplier.get(); 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..873f31876e 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 @@ -31,6 +31,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows; 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,19 +48,72 @@ 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_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 pmce_clientWindowBitsValidatedAgainstAdvertised15() { + // The offer always advertises client_max_window_bits=15 because the JDK Deflater + // cannot compress with a smaller window, so a response of 15 must be accepted even + // when the configured offer value was smaller. + final WebSocketClientConfig cfg = WebSocketClientConfig.custom() + .offerClientMaxWindowBits(10) + .build(); + final ExtensionChain chain = Http1UpgradeProtocol.buildExtensionChain(cfg, + "permessage-deflate; client_max_window_bits=15"); + 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, + Http1UpgradeProtocol.headerValue(response, "Sec-WebSocket-Extensions"))); + } + @Test void pmce_rejectedOnUnknownOrDuplicate() { final WebSocketClientConfig cfg = WebSocketClientConfig.custom().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..71eedb6299 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,43 @@ 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 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/WebSocketExtensionRegistryTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistryTest.java index f3b6b7418b..7cd5724670 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 @@ -30,6 +30,7 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Collections; import org.junit.jupiter.api.Test; @@ -102,4 +103,18 @@ 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"); + } } diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketFrameReaderTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketFrameReaderTest.java index c70921bbec..1cb04aead4 100644 --- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketFrameReaderTest.java +++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketFrameReaderTest.java @@ -160,4 +160,16 @@ void frame_too_large_throws() { final ByteBuffer f = maskedFrame(0x82, new byte[len]); assertThrows(WebSocketException.class, () -> readFrame(f, 1024)); } + + @Test + void negative_64bit_length_is_rejected() { + // FIN|BINARY, MASK|127, then a 64-bit length with the most significant bit set + // (RFC 6455 section 5.2 requires it to be 0). Rejected before any allocation. + final ByteBuffer f = ByteBuffer.allocate(2 + 8); + f.put((byte) 0x82); + f.put((byte) (0x80 | 127)); + f.putLong(Long.MIN_VALUE); + f.flip(); + assertThrows(WebSocketException.class, () -> readFrame(f, Integer.MAX_VALUE)); + } } 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..2626c11eb9 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 @@ -134,6 +134,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..6172d536a0 --- /dev/null +++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/PerMessageDeflateFragmentedRoundTripTest.java @@ -0,0 +1,102 @@ +/* + * ==================================================================== + * 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 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..1e78042edd 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 @@ -29,19 +29,30 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.List; +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.WebSocketExtensionRegistry; import org.apache.hc.core5.websocket.WebSocketHandler; +import org.apache.hc.core5.websocket.WebSocketSession; import org.junit.jupiter.api.Test; class WebSocketH2ServerExchangeHandlerTest { @@ -115,4 +126,113 @@ void rejectsUnknownProtocol() throws Exception { assertNotNull(channel.getResponse()); assertEquals(HttpStatus.SC_BAD_REQUEST, channel.getResponse().getCode()); } + + @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 final class CollectingDataStreamChannel implements DataStreamChannel { + private final ByteArrayOutputStream collected = new ByteArrayOutputStream(); + + @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() { + } + + @Override + public void endStream(final List trailers) { + } + + byte[] bytes() { + return collected.toByteArray(); + } + } + + @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..d88346fa9b 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 @@ -132,6 +132,9 @@ void returnsUpgradeRequiredForNonUpgradeRequest() throws Exception { handler.handle(request, trigger, HttpCoreContext.create()); assertEquals(HttpStatus.SC_UPGRADE_REQUIRED, trigger.response.getCode()); + // 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