websocket: fix PMD end-of-message tail partial consumption (permessage-deflate)

example/CMakeLists.txt

@@ -12,3 +12,6 @@ add_subdirectory (http)
 add_subdirectory (websocket)
 
 add_subdirectory (echo-op)
+
+# Repro tools for permessage-deflate receive path
+add_subdirectory (repro)

example/repro/CMakeLists.txt

@@ -0,0 +1,90 @@
+cmake_minimum_required(VERSION 3.16)
+project(beast_pmd_repro LANGUAGES CXX)
+set(CMAKE_CXX_STANDARD 17)
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
+
+# 这两个任选其一传入：
+# -DBOOST_SRC=<boost 超级仓库根目录>（已执行 b2 headers / 可选 b2 stage）
+# 或
+# -DBOOST_PREFIX=<安装前缀>
+set(BOOST_SRC "" CACHE PATH "Boost superproject root (with ./boost after `b2 headers`)")
+set(BOOST_PREFIX "" CACHE PATH "Boost install prefix")
+
+# 解析 include/lib 路径
+unset(BOOST_INC)
+unset(BOOST_LIB)
+
+if(BOOST_SRC)
+  # 源码树模式
+  set(BOOST_INC "${BOOST_SRC}")                 # 这里直接包含 $BOOST_SRC/boost/...
+  set(BOOST_LIB "${BOOST_SRC}/stage/lib")       # 库来自 stage/lib
+  if(NOT EXISTS "${BOOST_INC}/boost/beast/core.hpp")
+    message(FATAL_ERROR
+      "Boost headers not found at ${BOOST_INC}/boost/...; run `b2 headers` in ${BOOST_SRC}")
+  endif()
+elseif(BOOST_PREFIX)
+  # 安装前缀模式
+  set(BOOST_INC "${BOOST_PREFIX}/include")
+  set(BOOST_LIB "${BOOST_PREFIX}/lib")
+  if(NOT EXISTS "${BOOST_INC}/boost/beast/core.hpp")
+    message(FATAL_ERROR
+      "Boost headers not found at ${BOOST_INC}/boost/...; reinstall headers to the prefix")
+  endif()
+else()
+  # 自动尝试：优先用环境变量 BOOST_SRC，其次 $HOME/opt/boost-dev
+  if(DEFINED ENV{BOOST_SRC} AND EXISTS "$ENV{BOOST_SRC}/boost/beast/core.hpp")
+    set(BOOST_INC "$ENV{BOOST_SRC}")
+    set(BOOST_LIB "$ENV{BOOST_SRC}/stage/lib")
+  elseif(DEFINED ENV{HOME} AND EXISTS "$ENV{HOME}/opt/boost-dev/include/boost/beast/core.hpp")
+    set(BOOST_INC "$ENV{HOME}/opt/boost-dev/include")
+    set(BOOST_LIB "$ENV{HOME}/opt/boost-dev/lib")
+  else()
+    message(FATAL_ERROR
+      "Set -DBOOST_SRC=<boost superproject root> (after `b2 headers`) "
+      "or -DBOOST_PREFIX=<install prefix>.")
+  endif()
+endif()
+
+message(STATUS "Using BOOST_INC=${BOOST_INC}")
+message(STATUS "Using BOOST_LIB=${BOOST_LIB}")
+
+find_package(Threads REQUIRED)
+find_package(OpenSSL QUIET)
+
+# ---- 定义接口库，集中注入 include/lib/link 规则 ----
+add_library(lib-asio INTERFACE)
+target_include_directories(lib-asio INTERFACE "${BOOST_INC}")
+target_link_directories(lib-asio INTERFACE "${BOOST_LIB}")
+
+# Boost.System（Beast/Asio 需要）
+find_library(BOOST_SYSTEM_LIB NAMES boost_system PATHS "${BOOST_LIB}" NO_DEFAULT_PATH)
+if(BOOST_SYSTEM_LIB)
+  target_link_libraries(lib-asio INTERFACE "${BOOST_SYSTEM_LIB}")
+else()
+  # 若上面没找到，依赖 -L${BOOST_LIB} 后用普通名解析
+  target_link_libraries(lib-asio INTERFACE boost_system)
+endif()
+target_link_libraries(lib-asio INTERFACE Threads::Threads)
+if(OpenSSL_FOUND)
+  target_link_libraries(lib-asio INTERFACE OpenSSL::SSL OpenSSL::Crypto)
+endif()
+
+# Beast 是 header-only，复用 asio 的设置
+add_library(lib-beast INTERFACE)
+target_link_libraries(lib-beast INTERFACE lib-asio)
+
+# 让可执行文件运行时能直接找到 stage/lib 或 prefix/lib
+add_link_options("-Wl,-rpath,${BOOST_LIB}")
+
+# ---- 你的四个可执行文件，保持原样 ----
+add_executable (beast_pmd_server server_smallbuf.cpp)
+target_link_libraries(beast_pmd_server PRIVATE lib-beast lib-asio)
+
+add_executable (beast_pmd_client_raw client_raw_replay.cpp)
+target_link_libraries(beast_pmd_client_raw PRIVATE lib-asio)
+
+add_executable (beast_pmd_server_writer server_pmd_writer.cpp)
+target_link_libraries(beast_pmd_server_writer PRIVATE lib-beast lib-asio)
+
+add_executable (beast_pmd_client_smallbuf client_smallbuf.cpp)
+target_link_libraries(beast_pmd_client_smallbuf PRIVATE lib-beast lib-asio)

