From 6c9ac88f50006e1dc649b01866fb17e79f71059c Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Thu, 29 Nov 2018 08:34:37 +0100 Subject: [PATCH] Add basic permessage-deflate, pass entire Autobahn --- README.md | 2 +- misc/15.pro | 6 +- misc/main.cpp | 3 + src/App.h | 43 ++++++++-- src/LoopData.h | 5 ++ src/PerMessageDeflate.h | 85 ++++++++++++++++++ src/WebSocket.h | 4 +- src/WebSocketContext.h | 55 ++++++++++-- src/WebSocketData.h | 10 ++- src/WebSocketExtensions.h | 176 ++++++++++++++++++++++++++++++++++++++ src/WebSocketProtocol.h | 2 +- 11 files changed, 372 insertions(+), 19 deletions(-) create mode 100644 src/PerMessageDeflate.h create mode 100644 src/WebSocketExtensions.h diff --git a/README.md b/README.md index 4c1915a..8baca76 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ #### Tread lightly. -This is the **development branch** of v0.15. Track progress & issues [here](https://github.com/uNetworking/issues/issues). Checkout [v0.14](https://github.com/uNetworking/uWebSockets/tree/v0.14) for something stable. +This is the **completely broken development branch** of v0.15. Track progress & issues [here](https://github.com/uNetworking/issues/issues). Checkout [v0.14](https://github.com/uNetworking/uWebSockets/tree/v0.14) for something stable. #### Express yourself briefly. ```c++ diff --git a/misc/15.pro b/misc/15.pro index 19d3f1f..97bec54 100644 --- a/misc/15.pro +++ b/misc/15.pro @@ -31,8 +31,10 @@ HEADERS += \ ../src/WebSocket.h \ ../src/WebSocketData.h \ ../src/WebSocketContext.h \ - ../src/WebSocketContextData.h + ../src/WebSocketContextData.h \ + ../src/WebSocketExtensions.h \ + ../src/PerMessageDeflate.h INCLUDEPATH += ../uSockets/src ../src QMAKE_CXXFLAGS += -fsanitize=address -LIBS += -lasan -pthread -lssl -lcrypto -lstdc++fs +LIBS += -lasan -pthread -lssl -lcrypto -lz -lstdc++fs diff --git a/misc/main.cpp b/misc/main.cpp index ba428e8..7b50107 100644 --- a/misc/main.cpp +++ b/misc/main.cpp @@ -12,6 +12,9 @@ int main(int argc, char **argv) { uWS::App().get("/hello", [](auto *res, auto *req) { res->end("Hello HTTP!"); }).ws("/*", { + /*.compression = */true, + /*.maxPayloadLength*/ + /*.open = */[](auto *ws, auto *req) { }, diff --git a/src/App.h b/src/App.h index 6b1ee90..648e054 100644 --- a/src/App.h +++ b/src/App.h @@ -24,6 +24,7 @@ #include "HttpResponse.h" #include "WebSocketContext.h" #include "WebSocket.h" +#include "WebSocketExtensions.h" #include "libwshandshake.hpp" @@ -49,6 +50,7 @@ public: } struct WebSocketBehavior { + bool compression = false; std::function *, HttpRequest *)> open = nullptr; std::function *, std::string_view, uWS::OpCode)> message = nullptr; }; @@ -58,6 +60,15 @@ public: /* Every route has its own websocket context with its own behavior and user data type */ auto *webSocketContext = WebSocketContext::create(Loop::defaultLoop(), (typename StaticDispatch::SOCKET_CONTEXT_TYPE *) httpContext); + /* If we are the first one to use compression, initialize it */ + if (behavior.compression) { + LoopData *loopData = (LoopData *) us_loop_ext(us_socket_context_loop(webSocketContext->getSocketContext())); + + if (!loopData->inflationStream) { + loopData->inflationStream = new InflationStream; + } + } + /* Copy all handlers */ webSocketContext->getExt()->messageHandler = behavior.message; @@ -65,9 +76,6 @@ public: /* If we have this header set, it's a websocket */ std::string_view secWebSocketKey = req->getHeader("sec-websocket-key"); if (secWebSocketKey.length()) { - - // todo: negotiate extensions such as compression here and pass autobahn fully - // note: OpenSSL can be used here to speed this up somewhat char secWebSocketAccept[29] = {}; WebSocketHandshake::generate(secWebSocketKey.data(), secWebSocketAccept); @@ -75,15 +83,38 @@ public: res->writeStatus("101 Switching Protocols") ->writeHeader("Upgrade", "websocket") ->writeHeader("Connection", "Upgrade") - ->writeHeader("Sec-WebSocket-Accept", secWebSocketAccept) - ->end(); + ->writeHeader("Sec-WebSocket-Accept", secWebSocketAccept); + + /* Negotiate compression */ + bool perMessageDeflate = false; + if (behavior.compression) { + std::string_view extensions = req->getHeader("sec-websocket-extensions"); + if (extensions.length()) { + // basically: parse(options, extensions) + ExtensionsNegotiator extensionsNegotiator(PERMESSAGE_DEFLATE | CLIENT_NO_CONTEXT_TAKEOVER); // take options + extensionsNegotiator.readOffer(extensions); + + //std::cout << extensions << " => " << extensionsNegotiator.generateOffer() << std::endl; + + /* Todo: remove these mid string copies */ + res->writeHeader("Sec-WebSocket-Extensions", extensionsNegotiator.generateOffer()); + + /* Did we negotiate permessage-deflate? */ + if (extensionsNegotiator.getNegotiatedOptions() & PERMESSAGE_DEFLATE) { + perMessageDeflate = true; + } + } + } + + /* Add mark, we don't want to end anything */ + res->writeHeader("WebSocket-Server", "uWebSockets")->end(); /* Adopting a socket invalidates it, do not rely on it directly to carry any data */ WebSocket *webSocket = (WebSocket *) StaticDispatch::static_dispatch(us_ssl_socket_context_adopt_socket, us_socket_context_adopt_socket)( (typename StaticDispatch::SOCKET_CONTEXT_TYPE *) webSocketContext, (typename StaticDispatch::SOCKET_TYPE *) res, /*sizeof(WebSocketData)*/ 150); httpContext->upgradeToWebSocket( - webSocket->init() + webSocket->init(perMessageDeflate) ); if (behavior.open) { diff --git a/src/LoopData.h b/src/LoopData.h index 443bc73..4ed035b 100644 --- a/src/LoopData.h +++ b/src/LoopData.h @@ -22,6 +22,8 @@ #include #include +#include "PerMessageDeflate.h" + namespace uWS { struct Loop; @@ -43,6 +45,9 @@ public: char *corkBuffer = new char[CORK_BUFFER_SIZE]; int corkOffset = 0; bool corked = false; + + /* Compression data */ + InflationStream *inflationStream = nullptr; }; } diff --git a/src/PerMessageDeflate.h b/src/PerMessageDeflate.h new file mode 100644 index 0000000..ccb929f --- /dev/null +++ b/src/PerMessageDeflate.h @@ -0,0 +1,85 @@ +/* + * Copyright 2018 Alex Hultman and contributors. + + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + + * http://www.apache.org/licenses/LICENSE-2.0 + + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// inflationStream? Ciompression + +#ifndef PERMESSAGEDEFLATE_H +#define PERMESSAGEDEFLATE_H +/* Do not compile this module if we don't want it */ +#ifndef UWS_NO_ZLIB + +#include + +#include +#include + +#define LARGE_BUFFER_SIZE 16000 // fix this + +// we also need DeflationStream + +// the loop holds one of these +struct InflationStream { + + std::string dynamicZlibBuffer; + z_stream inflationStream = {}; + char *zlibBuffer; + + InflationStream() { + std::cout << "Initliazing shared compression" << std::endl; + zlibBuffer = (char *) malloc(LARGE_BUFFER_SIZE); + + inflateInit2(&inflationStream, -15); + } + + std::string_view inflate(std::string_view compressed) { + + int maxPayload = 160000; // todo: fix this + + dynamicZlibBuffer.clear(); + + inflationStream.next_in = (Bytef *) compressed.data(); + inflationStream.avail_in = (unsigned int) compressed.length(); + + int err; + do { + inflationStream.next_out = (Bytef *) zlibBuffer; + inflationStream.avail_out = LARGE_BUFFER_SIZE; + err = ::inflate(&inflationStream, Z_FINISH); + if (!inflationStream.avail_in) { + break; + } + + dynamicZlibBuffer.append(zlibBuffer, LARGE_BUFFER_SIZE - inflationStream.avail_out); + } while (err == Z_BUF_ERROR && dynamicZlibBuffer.length() <= maxPayload); + + inflateReset(&inflationStream); + + if ((err != Z_BUF_ERROR && err != Z_OK) || dynamicZlibBuffer.length() > maxPayload) { + return {nullptr, 0}; + } + + if (dynamicZlibBuffer.length()) { + dynamicZlibBuffer.append(zlibBuffer, LARGE_BUFFER_SIZE - inflationStream.avail_out); + return {dynamicZlibBuffer.data(), dynamicZlibBuffer.length()}; + } + + return {zlibBuffer, LARGE_BUFFER_SIZE - inflationStream.avail_out}; + } + +}; + +#endif +#endif // PERMESSAGEDEFLATE_H diff --git a/src/WebSocket.h b/src/WebSocket.h index 36afea0..9802115 100644 --- a/src/WebSocket.h +++ b/src/WebSocket.h @@ -31,8 +31,8 @@ struct WebSocket : AsyncSocket { private: typedef AsyncSocket Super; - void *init() { - new (us_socket_ext((us_socket *) this)) WebSocketData; + void *init(bool perMessageDeflate) { + new (us_socket_ext((us_socket *) this)) WebSocketData(perMessageDeflate); return this; } public: diff --git a/src/WebSocketContext.h b/src/WebSocketContext.h index 2a22f7f..dc2c70f 100644 --- a/src/WebSocketContext.h +++ b/src/WebSocketContext.h @@ -46,8 +46,16 @@ private: return (WebSocketContextData *) us_socket_context_ext((SOCKET_CONTEXT_TYPE *) this); } - static bool setCompressed(uWS::WebSocketState *wState) { - return false; // do not support it + /* If we have negotiated compression, set this frame compressed */ + static bool setCompressed(uWS::WebSocketState *wState, void *s) { + WebSocketData *webSocketData = (WebSocketData *) us_socket_ext((us_socket *) s); + + if (webSocketData->compressionStatus == WebSocketData::CompressionStatus::ENABLED) { + webSocketData->compressionStatus = WebSocketData::CompressionStatus::COMPRESSED_FRAME; + return true; + } else { + return false; + } } static void forceClose(uWS::WebSocketState *wState, void *s) { @@ -65,6 +73,22 @@ private: /* Did we get everything in one go? */ if (!remainingBytes && fin && !webSocketData->fragmentBuffer.length()) { + /* Handle compressed frame */ + if (webSocketData->compressionStatus == WebSocketData::CompressionStatus::COMPRESSED_FRAME) { + webSocketData->compressionStatus = WebSocketData::CompressionStatus::ENABLED; + + LoopData *loopData = (LoopData *) us_loop_ext(us_socket_context_loop(us_socket_get_context((us_socket *) s))); + + std::string_view inflatedFrame = loopData->inflationStream->inflate({data, length}); + if (!inflatedFrame.length()) { + forceClose(webSocketState, s); + return true; + } else { + data = (char *) inflatedFrame.data(); + length = inflatedFrame.length(); + } + } + /* Check text messages for Utf-8 validity */ if (opCode == 1 && !WebSocketProtocol>::isValidUtf8((unsigned char *) data, length)) { forceClose(webSocketState, s); @@ -87,9 +111,30 @@ private: // what if we don't have any remaining bytes yet we are not fin? forceclose! if (!remainingBytes && fin) { - // reset length and data ptrs - length = webSocketData->fragmentBuffer.length(); - data = webSocketData->fragmentBuffer.data(); + /* Handle compression */ + if (webSocketData->compressionStatus == WebSocketData::CompressionStatus::COMPRESSED_FRAME) { + webSocketData->compressionStatus = WebSocketData::CompressionStatus::ENABLED; + + // what's really the story here? + webSocketData->fragmentBuffer.append("...."); + + LoopData *loopData = (LoopData *) us_loop_ext(us_socket_context_loop(us_socket_get_context((us_socket *) s))); + + std::string_view inflatedFrame = loopData->inflationStream->inflate({webSocketData->fragmentBuffer.data(), webSocketData->fragmentBuffer.length() - 4}); + if (!inflatedFrame.length()) { + forceClose(webSocketState, s); + return true; + } else { + data = (char *) inflatedFrame.data(); + length = inflatedFrame.length(); + } + + + } else { + // reset length and data ptrs + length = webSocketData->fragmentBuffer.length(); + data = webSocketData->fragmentBuffer.data(); + } /* Check text messages for Utf-8 validity */ if (opCode == 1 && !WebSocketProtocol>::isValidUtf8((unsigned char *) data, length)) { diff --git a/src/WebSocketData.h b/src/WebSocketData.h index 2231463..4fb8c20 100644 --- a/src/WebSocketData.h +++ b/src/WebSocketData.h @@ -32,9 +32,15 @@ private: std::string fragmentBuffer; int controlTipLength = 0; bool isShuttingDown = 0; + enum CompressionStatus : char { + DISABLED, + ENABLED, + COMPRESSED_FRAME + } compressionStatus; public: - WebSocketData() : WebSocketState() { - //std::cout << "init websocket data!" << std::endl; + WebSocketData(bool perMessageDeflate) : WebSocketState() { + std::cout << "perMessageDeflate: " << perMessageDeflate << std::endl; + compressionStatus = perMessageDeflate ? ENABLED : DISABLED; } }; diff --git a/src/WebSocketExtensions.h b/src/WebSocketExtensions.h new file mode 100644 index 0000000..43afe35 --- /dev/null +++ b/src/WebSocketExtensions.h @@ -0,0 +1,176 @@ +/* + * Copyright 2018 Alex Hultman and contributors. + + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + + * http://www.apache.org/licenses/LICENSE-2.0 + + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef WEBSOCKETEXTENSIONS_H +#define WEBSOCKETEXTENSIONS_H + +#include + +namespace uWS { + +enum Options : unsigned int { + NO_OPTIONS = 0, + PERMESSAGE_DEFLATE = 1, + SERVER_NO_CONTEXT_TAKEOVER = 2, // remove this + CLIENT_NO_CONTEXT_TAKEOVER = 4, // remove this + NO_DELAY = 8, + SLIDING_DEFLATE_WINDOW = 16 +}; + +template +class ExtensionsNegotiator { +protected: + int options; +public: + ExtensionsNegotiator(int wantedOptions); + std::string generateOffer(); + void readOffer(std::string_view offer); + int getNegotiatedOptions(); +}; + +enum ExtensionTokens { + TOK_PERMESSAGE_DEFLATE = 1838, + TOK_SERVER_NO_CONTEXT_TAKEOVER = 2807, + TOK_CLIENT_NO_CONTEXT_TAKEOVER = 2783, + TOK_SERVER_MAX_WINDOW_BITS = 2372, + TOK_CLIENT_MAX_WINDOW_BITS = 2348 +}; + +class ExtensionsParser { +private: + int *lastInteger = nullptr; + +public: + bool perMessageDeflate = false; + bool serverNoContextTakeover = false; + bool clientNoContextTakeover = false; + int serverMaxWindowBits = 0; + int clientMaxWindowBits = 0; + + int getToken(const char *&in, const char *stop); + ExtensionsParser(const char *data, size_t length); +}; + +int ExtensionsParser::getToken(const char *&in, const char *stop) { + while (!isalnum(*in) && in != stop) { + in++; + } + + int hashedToken = 0; + while (isalnum(*in) || *in == '-' || *in == '_') { + if (isdigit(*in)) { + hashedToken = hashedToken * 10 - (*in - '0'); + } else { + hashedToken += *in; + } + in++; + } + return hashedToken; +} + +ExtensionsParser::ExtensionsParser(const char *data, size_t length) { + const char *stop = data + length; + int token = 1; + for (; token && token != TOK_PERMESSAGE_DEFLATE; token = getToken(data, stop)); + + perMessageDeflate = (token == TOK_PERMESSAGE_DEFLATE); + while ((token = getToken(data, stop))) { + switch (token) { + case TOK_PERMESSAGE_DEFLATE: + return; + case TOK_SERVER_NO_CONTEXT_TAKEOVER: + serverNoContextTakeover = true; + break; + case TOK_CLIENT_NO_CONTEXT_TAKEOVER: + clientNoContextTakeover = true; + break; + case TOK_SERVER_MAX_WINDOW_BITS: + serverMaxWindowBits = 1; + lastInteger = &serverMaxWindowBits; + break; + case TOK_CLIENT_MAX_WINDOW_BITS: + clientMaxWindowBits = 1; + lastInteger = &clientMaxWindowBits; + break; + default: + if (token < 0 && lastInteger) { + *lastInteger = -token; + } + break; + } + } +} + +template +ExtensionsNegotiator::ExtensionsNegotiator(int wantedOptions) { + options = wantedOptions; +} + +template +std::string ExtensionsNegotiator::generateOffer() { + std::string extensionsOffer; + if (options & Options::PERMESSAGE_DEFLATE) { + extensionsOffer += "permessage-deflate"; + + if (options & Options::CLIENT_NO_CONTEXT_TAKEOVER) { + extensionsOffer += "; client_no_context_takeover"; + } + + // we do not support accepting this yet + // todo: if we agree on this, do not allocate a compressor + // per socket! + + // It is RECOMMENDED that a server supports the + // "server_no_context_takeover" extension parameter in an extension + // negotiation offer. + if (options & Options::SERVER_NO_CONTEXT_TAKEOVER) { + //extensionsOffer += "; server_no_context_takeover"; + } + } + + return extensionsOffer; +} + +template +void ExtensionsNegotiator::readOffer(std::string_view offer) { + if (isServer) { + ExtensionsParser extensionsParser(offer.data(), offer.length()); + if ((options & PERMESSAGE_DEFLATE) && extensionsParser.perMessageDeflate) { + if (extensionsParser.clientNoContextTakeover || (options & CLIENT_NO_CONTEXT_TAKEOVER)) { + options |= CLIENT_NO_CONTEXT_TAKEOVER; + } + + if (extensionsParser.serverNoContextTakeover) { + options |= SERVER_NO_CONTEXT_TAKEOVER; + } else { + options &= ~SERVER_NO_CONTEXT_TAKEOVER; + } + } else { + options &= ~PERMESSAGE_DEFLATE; + } + } else { + // todo! + } +} + +template +int ExtensionsNegotiator::getNegotiatedOptions() { + return options; +} + +} + +#endif // WEBSOCKETEXTENSIONS_H diff --git a/src/WebSocketProtocol.h b/src/WebSocketProtocol.h index b24a61f..a0ef600 100644 --- a/src/WebSocketProtocol.h +++ b/src/WebSocketProtocol.h @@ -370,7 +370,7 @@ public: while (length >= SHORT_MESSAGE_HEADER) { // invalid reserved bits / invalid opcodes / invalid control frames / set compressed frame - if ((rsv1(src) && !Impl::setCompressed(wState)) || rsv23(src) || (getOpCode(src) > 2 && getOpCode(src) < 8) || + if ((rsv1(src) && !Impl::setCompressed(wState, user)) || rsv23(src) || (getOpCode(src) > 2 && getOpCode(src) < 8) || getOpCode(src) > 10 || (getOpCode(src) > 2 && (!isFin(src) || payloadLength(src) > 125))) { Impl::forceClose(wState, user); return;