Fix Autobahn compliance while preserving close lifecycle guarantees

ixwebsocket/IXWebSocketHandshake.cpp

@@ -386,7 +386,7 @@ namespace ix
         {
             _enablePerMessageDeflate = true;
 
-            if (!_perMessageDeflate->init(webSocketPerMessageDeflateOptions))
+            if (!_perMessageDeflate->init(webSocketPerMessageDeflateOptions, true))
             {
                 return WebSocketInitResult(
                     false, 0, "Failed to initialize per message deflate engine");

ixwebsocket/IXWebSocketPerMessageDeflate.cpp

@@ -68,16 +68,25 @@ namespace ix
         ;
     }
 
-    bool WebSocketPerMessageDeflate::init(
-        const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions)
+    bool WebSocketPerMessageDeflate::init(const WebSocketPerMessageDeflateOptions& options)
     {
-        bool clientNoContextTakeover = perMessageDeflateOptions.getClientNoContextTakeover();
+        return init(options, false);
+    }
 
-        uint8_t deflateBits = perMessageDeflateOptions.getClientMaxWindowBits();
-        uint8_t inflateBits = perMessageDeflateOptions.getServerMaxWindowBits();
+    bool WebSocketPerMessageDeflate::init(const WebSocketPerMessageDeflateOptions& options,
+                                          bool isServer)
+    {
+        const uint8_t deflateBits =
+            isServer ? options.getServerMaxWindowBits() : options.getClientMaxWindowBits();
+        const uint8_t inflateBits =
+            isServer ? options.getClientMaxWindowBits() : options.getServerMaxWindowBits();
+        const bool noContextTakeover =
+            isServer ? options.getServerNoContextTakeover() : options.getClientNoContextTakeover();
 
-        return _compressor->init(deflateBits, clientNoContextTakeover) &&
-               _decompressor->init(inflateBits, clientNoContextTakeover);
+        const bool peerNoContextTakeover =
+            isServer ? options.getClientNoContextTakeover() : options.getServerNoContextTakeover();
+        return _compressor->init(deflateBits, noContextTakeover) &&
+               _decompressor->init(inflateBits, peerNoContextTakeover);
     }
 
     bool WebSocketPerMessageDeflate::compress(const IXWebSocketSendData& in, std::string& out)

ixwebsocket/IXWebSocketPerMessageDeflate.h

@@ -51,6 +51,7 @@ namespace ix
         ~WebSocketPerMessageDeflate();
 
         bool init(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions);
+        bool init(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions, bool isServer);
         bool compress(const IXWebSocketSendData& in, std::string& out);
         bool compress(const std::string& in, std::string& out);
         bool decompress(const std::string& in, std::string& out);

ixwebsocket/IXWebSocketTransport.cpp

@@ -41,6 +41,7 @@
 #include "IXUtf8Validator.h"
 #include "IXWebSocketHandshake.h"
 #include "IXWebSocketHttpHeaders.h"
+#include <algorithm>
 #include <chrono>
 #include <cstdarg>
 #include <cstdlib>
@@ -420,6 +421,8 @@ namespace ix
         _rxbuf.clear();
         _rxbufWanted = 0;
         _chunks.clear();
+        _utf8Validator.reset();
+        _validatedBytes = 0;
         _fragmentedMessageKind = MessageKind::MSG_TEXT;
         _receivedMessageCompressed = false;
     }
@@ -462,6 +465,28 @@ namespace ix
         }
     }
 
+    bool WebSocketTransport::validateFramePayload(const wsheader_type& ws)
+    {
+        if (_readyState != ReadyState::OPEN) return true;
+        const bool text = ws.opcode == wsheader_type::TEXT_FRAME && _chunks.empty() &&
+                          !(_enablePerMessageDeflate && ws.rsv1);
+        const bool continuation = ws.opcode == wsheader_type::CONTINUATION && !_chunks.empty() &&
+                                  _fragmentedMessageKind == MessageKind::MSG_TEXT &&
+                                  !_receivedMessageCompressed;
+        if (!text && !continuation) return true;
+        if (text && _validatedBytes == 0) _utf8Validator.reset();
+
+        const auto available =
+            (std::min) (static_cast<uint64_t>(_rxbuf.size() - ws.header_size), ws.N);
+        for (; _validatedBytes < available; ++_validatedBytes)
+        {
+            uint8_t byte = _rxbuf[ws.header_size + _validatedBytes];
+            if (ws.mask) byte ^= ws.masking_key[_validatedBytes & 3];
+            if (!_utf8Validator.consume(byte)) return false;
+        }
+        return !ws.fin || available != ws.N || _utf8Validator.complete();
+    }
+
     //
     // http://tools.ietf.org/html/rfc6455#section-5.2  Base Framing Protocol
     //