example/repro/README.md

@@ -0,0 +1,42 @@
+# PMD Receive Path Repro (Client-Side)
+
+This pair of tools reproduces a client-side decompression fault when using `permessage-deflate` (with context takeover) by stressing the WebSocket receive path with a 1-byte read buffer.
+
+## Build
+
+```bash
+cmake -S . -B build -DBeast_BUILD_EXAMPLES=ON
+cmake --build build -j
+```
+
+## Run (One-Click Repro)
+
+1) Start the server that negotiates PMD and sends many messages:
+
+```bash
+./build/example/repro/beast_pmd_server_writer 0.0.0.0 8080 200 256
+```
+
+2) Start the client that enables PMD and reads with a tiny buffer (default 1 byte):
+
+```bash
+./build/example/repro/beast_pmd_client_smallbuf 127.0.0.1 8080 /
+```
+
+You should soon observe `read_some ec=...` on the client side (e.g., inflate invalid data), along with a summary line reporting counts of read calls, total bytes, and completed messages.
+
+## Notes
+
+- Both peers enable `permessage-deflate` without `*_no_context_takeover`, window bits 15.
+- The server sends many large, compressible messages to exercise cross-message context.
+- The client constrains `avail_out` by reading 1 byte at a time to stress the end-of-message inflate phase.
+- If it doesn’t reproduce quickly, increase message count/size on the server (e.g., `1000 512`).
+
+## Optional
+
+- The client accepts an optional `buf_bytes` argument to adjust read buffer (default 1):
+
+```bash
+./beast_pmd_client_smallbuf <host> <port> [target] [buf_bytes]
+```
+

example/repro/client_raw_replay.cpp

@@ -0,0 +1,110 @@
+// TCP raw replay client: connects to a WebSocket server and writes bytes
+// from a file to the socket, optionally in fixed-size chunks with delay.
+// Intended to replay application-layer bytes extracted from a plaintext
+// WebSocket PCAP (client->server direction), including the HTTP upgrade
+// and subsequent frames.
+
+#include <boost/asio.hpp>
+#include <fstream>
+#include <iostream>
+#include <thread>
+
+namespace net = boost::asio;
+using tcp = net::ip::tcp;
+
+int main(int argc, char** argv)
+{
+    if(argc < 4)
+    {
+        std::cerr << "Usage: beast_pmd_client_raw <host> <port> <binfile> [chunk_bytes] [delay_ms]\n";
+        std::cerr << "  binfile: binary blob of client->server TCP payload (concatenated)\n";
+        std::cerr << "  chunk_bytes: send in fixed-size chunks (default: all at once)\n";
+        std::cerr << "  delay_ms: sleep between chunks (default: 0)\n";
+        return 64;
+    }
+
+    std::string host = argv[1];
+    std::string port = argv[2];
+    std::string path = argv[3];
+    std::size_t chunk = (argc > 4) ? static_cast<std::size_t>(std::stoul(argv[4])) : 0;
+    unsigned delay_ms = (argc > 5) ? static_cast<unsigned>(std::stoul(argv[5])) : 0;
+
+    try
+    {
+        // Load file
+        std::ifstream ifs(path, std::ios::binary);
+        if(!ifs)
+        {
+            std::cerr << "Cannot open file: " << path << std::endl;
+            return 66;
+        }
+        std::vector<unsigned char> data((std::istreambuf_iterator<char>(ifs)), {});
+        std::cout << "Loaded " << data.size() << " bytes from " << path << std::endl;
+        if(data.empty())
+        {
+            std::cerr << "File is empty." << std::endl;
+            return 65;
+        }
+
+        net::io_context ioc;
+        tcp::resolver res{ioc};
+        tcp::socket sock{ioc};
+
+        auto const results = res.resolve(host, port);
+        net::connect(sock, results);
+        sock.set_option(tcp::no_delay(true));
+        std::cout << "Connected to " << host << ":" << port << std::endl;
+
+        // Reader thread to drain server responses to avoid backpressure
+        std::atomic<bool> stop{false};
+        std::thread reader([&]
+        {
+            std::array<unsigned char, 4096> buf{};
+            boost::system::error_code ec;
+            while(!stop.load())
+            {
+                auto n = sock.read_some(net::buffer(buf), ec);
+                if(ec)
+                    break;
+                (void)n; // discard
+            }
+        });
+
+        // Write data
+        boost::system::error_code ec;
+        if(chunk == 0)
+        {
+            net::write(sock, net::buffer(data), ec);
+            if(ec) std::cerr << "write error: " << ec.message() << std::endl;
+        }
+        else
+        {
+            std::size_t off = 0;
+            while(off < data.size())
+            {
+                auto n = std::min(chunk, data.size() - off);
+                auto sent = net::write(sock, net::buffer(&data[off], n), ec);
+                if(ec)
+                {
+                    std::cerr << "write error: " << ec.message() << std::endl;
+                    break;
+                }
+                off += sent;
+                if(delay_ms) std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));
+            }
+        }
+
+        stop = true;
+        sock.shutdown(tcp::socket::shutdown_both, ec);
+        sock.close(ec);
+        if(reader.joinable()) reader.join();
+    }
+    catch(std::exception const& e)
+    {
+        std::cerr << "FATAL: " << e.what() << std::endl;
+        return 1;
+    }
+
+    return 0;
+}
+

