Loading src/http/WebSocketProxy.cpp +76 −39 Changes for src/http/WebSocketProxy.cpp: 76 added lines, 39 removed lines. Original line number Diff line number Diff line Loading @@ -151,6 +151,31 @@ void sendAll(netplus::socket &sock, const std::string &data) { } } // Non-blocking counterpart used by the long-lived relay. A TLS send can temporarily need the // fd to become writable while flushing encrypted records; that is not a failed tunnel and must // not discard the WebSocket frame currently being forwarded. void sendAllNonBlocking(netplus::socket &sock, const char *data, size_t size) { size_t sent = 0; netplus::socketwait sw; while (sent < size) { try { netplus::buffer buf(data + sent, size - sent); size_t n = sock.sendData(buf, 0); if (n == 0) { netplus::NetException e; e[netplus::NetException::Error] << "WebSocket relay: peer write failed"; throw e; } sent += n; } catch (const netplus::NetException &e) { if (e.getErrorType() != netplus::NetException::Note || !sw.waitWrite(sock, kWsSocketTimeoutSeconds * 1000)) { throw; } } } } // Reads from sock until "\r\n\r\n" is seen (i.e. the full status line + // headers), capped so a misbehaving upstream can't grow this unbounded. // Returns the raw bytes read, which may include some response body / Loading Loading @@ -205,19 +230,22 @@ void relayLoop(std::unique_ptr<netplus::socket> client, std::string upstreamLeftover, ScopedTunnelSlot slot) { try { // This thread owns both sockets exclusively for its whole lifetime (nothing else will // ever touch client's fd again -- see detachConnection()'s contract), so plain blocking // reads/writes are simplest and correct -- a hardenTunnelSocket()-induced EAGAIN/Note // from a timed-out call is just another way this loop ends, caught below like any other // disconnect. client arrives non-blocking (the event loop always runs non-blocking // sockets); upstream is fresh from connectUpstream() and likely already blocking, but // set explicitly rather than assume. // Keep the handshake handoff blocking so its small, ordered leftovers are completely // flushed before normal duplex traffic begins. client->setBlock(); upstream->setBlock(); if (!upstreamLeftover.empty()) sendAll(*client, upstreamLeftover); if (!clientLeftover.empty()) sendAll(*upstream, clientLeftover); // Relay non-blocking. In particular, ssl::recvData() can consume TLS control records or // partial raw records and report Note without yielding application bytes. With blocking // sockets that can hold the only relay thread inside one direction while the opposite // direction (Home Assistant's auth response) is ready. Non-blocking pumping lets both // directions make progress on every wake. client->setNonBlock(); upstream->setNonBlock(); netplus::socketwait sw; std::vector<netplus::socket *> socks{client.get(), upstream.get()}; std::vector<bool> ready; Loading @@ -234,47 +262,56 @@ void relayLoop(std::unique_ptr<netplus::socket> client, time_t lastActivity = std::time(nullptr); for (;;) { // A TLS socket may already hold one or more decrypted records in its own receive // buffer even though the kernel fd is no longer readable. Waiting only in // socketwait in that state strands those records forever: no new edge has to // arrive, and Home Assistant's tiny auth_ok/auth_invalid frame is a particularly // easy victim. Check transport-owned buffering before polling the fds. ready.assign(socks.size(), false); ready[0] = client->hasBufferedData(); ready[1] = upstream->hasBufferedData(); if (!ready[0] && !ready[1]) { int n = sw.waitReadMulti(socks, ready, /*timeout_ms=*/60000); if (n <= 0) { if (std::time(nullptr) - lastActivity > kWsIdleTimeoutSeconds) break; continue; // timeout -- keep waiting, WebSocket pings keep this alive } } bool progressed = false; bool closed = false; if (ready[0]) { auto pumpOne = [&](netplus::socket &source, netplus::socket &destination) { try { netplus::buffer buf(16384); size_t r = client->recvData(buf, 0); if (r == 0) break; // client closed sendAll(*upstream, std::string(buf.data.buf, r)); lastActivity = std::time(nullptr); // Accepted/server-side netplus::ssl sockets deliberately do not pull raw // TCP bytes from ssl::recvData(): while attached, the event loop owns that // step and feeds TLS through prime_recv(). This WebSocket socket has been // detached from that event loop, so the relay must take over the same raw // read before asking TLS for decrypted application data. For plain TCP the // base implementation is a harmless no-op; for client-side TLS this merely // pre-fills the same buffer recvData() would otherwise fill itself. try { netplus::buffer raw(16384); source.prime_recv(raw, 0); } catch (const netplus::NetException &e) { // TLS recvData() uses Note for a live connection whose readable record was // control traffic (e.g. KeyUpdate/NewSessionTicket) rather than application // data. That is a retry condition, not a tunnel disconnect. if (e.getErrorType() != netplus::NetException::Note) break; if (e.getErrorType() != netplus::NetException::Note) { closed = true; return; } } if (ready[1]) { try { netplus::buffer buf(16384); size_t r = upstream->recvData(buf, 0); if (r == 0) break; // upstream closed sendAll(*client, std::string(buf.data.buf, r)); size_t r = source.recvData(buf, 0); if (r == 0) { closed = true; return; } sendAllNonBlocking(destination, buf.data.buf, r); progressed = true; lastActivity = std::time(nullptr); } catch (const netplus::NetException &e) { if (e.getErrorType() != netplus::NetException::Note) break; // Note means either EAGAIN or a TLS record that produced no application // bytes. Both are normal retry conditions; every other error ends the // tunnel. if (e.getErrorType() != netplus::NetException::Note) closed = true; } }; // Always try both directions. This drains transport-owned TLS buffering even when // the corresponding kernel fd has no remaining readable edge. pumpOne(*client, *upstream); if (!closed) pumpOne(*upstream, *client); if (closed) break; if (progressed) continue; int n = sw.waitReadMulti(socks, ready, /*timeout_ms=*/60000); if (n <= 0) { if (std::time(nullptr) - lastActivity > kWsIdleTimeoutSeconds) break; continue; // timeout -- WebSocket pings normally prevent this path } } } catch (const netplus::NetException &) { Loading Loading
src/http/WebSocketProxy.cpp +76 −39 Changes for src/http/WebSocketProxy.cpp: 76 added lines, 39 removed lines. Original line number Diff line number Diff line Loading @@ -151,6 +151,31 @@ void sendAll(netplus::socket &sock, const std::string &data) { } } // Non-blocking counterpart used by the long-lived relay. A TLS send can temporarily need the // fd to become writable while flushing encrypted records; that is not a failed tunnel and must // not discard the WebSocket frame currently being forwarded. void sendAllNonBlocking(netplus::socket &sock, const char *data, size_t size) { size_t sent = 0; netplus::socketwait sw; while (sent < size) { try { netplus::buffer buf(data + sent, size - sent); size_t n = sock.sendData(buf, 0); if (n == 0) { netplus::NetException e; e[netplus::NetException::Error] << "WebSocket relay: peer write failed"; throw e; } sent += n; } catch (const netplus::NetException &e) { if (e.getErrorType() != netplus::NetException::Note || !sw.waitWrite(sock, kWsSocketTimeoutSeconds * 1000)) { throw; } } } } // Reads from sock until "\r\n\r\n" is seen (i.e. the full status line + // headers), capped so a misbehaving upstream can't grow this unbounded. // Returns the raw bytes read, which may include some response body / Loading Loading @@ -205,19 +230,22 @@ void relayLoop(std::unique_ptr<netplus::socket> client, std::string upstreamLeftover, ScopedTunnelSlot slot) { try { // This thread owns both sockets exclusively for its whole lifetime (nothing else will // ever touch client's fd again -- see detachConnection()'s contract), so plain blocking // reads/writes are simplest and correct -- a hardenTunnelSocket()-induced EAGAIN/Note // from a timed-out call is just another way this loop ends, caught below like any other // disconnect. client arrives non-blocking (the event loop always runs non-blocking // sockets); upstream is fresh from connectUpstream() and likely already blocking, but // set explicitly rather than assume. // Keep the handshake handoff blocking so its small, ordered leftovers are completely // flushed before normal duplex traffic begins. client->setBlock(); upstream->setBlock(); if (!upstreamLeftover.empty()) sendAll(*client, upstreamLeftover); if (!clientLeftover.empty()) sendAll(*upstream, clientLeftover); // Relay non-blocking. In particular, ssl::recvData() can consume TLS control records or // partial raw records and report Note without yielding application bytes. With blocking // sockets that can hold the only relay thread inside one direction while the opposite // direction (Home Assistant's auth response) is ready. Non-blocking pumping lets both // directions make progress on every wake. client->setNonBlock(); upstream->setNonBlock(); netplus::socketwait sw; std::vector<netplus::socket *> socks{client.get(), upstream.get()}; std::vector<bool> ready; Loading @@ -234,47 +262,56 @@ void relayLoop(std::unique_ptr<netplus::socket> client, time_t lastActivity = std::time(nullptr); for (;;) { // A TLS socket may already hold one or more decrypted records in its own receive // buffer even though the kernel fd is no longer readable. Waiting only in // socketwait in that state strands those records forever: no new edge has to // arrive, and Home Assistant's tiny auth_ok/auth_invalid frame is a particularly // easy victim. Check transport-owned buffering before polling the fds. ready.assign(socks.size(), false); ready[0] = client->hasBufferedData(); ready[1] = upstream->hasBufferedData(); if (!ready[0] && !ready[1]) { int n = sw.waitReadMulti(socks, ready, /*timeout_ms=*/60000); if (n <= 0) { if (std::time(nullptr) - lastActivity > kWsIdleTimeoutSeconds) break; continue; // timeout -- keep waiting, WebSocket pings keep this alive } } bool progressed = false; bool closed = false; if (ready[0]) { auto pumpOne = [&](netplus::socket &source, netplus::socket &destination) { try { netplus::buffer buf(16384); size_t r = client->recvData(buf, 0); if (r == 0) break; // client closed sendAll(*upstream, std::string(buf.data.buf, r)); lastActivity = std::time(nullptr); // Accepted/server-side netplus::ssl sockets deliberately do not pull raw // TCP bytes from ssl::recvData(): while attached, the event loop owns that // step and feeds TLS through prime_recv(). This WebSocket socket has been // detached from that event loop, so the relay must take over the same raw // read before asking TLS for decrypted application data. For plain TCP the // base implementation is a harmless no-op; for client-side TLS this merely // pre-fills the same buffer recvData() would otherwise fill itself. try { netplus::buffer raw(16384); source.prime_recv(raw, 0); } catch (const netplus::NetException &e) { // TLS recvData() uses Note for a live connection whose readable record was // control traffic (e.g. KeyUpdate/NewSessionTicket) rather than application // data. That is a retry condition, not a tunnel disconnect. if (e.getErrorType() != netplus::NetException::Note) break; if (e.getErrorType() != netplus::NetException::Note) { closed = true; return; } } if (ready[1]) { try { netplus::buffer buf(16384); size_t r = upstream->recvData(buf, 0); if (r == 0) break; // upstream closed sendAll(*client, std::string(buf.data.buf, r)); size_t r = source.recvData(buf, 0); if (r == 0) { closed = true; return; } sendAllNonBlocking(destination, buf.data.buf, r); progressed = true; lastActivity = std::time(nullptr); } catch (const netplus::NetException &e) { if (e.getErrorType() != netplus::NetException::Note) break; // Note means either EAGAIN or a TLS record that produced no application // bytes. Both are normal retry conditions; every other error ends the // tunnel. if (e.getErrorType() != netplus::NetException::Note) closed = true; } }; // Always try both directions. This drains transport-owned TLS buffering even when // the corresponding kernel fd has no remaining readable edge. pumpOne(*client, *upstream); if (!closed) pumpOne(*upstream, *client); if (closed) break; if (progressed) continue; int n = sw.waitReadMulti(socks, ready, /*timeout_ms=*/60000); if (n <= 0) { if (std::time(nullptr) - lastActivity > kWsIdleTimeoutSeconds) break; continue; // timeout -- WebSocket pings normally prevent this path } } } catch (const netplus::NetException &) { Loading