@@ -572,13 +597,38 @@ namespace ix
                 break;
             }
 
+            // Check available text bytes even when TCP has not delivered the whole frame.
+            if (!validateFramePayload(ws))
+            {
+                close(WebSocketCloseConstants::kInvalidFramePayloadData,
+                      WebSocketCloseConstants::kInvalidFramePayloadDataMessage);
+                break;
+            }
+
             if (_rxbuf.size() < ws.header_size + ws.N)
             {
                 _rxbufWanted = ws.header_size + ws.N;
                 break; /* Need: ws.header_size+ws.N - _rxbuf.size() */
             }
 
             _rxbufWanted = 0;
+            _validatedBytes = 0;
+
+            // After sending Close, consume buffered frames only to find the peer's Close.
+            // In particular, never send a Pong or deliver more data while CLOSING (#620).
+            if (_readyState == ReadyState::CLOSING && ws.opcode != wsheader_type::CLOSE)
+            {
+                _rxbuf.erase(_rxbuf.begin(), _rxbuf.begin() + ws.header_size + (size_t) ws.N);
+                continue;
+            }
+
+            if (ws.opcode == wsheader_type::CLOSE && ws.N > 125)
+            {
+                close(WebSocketCloseConstants::kProtocolErrorCode,
+                      WebSocketCloseConstants::kProtocolErrorMessage);
+                _rxbuf.erase(_rxbuf.begin(), _rxbuf.begin() + ws.header_size + (size_t) ws.N);
+                continue;
+            }
 
             if (!ws.fin && (ws.opcode == wsheader_type::PING || ws.opcode == wsheader_type::PONG ||
                             ws.opcode == wsheader_type::CLOSE))
@@ -746,15 +796,10 @@ namespace ix
                 }
                 else
                 {
-                    // we got the CLOSE frame answer from our close, so we can close the connection
-                    // if the code/reason are the same
-                    bool identicalReason = _closeCode == code && getCloseReason() == reason;
-
-                    if (identicalReason)
-                    {
-                        bool remote = false;
-                        closeSocketAndSwitchToClosedState(code, reason, _rxbuf.size(), remote);
-                    }
+                    // A peer may reply with a different status/reason. Receiving its Close
+                    // completes the handshake; retain our original local close notification.
+                    closeSocketAndSwitchToClosedState(
+                        _closeCode, getCloseReason(), _rxbuf.size(), false);
                 }
             }
             else
@@ -839,15 +884,7 @@ namespace ix
         }
         else
         {
-            if (messageKind == MessageKind::MSG_TEXT && !validateUtf8(message))
-            {
-                close(WebSocketCloseConstants::kInvalidFramePayloadData,
-                      WebSocketCloseConstants::kInvalidFramePayloadDataMessage);
-            }
-            else
-            {
-                onMessageCallback(message, wireSize, false, messageKind);
-            }
+            onMessageCallback(message, wireSize, false, messageKind);
         }
     }
 

ixwebsocket/IXWebSocketTransport.h

@@ -13,6 +13,7 @@
 #include "IXCancellationRequest.h"
 #include "IXProgressCallback.h"
 #include "IXSocketTLSOptions.h"
+#include "IXUtf8Validator.h"
 #include "IXWebSocketCloseConstants.h"
 #include "IXWebSocketHandshake.h"
 #include "IXWebSocketHttpHeaders.h"
@@ -123,6 +124,8 @@ namespace ix
     private:
         friend struct WebSocketTransportTestAccess;
         std::string _url;
