Protect websockets_ by mutex

Author: andrew-guz

CMakeLists.txt

@@ -57,6 +57,7 @@ option(CROW_RETURNS_OK_ON_HTTP_OPTIONS_REQUEST
 
 option(CROW_ENABLE_SSL "Enable Crow's SSL feature for supporting https" OFF)
 option(CROW_ENABLE_COMPRESSION "Enable Crow's Compression feature for supporting compressed http content" OFF)
+option(ENABLE_TSAN "Enable ThreadSanitizer" OFF)
 
 if(CROW_GENERATE_SBOM OR CROW_BUILD_TESTS)
 	include(${CMAKE_CURRENT_SOURCE_DIR}/cmake/CPM.cmake)
@@ -82,6 +83,17 @@ if(CROW_GENERATE_SBOM)
 	)
 endif()
 
+if(ENABLE_TSAN)
+  message(STATUS "ThreadSanitizer enabled")
+  add_compile_options(
+    -fsanitize=thread
+    -fno-omit-frame-pointer
+    -O1
+    -g
+  )
+  add_link_options(-fsanitize=thread)
+endif()
+
 #####################################
 # Define Targets
 #####################################

include/crow/app.h

@@ -619,6 +619,7 @@ namespace crow
 
         void close_websockets()
         {
+            std::lock_guard<std::mutex> lock{websockets_mutex_};
             for (auto websocket : websockets_)
             {
                 CROW_LOG_INFO << "Quitting Websocket: " << websocket;
@@ -629,11 +630,13 @@ namespace crow
 
         void add_websocket(std::shared_ptr<websocket::connection> conn)
         {
+            std::lock_guard<std::mutex> lock{websockets_mutex_};
             websockets_.push_back(conn);
         }
 
         void remove_websocket(std::shared_ptr<websocket::connection> conn)
         {
+            std::lock_guard<std::mutex> lock{websockets_mutex_};
             websockets_.erase(std::remove(websockets_.begin(), websockets_.end(), conn), websockets_.end());
         }
 
@@ -846,6 +849,7 @@ namespace crow
         bool server_started_{false};
         std::condition_variable cv_started_;
         std::mutex start_mutex_;
+        std::mutex websockets_mutex_; ///< \brief mutex to protect websockets_
         std::vector<std::shared_ptr<websocket::connection>> websockets_;
     };
 

tests/unit_tests/test_websocket.cpp

@@ -1,6 +1,7 @@
 #include "catch2/catch_all.hpp"
 
 #include "crow.h"
+#include <cstddef>
 
 using namespace std;
 using namespace crow;
@@ -649,4 +650,100 @@ TEST_CASE("mirror_websocket_subprotocols", "[websocket]")
     }
 
     app.stop();
-}
+}
+
+TEST_CASE("multithreaded_websockets_open_close", "[websocket]")
+{
+    static std::string http_message = 
+      "GET /ws HTTP/1.1\r\n"
+      "Connection: keep-alive, Upgrade\r\n"
+      "upgrade: websocket\r\n"
+      "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n"
+      "Sec-WebSocket-Version: 13\r\n"
+      "Host: localhost\r\n"
+      "\r\n";
+
+    CROW_LOG_INFO << "Setting up app!\n";
+    SimpleApp app;
+
+    CROW_WEBSOCKET_ROUTE(app, "/ws")
+      .onaccept([&](const crow::request& req, void**) {
+          CROW_LOG_INFO << "Accepted websocket with URL " << req.url;
+          return true;
+      })
+      .onopen([&](websocket::connection&) {
+          CROW_LOG_INFO << "Connected websocket";
+      })
+      .onmessage([&](websocket::connection& conn, const std::string& message, bool) {
+          CROW_LOG_INFO << "Message is \"" << message << '\"';
+          if (message == "quit-default")
+              conn.close();
+          else if (message == "quit-custom")
+              conn.close("custom", crow::websocket::StartStatusCodesForPrivateUse + 10u);
+      })
+      .onclose([&](websocket::connection& conn, const std::string&, uint16_t) {
+          // There should just be one connection
+          CHECK_FALSE(conn.get_remote_ip().empty());
+          CROW_LOG_INFO << "Closing websocket";
+      });
+
+    app.validate();
+
+    CROW_LOG_WARNING << "Starting app!\n";
+    auto _ = app.bindaddr(LOCALHOST_ADDRESS).port(45453).run_async();
+    app.wait_for_server_start();
+    CROW_LOG_WARNING << "App started!\n";
+    asio::io_context ic;
+
+    const auto thread_function = [&]()
+    {
+        asio::ip::tcp::socket c(ic);
+        c.connect(asio::ip::tcp::endpoint(
+        asio::ip::make_address(LOCALHOST_ADDRESS), 45453));
+
+        CROW_LOG_WARNING << "Connected!\n";
+
+        char buf[2048];
+
+        //----------Handshake----------
+        {
+            std::fill_n(buf, 2048, 0);
+            c.send(asio::buffer(http_message));
+
+            c.receive(asio::buffer(buf, 2048));
+            std::this_thread::sleep_for(std::chrono::milliseconds(5));
+        }
+
+        //----------Close websocket----------
+        std::fill_n(buf, 2048, 0);
+        // Close message with, len = 2, status code = 1000
+        char close_message[5]("\x88\x02\x03\xE8");
+        c.send(asio::buffer(close_message, 4));
+        c.receive(asio::buffer(buf, 2048));
+        std::this_thread::sleep_for(std::chrono::milliseconds(5));
+        CHECK((int)(unsigned char)buf[0] == 0x88);
+        CHECK((int)(unsigned char)buf[1] == 0x02);
+        CHECK((int)(unsigned char)buf[2] == 0x03);
+        CHECK((int)(unsigned char)buf[3] == 0xE8);
+    };
+
+    const auto multiple_run_thread_function = [&](const std::size_t count){
+        for(std::size_t i = 0; i < count; ++i) {
+            thread_function();
+        }
+    };
+
+    constexpr std::size_t threads_count = 3;
+    constexpr std::size_t thread_run = 10;
+
+    std::vector<std::thread> threads;
+    for(std::size_t i = 0; i < threads_count; ++i) {    
+        threads.emplace_back(multiple_run_thread_function, thread_run);
+    }
+    for(std::thread& thread : threads) {
+        thread.join();
+    }
+
+    CROW_LOG_WARNING << "Stopping app!\n";
+    app.stop();
+}