Merge #28649: Do the SOCKS5 handshake reliably

af0fca5 netbase: use reliable send() during SOCKS5 handshake (Vasil Dimov)
1b19d11 sock: change Sock::SendComplete() to take Span (Vasil Dimov)

Pull request description:

  The `Socks5()` function which does the SOCKS5 handshake with the SOCKS5 proxy sends bytes to the socket without retrying partial writes.

  `send(2)` may write only part of the provided data and return. In this case the caller is responsible for retrying the operation with the remaining data. Change `Socks5()` to do that. There is already a method `Sock::SendComplete()` which does exactly that, so use it in `Socks5()`.

  A minor complication for this PR is that `Sock::SendComplete()` takes `std::string` argument whereas `Socks5()` has `std::vector<uint8_t>`. Thus the necessity for the first commit. It is possible to do also in other ways - convert the data in `Socks5()` to `std::string` or have just one `Sock::SendComplete()` that takes `void*` and change the callers to pass `str.data(), str.size()` or `vec.data(), vec.size()`.

  This came up while testing #27375.

ACKs for top commit:
  achow101:
    ACK af0fca5
  jonatack:
    ACK af0fca5
  pinheadmz:
    ACK af0fca5

Tree-SHA512: 1d4a53d0628f7607378038ac56dc3b8624ce9322b034c9547a0c3ce052eafb4b18213f258aa3b57bcb4d990a5e0548a37ec70af2bd55f6e8e6399936f1ce047a

src/net.cpp

@@ -3219,7 +3219,6 @@ bool CConnman::Start(CScheduler& scheduler, const Options& connOptions)
     // Start threads
     //
     assert(m_msgproc);
-    InterruptSocks5(false);
     interruptNet.reset();
     flagInterruptMsgProc = false;
 
@@ -3291,7 +3290,7 @@ void CConnman::Interrupt()
     condMsgProc.notify_all();
 
     interruptNet();
