From 39c248b4b7dd3f1946efe5707d537e56cda4b72c Mon Sep 17 00:00:00 2001 From: Jarno Rajahalme Date: Tue, 25 Aug 2026 11:44:24 +0200 Subject: [PATCH 1/9] tests: leave buffer headroom for framing in WebSocket flush tests The tests currently send a payload equal to the configured buffer limit. WebSocket framing makes the encoded data exceed that limit, potentially disabling reads before the FIN is observed and making the test timing dependent. Double the buffer limits while retaining the 50 MiB payload. The payload remains larger than the kernel socket buffer, so the tests still exercise upstream flushing without crossing the Envoy buffer limit solely because of WebSocket framing overhead. Signed-off-by: Jarno Rajahalme --- tests/cilium_websocket_encap_integration_test.cc | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tests/cilium_websocket_encap_integration_test.cc b/tests/cilium_websocket_encap_integration_test.cc index eb2fc758d..90e889876 100644 --- a/tests/cilium_websocket_encap_integration_test.cc +++ b/tests/cilium_websocket_encap_integration_test.cc @@ -472,10 +472,10 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamFlush) { // Test that an upstream flush works correctly (all data is flushed) TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlush) { - // Use a very large size to make sure it is larger than the kernel socket read - // buffer. + // Keep the payload larger than the kernel socket read buffer so that an upstream flush is needed, + // while leaving buffer-limit headroom for WebSocket framing and control traffic. const uint32_t size = 50 * 1024 * 1024; - config_helper_.setBufferLimits(size, size); + config_helper_.setBufferLimits(2 * size, 2 * size); initialize(); std::string data(size, 'a'); @@ -519,10 +519,10 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlush) { // Test that Envoy doesn't crash or assert when shutting down with an upstream // flush active TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlushEnvoyExit) { - // Use a very large size to make sure it is larger than the kernel socket read - // buffer. + // Keep the payload larger than the kernel socket read buffer so that an upstream flush is needed, + // while leaving buffer-limit headroom for WebSocket framing and control traffic. const uint32_t size = 50 * 1024 * 1024; - config_helper_.setBufferLimits(size, size); + config_helper_.setBufferLimits(2 * size, 2 * size); initialize(); std::string data(size, 'a'); From dc27a719b20dfe0c6b68d4a78c2960369002fc49 Mon Sep 17 00:00:00 2001 From: Jarno Rajahalme Date: Mon, 24 Aug 2026 20:53:41 +0200 Subject: [PATCH 2/9] tests: use WebSocket CLOSE for orderly shutdown Update the WebSocket encapsulation tests to perform an orderly CLOSE exchange instead of relying on the outer TCP transport FIN to delimit the tunneled stream. Keep the TCP client open after sending its write-side FIN, wait for the resulting WebSocket CLOSE, and have the peer send its remaining data followed by its own CLOSE. This more accurately exercises TCP half-close semantics and verifies that reverse-direction data can still be delivered after the first direction has ended. Also fix the downstream flush test's 64-bit WebSocket payload header. The old header provided only four of the required eight extended-length bytes. The decoder consequently consumed the beginning of the payload as part of the frame header, but the test happened to pass because the outer transport FIN terminated processing. Encode the full 10-byte frame header so that the following CLOSE is parsed at the correct frame boundary. These changes make the tests use valid WebSocket framing and explicit CLOSE messages for logical stream termination, without requiring any codec behavior changes. Signed-off-by: Jarno Rajahalme --- ...cilium_websocket_encap_integration_test.cc | 25 +++++++++++++------ 1 file changed, 17 insertions(+), 8 deletions(-) diff --git a/tests/cilium_websocket_encap_integration_test.cc b/tests/cilium_websocket_encap_integration_test.cc index 90e889876..9643bc58e 100644 --- a/tests/cilium_websocket_encap_integration_test.cc +++ b/tests/cilium_websocket_encap_integration_test.cc @@ -350,8 +350,9 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamDisconnect) { ASSERT_TRUE(frame_offset > 0); ASSERT_EQ(frame_offset, 6); + ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x82\x4last\x88\0", 8}, true)); + tcp_client->waitForData("worldlast"); ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); - ASSERT_TRUE(fake_upstream_connection->write("", true)); ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); tcp_client->waitForDisconnect(); } @@ -396,9 +397,13 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketLargeWrite) { ASSERT_TRUE(fake_upstream_connection->write("\x82\x7e\x80\x00"s)); ASSERT_TRUE(fake_upstream_connection->write(data)); tcp_client->waitForData(data); - tcp_client->close(); + ASSERT_TRUE(tcp_client->write("", true)); + ASSERT_TRUE(fake_upstream_connection->waitForData( + expected_handshake.length() + 2 * 8 + data.size() + 6, &received_data)); + ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x88\0", 2})); + tcp_client->waitForHalfClose(); ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); - ASSERT_TRUE(fake_upstream_connection->close()); + ASSERT_TRUE(fake_upstream_connection->write("", true)); ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); uint32_t upstream_pauses = @@ -443,14 +448,16 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamFlush) { tcp_client->readDisable(true); ASSERT_TRUE(tcp_client->write("", true)); - // This ensures that readDisable(true) has been run on it's thread - // before tcp_client starts writing. - ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); + // Waiting for the encoded CLOSE ensures that readDisable(true) has run on its thread before the + // peer starts writing. + ASSERT_TRUE( + fake_upstream_connection->waitForData(expected_handshake.length() + 6, &received_data)); // writing data in one large chunk - ASSERT_TRUE(fake_upstream_connection->write("\x82\x7f\x03\x20\0\0"s)); - ASSERT_TRUE(fake_upstream_connection->write(data, true)); + ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x82\x7f\0\0\0\0\x03\x20\0\0", 10})); + ASSERT_TRUE(fake_upstream_connection->write(data)); + ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x88\0", 2})); test_server_->waitForCounterGe("cluster.cluster1.upstream_flow_control_paused_reading_total", 1); EXPECT_EQ(test_server_->counter("cluster.cluster1.upstream_flow_control_resumed_reading_total") @@ -460,6 +467,8 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamFlush) { tcp_client->waitForData(data); tcp_client->waitForHalfClose(); ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); + ASSERT_TRUE(fake_upstream_connection->write("", true)); + ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); uint32_t upstream_pauses = test_server_->counter("cluster.cluster1.upstream_flow_control_paused_reading_total")->value(); From 342e364a618c565b83053424513992ca6223267a Mon Sep 17 00:00:00 2001 From: Jarno Rajahalme Date: Mon, 24 Aug 2026 20:41:44 +0200 Subject: [PATCH 3/9] websocket: defer transport FIN to server after both tunnel directions end A WebSocket CLOSE represents a directional FIN for the tunneled TCP connection. The codec previously forwarded CLOSE with end_stream=true, coupling the tunneled FIN to a FIN on the underlying WebSocket transport. When both tunnel directions ended close together, final data, CLOSE, and transport FIN could be delivered together. Transport teardown could then race flushing the final frames. Sanitizer instrumentation made this timing race visible in CI as intermittent loss of the last tunneled data. Keep the WebSocket transport open after the first CLOSE. Only the server codec explicitly sends transport end-stream, and only after both the encoder and decoder have ended. This follows the closing asymmetry recommended by RFC 6455 section 7.1.1 and allows client-generated final frames to enter the transport without an attached FIN, so normal transport flushing can complete before shutdown. Handle either CLOSE ordering and guard against sending transport end-stream twice when propagating a tunneled FIN synchronously invokes the reverse codec path. Buffered client data released after the handshake likewise does not end the transport. Extend the integration tests to verify that the first CLOSE leaves the reverse direction usable and that final reverse-direction data is delivered before the WebSocket transport closes. Signed-off-by: Jarno Rajahalme --- cilium/websocket_codec.cc | 33 +++++++++++++++---- cilium/websocket_codec.h | 1 + ...cilium_websocket_encap_integration_test.cc | 6 ++++ ...ilium_websocket_policy_integration_test.cc | 3 +- 4 files changed, 35 insertions(+), 8 deletions(-) diff --git a/cilium/websocket_codec.cc b/cilium/websocket_codec.cc index b7106b2e2..461bc32f7 100644 --- a/cilium/websocket_codec.cc +++ b/cilium/websocket_codec.cc @@ -381,7 +381,14 @@ void Codec::encode(Buffer::Instance& data, bool end_stream) { if (encoder_.hasData()) { resetPingTimer(); } - parent_->injectEncoded(encoder_.data(), encoder_.endStream()); + // A WebSocket CLOSE represents the FIN for this direction of the tunneled TCP connection. + // RFC 6455 section 7.1.1 recommends that the server close the underlying transport after both + // sides have sent CLOSE. Keep the client transport open so that its final frames are flushed + // before the server closes the connection. + const bool transport_end_stream = + !config()->client_ && encoder_.endStream() && decoder_.endStream(); + encoded_end_stream_sent_ |= transport_end_stream; + parent_->injectEncoded(encoder_.data(), transport_end_stream); } } @@ -586,9 +593,9 @@ void Codec::decode(Buffer::Instance& data, bool end_stream) { return closeOnError(handshake_buffer_, "Invalid WebSocket response"); } - // Kick write on the other direction - parent_->injectEncoded(encoder_.data(), encoder_.endStream()); - + // Kick the buffered write. This is the client codec, so the server remains responsible for + // ending the underlying WebSocket transport. + parent_->injectEncoded(encoder_.data(), false); } else { // Server needs to wait for a valid handshake request before accepting any data RequestParser parser; @@ -659,7 +666,19 @@ void Codec::decode(Buffer::Instance& data, bool end_stream) { resetPingTimer(); } - parent_->injectDecoded(decoder_.data(), decoder_.endStream()); + // Record decoder end_stream before injecting the corresponding TCP FIN (via + // end_stream=true). That injection may synchronously report the reverse-direction FIN through + // encode(..., end_stream=true). + const bool decoded_end_stream = decoder_.endStream(); + parent_->injectDecoded(decoder_.data(), decoded_end_stream); + + // Complete the CLOSE handshake by ending the underlying transport from the server side. RFC 6455 + // section 7.1.1 recommends that the client wait for the server to close the connection. + if (!config->client_ && decoded_end_stream && encoder_.endStream() && !encoded_end_stream_sent_) { + Buffer::OwnedImpl empty; + encoded_end_stream_sent_ = true; + parent_->injectEncoded(empty, true); + } } bool Codec::ping(const void* payload, size_t len) { @@ -669,7 +688,7 @@ bool Codec::ping(const void* payload, size_t len) { Buffer::OwnedImpl buf(payload, len); encoder_.encode(buf, false, OPCODE_PING); parent_->config()->stats_.ping_sent_count_.inc(); - parent_->injectEncoded(encoder_.data(), encoder_.endStream()); + parent_->injectEncoded(encoder_.data(), false); return true; } @@ -679,7 +698,7 @@ bool Codec::pong(const void* payload, size_t len) { } Buffer::OwnedImpl buf(payload, len); encoder_.encode(buf, false, OPCODE_PONG); - parent_->injectEncoded(encoder_.data(), encoder_.endStream()); + parent_->injectEncoded(encoder_.data(), false); return true; } diff --git a/cilium/websocket_codec.h b/cilium/websocket_codec.h index 817940d5f..60e0002dc 100644 --- a/cilium/websocket_codec.h +++ b/cilium/websocket_codec.h @@ -126,6 +126,7 @@ class Codec : Logger::Loggable { Event::TimerPtr handshake_timer_{nullptr}; Buffer::OwnedImpl handshake_buffer_; bool accepted_{false}; + bool encoded_end_stream_sent_{false}; }; using CodecPtr = std::unique_ptr; diff --git a/tests/cilium_websocket_encap_integration_test.cc b/tests/cilium_websocket_encap_integration_test.cc index 9643bc58e..cb421e983 100644 --- a/tests/cilium_websocket_encap_integration_test.cc +++ b/tests/cilium_websocket_encap_integration_test.cc @@ -350,6 +350,12 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamDisconnect) { ASSERT_TRUE(frame_offset > 0); ASSERT_EQ(frame_offset, 6); + // The locally generated CLOSE represents only the tunneled downstream FIN. It must not also send + // FIN on the outer WebSocket transport while the peer's tunneled direction remains open. + EXPECT_FALSE(fake_upstream_connection->waitForHalfClose(std::chrono::milliseconds(100))); + + // The fake WebSocket server sends its final data and CLOSE, then ends the transport as + // recommended by RFC 6455 section 7.1.1. ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x82\x4last\x88\0", 8}, true)); tcp_client->waitForData("worldlast"); ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); diff --git a/tests/cilium_websocket_policy_integration_test.cc b/tests/cilium_websocket_policy_integration_test.cc index 17b682c5a..84d120b7f 100644 --- a/tests/cilium_websocket_policy_integration_test.cc +++ b/tests/cilium_websocket_policy_integration_test.cc @@ -326,7 +326,8 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamDisconnect) { ASSERT_TRUE(fake_upstream_connection->waitForData(10, &received)); ASSERT_EQ(received, "hellohello"); ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); - ASSERT_TRUE(fake_upstream_connection->write("", true)); + ASSERT_TRUE(fake_upstream_connection->write("upstream final", true)); + tcp_client->waitForData("worldupstream final"); ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); tcp_client->waitForDisconnect(); } From 77fef7bcae0ff616d7178561defdd8489468e010 Mon Sep 17 00:00:00 2001 From: Jarno Rajahalme Date: Mon, 24 Aug 2026 11:28:37 +0200 Subject: [PATCH 4/9] envoy: allow write filters to consume data after end stream A protocol filter may expose an end stream for its decoded data while the underlying protocol connection remains active for control traffic. Envoy currently rejects writes after ConnectionImpl::write_end_stream_ is set before invoking the write filter chain. This prevents the WebSocket codec from receiving upstream WebSocket control frames after it has translated a CLOSE into a FIN on the decoded downstream TCP connection. Carry an Envoy patch that gives write filters an opportunity to consume non-empty data after end stream. Any data left unconsumed is still rejected and cannot be written to the socket after FIN, preserving the underlying TCP invariant. This is the write-side counterpart to the receive_before_connect support we previously contributed upstream in envoyproxy/envoy#38189. That change allows protocol filters to receive downstream data before TCP proxy has established its upstream connection. This patch allows a protocol filter to continue receiving upstream protocol traffic after its decoded downstream stream has ended, and consequently to send control responses such as PONG back upstream. Together they decouple the lifetime of a protocol filter's control plane from the lifetime of the proxied TCP data stream. Signed-off-by: Jarno Rajahalme --- WORKSPACE | 1 + ...rite-filters-to-consume-data-after-e.patch | 65 +++++++++++++++++++ 2 files changed, 66 insertions(+) create mode 100644 patches/0009-network-Allow-write-filters-to-consume-data-after-e.patch diff --git a/WORKSPACE b/WORKSPACE index 6efad5d57..05f984c41 100644 --- a/WORKSPACE +++ b/WORKSPACE @@ -45,6 +45,7 @@ git_repository( "@//patches:0006-test-integration-Defer-fake-upstream-read-enable-un.patch", "@//patches:0007-config-add-grpc-mux-stream-event-callback.patch", "@//patches:0008-repo-Make-yq-dependency-optional-for-CI-config-parsi.patch", + "@//patches:0009-network-Allow-write-filters-to-consume-data-after-e.patch", ], # // clang-format off: Envoy's format check: Only repository_locations.bzl may contains URL references remote = "https://github.com/envoyproxy/envoy.git", diff --git a/patches/0009-network-Allow-write-filters-to-consume-data-after-e.patch b/patches/0009-network-Allow-write-filters-to-consume-data-after-e.patch new file mode 100644 index 000000000..81f1e55e6 --- /dev/null +++ b/patches/0009-network-Allow-write-filters-to-consume-data-after-e.patch @@ -0,0 +1,65 @@ +From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001 +From: Cilium Authors +Date: Mon, 24 Aug 2026 08:00:00 +0200 +Subject: [PATCH] network: Allow write filters to consume data after end stream + +A protocol filter may synthesize an end stream for the data it exposes while +the underlying transport remains open for protocol control frames. Give write +filters an opportunity to consume such frames after the connection write side +has been half-closed. + +The connection still rejects any data that a filter does not consume, so this +does not permit application data to be written to the transport after FIN. +--- + source/common/network/connection_impl.cc | 22 ++++++++++++---------- + 1 file changed, 12 insertions(+), 10 deletions(-) + +diff --git a/source/common/network/connection_impl.cc b/source/common/network/connection_impl.cc +index 21a263323b..042f12d65d 100644 +--- a/source/common/network/connection_impl.cc ++++ b/source/common/network/connection_impl.cc +@@ -596,30 +596,32 @@ void ConnectionImpl::write(Buffer::Instance& data, bool end_stream) { + void ConnectionImpl::write(Buffer::Instance& data, bool end_stream, bool through_filter_chain) { + ASSERT(!end_stream || enable_half_close_); + ASSERT(dispatcher_.isThreadSafe()); + +- if (write_end_stream_) { +- // It is an API violation to write more data after writing end_stream, but a duplicate +- // end_stream with no data is harmless. This catches misuse of the API that could result in data +- // being lost. +- ASSERT(data.length() == 0 && end_stream); +- +- return; +- } +- +- if (through_filter_chain) { ++ // A filter may synthesize end_stream for its decoded data while its underlying protocol remains ++ // open for control frames. In that case, give write filters a chance to consume subsequent data. ++ // Any data left unconsumed is still rejected below and cannot reach the transport after FIN. ++ if (through_filter_chain && (!write_end_stream_ || data.length() > 0)) { + // NOTE: This is kind of a hack, but currently we don't support restart/continue on the write + // path, so we just pass around the buffer passed to us in this function. If we ever + // support buffer/restart/continue on the write path this needs to get more complicated. + current_write_buffer_ = &data; + current_write_end_stream_ = end_stream; + FilterStatus status = filter_manager_.onWrite(); + current_write_buffer_ = nullptr; + + if (FilterStatus::StopIteration == status) { + return; + } + } + ++ if (write_end_stream_) { ++ // Any data left after the filters ran is an API violation and cannot be written after FIN. An ++ // empty write is harmless regardless of whether it repeats end_stream. ++ ASSERT(data.length() == 0); ++ ++ return; ++ } ++ + write_end_stream_ = end_stream; + if (data.length() > 0 || end_stream) { + ENVOY_CONN_LOG(trace, "writing {} bytes, end_stream {}", *this, data.length(), end_stream); +-- +2.43.0 From 97a73b56a7a8c8b54aaef27633310e73d4b8858b Mon Sep 17 00:00:00 2001 From: Jarno Rajahalme Date: Tue, 25 Aug 2026 20:15:31 +0200 Subject: [PATCH 5/9] envoy: Expose close delay to TCPProxy filters Normally TCPProxy upstream is closed without delay when the downstream connection gets closed. This can lead to droppoing data already received from the downstream TCP connection, but not yet sent to the upstream connection (e.g., being buffered at the filter chain). Similarly, an abrupt closure without flushing the data on the upstream socket receive buffer, or any receive direction data that happens to be in-flight, can cause a TCP RST. Add an Envoy patch to exposes the existing delayed connection close functionality to TCPProxy filter chains. This allows urderly delivery of any buffered data to be sent to the upstream socket, and allows protocols implemented in network filters to continue processing after downstream closes, for example, to perform a graceful shutdown. Signed-off-by: Jarno Rajahalme --- WORKSPACE | 1 + ...w-filters-to-wait-for-upstream-close.patch | 260 ++++++++++++++++++ 2 files changed, 261 insertions(+) create mode 100644 patches/0010-tcp_proxy-Allow-filters-to-wait-for-upstream-close.patch diff --git a/WORKSPACE b/WORKSPACE index 05f984c41..153ac6c5e 100644 --- a/WORKSPACE +++ b/WORKSPACE @@ -46,6 +46,7 @@ git_repository( "@//patches:0007-config-add-grpc-mux-stream-event-callback.patch", "@//patches:0008-repo-Make-yq-dependency-optional-for-CI-config-parsi.patch", "@//patches:0009-network-Allow-write-filters-to-consume-data-after-e.patch", + "@//patches:0010-tcp_proxy-Allow-filters-to-wait-for-upstream-close.patch", ], # // clang-format off: Envoy's format check: Only repository_locations.bzl may contains URL references remote = "https://github.com/envoyproxy/envoy.git", diff --git a/patches/0010-tcp_proxy-Allow-filters-to-wait-for-upstream-close.patch b/patches/0010-tcp_proxy-Allow-filters-to-wait-for-upstream-close.patch new file mode 100644 index 000000000..e6bf1f8a9 --- /dev/null +++ b/patches/0010-tcp_proxy-Allow-filters-to-wait-for-upstream-close.patch @@ -0,0 +1,260 @@ +From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001 +From: Cilium Authors +Date: Tue, 25 Aug 2026 16:00:00 +0200 +Subject: [PATCH] tcp_proxy: Allow filters to wait for upstream close + +A protocol filter may need the upstream transport to remain open after its +downstream connection has ended so the peer can finish a protocol-level close +handshake. Add a per-connection filter-state timeout which makes TCP proxy +flush its pending writes and wait for the upstream peer to close. + +Data received during this final drain interval cannot be delivered to the +closed downstream connection and is discarded. Preserve the existing reset +behavior for all connections which do not opt in. +--- + envoy/tcp/upstream.h | 7 ++++++- + source/common/tcp_proxy/tcp_proxy.cc | 31 +++++++++++++++++++++++++------ + source/common/tcp_proxy/tcp_proxy.h | 17 +++++++++++++++-- + source/common/tcp_proxy/upstream.cc | 17 +++++++++++++---- + source/common/tcp_proxy/upstream.h | 12 +++++++++--- + 5 files changed, 68 insertions(+), 16 deletions(-) + +diff --git a/envoy/tcp/upstream.h b/envoy/tcp/upstream.h +index 9a96694e41..2f79b3bca5 100644 +--- a/envoy/tcp/upstream.h ++++ b/envoy/tcp/upstream.h +@@ -1,5 +1,7 @@ + #pragma once + ++#include ++ + #include "envoy/buffer/buffer.h" + #include "envoy/extensions/filters/network/tcp_proxy/v3/tcp_proxy.pb.h" + #include "envoy/http/filter.h" +@@ -157,11 +159,14 @@ public: + * Called when an event is received on the downstream connection + * @param event supplies the event which occurred. + * @param details supplies the details of the event, e.g. the local close reason. ++ * @param flush_wait_timeout supplies how long a TCP upstream should wait for its peer to close ++ * after pending writes have been flushed. Zero preserves the normal immediate close. + * @return the underlying ConnectionData if the event is not "Connected" and draining + is supported for this upstream. + */ + virtual Tcp::ConnectionPool::ConnectionData* +- onDownstreamEvent(Network::ConnectionEvent event, absl::string_view details = "") PURE; ++ onDownstreamEvent(Network::ConnectionEvent event, absl::string_view details = "", ++ std::chrono::milliseconds flush_wait_timeout = {}) PURE; + + /* Called to convert underlying transport socket from non-secure mode + * to secure mode. Implemented only by start_tls transport socket. +diff --git a/source/common/tcp_proxy/tcp_proxy.cc b/source/common/tcp_proxy/tcp_proxy.cc +index e5d23f2199..405ef9c1bc 100644 +--- a/source/common/tcp_proxy/tcp_proxy.cc ++++ b/source/common/tcp_proxy/tcp_proxy.cc +@@ -457,6 +457,16 @@ void Filter::initialize(Network::ReadFilterCallbacks& callbacks, bool set_connect + } + } + ++ const StreamInfo::UInt64Accessor* upstream_flush_wait_timeout = ++ read_callbacks_->connection() ++ .streamInfo() ++ .filterState() ++ ->getDataReadOnly(UpstreamFlushWaitTimeoutMs); ++ if (upstream_flush_wait_timeout != nullptr) { ++ upstream_flush_wait_timeout_ = ++ std::chrono::milliseconds(upstream_flush_wait_timeout->value()); ++ } ++ + // Handle TLS handshake wait mode. + if (connect_mode_ == UpstreamConnectMode::ON_DOWNSTREAM_TLS_HANDSHAKE) { + const auto ssl_connection = read_callbacks_->connection().ssl(); +@@ -1204,12 +1214,14 @@ void Filter::onDownstreamEvent(Network::ConnectionEvent event) { + if (upstream_) { + absl::string_view downstream_close_details = read_callbacks_->connection().localCloseReason(); + Tcp::ConnectionPool::ConnectionDataPtr conn_data( +- upstream_->onDownstreamEvent(event, downstream_close_details)); ++ upstream_->onDownstreamEvent(event, downstream_close_details, ++ upstream_flush_wait_timeout_)); + if (conn_data != nullptr && + conn_data->connection().state() != Network::Connection::State::Closed) { + config_->drainManager().add(config_->sharedConfig(), std::move(conn_data), + std::move(upstream_callbacks_), std::move(idle_timer_), +- idle_timeout_, read_callbacks_->upstreamHost()); ++ idle_timeout_, read_callbacks_->upstreamHost(), ++ upstream_flush_wait_timeout_.count() > 0); + } + if (event == Network::ConnectionEvent::LocalClose || + event == Network::ConnectionEvent::RemoteClose) { +@@ -1508,8 +1520,10 @@ void UpstreamDrainManager::add(const Config::SharedConfigSharedPtr& config, + Event::TimerPtr&& idle_timer, + absl::optional idle_timeout, +- const Upstream::HostDescriptionConstSharedPtr& upstream_host) { ++ const Upstream::HostDescriptionConstSharedPtr& upstream_host, ++ bool discard_upstream_data) { + DrainerPtr drainer(new Drainer(*this, config, callbacks, std::move(upstream_conn_data), +- std::move(idle_timer), idle_timeout, upstream_host)); ++ std::move(idle_timer), idle_timeout, upstream_host, ++ discard_upstream_data)); + callbacks->drain(*drainer); + + // Use temporary to ensure we get the pointer before we move it out of drainer +@@ -1530,9 +1544,10 @@ Drainer::Drainer(UpstreamDrainManager& parent, const Config::SharedConfigSharedPt + Tcp::ConnectionPool::ConnectionDataPtr&& conn_data, Event::TimerPtr&& idle_timer, + absl::optional idle_timeout, +- const Upstream::HostDescriptionConstSharedPtr& upstream_host) ++ const Upstream::HostDescriptionConstSharedPtr& upstream_host, ++ bool discard_upstream_data) + : parent_(parent), callbacks_(callbacks), upstream_conn_data_(std::move(conn_data)), + idle_timer_(std::move(idle_timer)), idle_timeout_(idle_timeout), +- upstream_host_(upstream_host), config_(config) { ++ upstream_host_(upstream_host), config_(config), discard_upstream_data_(discard_upstream_data) { + ENVOY_CONN_LOG(trace, "draining the upstream connection", upstream_conn_data_->connection()); + config_->stats().upstream_flush_total_.inc(); + config_->stats().upstream_flush_active_.inc(); +@@ -1551,6 +1564,10 @@ void Drainer::onEvent(Network::ConnectionEvent event) { + } + + void Drainer::onData(Buffer::Instance& data, bool) { ++ if (discard_upstream_data_) { ++ data.drain(data.length()); ++ return; ++ } + if (data.length() > 0) { + // There is no downstream connection to send any data to, but the upstream + // sent some data. Try to behave similar to what the kernel would do +diff --git a/source/common/tcp_proxy/tcp_proxy.h b/source/common/tcp_proxy/tcp_proxy.h +index b10971523a..39540726b7 100644 +--- a/source/common/tcp_proxy/tcp_proxy.h ++++ b/source/common/tcp_proxy/tcp_proxy.h +@@ -55,6 +55,15 @@ constexpr absl::string_view PerConnectionIdleTimeoutMs = + */ + constexpr absl::string_view ReceiveBeforeConnectKey = "envoy.tcp_proxy.receive_before_connect"; + ++/** ++ * UpstreamFlushWaitTimeoutMs is the key for a per-connection upstream flush wait timeout. A ++ * non-zero ``StreamInfo::UInt64Accessor`` value makes a raw TCP upstream wait for its peer to ++ * close after pending writes have been flushed when the downstream connection closes. Data ++ * received during this interval is discarded because the downstream connection no longer exists. ++ */ ++constexpr absl::string_view UpstreamFlushWaitTimeoutMs = ++ "envoy.tcp_proxy.upstream_flush_wait_timeout_ms"; ++ + /** + * All tcp proxy stats. @see stats_macros.h + */ +@@ -755,4 +764,5 @@ protected: + bool receive_before_connect_{false}; ++ std::chrono::milliseconds upstream_flush_wait_timeout_{0}; + bool early_data_end_stream_{false}; + Buffer::OwnedImpl early_data_buffer_{}; + HttpStreamDecoderFilterCallbacks upstream_decoder_filter_callbacks_; +@@ -778,6 +788,7 @@ public: + Tcp::ConnectionPool::ConnectionDataPtr&& conn_data, Event::TimerPtr&& idle_timer, + absl::optional idle_timeout, +- const Upstream::HostDescriptionConstSharedPtr& upstream_host); ++ const Upstream::HostDescriptionConstSharedPtr& upstream_host, ++ bool discard_upstream_data); + + void onEvent(Network::ConnectionEvent event); + void onData(Buffer::Instance& data, bool end_stream); +@@ -794,5 +805,6 @@ private: + Upstream::HostDescriptionConstSharedPtr upstream_host_; + Config::SharedConfigSharedPtr config_; ++ const bool discard_upstream_data_; + }; + + using DrainerPtr = std::unique_ptr; +@@ -806,6 +818,7 @@ public: + const std::shared_ptr& callbacks, + Event::TimerPtr&& idle_timer, absl::optional idle_timeout, +- const Upstream::HostDescriptionConstSharedPtr& upstream_host); ++ const Upstream::HostDescriptionConstSharedPtr& upstream_host, ++ bool discard_upstream_data); + void remove(Drainer& drainer, Event::Dispatcher& dispatcher); + + private: +diff --git a/source/common/tcp_proxy/upstream.cc b/source/common/tcp_proxy/upstream.cc +index f466703e25..9d12c690d4 100644 +--- a/source/common/tcp_proxy/upstream.cc ++++ b/source/common/tcp_proxy/upstream.cc +@@ -125,14 +125,21 @@ StreamInfo::DetectedCloseType TcpUpstream::detectedCloseType() const { + } + + Tcp::ConnectionPool::ConnectionData* TcpUpstream::onDownstreamEvent(Network::ConnectionEvent event, +- absl::string_view details) { ++ absl::string_view details, ++ std::chrono::milliseconds ++ flush_wait_timeout) { + // TODO(botengyao): propagate RST back to upstream connection if RST is received from downstream. + if (event == Network::ConnectionEvent::RemoteClose) { + // The close call may result in this object being deleted. Latch the + // connection locally so it can be returned for potential draining. + auto* conn_data = upstream_conn_data_.release(); ++ const bool wait_for_peer_close = flush_wait_timeout.count() > 0; ++ if (wait_for_peer_close) { ++ conn_data->connection().setDelayedCloseTimeout(flush_wait_timeout); ++ } + conn_data->connection().close( +- Network::ConnectionCloseType::FlushWrite, ++ wait_for_peer_close ? Network::ConnectionCloseType::FlushWriteAndDelay ++ : Network::ConnectionCloseType::FlushWrite, + StreamInfo::LocalCloseReasons::get().ClosingUpstreamTcpDueToDownstreamRemoteClose); + return conn_data; + } else if (event == Network::ConnectionEvent::LocalClose) { +@@ -235,7 +242,8 @@ void HttpUpstream::addBytesSentCallback(Network::Connection::BytesSentCb) { + } + + Tcp::ConnectionPool::ConnectionData* +-HttpUpstream::onDownstreamEvent(Network::ConnectionEvent event, absl::string_view /*details*/) { ++HttpUpstream::onDownstreamEvent(Network::ConnectionEvent event, absl::string_view /*details*/, ++ std::chrono::milliseconds /*flush_wait_timeout*/) { + if (event == Network::ConnectionEvent::LocalClose || + event == Network::ConnectionEvent::RemoteClose) { + resetEncoder(Network::ConnectionEvent::LocalClose, false); +@@ -563,7 +571,8 @@ bool CombinedUpstream::readDisable(bool disable) { + } + + Tcp::ConnectionPool::ConnectionData* +-CombinedUpstream::onDownstreamEvent(Network::ConnectionEvent event, absl::string_view /*details*/) { ++CombinedUpstream::onDownstreamEvent(Network::ConnectionEvent event, absl::string_view /*details*/, ++ std::chrono::milliseconds /*flush_wait_timeout*/) { + if (!upstream_request_) { + return nullptr; + } +diff --git a/source/common/tcp_proxy/upstream.h b/source/common/tcp_proxy/upstream.h +index 4ce796b563..090856885e 100644 +--- a/source/common/tcp_proxy/upstream.h ++++ b/source/common/tcp_proxy/upstream.h +@@ -177,7 +177,9 @@ public: + void encodeData(Buffer::Instance& data, bool end_stream) override; + void addBytesSentCallback(Network::Connection::BytesSentCb cb) override; + Tcp::ConnectionPool::ConnectionData* onDownstreamEvent(Network::ConnectionEvent event, +- absl::string_view details = "") override; ++ absl::string_view details = "", ++ std::chrono::milliseconds ++ flush_wait_timeout = {}) override; + bool startUpstreamSecureTransport() override; + Ssl::ConnectionInfoConstSharedPtr getUpstreamConnectionSslInfo() override; + StreamInfo::DetectedCloseType detectedCloseType() const override; +@@ -207,7 +209,9 @@ public: + void encodeData(Buffer::Instance& data, bool end_stream) override; + void addBytesSentCallback(Network::Connection::BytesSentCb cb) override; + Tcp::ConnectionPool::ConnectionData* onDownstreamEvent(Network::ConnectionEvent event, +- absl::string_view details = "") override; ++ absl::string_view details = "", ++ std::chrono::milliseconds ++ flush_wait_timeout = {}) override; + // HTTP upstream must not implement converting upstream transport + // socket from non-secure to secure mode. + bool startUpstreamSecureTransport() override { return false; } +@@ -299,7 +303,9 @@ public: + void newStream(GenericConnectionPoolCallbacks& callbacks); + void encodeData(Buffer::Instance& data, bool end_stream) override; + Tcp::ConnectionPool::ConnectionData* onDownstreamEvent(Network::ConnectionEvent event, +- absl::string_view details = "") override; ++ absl::string_view details = "", ++ std::chrono::milliseconds ++ flush_wait_timeout = {}) override; + bool isValidResponse(const Http::ResponseHeaderMap&); + bool readDisable(bool disable) override; + void setConnPoolCallbacks(std::unique_ptr&& callbacks) { From 0dddca076427b6fe13ba14f93b5a85058479fa5f Mon Sep 17 00:00:00 2001 From: Jarno Rajahalme Date: Tue, 25 Aug 2026 20:17:54 +0200 Subject: [PATCH 6/9] websocket: Delay TCPProxy upstream close after downstream closes Configure TcpProxy::UpstreamFlushWaitTimeoutMs filter state to enable delayed close on the upstream TCP connection used for WebSocket tunnel, so that WebSocket codec can cleanly encode any pending data and finish WebSocket CLOSE handshake after the downstream connection closes. Signed-off-by: Jarno Rajahalme --- cilium/BUILD | 1 + cilium/websocket.cc | 14 ++++++++++++++ 2 files changed, 15 insertions(+) diff --git a/cilium/BUILD b/cilium/BUILD index cc2fac167..db2e7c34e 100644 --- a/cilium/BUILD +++ b/cilium/BUILD @@ -171,6 +171,7 @@ envoy_cc_library( "@envoy//source/common/http:utility_lib", "@envoy//source/common/network:filter_manager_lib", "@envoy//source/common/stream_info:bool_accessor_lib", + "@envoy//source/common/stream_info:uint64_accessor_lib", "@envoy//source/common/tcp_proxy", "@envoy_api//envoy/extensions/request_id/uuid/v3:pkg_cc_proto", ], diff --git a/cilium/websocket.cc b/cilium/websocket.cc index c84a2b279..47ce3b233 100644 --- a/cilium/websocket.cc +++ b/cilium/websocket.cc @@ -1,5 +1,6 @@ #include "cilium/websocket.h" +#include #include #include #include @@ -19,6 +20,7 @@ #include "source/common/protobuf/protobuf.h" #include "source/common/protobuf/utility.h" #include "source/common/stream_info/bool_accessor_impl.h" +#include "source/common/stream_info/uint64_accessor_impl.h" #include "source/common/tcp_proxy/tcp_proxy.h" #include "absl/status/statusor.h" @@ -34,6 +36,8 @@ namespace WebSocket { namespace { +constexpr std::chrono::milliseconds WebSocketTransportCloseTimeout{1000}; + Http::RegisterCustomInlineHeader origin_handle(Http::CustomHeaders::get().Origin); Http::RegisterCustomInlineHeader @@ -128,6 +132,16 @@ void Instance::initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callb callbacks_->connection().streamInfo().filterState()->setData( TcpProxy::ReceiveBeforeConnectKey, std::make_unique(true), StreamInfo::FilterState::StateType::Mutable, StreamInfo::FilterState::LifeSpan::Connection); + + // After both directions of a client-side WebSocket tunnel have ended, TcpProxy must flush the + // final data and CLOSE frame and wait for the peer's transport FIN. Closing immediately can + // generate an RST if a peer control frame is still unread, losing the just-flushed frames. + if (config_->client_) { + callbacks_->connection().streamInfo().filterState()->setData( + TcpProxy::UpstreamFlushWaitTimeoutMs, + std::make_unique(WebSocketTransportCloseTimeout.count()), + StreamInfo::FilterState::StateType::Mutable, StreamInfo::FilterState::LifeSpan::Connection); + } } Network::FilterStatus Instance::onNewConnection() { From e573706c5cccb3916a05c291b093ae2b65c2e156 Mon Sep 17 00:00:00 2001 From: Jarno Rajahalme Date: Tue, 25 Aug 2026 07:31:03 +0200 Subject: [PATCH 7/9] websocket: Properly mask client frames in decoding tests RFC 6455 requires every client-to-server WebSocket frame to be masked, but the decapsulation integration tests sent most of their client frames unmasked. The decoder accepted these frames, causing otherwise successful tests to depend on invalid protocol traffic. Add a helper for constructing masked client binary frames and use it throughout the existing positive test coverage. Preserve coverage for non-minimal 16-bit and 64-bit payload-length encodings, frames received back-to-back, and frames split across multiple writes. This prepares the tests for enforcing client masking without changing production behavior. Signed-off-by: Jarno Rajahalme --- ...cilium_websocket_decap_integration_test.cc | 102 ++++++++++-------- 1 file changed, 60 insertions(+), 42 deletions(-) diff --git a/tests/cilium_websocket_decap_integration_test.cc b/tests/cilium_websocket_decap_integration_test.cc index 46f9411fe..1d0ce0bcd 100644 --- a/tests/cilium_websocket_decap_integration_test.cc +++ b/tests/cilium_websocket_decap_integration_test.cc @@ -3,7 +3,9 @@ #include #include +#include #include +#include #include #include "envoy/event/dispatcher.h" @@ -23,6 +25,42 @@ using namespace std::literals; namespace Envoy { +namespace { + +enum class PayloadLengthEncoding { Minimal, Uint16, Uint64 }; + +std::string maskedClientFrame(absl::string_view payload, + PayloadLengthEncoding encoding = PayloadLengthEncoding::Minimal) { + constexpr std::array mask = {0x12, 0x34, 0x56, 0x78}; + std::string frame; + frame.reserve(14 + payload.size()); + frame.push_back('\x82'); // FIN and binary opcode + + if (encoding == PayloadLengthEncoding::Minimal && payload.size() < 126) { + frame.push_back(static_cast(0x80 | payload.size())); + } else if (encoding == PayloadLengthEncoding::Uint16 || + (encoding == PayloadLengthEncoding::Minimal && payload.size() <= UINT16_MAX)) { + frame.push_back(static_cast(0x80 | 126)); + frame.push_back(static_cast(payload.size() >> 8)); + frame.push_back(static_cast(payload.size())); + } else { + frame.push_back(static_cast(0x80 | 127)); + for (int shift = 56; shift >= 0; shift -= 8) { + frame.push_back(static_cast(payload.size() >> shift)); + } + } + + for (uint8_t byte : mask) { + frame.push_back(static_cast(byte)); + } + for (size_t i = 0; i < payload.size(); ++i) { + frame.push_back(static_cast(static_cast(payload[i]) ^ mask[i % mask.size()])); + } + return frame; +} + +} // namespace + // params: is_ingress ("true", "false") const std::string cilium_tcp_proxy_config_fmt = R"EOF( admin: @@ -158,9 +196,8 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) { auto client_conn = codec_client_->connection(); - // Create websocket framed data & write it on the client connection - Buffer::OwnedImpl buf{"\x82\x5" - "hello"}; + // Create masked WebSocket framed data and write it on the client connection. + Buffer::OwnedImpl buf{maskedClientFrame("hello")}; client_conn->write(buf, false); // Run the dispatcher so that the write event is handled client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock); @@ -181,12 +218,9 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) { // Send multiple frames back-to-back ASSERT_EQ(buf.length(), 0); - buf.add("\x82\x6" - "hello2" - "\x82\x7" - "hello21" - "\x82\x3" - "foo"); + buf.add(maskedClientFrame("hello2")); + buf.add(maskedClientFrame("hello21")); + buf.add(maskedClientFrame("foo")); client_conn->write(buf, false); // Run the dispatcher so that the write event is handled client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock); @@ -207,9 +241,7 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) { // Officially optimal length formats must be used, but our implementation // accepts larger formats with less data, which makes testing easier. ASSERT_EQ(buf.length(), 0); - absl::string_view frame16{"\x82\x7e\0\x5" - "len16", - 9}; + const std::string frame16 = maskedClientFrame("len16", PayloadLengthEncoding::Uint16); buf.add(frame16); client_conn->write(buf, false); // Run the dispatcher so that the write event is handled @@ -234,9 +266,7 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) { // Officially optimal length formats must be used, but our implementation // accepts larger formats with less data, which makes testing easier. ASSERT_EQ(buf.length(), 0); - absl::string_view frame64{"\x82\x7f\0\0\0\0\0\0\0\x5" - "len64", - 15}; + const std::string frame64 = maskedClientFrame("len64", PayloadLengthEncoding::Uint64); buf.add(frame64); client_conn->write(buf, false); // Run the dispatcher so that the write event is handled @@ -255,10 +285,11 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) { // Gaps within a frame ASSERT_EQ(buf.length(), 0); - buf.add("\x82\x5" - "hello" - "\x82\xe" - "gap "); + const std::string hello_frame = maskedClientFrame("hello"); + const std::string gap_frame = maskedClientFrame("gap in between"); + buf.add(hello_frame); + // Send the second frame's header, masking key, and first four payload bytes. + buf.add(gap_frame.substr(0, 10)); client_conn->write(buf, false); // Run the dispatcher so that the write event is handled client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock); @@ -270,9 +301,8 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) { ASSERT_TRUE(fake_upstream_connection->write("bar42")); ASSERT_EQ(buf.length(), 0); - buf.add("in between" - "\x82\x3" - "foo"); + buf.add(gap_frame.substr(10)); + buf.add(maskedClientFrame("foo")); client_conn->write(buf, false); // Run the dispatcher so that the write event is handled client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock); @@ -287,17 +317,10 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) { "bar42"); response->clearBody(); - // Masked frames + // Masked frame ASSERT_EQ(buf.length(), 0); auto msg = "heello there\r\n"s; - unsigned char mask[4] = {0x12, 0x34, 0x56, 0x78}; - auto masked = msg; - for (size_t i = 0; i < msg.length(); i++) { - masked[i] = msg[i] ^ mask[i % 4]; - } - buf.add("\x82\x8e"); - buf.add(mask, 4); - buf.add(masked.data(), masked.length()); + buf.add(maskedClientFrame(msg)); client_conn->write(buf, false); // Run the dispatcher so that the write event is handled client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock); @@ -315,23 +338,18 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) { "heello there\r\n"); response->clearBody(); - // 2nd masked frame + // Split masked frame ASSERT_EQ(buf.length(), 0); auto msg2 = "hello there\r\n"s; - unsigned char mask2[4] = {0x90, 0xab, 0xcd, 0xef}; - auto masked2 = msg2; - for (size_t i = 0; i < msg2.length(); i++) { - masked2[i] = msg2[i] ^ mask2[i % 4]; - } - // Write frame header - buf.add("\x82\x8d"); - buf.add(mask2, 4); + const std::string masked2 = maskedClientFrame(msg2); + // Write the frame header and masking key. + buf.add(masked2.substr(0, 6)); client_conn->write(buf, false); // Run the dispatcher so that the write event is handled client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock); // Write 5 first bytes - buf.add(masked2.data(), 5); + buf.add(masked2.substr(6, 5)); client_conn->write(buf, false); // Run the dispatcher so that the write event is handled client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock); @@ -341,7 +359,7 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) { seen_data_len = data.length(); // Write remaining bytes - buf.add(masked2.data() + 5, masked2.length() - 5); + buf.add(masked2.substr(11)); client_conn->write(buf, false); // Run the dispatcher so that the write event is handled client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock); From c3b82c8c5f6cbece0e3976fe00c47c685462f45b Mon Sep 17 00:00:00 2001 From: Jarno Rajahalme Date: Tue, 25 Aug 2026 08:05:49 +0200 Subject: [PATCH 8/9] websocket: complete tunneled TCP half-closed support Envoy TCP Proxy supports TCP half-close semantics, allowing one direction to end while data continues flowing in the reverse direction. Complete the corresponding behavior for TCP connections tunneled over WebSocket. Treat WebSocket CLOSE as a directional FIN. Track sent and received CLOSE independently, and emit end stream toward the corresponding tunneled TCP connection exactly once while leaving the reverse direction available. Delay a CLOSE response until the reverse TCP direction reaches end stream, allowing all pending reverse-direction data to be encoded first. Preserve the received CLOSE payload in the response, including its status-code bytes and optional reason. If the local TCP direction ends first, send CLOSE and continue processing peer control traffic until the peer's CLOSE arrives. Keep PING and PONG processing active while only one tunnel direction has closed. After receiving CLOSE, discard data-frame payloads while retaining frame synchronization so that subsequent control frames can still be processed. Refactor control-frame encoding so that CLOSE, PING, and PONG, including empty control frames, use the same encoder path. Client-generated control frames remain masked as required by the protocol. Extend integration coverage for client-first and server-first half-close, reverse-direction data after CLOSE, early data with FIN, buffer flushing, TCP resets, control frames during the half-closed interval, policy-filtered connections, and CLOSE payload propagation. Signed-off-by: Jarno Rajahalme --- cilium/websocket_codec.cc | 353 ++++++++++-------- cilium/websocket_codec.h | 12 +- ...cilium_websocket_codec_integration_test.cc | 60 ++- ...cilium_websocket_decap_integration_test.cc | 46 +++ ...cilium_websocket_encap_integration_test.cc | 185 +++++++-- ...ilium_websocket_policy_integration_test.cc | 7 +- 6 files changed, 448 insertions(+), 215 deletions(-) diff --git a/cilium/websocket_codec.cc b/cilium/websocket_codec.cc index 461bc32f7..38c567e58 100644 --- a/cilium/websocket_codec.cc +++ b/cilium/websocket_codec.cc @@ -373,22 +373,45 @@ void Codec::handshake() { void Codec::encode(Buffer::Instance& data, bool end_stream) { ENVOY_LOG(debug, "websocket: encode {} bytes, end_stream: {}", data.length(), end_stream); - encoder_.encode(data, end_stream, OPCODE_BIN); + // RFC 6455 section 5.5.1 forbids sending data frames after a CLOSE frame. + if (close_sent_) { + const auto len = data.length(); + if (len > 0) { + ENVOY_LOG(debug, "websocket encoder: data received after CLOSE: {} bytes", len); + data.drain(len); + } + return; + } + + encoder_.encode(data, OPCODE_BIN); + + if (end_stream) { + // CLOSE represents the FIN for this direction of the tunneled TCP connection. If the peer + // half-closed first, echo its validated CLOSE payload only after all reverse-direction data + // has been encoded. This deliberately delays the CLOSE response to preserve TCP half-close + // semantics when the WebSocket infrastructure permits it. + Buffer::OwnedImpl close_payload; + close_payload.add(decoder_.close_payload_); + encoder_.encode(close_payload, OPCODE_CLOSE); + close_sent_ = true; + + // Only end stream if CLOSE has been received already and if we are the server, as additional + // control frames may still be sent, and RFC 6455 section 7.1.1 recommends that the server close + // the underlying transport after both sides have sent CLOSE. Keep the client transport open so + // that its final frames are flushed before the server closes the connection. + end_stream = !config()->client_ && decoder_.close_received_; + } // Only forward data if handshake has completed if (accepted_) { - // Reset idle timer on data + // Reset idle timer on any frame content if (encoder_.hasData()) { resetPingTimer(); } - // A WebSocket CLOSE represents the FIN for this direction of the tunneled TCP connection. - // RFC 6455 section 7.1.1 recommends that the server close the underlying transport after both - // sides have sent CLOSE. Keep the client transport open so that its final frames are flushed - // before the server closes the connection. - const bool transport_end_stream = - !config()->client_ && encoder_.endStream() && decoder_.endStream(); - encoded_end_stream_sent_ |= transport_end_stream; - parent_->injectEncoded(encoder_.data(), transport_end_stream); + if (end_stream) { + encoded_end_stream_sent_ = end_stream; + } + parent_->injectEncoded(encoder_.data(), end_stream); } } @@ -661,130 +684,114 @@ void Codec::decode(Buffer::Instance& data, bool end_stream) { // Handshake done, process data. decoder_.decode(data, end_stream); + if (end_stream && !decoder_.close_received_) { + decoder_.drain(); + return closeOnError("websocket transport closed without CLOSE"); + } + // Reset idle timer on data if (decoder_.hasData()) { resetPingTimer(); } - // Record decoder end_stream before injecting the corresponding TCP FIN (via - // end_stream=true). That injection may synchronously report the reverse-direction FIN through - // encode(..., end_stream=true). - const bool decoded_end_stream = decoder_.endStream(); + // The decoder records CLOSE before injecting the corresponding TCP FIN. That injection may + // synchronously report the reverse-direction FIN through encode(..., end_stream=true). + const bool decoded_end_stream = decoder_.close_received_ && !decoded_end_stream_sent_; parent_->injectDecoded(decoder_.data(), decoded_end_stream); + decoded_end_stream_sent_ |= decoded_end_stream; - // Complete the CLOSE handshake by ending the underlying transport from the server side. RFC 6455 - // section 7.1.1 recommends that the client wait for the server to close the connection. - if (!config->client_ && decoded_end_stream && encoder_.endStream() && !encoded_end_stream_sent_) { + // Complete CLOSE handshake by encoding end_stream from the server side, if closed in both + // directions. RFC 6455 section 7.1.1 recommends that the client wait for the server to close the + // connection. + if (!config->client_ && decoder_.close_received_ && close_sent_ && !encoded_end_stream_sent_) { Buffer::OwnedImpl empty; encoded_end_stream_sent_ = true; parent_->injectEncoded(empty, true); } + // Otherwise defer the CLOSE response until encode(..., end_stream=true) reports the FIN from + // the other direction of the tunneled TCP connection. Data remains legal to encode until that + // response is sent. RFC 6455 section 5.5.1 permits delaying a CLOSE response, but warns that a + // generic peer is not guaranteed to keep processing data after sending CLOSE; this behavior is + // therefore a tunnel convention between cooperating endpoints. } bool Codec::ping(const void* payload, size_t len) { - if (encoder_.endStream()) { + // RFC 6455 section 5.5.2 permits PING until the connection is closed. For this tunnel, one CLOSE + // is only a directional FIN, so keepalive PINGs continue through the half-closed interval. + if (close_sent_ && decoder_.close_received_) { return false; } Buffer::OwnedImpl buf(payload, len); - encoder_.encode(buf, false, OPCODE_PING); + encoder_.encode(buf, OPCODE_PING); parent_->config()->stats_.ping_sent_count_.inc(); parent_->injectEncoded(encoder_.data(), false); return true; } bool Codec::pong(const void* payload, size_t len) { - if (encoder_.endStream()) { + // RFC 6455 section 5.5.2 does not require PONG after receiving CLOSE, but permits it. Continue to + // answer PING while the tunneled TCP connection is only half-closed. + if (close_sent_ && decoder_.close_received_) { return false; } Buffer::OwnedImpl buf(payload, len); - encoder_.encode(buf, false, OPCODE_PONG); + encoder_.encode(buf, OPCODE_PONG); parent_->injectEncoded(encoder_.data(), false); return true; } // Encoder -// Encode 'data' and 'end_stream' as websocket frames into 'encoded_'. Uses 'opcode' as the -// websocket frame type for the data frames. -void Codec::Encoder::encode(Buffer::Instance& data, bool end_stream, uint8_t opcode) { +// Encode 'data' as one websocket frame into 'encoded_'. +void Codec::Encoder::encode(Buffer::Instance& data, uint8_t opcode) { auto hex_len = std::min(data.length(), 20UL); const uint8_t* hex_data = reinterpret_cast(data.linearize(hex_len)); - ENVOY_LOG(debug, "websocket encoder: {} bytes: 0x{}, end_stream: {}, opcode: {}", data.length(), - Hex::encode(hex_data, hex_len), end_stream, opcode); + ENVOY_LOG(debug, "websocket encoder: {} bytes: 0x{}, opcode: {}", data.length(), + Hex::encode(hex_data, hex_len), opcode); auto& config = parent_.config(); - // - // Encode data as a single WebSocket frame - // - if (data.length() > 0) { - uint8_t frame_header[14]; - size_t frame_header_length = 2; - size_t payload_len = data.length(); - - frame_header[0] = FIN_MASK | opcode; - if (payload_len < 126) { - frame_header[1] = payload_len; - } else if (payload_len < 65536) { - uint16_t len16; - - frame_header[1] = 126; - len16 = htobe16(payload_len); - memcpy(frame_header + frame_header_length, &len16, 2); // NOLINT(safe-memcpy) - frame_header_length += 2; - } else { - uint64_t len64; - - frame_header[1] = 127; - len64 = htobe64(payload_len); - memcpy(frame_header + frame_header_length, &len64, 8); // NOLINT(safe-memcpy) - frame_header_length += 8; - } - - // Client must mask the payload - if (config->client_) { - frame_header[1] |= MASK_MASK; - - union { - uint8_t bytes[4]; - uint32_t word; - } mask; - - mask.word = config->random_.random(); - memcpy(frame_header + frame_header_length, &mask, 4); // NOLINT(safe-memcpy) - frame_header_length += 4; - uint8_t* buf = reinterpret_cast(data.linearize(payload_len)); - maskData(buf, payload_len, mask.bytes); - } - - // Add frame header and (masked) data - encoded_.add(absl::string_view{reinterpret_cast(frame_header), frame_header_length}); - encoded_.move(data, payload_len); + const size_t payload_len = data.length(); + const bool control_frame = opcode >= OPCODE_CLOSE; + if (payload_len == 0 && !control_frame) { + return; } - // - // Append closing frame if 'end_stream' - // - if (end_stream) { - uint8_t frame_header[14]; - size_t frame_header_length = 2; - size_t payload_len = 0; + RELEASE_ASSERT(!control_frame || payload_len <= WEBSOCKET_CONTROL_FRAME_MAX_SIZE, + "websocket control frame too large"); - frame_header[0] = FIN_MASK | OPCODE_CLOSE; + uint8_t frame_header[14]; + size_t frame_header_length = 2; + frame_header[0] = FIN_MASK | opcode; + if (payload_len < 126) { frame_header[1] = payload_len; - // Client must mask the payload - if (config->client_) { - frame_header[1] |= MASK_MASK; - - uint32_t mask = config->random_.random(); - memcpy(frame_header + frame_header_length, &mask, 4); // NOLINT(safe-memcpy) - frame_header_length += 4; - // No data to mask + } else if (payload_len < 65536) { + frame_header[1] = 126; + frame_header[frame_header_length++] = static_cast(payload_len >> 8); + frame_header[frame_header_length++] = static_cast(payload_len); + } else { + frame_header[1] = 127; + const uint64_t encoded_payload_len = payload_len; + for (int shift = 56; shift >= 0; shift -= 8) { + frame_header[frame_header_length++] = static_cast(encoded_payload_len >> shift); } - encoded_.add(reinterpret_cast(frame_header), frame_header_length); - end_stream_ = true; + } - ENVOY_LOG(debug, "websocket encoder: sent WebSocket CLOSE message, end_stream: {}", end_stream); + // WebSocket clients must mask all frames, including control frames. + if (config->client_) { + frame_header[1] |= MASK_MASK; + const uint32_t mask_word = config->random_.random(); + uint8_t mask[4]; + for (size_t i = 0; i < sizeof(mask); ++i) { + mask[i] = static_cast(mask_word >> (i * 8)); + frame_header[frame_header_length++] = mask[i]; + } + uint8_t* buf = reinterpret_cast(data.linearize(payload_len)); + maskData(buf, payload_len, mask); } + + // Add frame header and (maybe masked) data + encoded_.add(absl::string_view{reinterpret_cast(frame_header), frame_header_length}); + encoded_.move(data, payload_len); } // Decoder @@ -811,40 +818,39 @@ void Codec::Decoder::decode(Buffer::Instance& data, bool end_stream) { buffer_.move(data); - if (end_stream_ && buffer_.length() > 0) { - ENVOY_LOG(debug, "websocket decoder: data received after CLOSE: {} bytes", buffer_.length()); - buffer_.drain(buffer_.length()); - return; - } - - if (end_stream) { - end_stream_ = true; - } - while (buffer_.length() > 0) { - // Try finish any frame in progress + // Try finish any data frame in progress while (payload_remaining_ > 0) { auto slice = buffer_.frontSlice(); size_t n_bytes = std::min(slice.len_, payload_remaining_); - // Unmask data in place - uint8_t* buf = static_cast(slice.mem_); - auto hex_len = std::min(n_bytes, 20UL); - if (unmasking_) { - ENVOY_LOG( - trace, - "websocket decoder: unmasking payload remaining: {}, offset: {}, processing: {}: 0x{}", - payload_remaining_, payload_offset_, n_bytes, Hex::encode(buf, hex_len)); - payload_offset_ = maskData(buf, n_bytes, mask_, payload_offset_); + if (n_bytes > 0) { + // Unmask data in place + uint8_t* buf = static_cast(slice.mem_); + auto hex_len = std::min(n_bytes, 20UL); + if (unmasking_) { + ENVOY_LOG(trace, + "websocket decoder: unmasking payload remaining: {}, offset: {}, processing: " + "{}: 0x{}", + payload_remaining_, payload_offset_, n_bytes, Hex::encode(buf, hex_len)); + payload_offset_ = maskData(buf, n_bytes, mask_, payload_offset_); + } + ENVOY_LOG(trace, + "websocket decoder: payload remaining: {}, offset: {}, processing: {}: 0x{}", + payload_remaining_, payload_offset_, n_bytes, Hex::encode(buf, hex_len)); + + if (close_received_) { + // No data is accepted after CLOSE has been received. + ENVOY_LOG(debug, "websocket decoder: data received after CLOSE: {} bytes", n_bytes); + buffer_.drain(n_bytes); + } else { + decoded_.move(buffer_, n_bytes); + } + payload_remaining_ -= n_bytes; } - ENVOY_LOG(trace, "websocket decoder: payload remaining: {}, offset: {}, processing: {}: 0x{}", - payload_remaining_, payload_offset_, n_bytes, Hex::encode(buf, hex_len)); - - decoded_.move(buffer_, n_bytes); - payload_remaining_ -= n_bytes; if (buffer_.length() == 0) { - return; + return; // continue when there is more data } } // @@ -856,27 +862,24 @@ void Codec::Decoder::decode(Buffer::Instance& data, bool end_stream) { uint8_t frame_header[2]; size_t frame_offset = 0; - uint8_t opcode; - uint64_t payload_len; ENVOY_LOG(trace, "websocket decoder: remaining buffer: {} bytes", buffer_.length()); TRY_READ_NETWORK(&frame_header); - opcode = frame_header[0] & OPCODE_MASK; - payload_len = frame_header[1] & PAYLOAD_LEN_MASK; + const uint8_t opcode = frame_header[0] & OPCODE_MASK; + const bool masked = (frame_header[1] & MASK_MASK) != 0; + uint64_t payload_len = frame_header[1] & PAYLOAD_LEN_MASK; if (payload_len == 126) { uint16_t len16; - TRY_READ_NETWORK(&len16); payload_len = be16toh(len16); } else if (payload_len == 127) { uint64_t len64; - TRY_READ_NETWORK(&len64); payload_len = be64toh(len64); } - if (frame_header[1] & MASK_MASK) { + if (masked) { TRY_READ_NETWORK(&mask_); unmasking_ = true; } @@ -885,57 +888,75 @@ void Codec::Decoder::decode(Buffer::Instance& data, bool end_stream) { // Whole header received and decoded // + if (opcode < OPCODE_CLOSE) { + // Unframe and forward all non-control frames + ENVOY_LOG(trace, "websocket decoder: received websocket data: header {} bytes, data {} bytes", + frame_offset, payload_len); + buffer_.drain(frame_offset); + payload_remaining_ = payload_len; + continue; // loop back to frame data decoding + } + // Terminate and respond to any control frames - if (opcode >= OPCODE_CLOSE) { - // Protect against too large control frames that could happen if the decoder ever loses - // sync with the data stream. - if (payload_len > WEBSOCKET_CONTROL_FRAME_MAX_SIZE) { - ENVOY_LOG(debug, "websocket decoder: too large control frame: {} bytes", payload_len); - buffer_.drain(buffer_.length()); - end_stream_ = true; - return; - } - // Buffer until whole control frame has been received - if (buffer_.length() < frame_offset + payload_len) { - return; - } - // Drain control frame header, get the payload - buffer_.drain(frame_offset); - uint8_t* payload = reinterpret_cast(buffer_.linearize(payload_len)); + // Protect against too large control frames that could happen if the decoder ever loses + // sync with the data stream. + if (payload_len > WEBSOCKET_CONTROL_FRAME_MAX_SIZE) { + ENVOY_LOG(debug, "websocket decoder: too large control frame: {} bytes", payload_len); + goto protocol_error; + } - // Unmask the control frame payload - if (unmasking_) { - maskData(payload, payload_len, mask_); - } + // Buffer until whole control frame has been received + if (buffer_.length() < frame_offset + payload_len) { + return; + } + + // Drain control frame header, get the payload + buffer_.drain(frame_offset); + uint8_t* payload = reinterpret_cast(buffer_.linearize(payload_len)); + + // Unmask the control frame payload + if (unmasking_) { + maskData(payload, payload_len, mask_); + } - switch (opcode) { - case OPCODE_CLOSE: - ENVOY_LOG(trace, "websocket decoder: CLOSE received"); - end_stream_ = true; - break; - case OPCODE_PING: { - ENVOY_LOG(trace, "websocket decoder: PING received"); - // Reply with a PONG with the same payload - parent_.pong(payload, payload_len); - break; + // unknown opcodes are ignored, but their data is drained to retain stream sync + switch (opcode) { + case OPCODE_CLOSE: + ENVOY_LOG(trace, "websocket decoder: CLOSE received"); + + // validate CLOSE payload if any + if (payload_len > 0) { + // store for sending the frame back + if (!close_received_) { + close_payload_.assign(reinterpret_cast(payload), payload_len); + } } - case OPCODE_PONG: - ENVOY_LOG(trace, "websocket decoder: PONG received"); - break; + close_received_ = true; + if (parent_.close_sent_) { + // Both CLOSE frames have now been exchanged; no later frames need to be processed. + buffer_.drain(buffer_.length()); + return; } - // Drain control plane payload - buffer_.drain(payload_len); - } else { - // Unframe and forward all non-control frames - ENVOY_LOG(trace, "websocket decoder: received websocket data: header {} bytes, data {} bytes", - frame_offset, payload_len); - - buffer_.drain(frame_offset); - payload_remaining_ = payload_len; + break; + case OPCODE_PING: + ENVOY_LOG(trace, "websocket decoder: PING received"); + // Reply with a PONG with the same payload + parent_.pong(payload, payload_len); + break; + case OPCODE_PONG: + ENVOY_LOG(trace, "websocket decoder: PONG received"); + break; } + // Drain control plane payload + buffer_.drain(payload_len); } + + return; + +protocol_error: + buffer_.drain(buffer_.length()); } } // namespace WebSocket diff --git a/cilium/websocket_codec.h b/cilium/websocket_codec.h index 60e0002dc..3e8f09e4d 100644 --- a/cilium/websocket_codec.h +++ b/cilium/websocket_codec.h @@ -54,15 +54,13 @@ class Codec : Logger::Loggable { public: Encoder(Codec& parent) : parent_(parent) {} - void encode(Buffer::Instance&, bool end_stream, uint8_t opcode); + void encode(Buffer::Instance&, uint8_t opcode); size_t hasData() { return encoded_.length() > 0; } Buffer::Instance& data() { return encoded_; } - bool endStream() { return end_stream_; } void drain() { encoded_.drain(encoded_.length()); } Codec& parent_; - bool end_stream_{false}; Buffer::OwnedImpl encoded_; // Buffer for encoded websocket frames }; @@ -74,11 +72,11 @@ class Codec : Logger::Loggable { size_t hasData() { return decoded_.length() > 0; } Buffer::Instance& data() { return decoded_; } - bool endStream() { return end_stream_; } void drain() { decoded_.drain(decoded_.length()); } Codec& parent_; - bool end_stream_{false}; + bool close_received_{false}; + std::string close_payload_; Buffer::OwnedImpl buffer_; // Buffer for partial websocket frames Buffer::OwnedImpl decoded_; // Buffer for decoded websocket frames @@ -90,7 +88,7 @@ class Codec : Logger::Loggable { void startPingTimer(); void resetPingTimer() { - if (ping_timer_ != nullptr) { + if (ping_timer_ != nullptr && !(close_sent_ && decoder_.close_received_)) { auto config = parent_->config(); if (config->ping_when_idle_) { ping_timer_->enableTimer(config->ping_interval_); @@ -126,6 +124,8 @@ class Codec : Logger::Loggable { Event::TimerPtr handshake_timer_{nullptr}; Buffer::OwnedImpl handshake_buffer_; bool accepted_{false}; + bool close_sent_{false}; + bool decoded_end_stream_sent_{false}; bool encoded_end_stream_sent_{false}; }; using CodecPtr = std::unique_ptr; diff --git a/tests/cilium_websocket_codec_integration_test.cc b/tests/cilium_websocket_codec_integration_test.cc index c4d1f70bf..50e619184 100644 --- a/tests/cilium_websocket_codec_integration_test.cc +++ b/tests/cilium_websocket_codec_integration_test.cc @@ -7,6 +7,9 @@ #include #include +#include "envoy/common/platform.h" +#include "envoy/network/connection.h" + #include "test/integration/fake_upstream.h" #include "test/integration/integration_tcp_client.h" #include "test/test_common/environment.h" @@ -134,9 +137,22 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamWritesFirst) { ASSERT_TRUE(fake_upstream_connection->waitForData(5, &received)); ASSERT_EQ(received, "hello"); - ASSERT_TRUE(fake_upstream_connection->write("", true)); + // A FIN in one direction must not prevent data from flowing in the reverse direction. The first + // CLOSE crosses both WebSocket codecs and becomes a half-close at the TCP client. + ASSERT_TRUE(fake_upstream_connection->write("upstream final", true)); + tcp_client->waitForData("helloupstream final"); tcp_client->waitForHalfClose(); - ASSERT_TRUE(tcp_client->write("", true)); + + // The first CLOSE is only a directional FIN. Keepalive PING/PONG processing must continue while + // the reverse direction remains open. + const uint64_t ping_count = test_server_->counter("websocket.ping_sent_count")->value(); + test_server_->waitForCounterGe("websocket.ping_sent_count", ping_count + 1); + + // The TCP client can still send after receiving that FIN. Its own FIN produces the CLOSE response + // only after the reverse-direction data has crossed the tunnel. + ASSERT_TRUE(tcp_client->write("downstream final", true)); + ASSERT_TRUE(fake_upstream_connection->waitForData(5 + sizeof("downstream final") - 1, &received)); + ASSERT_EQ(received, "hellodownstream final"); ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); } @@ -165,6 +181,23 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamDisconnect) { EXPECT_EQ("world", tcp_client->data()); } +#if ENVOY_PLATFORM_ENABLE_SEND_RST +// A TCP reset is an abort, not a directional FIN, and must tear down the tunnel immediately. +TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamReset) { + initialize(); + IntegrationTcpClientPtr tcp_client = makeTcpConnection(lookupPort("tcp_proxy")); + ASSERT_TRUE(tcp_client->write("hello")); + FakeRawConnectionPtr fake_upstream_connection; + ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection)); + ASSERT_TRUE(fake_upstream_connection->waitForData(5)); + + ASSERT_TRUE(fake_upstream_connection->close(Network::ConnectionCloseType::AbortReset)); + ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); + tcp_client->waitForDisconnect(); +} + +#endif + // Test proxying data in both directions, and that all data is flushed properly // when the client disconnects. TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamDisconnect) { @@ -186,7 +219,12 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamDisconnect) { ASSERT_TRUE(fake_upstream_connection->waitForData(10, &received)); ASSERT_EQ(received, "hellohello"); ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); - ASSERT_TRUE(fake_upstream_connection->write("", true)); + + const uint64_t ping_count = test_server_->counter("websocket.ping_sent_count")->value(); + test_server_->waitForCounterGe("websocket.ping_sent_count", ping_count + 1); + + ASSERT_TRUE(fake_upstream_connection->write("upstream final", true)); + tcp_client->waitForData("worldupstream final"); ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); tcp_client->waitForDisconnect(); } @@ -252,8 +290,7 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamFlush) { tcp_client->readDisable(true); ASSERT_TRUE(tcp_client->write("", true)); - // This ensures that readDisable(true) has been run on it's thread - // before tcp_client starts writing. + // Confirm that the downstream FIN crossed the WebSocket tunnel before sending a large response. ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); ASSERT_TRUE(fake_upstream_connection->write(data, true)); @@ -265,7 +302,6 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamFlush) { tcp_client->readDisable(false); tcp_client->waitForData(data); tcp_client->waitForHalfClose(); - ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); uint32_t upstream_pauses = test_server_->counter("cluster.cluster1.upstream_flow_control_paused_reading_total")->value(); @@ -295,8 +331,7 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlush) { ASSERT_TRUE(fake_upstream_connection->readDisable(true)); ASSERT_TRUE(fake_upstream_connection->write("", true)); - // This ensures that fake_upstream_connection->readDisable has been run on - // it's thread before tcp_client starts writing. + // Confirm that the upstream FIN crossed the WebSocket tunnel before sending a large request. tcp_client->waitForHalfClose(); ASSERT_TRUE(tcp_client->write(data, true, true, std::chrono::milliseconds(30000))); @@ -309,7 +344,6 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlush) { ASSERT_EQ(received, data); ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); - tcp_client->waitForHalfClose(); EXPECT_EQ(test_server_->counter("tcp.tcp_stats.upstream_flush_total")->value(), 1); test_server_->waitForGaugeEq("tcp.tcp_stats.upstream_flush_active", 0); @@ -329,15 +363,15 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlushEnvoyExit) { FakeRawConnectionPtr fake_upstream_connection; ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection)); + // Confirm that the WebSocket handshake and keepalive timer are active. + test_server_->waitForCounterGe("websocket.ping_sent_count", 1); + ASSERT_TRUE(fake_upstream_connection->readDisable(true)); ASSERT_TRUE(fake_upstream_connection->write("", true)); - // This ensures that fake_upstream_connection->readDisable has been run on - // it's thread before tcp_client starts writing. + // Confirm that the upstream FIN crossed the WebSocket tunnel before filling the write buffer. tcp_client->waitForHalfClose(); - test_server_->waitForCounterGe("websocket.ping_sent_count", 1); - ASSERT_TRUE(tcp_client->write(data, true)); test_server_->waitForGaugeEq("tcp.tcp_stats.upstream_flush_active", 1); diff --git a/tests/cilium_websocket_decap_integration_test.cc b/tests/cilium_websocket_decap_integration_test.cc index 1d0ce0bcd..44158f648 100644 --- a/tests/cilium_websocket_decap_integration_test.cc +++ b/tests/cilium_websocket_decap_integration_test.cc @@ -388,4 +388,50 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) { codec_client_->close(); } +TEST_P(CiliumWebSocketIntegrationTest, CloseResponseWaitsForReverseFin) { + enableHalfClose(true); + initialize(); + auto request_headers = Http::TestRequestHeaderMapImpl{ + {":method", "GET"}, + {":path", "/"}, + {":authority", "host"}, + {"Upgrade", "websocket"}, + {"Connection", "Upgrade"}, + {"Origin", "jarno.cilium.rocks"}, + {"Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ=="}, + {"Sec-WebSocket-Version", "13"}, + {"x-request-id", "000000ff-0000-0000-0000-000000000001"}, + {"x-envoy-original-dst-host", original_dst_address->asString()}}; + codec_client_ = makeHttpConnection(lookupPort("http")); + + IntegrationStreamDecoderPtr response = codec_client_->makeHeaderOnlyRequest(request_headers); + FakeRawConnectionPtr fake_upstream_connection; + ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection)); + response->waitForHeaders(); + ASSERT_EQ("101", response->headers().getStatusValue()); + + // A masked CLOSE carrying status 1000 and reason "done". + const uint8_t masked_close[] = {0x88, 0x86, 0, 0, 0, 0, 0x03, 0xe8, 'd', 'o', 'n', 'e'}; + Buffer::OwnedImpl close_buffer(masked_close, sizeof(masked_close)); + auto* client_connection = codec_client_->connection(); + client_connection->write(close_buffer, false); + client_connection->dispatcher().run(Event::Dispatcher::RunType::NonBlock); + + // The received CLOSE represents a FIN toward the upstream TCP connection, but that connection's + // reverse direction remains usable. + ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); + ASSERT_TRUE(fake_upstream_connection->write("reverse", true)); + + // The reverse data frame precedes the delayed, unmasked CLOSE response, which echoes the status + // and reason from the peer's CLOSE. + const uint8_t expected_response[] = {0x82, 0x07, 'r', 'e', 'v', 'e', 'r', 's', 'e', + 0x88, 0x06, 0x03, 0xe8, 'd', 'o', 'n', 'e'}; + response->waitForBodyData(sizeof(expected_response)); + ASSERT_EQ(response->body(), absl::string_view(reinterpret_cast(expected_response), + sizeof(expected_response))); + ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); + + codec_client_->close(); +} + } // namespace Envoy diff --git a/tests/cilium_websocket_encap_integration_test.cc b/tests/cilium_websocket_encap_integration_test.cc index cb421e983..7a7371778 100644 --- a/tests/cilium_websocket_encap_integration_test.cc +++ b/tests/cilium_websocket_encap_integration_test.cc @@ -263,16 +263,55 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketHandshakeSuccess) { ASSERT_TRUE(frame_offset > 0); ASSERT_EQ(received_data.substr(frame_offset, 5), "hello"); - ASSERT_TRUE(fake_upstream_connection->write("\x82\x5" - "world")); - ASSERT_TRUE(fake_upstream_connection->close()); - ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); + ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x82\x5world\x88\0", 9})); tcp_client->waitForHalfClose(); - tcp_client->close(); + ASSERT_TRUE(tcp_client->write("", true)); + ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); + ASSERT_TRUE(fake_upstream_connection->write("", true)); + ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); EXPECT_EQ("world", tcp_client->data()); } +// Adapted from TcpProxyIntegrationTest.UpstreamConnectModeEarlyDataWithHalfClose: data and FIN +// arriving before the upstream handshake must be preserved, including reverse data after the FIN. +TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketHandshakeWithEarlyDataAndFin) { + initialize(); + IntegrationTcpClientPtr tcp_client = makeTcpConnection(lookupPort("tcp_proxy")); + ASSERT_TRUE(tcp_client->write("final_data", true)); + FakeRawConnectionPtr fake_upstream_connection; + ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection)); + + std::string expected_handshake = + fmt::format(fmt::runtime(EXPECTED_HANDSHAKE_FMT), original_dst_address->asString()); + std::string received_data; + ASSERT_TRUE(fake_upstream_connection->waitForData(expected_handshake.length(), &received_data)); + ASSERT_EQ(normalizeXRequestId(received_data), sizeof(X_REQUEST_ID_VALUE) - 1); + ASSERT_EQ(received_data, expected_handshake); + + std::string handshake_response = + fmt::format(fmt::runtime(HANDSHAKE_RESPONSE_FMT), "GjgmQ9MzNsn3h7+vuIzY25rbQ9M="); + ASSERT_TRUE(fake_upstream_connection->write(handshake_response)); + + // A ten-byte masked data frame is followed by the masked CLOSE representing the early FIN. + ASSERT_TRUE( + fake_upstream_connection->waitForData(expected_handshake.length() + 16 + 6, &received_data)); + received_data.erase(0, expected_handshake.length()); + auto frame_offset = unmaskData(received_data.data(), 16); + ASSERT_TRUE(frame_offset > 0); + ASSERT_EQ(received_data.substr(frame_offset, 10), "final_data"); + received_data.erase(0, 16); + frame_offset = unmaskData(received_data.data(), 6, OPCODE_CLOSE); + ASSERT_TRUE(frame_offset > 0); + + EXPECT_FALSE(fake_upstream_connection->waitForHalfClose(std::chrono::milliseconds(100))); + ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x82\x8response\x88\0", 12}, true)); + tcp_client->waitForData("response"); + tcp_client->waitForHalfClose(); + ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); + ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); +} + // Test successful handshake where client does not send any data, and the server side sends data // right after the handshake response. TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketHandshakeNoData) { @@ -293,16 +332,107 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketHandshakeNoData) { fmt::format(fmt::runtime(HANDSHAKE_RESPONSE_FMT), "GjgmQ9MzNsn3h7+vuIzY25rbQ9M="); ASSERT_TRUE(fake_upstream_connection->write(handshake_response)); - ASSERT_TRUE(fake_upstream_connection->write("\x82\x5" - "world")); - ASSERT_TRUE(fake_upstream_connection->close()); - ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); + ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x82\x5world\x88\0", 9})); tcp_client->waitForHalfClose(); - tcp_client->close(); + ASSERT_TRUE(tcp_client->write("", true)); + ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); + ASSERT_TRUE(fake_upstream_connection->write("", true)); + ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); EXPECT_EQ("world", tcp_client->data()); } +TEST_P(CiliumWebSocketIntegrationTest, ControlFramesAfterSendingClose) { + initialize(); + IntegrationTcpClientPtr tcp_client = makeTcpConnection(lookupPort("tcp_proxy")); + FakeRawConnectionPtr fake_upstream_connection; + ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection)); + + std::string expected_handshake = + fmt::format(fmt::runtime(EXPECTED_HANDSHAKE_FMT), original_dst_address->asString()); + std::string received_data; + ASSERT_TRUE(fake_upstream_connection->waitForData(expected_handshake.length(), &received_data)); + + std::string handshake_response = + fmt::format(fmt::runtime(HANDSHAKE_RESPONSE_FMT), "GjgmQ9MzNsn3h7+vuIzY25rbQ9M="); + ASSERT_TRUE(fake_upstream_connection->write(handshake_response)); + + // The local FIN sends a masked CLOSE but leaves the WebSocket transport open for reverse data. + ASSERT_TRUE(tcp_client->write("", true)); + ASSERT_TRUE( + fake_upstream_connection->waitForData(expected_handshake.length() + 6, &received_data)); + // Remove the already handled prefix that waitForData() reintroduces on each call. + received_data.erase(0, expected_handshake.length()); + ASSERT_GT(unmaskData(received_data.data(), 6, OPCODE_CLOSE), 0); + + // PING and PONG remain valid control frames after sending CLOSE. The PING response must echo its + // payload, while the unsolicited PONG requires no response. + ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x89\x04ping\x8a\x04pong", 12})); + ASSERT_TRUE( + fake_upstream_connection->waitForData(expected_handshake.length() + 6 + 10, &received_data)); + // Remove the already handled prefix that waitForData() reintroduces on each call. + received_data.erase(0, expected_handshake.length() + 6); + const size_t frame_offset = unmaskData(received_data.data(), 10, OPCODE_PONG); + ASSERT_GT(frame_offset, 0); + EXPECT_EQ(received_data.substr(frame_offset), "ping"); + + ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x82\x04done\x88\0", 8}, true)); + tcp_client->waitForData("done"); + tcp_client->waitForHalfClose(); + ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); + ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); +} + +TEST_P(CiliumWebSocketIntegrationTest, ControlFramesAfterReceivingClose) { + initialize(); + IntegrationTcpClientPtr tcp_client = makeTcpConnection(lookupPort("tcp_proxy")); + FakeRawConnectionPtr fake_upstream_connection; + ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection)); + + std::string expected_handshake = + fmt::format(fmt::runtime(EXPECTED_HANDSHAKE_FMT), original_dst_address->asString()); + std::string received_data; + ASSERT_TRUE(fake_upstream_connection->waitForData(expected_handshake.length(), &received_data)); + + std::string handshake_response = + fmt::format(fmt::runtime(HANDSHAKE_RESPONSE_FMT), "GjgmQ9MzNsn3h7+vuIzY25rbQ9M="); + ASSERT_TRUE(fake_upstream_connection->write(handshake_response)); + + // A peer CLOSE half-closes the local TCP receive side, but the reverse direction and WebSocket + // control plane remain active. + ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x88\0", 2})); + tcp_client->waitForHalfClose(); + // Data frames after CLOSE are prohibited and drained, but must not hide subsequent control + // frames in the same read. + ASSERT_TRUE( + fake_upstream_connection->write(std::string{"\x82\x07ignored\x89\x04ping\x8a\x04pong", 21})); + ASSERT_TRUE( + fake_upstream_connection->waitForData(expected_handshake.length() + 10, &received_data)); + // Remove the already handled prefix that waitForData() reintroduces on each call. + received_data.erase(0, expected_handshake.length()); + size_t frame_offset = unmaskData(received_data.data(), 10, OPCODE_PONG); + ASSERT_GT(frame_offset, 0); + EXPECT_EQ(received_data.substr(frame_offset), "ping"); + + // Sending the reverse data and FIN completes the two directional CLOSE messages. + ASSERT_TRUE(tcp_client->write("done", true)); + ASSERT_TRUE(fake_upstream_connection->waitForData(expected_handshake.length() + 10 + 10 + 6, + &received_data)); + // Remove the already handled prefix that waitForData() reintroduces on each call. + received_data.erase(0, expected_handshake.length() + 10); + frame_offset = unmaskData(received_data.data(), 10); + ASSERT_GT(frame_offset, 0); + EXPECT_EQ(received_data.substr(frame_offset, 4), "done"); + // Remove the already handled prefix that waitForData() reintroduces on each call. + received_data.erase(0, 10); + ASSERT_GT(unmaskData(received_data.data(), 6, OPCODE_CLOSE), 0); + + ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); + ASSERT_TRUE(fake_upstream_connection->write("", true)); + ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); +} + + // Test proxying data in both directions, and that all data is flushed properly // when the client disconnects. TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamDisconnect) { @@ -350,8 +480,8 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamDisconnect) { ASSERT_TRUE(frame_offset > 0); ASSERT_EQ(frame_offset, 6); - // The locally generated CLOSE represents only the tunneled downstream FIN. It must not also send - // FIN on the outer WebSocket transport while the peer's tunneled direction remains open. + // CLOSE represents the downstream TCP FIN, but must not close the WebSocket transport before + // reverse-direction data and its FIN have arrived. EXPECT_FALSE(fake_upstream_connection->waitForHalfClose(std::chrono::milliseconds(100))); // The fake WebSocket server sends its final data and CLOSE, then ends the transport as @@ -454,13 +584,15 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamFlush) { tcp_client->readDisable(true); ASSERT_TRUE(tcp_client->write("", true)); - // Waiting for the encoded CLOSE ensures that readDisable(true) has run on its thread before the - // peer starts writing. + // The downstream FIN is encoded as a masked CLOSE. Unlike a transport half-close, observing this + // frame does not prevent the WebSocket peer from sending a large response. ASSERT_TRUE( fake_upstream_connection->waitForData(expected_handshake.length() + 6, &received_data)); + received_data.erase(0, expected_handshake.length()); + auto frame_offset = unmaskData(received_data.data(), received_data.length(), OPCODE_CLOSE); + ASSERT_TRUE(frame_offset > 0); // writing data in one large chunk - ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x82\x7f\0\0\0\0\x03\x20\0\0", 10})); ASSERT_TRUE(fake_upstream_connection->write(data)); ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x88\0", 2})); @@ -485,7 +617,7 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamFlush) { EXPECT_GT(upstream_resumes, 0); } -// Test that an upstream flush works correctly (all data is flushed) +// Test that an upstream flush works correctly after the upstream direction has half-closed. TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlush) { // Keep the payload larger than the kernel socket read buffer so that an upstream flush is needed, // while leaving buffer-limit headroom for WebSocket framing and control traffic. @@ -511,28 +643,29 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlush) { ASSERT_TRUE(fake_upstream_connection->write(handshake_response)); ASSERT_TRUE(fake_upstream_connection->readDisable(true)); - ASSERT_TRUE(fake_upstream_connection->write("", true)); + ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x88\0", 2})); - // This ensures that fake_upstream_connection->readDisable has been run on - // it's thread before tcp_client starts writing. + // The peer's CLOSE becomes an upstream FIN, while the downstream-to-upstream direction remains + // available for the large request. tcp_client->waitForHalfClose(); ASSERT_TRUE(tcp_client->write(data, true, true, std::chrono::milliseconds(30000))); + test_server_->waitForGaugeEq("tcp.tcp_stats.upstream_flush_active", 1); + ASSERT_TRUE(fake_upstream_connection->readDisable(false)); size_t min_size = expected_handshake.length() + data.size() + 14 + 6; ASSERT_TRUE( fake_upstream_connection->waitForData(FakeRawConnection::waitForAtLeastBytes(min_size))); ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); + ASSERT_TRUE(fake_upstream_connection->write("", true)); ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); - tcp_client->waitForHalfClose(); test_server_->waitForGaugeEq("tcp.tcp_stats.upstream_flush_active", 0); EXPECT_EQ(test_server_->counter("tcp.tcp_stats.upstream_flush_total")->value(), 1); } -// Test that Envoy doesn't crash or assert when shutting down with an upstream -// flush active +// Test that Envoy doesn't crash or assert when shutting down with an upstream flush active. TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlushEnvoyExit) { // Keep the payload larger than the kernel socket read buffer so that an upstream flush is needed, // while leaving buffer-limit headroom for WebSocket framing and control traffic. @@ -558,16 +691,12 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlushEnvoyExit) { ASSERT_TRUE(fake_upstream_connection->write(handshake_response)); ASSERT_TRUE(fake_upstream_connection->readDisable(true)); - ASSERT_TRUE(fake_upstream_connection->write("", true)); - - // This ensures that fake_upstream_connection->readDisable has been run on - // it's thread before tcp_client starts writing. + ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x88\0", 2})); tcp_client->waitForHalfClose(); ASSERT_TRUE(tcp_client->write(data, true)); - // test_server_->waitForCounterGe("tcp.tcp_stats.upstream_flush_total", 1); - + test_server_->waitForGaugeEq("tcp.tcp_stats.upstream_flush_active", 1); test_server_.reset(); ASSERT_TRUE(fake_upstream_connection->close()); ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); diff --git a/tests/cilium_websocket_policy_integration_test.cc b/tests/cilium_websocket_policy_integration_test.cc index 84d120b7f..0fdddb72e 100644 --- a/tests/cilium_websocket_policy_integration_test.cc +++ b/tests/cilium_websocket_policy_integration_test.cc @@ -278,9 +278,12 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamWritesFirst) { ASSERT_TRUE(fake_upstream_connection->waitForData(5, &received)); ASSERT_EQ(received, "hello"); - ASSERT_TRUE(fake_upstream_connection->write("", true)); + ASSERT_TRUE(fake_upstream_connection->write("upstream final", true)); + tcp_client->waitForData("helloupstream final"); tcp_client->waitForHalfClose(); - ASSERT_TRUE(tcp_client->write("", true)); + ASSERT_TRUE(tcp_client->write("downstream final", true)); + ASSERT_TRUE(fake_upstream_connection->waitForData(5 + sizeof("downstream final") - 1, &received)); + ASSERT_EQ(received, "hellodownstream final"); ASSERT_TRUE(fake_upstream_connection->waitForHalfClose()); ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); } From 01d7864ee2393a073eb1d965133cbaea37cbdad2 Mon Sep 17 00:00:00 2001 From: Jarno Rajahalme Date: Tue, 25 Aug 2026 08:23:03 +0200 Subject: [PATCH 9/9] websocket: Tighten frame decoding to RFC 6455 requirements Reject WebSocket frames that violate RFC 6455 instead of accepting them or treating malformed input as an orderly tunneled TCP shutdown. Require every client-to-server frame to be masked and every server-to-client frame to be unmasked. Require control frames to be final and limit their payloads to 125 bytes. Validate CLOSE payload structure by rejecting one-byte payloads and requiring the optional reason following the status-code bytes to be valid UTF-8. Keep the status-code bytes opaque so the codec remains compatible with future protocol revisions. Distinguish a valid WebSocket CLOSE from termination of the underlying transport. Treat a transport FIN received without CLOSE, as well as other frame protocol errors, as an abort that closes both tunnel directions immediately. Add integration coverage showing that unmasked client frames, malformed CLOSE payloads, and transport termination without CLOSE are rejected. Signed-off-by: Jarno Rajahalme --- cilium/BUILD | 1 + cilium/websocket_codec.cc | 37 +++++++++++ cilium/websocket_codec.h | 1 + cilium/websocket_protocol.h | 2 +- ...cilium_websocket_decap_integration_test.cc | 66 +++++++++++++++++++ ...cilium_websocket_encap_integration_test.cc | 22 +++++++ 6 files changed, 128 insertions(+), 1 deletion(-) diff --git a/cilium/BUILD b/cilium/BUILD index db2e7c34e..ef1888b16 100644 --- a/cilium/BUILD +++ b/cilium/BUILD @@ -160,6 +160,7 @@ envoy_cc_library( "//cilium:accesslog_lib", "//cilium:filter_state_lib", "//cilium/api:websocket_cc_proto", + "@com_google_protobuf//third_party/utf8_range:utf8_validity", "@envoy//bazel/external/http_parser", "@envoy//envoy/common/crypto:crypto_interface", "@envoy//source/common/common:base64_lib", diff --git a/cilium/websocket_codec.cc b/cilium/websocket_codec.cc index 38c567e58..406b7d220 100644 --- a/cilium/websocket_codec.cc +++ b/cilium/websocket_codec.cc @@ -37,6 +37,7 @@ #include "absl/strings/string_view.h" #include "cilium/websocket_config.h" #include "cilium/websocket_protocol.h" +#include "third_party/utf8_range/utf8_validity.h" namespace Envoy { namespace Cilium { @@ -684,6 +685,12 @@ void Codec::decode(Buffer::Instance& data, bool end_stream) { // Handshake done, process data. decoder_.decode(data, end_stream); + if (decoder_.protocol_error_) { + config->stats_.protocol_error_.inc(); + decoder_.drain(); + return closeOnError("invalid WebSocket control frame"); + } + if (end_stream && !decoder_.close_received_) { decoder_.drain(); return closeOnError("websocket transport closed without CLOSE"); @@ -867,6 +874,7 @@ void Codec::Decoder::decode(Buffer::Instance& data, bool end_stream) { TRY_READ_NETWORK(&frame_header); const uint8_t opcode = frame_header[0] & OPCODE_MASK; + const bool final_frame = (frame_header[0] & FIN_MASK) != 0; const bool masked = (frame_header[1] & MASK_MASK) != 0; uint64_t payload_len = frame_header[1] & PAYLOAD_LEN_MASK; @@ -888,6 +896,14 @@ void Codec::Decoder::decode(Buffer::Instance& data, bool end_stream) { // Whole header received and decoded // + // RFC 6455 section 5.1 requires client to mask all frames it sends to the server, + // and prohibits server from ever masking any frames is sends to the client. + const bool expected_masked = !parent_.config()->client_; + if (masked != expected_masked) { + ENVOY_LOG(debug, "websocket decoder: invalid frame masking"); + goto protocol_error; + } + if (opcode < OPCODE_CLOSE) { // Unframe and forward all non-control frames ENVOY_LOG(trace, "websocket decoder: received websocket data: header {} bytes, data {} bytes", @@ -899,6 +915,11 @@ void Codec::Decoder::decode(Buffer::Instance& data, bool end_stream) { // Terminate and respond to any control frames + // control frames are always final + if (!final_frame) { + ENVOY_LOG(debug, "websocket decoder: invalid control frame"); + goto protocol_error; + } // Protect against too large control frames that could happen if the decoder ever loses // sync with the data stream. @@ -928,6 +949,21 @@ void Codec::Decoder::decode(Buffer::Instance& data, bool end_stream) { // validate CLOSE payload if any if (payload_len > 0) { + if (payload_len == 1) { + ENVOY_LOG(debug, "websocket decoder: invalid CLOSE payload"); + goto protocol_error; + } + + // we do not interpret the status code to remain compatible with future revisions of the + // WebSocket protocol specification, but we validate that any reason is UTF-8 encoded as + // required. + const absl::string_view reason{reinterpret_cast(payload + 2), + static_cast(payload_len - 2)}; + if (!utf8_range::IsStructurallyValid(reason)) { + ENVOY_LOG(debug, "websocket decoder: invalid CLOSE reason"); + goto protocol_error; + } + // store for sending the frame back if (!close_received_) { close_payload_.assign(reinterpret_cast(payload), payload_len); @@ -957,6 +993,7 @@ void Codec::Decoder::decode(Buffer::Instance& data, bool end_stream) { protocol_error: buffer_.drain(buffer_.length()); + protocol_error_ = true; } } // namespace WebSocket diff --git a/cilium/websocket_codec.h b/cilium/websocket_codec.h index 3e8f09e4d..5739910eb 100644 --- a/cilium/websocket_codec.h +++ b/cilium/websocket_codec.h @@ -76,6 +76,7 @@ class Codec : Logger::Loggable { Codec& parent_; bool close_received_{false}; + bool protocol_error_{false}; std::string close_payload_; Buffer::OwnedImpl buffer_; // Buffer for partial websocket frames Buffer::OwnedImpl decoded_; // Buffer for decoded websocket frames diff --git a/cilium/websocket_protocol.h b/cilium/websocket_protocol.h index baf9af5fe..d5b4cbea3 100644 --- a/cilium/websocket_protocol.h +++ b/cilium/websocket_protocol.h @@ -4,7 +4,7 @@ // Some sensible limits to protect against excess resource use #define WEBSOCKET_HANDSHAKE_MAX_SIZE 4096 -#define WEBSOCKET_CONTROL_FRAME_MAX_SIZE 256 +#define WEBSOCKET_CONTROL_FRAME_MAX_SIZE 125 // RFC 6455 ยง5.5 /* Ref. RFC 6455 */ diff --git a/tests/cilium_websocket_decap_integration_test.cc b/tests/cilium_websocket_decap_integration_test.cc index 44158f648..d02f7adc3 100644 --- a/tests/cilium_websocket_decap_integration_test.cc +++ b/tests/cilium_websocket_decap_integration_test.cc @@ -388,6 +388,39 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) { codec_client_->close(); } +TEST_P(CiliumWebSocketIntegrationTest, UnmaskedClientFrameRejected) { + initialize(); + auto request_headers = Http::TestRequestHeaderMapImpl{ + {":method", "GET"}, + {":path", "/"}, + {":authority", "host"}, + {"Upgrade", "websocket"}, + {"Connection", "Upgrade"}, + {"Origin", "jarno.cilium.rocks"}, + {"Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ=="}, + {"Sec-WebSocket-Version", "13"}, + {"x-request-id", "000000ff-0000-0000-0000-000000000001"}, + {"x-envoy-original-dst-host", original_dst_address->asString()}}; + codec_client_ = makeHttpConnection(lookupPort("http")); + + IntegrationStreamDecoderPtr response = codec_client_->makeHeaderOnlyRequest(request_headers); + FakeRawConnectionPtr fake_upstream_connection; + ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection)); + response->waitForHeaders(); + ASSERT_EQ("101", response->headers().getStatusValue()); + + // RFC 6455 section 5.1 requires every client-to-server frame to be masked. + const uint8_t unmasked_frame[] = {0x82, 0x05, 'h', 'e', 'l', 'l', 'o'}; + Buffer::OwnedImpl frame_buffer(unmasked_frame, sizeof(unmasked_frame)); + auto* client_connection = codec_client_->connection(); + client_connection->write(frame_buffer, false); + client_connection->dispatcher().run(Event::Dispatcher::RunType::NonBlock); + + test_server_->waitForCounterGe("websocket.protocol_error", 1); + ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); + ASSERT_TRUE(codec_client_->waitForDisconnect()); +} + TEST_P(CiliumWebSocketIntegrationTest, CloseResponseWaitsForReverseFin) { enableHalfClose(true); initialize(); @@ -434,4 +467,37 @@ TEST_P(CiliumWebSocketIntegrationTest, CloseResponseWaitsForReverseFin) { codec_client_->close(); } +TEST_P(CiliumWebSocketIntegrationTest, InvalidCloseAbortsImmediately) { + initialize(); + auto request_headers = Http::TestRequestHeaderMapImpl{ + {":method", "GET"}, + {":path", "/"}, + {":authority", "host"}, + {"Upgrade", "websocket"}, + {"Connection", "Upgrade"}, + {"Origin", "jarno.cilium.rocks"}, + {"Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ=="}, + {"Sec-WebSocket-Version", "13"}, + {"x-request-id", "000000ff-0000-0000-0000-000000000001"}, + {"x-envoy-original-dst-host", original_dst_address->asString()}}; + codec_client_ = makeHttpConnection(lookupPort("http")); + + IntegrationStreamDecoderPtr response = codec_client_->makeHeaderOnlyRequest(request_headers); + FakeRawConnectionPtr fake_upstream_connection; + ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection)); + response->waitForHeaders(); + ASSERT_EQ("101", response->headers().getStatusValue()); + + // RFC 6455 forbids a CLOSE payload of exactly one byte. A protocol error aborts both TCP sides + // instead of waiting for the upstream FIN used by the normal tunnel half-close path. + const uint8_t invalid_masked_close[] = {0x88, 0x81, 0, 0, 0, 0, 0}; + Buffer::OwnedImpl close_buffer(invalid_masked_close, sizeof(invalid_masked_close)); + auto* client_connection = codec_client_->connection(); + client_connection->write(close_buffer, false); + client_connection->dispatcher().run(Event::Dispatcher::RunType::NonBlock); + + ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); + ASSERT_TRUE(codec_client_->waitForDisconnect()); +} + } // namespace Envoy diff --git a/tests/cilium_websocket_encap_integration_test.cc b/tests/cilium_websocket_encap_integration_test.cc index 7a7371778..b0839f7e9 100644 --- a/tests/cilium_websocket_encap_integration_test.cc +++ b/tests/cilium_websocket_encap_integration_test.cc @@ -432,6 +432,28 @@ TEST_P(CiliumWebSocketIntegrationTest, ControlFramesAfterReceivingClose) { ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); } +TEST_P(CiliumWebSocketIntegrationTest, WebSocketTransportFinWithoutCloseAborts) { + initialize(); + IntegrationTcpClientPtr tcp_client = makeTcpConnection(lookupPort("tcp_proxy")); + FakeRawConnectionPtr fake_upstream_connection; + ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection)); + + std::string expected_handshake = + fmt::format(fmt::runtime(EXPECTED_HANDSHAKE_FMT), original_dst_address->asString()); + ASSERT_TRUE(fake_upstream_connection->waitForData(expected_handshake.length())); + + std::string handshake_response = + fmt::format(fmt::runtime(HANDSHAKE_RESPONSE_FMT), "GjgmQ9MzNsn3h7+vuIzY25rbQ9M="); + ASSERT_TRUE(fake_upstream_connection->write(handshake_response)); + + // An outer transport FIN without a WebSocket CLOSE is an abort, not a tunneled TCP FIN. + ASSERT_TRUE(fake_upstream_connection->write("", true)); + // The raw downstream observes EOF under half-close even though the WebSocket transport was + // aborted immediately. + tcp_client->waitForHalfClose(); + ASSERT_TRUE(fake_upstream_connection->waitForDisconnect()); + tcp_client->close(); +} // Test proxying data in both directions, and that all data is flushed properly // when the client disconnects.