+        Utf8Validator _utf8Validator;
+        size_t _validatedBytes = 0;
 
         struct wsheader_type
         {
@@ -146,6 +149,8 @@ namespace ix
             uint8_t masking_key[4];
         };
 
+        bool validateFramePayload(const wsheader_type& ws);
+
         // Tells whether we should mask the data we send.
         // client should mask but server should not
         std::atomic<bool> _useMask;

test/IXWebSocketDispatchTest.cpp

@@ -321,3 +321,166 @@ TEST_CASE("Parsing errors after socket failure retain their protocol close code"
     CHECK(client.getReadyState() == Transport::ReadyState::CLOSED);
     CHECK(closes == 1);
 }
+
+namespace ix
+{
+    struct WebSocketTransportTestAccess
+    {
+        static void install(WebSocketTransport& transport, std::unique_ptr<Socket> socket)
+        {
+            transport._socket = std::move(socket);
+            transport._useMask = false;
+            transport.setReadyState(WebSocketTransport::ReadyState::OPEN);
+        }
+
+        static void receive(WebSocketTransport& transport, const std::string& bytes)
+        {
+            transport._rxbuf.insert(transport._rxbuf.end(), bytes.begin(), bytes.end());
+        }
+    };
+} // namespace ix
+
+namespace
+{
+    class RecordingSocket final : public ix::Socket
+    {
+    public:
+        std::ptrdiff_t send(char* buffer, size_t size) override
+        {
+            sent.append(buffer, size);
+            return static_cast<std::ptrdiff_t>(size);
+        }
+
+        void close() override
+        {
+        }
+        std::string sent;
+    };
+
+    void dispatchBytes(Transport& transport,
+                       const std::string& bytes,
+                       const Transport::OnMessageCallback& callback)
+    {
+        ix::WebSocketTransportTestAccess::receive(transport, bytes);
+        transport.dispatch(Transport::PollResult::Succeeded, callback);
+    }
+} // namespace
+
+TEST_CASE("Protocol closure never responds to buffered Pings", "[websocket][close][dispatch]")
+{
+    const unsigned char opcode = GENERATE(0x85, 0x86, 0x87, 0x8b, 0x8c, 0x8f);
+    Transport transport;
+    auto* socket = new RecordingSocket;
+    ix::WebSocketTransportTestAccess::install(transport, std::unique_ptr<ix::Socket>(socket));
+    unsigned callbacks = 0;
+    unsigned closes = 0;
+    transport.setOnCloseCallback(
+        [&](uint16_t code, const std::string&, size_t, bool remote)
+        {
+            CHECK(code == 1002);
+            CHECK_FALSE(remote);
+            ++closes;
+        });
+    const auto callback = [&](const std::string&, size_t, bool, Transport::MessageKind)
+    { ++callbacks; };
+    dispatchBytes(transport, frame(opcode, "") + frame(0x89, ""), callback);
+    REQUIRE(transport.getReadyState() == Transport::ReadyState::CLOSING);
+    REQUIRE_FALSE(socket->sent.empty());
+    const auto closeFrame = socket->sent;
+    CHECK(static_cast<unsigned char>(closeFrame[0]) == 0x88);
+    CHECK(closeFrame.size() == 2 + static_cast<unsigned char>(closeFrame[1]));
+    dispatchBytes(
+        transport, frame(0x89, "late") + frame(0x88, std::string("\x03\xe8", 2)), callback);
+    CHECK(socket->sent == closeFrame);
+    CHECK(callbacks == 0);
+    CHECK(closes == 1);
+    CHECK(transport.getReadyState() == Transport::ReadyState::CLOSED);
+}
+
+TEST_CASE("Fragmented UTF8 is checked before the final fragment", "[websocket][dispatch][utf8]")
+{
+    const bool invalid = GENERATE(false, true);
+    Transport transport;
+    auto* socket = new RecordingSocket;
+    ix::WebSocketTransportTestAccess::install(transport, std::unique_ptr<ix::Socket>(socket));
+    std::vector<std::string> messages;
+    const auto callback = [&](const std::string& value, size_t, bool, Transport::MessageKind kind)
+    {
+        if (kind == Transport::MessageKind::MSG_TEXT) messages.push_back(value);
+    };
+    dispatchBytes(transport, frame(0x01, std::string("\xf0\x9f", 2)), callback);
+    CHECK(transport.getReadyState() == Transport::ReadyState::OPEN);
+    dispatchBytes(transport, frame(0x89, "ping"), callback);
+    dispatchBytes(transport, frame(0x00, invalid ? "x" : std::string("\x98\x80", 2)), callback);
+    if (invalid)
+    {
+        CHECK(transport.getReadyState() == Transport::ReadyState::CLOSING);
+        CHECK(messages.empty());
+    }
+    else
+    {
+        REQUIRE(transport.getReadyState() == Transport::ReadyState::OPEN);
+        dispatchBytes(transport, frame(0x80, "!") + frame(0x81, "next"), callback);
+        CHECK(messages == std::vector<std::string> {std::string("\xf0\x9f\x98\x80!", 5), "next"});
+    }
+}
+
+TEST_CASE("Oversized close frames receive a protocol error", "[websocket][close][dispatch]")
+{
+    Transport transport;
+    auto* socket = new RecordingSocket;
+    ix::WebSocketTransportTestAccess::install(transport, std::unique_ptr<ix::Socket>(socket));
+    const auto callback = [](const std::string&, size_t, bool, Transport::MessageKind)
+    { FAIL("Close must not deliver data"); };
+    dispatchBytes(
+        transport, std::string("\x88\x7e\x00\x7e\x03\xe8", 6) + std::string(124, 'x'), callback);
+    REQUIRE(socket->sent.size() >= 4);
+    CHECK(socket->sent.substr(2, 2) == std::string("\x03\xea", 2));
+    CHECK(static_cast<unsigned char>(socket->sent[1]) <= 125);
+    CHECK(transport.getReadyState() == Transport::ReadyState::CLOSING);
+}
+
+TEST_CASE("UTF8 validation handles partial masked and unmasked frames",
+          "[websocket][dispatch][utf8]")
+{
+    const bool masked = GENERATE(false, true);
+    const bool invalid = GENERATE(false, true);
+    Transport transport;
+    auto* socket = new RecordingSocket;
+    ix::WebSocketTransportTestAccess::install(transport, std::unique_ptr<ix::Socket>(socket));
+    std::vector<std::string> messages;
+    const auto callback = [&](const std::string& value, size_t, bool, Transport::MessageKind kind)
+    {
+        if (kind == Transport::MessageKind::MSG_TEXT) messages.push_back(value);
+    };
+    const std::string payload =
+        invalid ? std::string("\xf4\x90\x80\x80", 4) : std::string("\xf0\x9f\x98\x80", 4);
+    std::string bytes = frame(0x81, payload);
+    const std::string mask("\x11\x22\x33\x44", 4);
+    if (masked)
+    {
+        bytes[1] |= static_cast<char>(0x80);
+        bytes.insert(2, mask);
+        for (size_t i = 0; i < payload.size(); ++i)
+            bytes[6 + i] ^= mask[i & 3];
+    }
+    const auto headerSize = masked ? 6 : 2;
+    dispatchBytes(transport, bytes.substr(0, headerSize + 1), callback);
+    CHECK(transport.getReadyState() == Transport::ReadyState::OPEN);
+    // Re-dispatch without receiving bytes must not validate the prefix a second time.
+    dispatchBytes(transport, "", callback);
+    dispatchBytes(transport, bytes.substr(headerSize + 1, 1), callback);
+    if (invalid)
+    {
+        CHECK(transport.getReadyState() == Transport::ReadyState::CLOSING);
+        REQUIRE(socket->sent.size() >= 4);
+        CHECK(socket->sent.substr(2, 2) == std::string("\x03\xef", 2));
+        CHECK(messages.empty());
+    }
+    else
+    {
+        REQUIRE(transport.getReadyState() == Transport::ReadyState::OPEN);
+        dispatchBytes(transport, bytes.substr(headerSize + 2), callback);
+        CHECK(messages == std::vector<std::string> {payload});
+    }
+}