-    InterruptSocks5(true);
+    g_socks5_interrupt();
 
     if (semOutbound) {
         for (int i=0; i<m_max_outbound; i++) {

src/netbase.cpp

@@ -30,7 +30,7 @@ bool fNameLookup = DEFAULT_NAME_LOOKUP;
 
 // Need ample time for negotiation for very slow proxies such as Tor
 std::chrono::milliseconds g_socks5_recv_timeout = 20s;
-static std::atomic<bool> interruptSocks5Recv(false);
+CThreadInterrupt g_socks5_interrupt;
 
 ReachableNets g_reachable_nets;
 
@@ -271,7 +271,7 @@ enum class IntrRecvError {
  *          IntrRecvError::OK only if all of the specified number of bytes were
  *          read.
  *
- * @see This function can be interrupted by calling InterruptSocks5(bool).
+ * @see This function can be interrupted by calling g_socks5_interrupt().
  *      Sockets can be made non-blocking with Sock::SetNonBlocking().
  */
 static IntrRecvError InterruptibleRecv(uint8_t* data, size_t len, std::chrono::milliseconds timeout, const Sock& sock)
@@ -299,8 +299,9 @@ static IntrRecvError InterruptibleRecv(uint8_t* data, size_t len, std::chrono::m
                 return IntrRecvError::NetworkError;
             }
         }
-        if (interruptSocks5Recv)
+        if (g_socks5_interrupt) {
             return IntrRecvError::Interrupted;
+        }
         curTime = Now<SteadyMilliseconds>();
     }
     return len == 0 ? IntrRecvError::OK : IntrRecvError::Timeout;
@@ -333,103 +334,93 @@ static std::string Socks5ErrorString(uint8_t err)
 
 bool Socks5(const std::string& strDest, uint16_t port, const ProxyCredentials* auth, const Sock& sock)
 {
-    IntrRecvError recvr;
-    LogPrint(BCLog::NET, "SOCKS5 connecting %s\n", strDest);
-    if (strDest.size() > 255) {
-        return error("Hostname too long");
-    }
-    // Construct the version identifier/method selection message
-    std::vector<uint8_t> vSocks5Init;
-    vSocks5Init.push_back(SOCKSVersion::SOCKS5); // We want the SOCK5 protocol
-    if (auth) {
-        vSocks5Init.push_back(0x02); // 2 method identifiers follow...
-        vSocks5Init.push_back(SOCKS5Method::NOAUTH);
-        vSocks5Init.push_back(SOCKS5Method::USER_PASS);
-    } else {
-        vSocks5Init.push_back(0x01); // 1 method identifier follows...
-        vSocks5Init.push_back(SOCKS5Method::NOAUTH);
-    }
-    ssize_t ret = sock.Send(vSocks5Init.data(), vSocks5Init.size(), MSG_NOSIGNAL);
-    if (ret != (ssize_t)vSocks5Init.size()) {
-        return error("Error sending to proxy");
-    }
-    uint8_t pchRet1[2];
-    if (InterruptibleRecv(pchRet1, 2, g_socks5_recv_timeout, sock) != IntrRecvError::OK) {
-        LogPrintf("Socks5() connect to %s:%d failed: InterruptibleRecv() timeout or other failure\n", strDest, port);
-        return false;
-    }
-    if (pchRet1[0] != SOCKSVersion::SOCKS5) {
-        return error("Proxy failed to initialize");
-    }
-    if (pchRet1[1] == SOCKS5Method::USER_PASS && auth) {
-        // Perform username/password authentication (as described in RFC1929)
-        std::vector<uint8_t> vAuth;
-        vAuth.push_back(0x01); // Current (and only) version of user/pass subnegotiation
-        if (auth->username.size() > 255 || auth->password.size() > 255)
-            return error("Proxy username or password too long");
-        vAuth.push_back(auth->username.size());
-        vAuth.insert(vAuth.end(), auth->username.begin(), auth->username.end());
-        vAuth.push_back(auth->password.size());
-        vAuth.insert(vAuth.end(), auth->password.begin(), auth->password.end());
-        ret = sock.Send(vAuth.data(), vAuth.size(), MSG_NOSIGNAL);
-        if (ret != (ssize_t)vAuth.size()) {
-            return error("Error sending authentication to proxy");
-        }
-        LogPrint(BCLog::PROXY, "SOCKS5 sending proxy authentication %s:%s\n", auth->username, auth->password);
-        uint8_t pchRetA[2];
-        if (InterruptibleRecv(pchRetA, 2, g_socks5_recv_timeout, sock) != IntrRecvError::OK) {
-            return error("Error reading proxy authentication response");
+    try {
+        IntrRecvError recvr;
+        LogPrint(BCLog::NET, "SOCKS5 connecting %s\n", strDest);
+        if (strDest.size() > 255) {
+            return error("Hostname too long");
         }
-        if (pchRetA[0] != 0x01 || pchRetA[1] != 0x00) {
-            return error("Proxy authentication unsuccessful");
+        // Construct the version identifier/method selection message
+        std::vector<uint8_t> vSocks5Init;
+        vSocks5Init.push_back(SOCKSVersion::SOCKS5); // We want the SOCK5 protocol
+        if (auth) {
+            vSocks5Init.push_back(0x02); // 2 method identifiers follow...
+            vSocks5Init.push_back(SOCKS5Method::NOAUTH);
+            vSocks5Init.push_back(SOCKS5Method::USER_PASS);
+        } else {
+            vSocks5Init.push_back(0x01); // 1 method identifier follows...
+            vSocks5Init.push_back(SOCKS5Method::NOAUTH);
         }
-    } else if (pchRet1[1] == SOCKS5Method::NOAUTH) {
-        // Perform no authentication
-    } else {
-        return error("Proxy requested wrong authentication method %02x", pchRet1[1]);
-    }
-    std::vector<uint8_t> vSocks5;
-    vSocks5.push_back(SOCKSVersion::SOCKS5); // VER protocol version
-    vSocks5.push_back(SOCKS5Command::CONNECT); // CMD CONNECT
-    vSocks5.push_back(0x00); // RSV Reserved must be 0
-    vSocks5.push_back(SOCKS5Atyp::DOMAINNAME); // ATYP DOMAINNAME
-    vSocks5.push_back(strDest.size()); // Length<=255 is checked at beginning of function
-    vSocks5.insert(vSocks5.end(), strDest.begin(), strDest.end());
-    vSocks5.push_back((port >> 8) & 0xFF);
-    vSocks5.push_back((port >> 0) & 0xFF);
-    ret = sock.Send(vSocks5.data(), vSocks5.size(), MSG_NOSIGNAL);
-    if (ret != (ssize_t)vSocks5.size()) {
-        return error("Error sending to proxy");
-    }
-    uint8_t pchRet2[4];
-    if ((recvr = InterruptibleRecv(pchRet2, 4, g_socks5_recv_timeout, sock)) != IntrRecvError::OK) {
-        if (recvr == IntrRecvError::Timeout) {
-            /* If a timeout happens here, this effectively means we timed out while connecting
-             * to the remote node. This is very common for Tor, so do not print an
-             * error message. */
+        sock.SendComplete(vSocks5Init, g_socks5_recv_timeout, g_socks5_interrupt);
+        uint8_t pchRet1[2];
+        if (InterruptibleRecv(pchRet1, 2, g_socks5_recv_timeout, sock) != IntrRecvError::OK) {
+            LogPrintf("Socks5() connect to %s:%d failed: InterruptibleRecv() timeout or other failure\n", strDest, port);
             return false;
+        }
+        if (pchRet1[0] != SOCKSVersion::SOCKS5) {
+            return error("Proxy failed to initialize");
+        }
+        if (pchRet1[1] == SOCKS5Method::USER_PASS && auth) {
+            // Perform username/password authentication (as described in RFC1929)
+            std::vector<uint8_t> vAuth;
+            vAuth.push_back(0x01); // Current (and only) version of user/pass subnegotiation
+            if (auth->username.size() > 255 || auth->password.size() > 255)
+                return error("Proxy username or password too long");
+            vAuth.push_back(auth->username.size());
+            vAuth.insert(vAuth.end(), auth->username.begin(), auth->username.end());
+            vAuth.push_back(auth->password.size());
+            vAuth.insert(vAuth.end(), auth->password.begin(), auth->password.end());
+            sock.SendComplete(vAuth, g_socks5_recv_timeout, g_socks5_interrupt);
+            LogPrint(BCLog::PROXY, "SOCKS5 sending proxy authentication %s:%s\n", auth->username, auth->password);
+            uint8_t pchRetA[2];
+            if (InterruptibleRecv(pchRetA, 2, g_socks5_recv_timeout, sock) != IntrRecvError::OK) {
+                return error("Error reading proxy authentication response");
+            }
+            if (pchRetA[0] != 0x01 || pchRetA[1] != 0x00) {
+                return error("Proxy authentication unsuccessful");
+            }
+        } else if (pchRet1[1] == SOCKS5Method::NOAUTH) {
+            // Perform no authentication
         } else {
-            return error("Error while reading proxy response");
+            return error("Proxy requested wrong authentication method %02x", pchRet1[1]);
         }
-    }
-    if (pchRet2[0] != SOCKSVersion::SOCKS5) {
-        return error("Proxy failed to accept request");
-    }
-    if (pchRet2[1] != SOCKS5Reply::SUCCEEDED) {
-        // Failures to connect to a peer that are not proxy errors
-        LogPrintf("Socks5() connect to %s:%d failed: %s\n", strDest, port, Socks5ErrorString(pchRet2[1]));
-        return false;
-    }
-    if (pchRet2[2] != 0x00) { // Reserved field must be 0
-        return error("Error: malformed proxy response");
-    }
-    uint8_t pchRet3[256];
-    switch (pchRet2[3])
-    {
+        std::vector<uint8_t> vSocks5;
+        vSocks5.push_back(SOCKSVersion::SOCKS5);   // VER protocol version
+        vSocks5.push_back(SOCKS5Command::CONNECT); // CMD CONNECT
+        vSocks5.push_back(0x00);                   // RSV Reserved must be 0
+        vSocks5.push_back(SOCKS5Atyp::DOMAINNAME); // ATYP DOMAINNAME
+        vSocks5.push_back(strDest.size());         // Length<=255 is checked at beginning of function
+        vSocks5.insert(vSocks5.end(), strDest.begin(), strDest.end());
+        vSocks5.push_back((port >> 8) & 0xFF);
+        vSocks5.push_back((port >> 0) & 0xFF);
+        sock.SendComplete(vSocks5, g_socks5_recv_timeout, g_socks5_interrupt);
+        uint8_t pchRet2[4];
+        if ((recvr = InterruptibleRecv(pchRet2, 4, g_socks5_recv_timeout, sock)) != IntrRecvError::OK) {
+            if (recvr == IntrRecvError::Timeout) {
+                /* If a timeout happens here, this effectively means we timed out while connecting
+                 * to the remote node. This is very common for Tor, so do not print an
+                 * error message. */
+                return false;
+            } else {
+                return error("Error while reading proxy response");
+            }
+        }
+        if (pchRet2[0] != SOCKSVersion::SOCKS5) {
+            return error("Proxy failed to accept request");
+        }
+        if (pchRet2[1] != SOCKS5Reply::SUCCEEDED) {
+            // Failures to connect to a peer that are not proxy errors
+            LogPrintf("Socks5() connect to %s:%d failed: %s\n", strDest, port, Socks5ErrorString(pchRet2[1]));
+            return false;
+        }
+        if (pchRet2[2] != 0x00) { // Reserved field must be 0
+            return error("Error: malformed proxy response");
+        }
+        uint8_t pchRet3[256];
+        switch (pchRet2[3]) {
         case SOCKS5Atyp::IPV4: recvr = InterruptibleRecv(pchRet3, 4, g_socks5_recv_timeout, sock); break;
         case SOCKS5Atyp::IPV6: recvr = InterruptibleRecv(pchRet3, 16, g_socks5_recv_timeout, sock); break;
-        case SOCKS5Atyp::DOMAINNAME:
-        {
+        case SOCKS5Atyp::DOMAINNAME: {
             recvr = InterruptibleRecv(pchRet3, 1, g_socks5_recv_timeout, sock);
             if (recvr != IntrRecvError::OK) {
                 return error("Error reading from proxy");
@@ -439,15 +430,18 @@ bool Socks5(const std::string& strDest, uint16_t port, const ProxyCredentials* a
             break;
         }
         default: return error("Error: malformed proxy response");
+        }
+        if (recvr != IntrRecvError::OK) {
+            return error("Error reading from proxy");
+        }
+        if (InterruptibleRecv(pchRet3, 2, g_socks5_recv_timeout, sock) != IntrRecvError::OK) {
+            return error("Error reading from proxy");
+        }
+        LogPrint(BCLog::NET, "SOCKS5 connected %s\n", strDest);
+        return true;
+    } catch (const std::runtime_error& e) {
+        return error("Error during SOCKS5 proxy handshake: %s", e.what());
     }
-    if (recvr != IntrRecvError::OK) {
-        return error("Error reading from proxy");
-    }
-    if (InterruptibleRecv(pchRet3, 2, g_socks5_recv_timeout, sock) != IntrRecvError::OK) {
-        return error("Error reading from proxy");
-    }
-    LogPrint(BCLog::NET, "SOCKS5 connected %s\n", strDest);
-    return true;
 }
 
 std::unique_ptr<Sock> CreateSockTCP(const CService& address_family)
@@ -681,11 +675,6 @@ CSubNet LookupSubNet(const std::string& subnet_str)
     return subnet;
 }
 
-void InterruptSocks5(bool interrupt)
-{
-    interruptSocks5Recv = interrupt;
-}
-
 bool IsBadPort(uint16_t port)
 {
     /* Don't forget to update doc/p2p-bad-ports.md if you change this list. */

src/netbase.h

@@ -13,6 +13,7 @@
 #include <netaddress.h>
 #include <serialize.h>
 #include <util/sock.h>
+#include <util/threadinterrupt.h>
 
 #include <functional>
 #include <memory>
@@ -274,7 +275,10 @@ bool ConnectSocketDirectly(const CService &addrConnect, const Sock& sock, int nT
  */
 bool ConnectThroughProxy(const Proxy& proxy, const std::string& strDest, uint16_t port, const Sock& sock, int nTimeout, bool& outProxyConnectionFailed);
 
-void InterruptSocks5(bool interrupt);
+/**
+ * Interrupt SOCKS5 reads or writes.
+ */
+extern CThreadInterrupt g_socks5_interrupt;
 
 /**
  * Connect to a specified destination service through an already connected

src/test/fuzz/socks5.cpp

@@ -32,7 +32,9 @@ FUZZ_TARGET(socks5, .init = initialize_socks5)
     ProxyCredentials proxy_credentials;
     proxy_credentials.username = fuzzed_data_provider.ConsumeRandomLengthString(512);
     proxy_credentials.password = fuzzed_data_provider.ConsumeRandomLengthString(512);
-    InterruptSocks5(fuzzed_data_provider.ConsumeBool());
+    if (fuzzed_data_provider.ConsumeBool()) {
+        g_socks5_interrupt();
+    }
     // Set FUZZED_SOCKET_FAKE_LATENCY=1 to exercise recv timeout code paths. This
     // will slow down fuzzing.
     g_socks5_recv_timeout = (fuzzed_data_provider.ConsumeBool() && std::getenv("FUZZED_SOCKET_FAKE_LATENCY") != nullptr) ? 1ms : default_socks5_recv_timeout;

src/util/sock.cpp

@@ -242,7 +242,7 @@ bool Sock::WaitMany(std::chrono::milliseconds timeout, EventsPerSock& events_per
 #endif /* USE_POLL */
 }
 
-void Sock::SendComplete(const std::string& data,
+void Sock::SendComplete(Span<const unsigned char> data,
                         std::chrono::milliseconds timeout,
                         CThreadInterrupt& interrupt) const
 {
@@ -283,6 +283,13 @@ void Sock::SendComplete(const std::string& data,
     }
 }
 
+void Sock::SendComplete(Span<const char> data,
+                        std::chrono::milliseconds timeout,
+                        CThreadInterrupt& interrupt) const
+{
+    SendComplete(MakeUCharSpan(data), timeout, interrupt);
+}
+
 std::string Sock::RecvUntilTerminator(uint8_t terminator,
                                       std::chrono::milliseconds timeout,
                                       CThreadInterrupt& interrupt,

src/util/sock.h

@@ -228,7 +228,14 @@ class Sock
      * @throws std::runtime_error if the operation cannot be completed. In this case only some of
      * the data will be written to the socket.
      */
-    virtual void SendComplete(const std::string& data,
+    virtual void SendComplete(Span<const unsigned char> data,
+                              std::chrono::milliseconds timeout,
+                              CThreadInterrupt& interrupt) const;
+
+    /**
+     * Convenience method, equivalent to `SendComplete(MakeUCharSpan(data), timeout, interrupt)`.
+     */
+    virtual void SendComplete(Span<const char> data,
                               std::chrono::milliseconds timeout,
                               CThreadInterrupt& interrupt) const;