example/repro/client_smallbuf.cpp

@@ -0,0 +1,102 @@
+// Beast WebSocket client that enables permessage-deflate and reads using a
+// very small buffer (1 byte) to stress the receive/decompression path.
+
+#include <boost/beast/core.hpp>
+#include <boost/beast/websocket.hpp>
+#include <boost/asio.hpp>
+#include <iostream>
+
+namespace net = boost::asio;
+namespace beast = boost::beast;
+namespace websocket = beast::websocket;
+using tcp = net::ip::tcp;
+
+int main(int argc, char** argv)
+{
+    if(argc < 3)
+    {
+        std::cerr << "Usage: beast_pmd_client_smallbuf <host> <port> [target] [buf_bytes]\n";
+        return 64;
+    }
+    std::string host = argv[1];
+    std::string port = argv[2];
+    std::string target = argc > 3 ? argv[3] : "/";
+    std::size_t buf_bytes = argc > 4 ? static_cast<std::size_t>(std::stoul(argv[4])) : 1;
+
+    try
+    {
+        net::io_context ioc;
+        tcp::resolver res{ioc};
+        tcp::socket sock{ioc};
+        auto results = res.resolve(host, port);
+        net::connect(sock, results);
+        sock.set_option(tcp::no_delay(true));
+
+        websocket::stream<tcp::socket> ws{std::move(sock)};
+
+        // Enable PMD in client role (offer extension), with context takeover
+        websocket::permessage_deflate pmd;
+        pmd.client_enable = true;          // offer PMD from client
+        pmd.server_enable = false;         // irrelevant on client
+        pmd.server_no_context_takeover = false;
+        pmd.client_no_context_takeover = false;
+        pmd.server_max_window_bits = 15;
+        pmd.client_max_window_bits = 15;
+        pmd.msg_size_threshold = 0;
+        ws.set_option(pmd);
+
+        // Upgrade
+        ws.handshake(host, target);
+        std::cout << "Handshake complete. PMD offered:"
+                  << " client_enable=1 server_enable=0"
+                  << " client_no_context_takeover=0 server_no_context_takeover=0"
+                  << " win_bits(client/server)=15/15\n";
+        std::cout << "Reading with tiny buffer of " << buf_bytes << " bytes..." << std::endl;
+
+        // Log control frames
+        ws.control_callback([](websocket::frame_type ft, beast::string_view sv){
+            char const* name = ft==websocket::frame_type::ping?"PING":(ft==websocket::frame_type::pong?"PONG":"CLOSE");
+            std::cout << "[ctrl] " << name << " len=" << sv.size() << std::endl;
+        });
+
+        // Small buffer to constrain avail_out (1 byte)
+        std::vector<unsigned char> tiny(buf_bytes?buf_bytes:1);
+        std::size_t total_bytes = 0;
+        std::size_t read_calls = 0;
+        std::size_t messages_done = 0;
+        for(;;)
+        {
+            beast::error_code ec;
+            auto n = ws.read_some(net::buffer(tiny.data(), tiny.size()), ec);
+            if(ec)
+            {
+                std::cout << "read_some ec=" << ec.message() << " (after "
+                          << read_calls << " calls, " << total_bytes << " bytes, messages_done="
+                          << messages_done << ")" << std::endl;
+                break;
+            }
+            ++read_calls;
+            total_bytes += n;
+            if(ws.is_message_done())
+            {
+                ++messages_done;
+                std::cout << "[message done] count=" << messages_done
+                          << " total_bytes=" << total_bytes
+                          << " type=" << (ws.got_binary()?"binary":"text")
+                          << std::endl;
+            }
+        }
+
+        beast::error_code ec;
+        ws.close(websocket::close_code::normal, ec);
+        std::cout << "Summary: read_calls=" << read_calls
+                  << " total_bytes=" << total_bytes
+                  << " messages_done=" << messages_done << std::endl;
+    }
+    catch(std::exception const& e)
+    {
+        std::cerr << "FATAL: " << e.what() << std::endl;
+        return 1;
+    }
+    return 0;
+}