Merge #21387: p2p: Refactor sock to add I2P fuzz and unit tests

40316a3 test: add I2P test for a runaway SAM proxy (Vasil Dimov)
2d8ac77 fuzz: add tests for the I2P Session public interface (Vasil Dimov)
9947e44 i2p: use pointers to Sock to accommodate mocking (Vasil Dimov)
82d360b net: change ConnectSocketDirectly() to take a Sock argument (Vasil Dimov)
b586110 net: add connect() and getsockopt() wrappers to Sock (Vasil Dimov)
5a887d4 fuzz: avoid FuzzedSock::Recv() repeated errors with EAGAIN (Vasil Dimov)
3088f83 fuzz: extend FuzzedSock::Recv() to support MSG_PEEK (Vasil Dimov)
9b05c49 fuzz: implement unimplemented FuzzedSock methods (Vasil Dimov)

Pull request description:

  Change the networking code and the I2P code to be fully mockable and use `FuzzedSocket` to fuzz the I2P methods `Listen()`, `Accept()` and `Connect()`.

  Add a mocked `Sock` implementation that returns a predefined data on reads and use it for a regression unit test for the bug fixed in bitcoin/bitcoin#21407.

ACKs for top commit:
  practicalswift:
    Tested ACK 40316a3
  MarcoFalke:
    Concept ACK 40316a3
  jonatack:
    re-ACK 40316a3 reviewed `git range-diff 01bb3af 23c861d 40316a3` and the new unit test commit, debug built, ran unit tests, ran bitcoind with an I2P service and network operation with seven I2P peers (2 in, 5 out) is looking nominal
  laanwj:
    Code review ACK 40316a3

Tree-SHA512: 7fc4f129849e16e0c7e16662d9f4d35dfcc369bb31450ee369a2b97bdca95285533bee7787983e881e5a3d248f912afb42b4a2299d5860ace7129b0b19623cc8

Author: laanwj

src/Makefile.test.include

@@ -90,6 +90,7 @@ BITCOIN_TESTS =\
   test/fs_tests.cpp \
   test/getarg_tests.cpp \
   test/hash_tests.cpp \
+  test/i2p_tests.cpp \
   test/interfaces_tests.cpp \
   test/key_io_tests.cpp \
   test/key_tests.cpp \
@@ -240,6 +241,7 @@ test_fuzz_fuzz_SOURCES = \
  test/fuzz/golomb_rice.cpp \
  test/fuzz/hex.cpp \
  test/fuzz/http_request.cpp \
+ test/fuzz/i2p.cpp \
  test/fuzz/integer.cpp \
  test/fuzz/key.cpp \
  test/fuzz/key_io.cpp \

src/i2p.cpp

@@ -20,6 +20,7 @@
 #include <util/system.h>
 
 #include <chrono>
+#include <memory>
 #include <stdexcept>
 #include <string>
 
@@ -115,7 +116,8 @@ namespace sam {
 Session::Session(const fs::path& private_key_file,
                  const CService& control_host,
                  CThreadInterrupt* interrupt)
-    : m_private_key_file(private_key_file), m_control_host(control_host), m_interrupt(interrupt)
+    : m_private_key_file(private_key_file), m_control_host(control_host), m_interrupt(interrupt),
+      m_control_sock(std::make_unique<Sock>(INVALID_SOCKET))
 {
 }
 
@@ -145,15 +147,15 @@ bool Session::Accept(Connection& conn)
     try {
         while (!*m_interrupt) {
             Sock::Event occurred;
-            conn.sock.Wait(MAX_WAIT_FOR_IO, Sock::RECV, &occurred);
+            conn.sock->Wait(MAX_WAIT_FOR_IO, Sock::RECV, &occurred);
 
             if ((occurred & Sock::RECV) == 0) {
                 // Timeout, no incoming connections within MAX_WAIT_FOR_IO.
                 continue;
             }
 
             const std::string& peer_dest =
-                conn.sock.RecvUntilTerminator('\n', MAX_WAIT_FOR_IO, *m_interrupt, MAX_MSG_SIZE);
+                conn.sock->RecvUntilTerminator('\n', MAX_WAIT_FOR_IO, *m_interrupt, MAX_MSG_SIZE);
 
             conn.peer = CService(DestB64ToAddr(peer_dest), Params().GetDefaultPort());
 
@@ -171,7 +173,7 @@ bool Session::Connect(const CService& to, Connection& conn, bool& proxy_error)
     proxy_error = true;
 
     std::string session_id;
-    Sock sock;
+    std::unique_ptr<Sock> sock;
     conn.peer = to;
 
     try {
@@ -184,12 +186,12 @@ bool Session::Connect(const CService& to, Connection& conn, bool& proxy_error)
         }
 
         const Reply& lookup_reply =
-            SendRequestAndGetReply(sock, strprintf("NAMING LOOKUP NAME=%s", to.ToStringIP()));
+            SendRequestAndGetReply(*sock, strprintf("NAMING LOOKUP NAME=%s", to.ToStringIP()));
 
         const std::string& dest = lookup_reply.Get("VALUE");
 
         const Reply& connect_reply = SendRequestAndGetReply(
-            sock, strprintf("STREAM CONNECT ID=%s DESTINATION=%s SILENT=false", session_id, dest),
+            *sock, strprintf("STREAM CONNECT ID=%s DESTINATION=%s SILENT=false", session_id, dest),
             false);
 
         const std::string& result = connect_reply.Get("RESULT");
@@ -271,29 +273,29 @@ Session::Reply Session::SendRequestAndGetReply(const Sock& sock,
     return reply;
 }
 
-Sock Session::Hello() const
+std::unique_ptr<Sock> Session::Hello() const
 {
     auto sock = CreateSock(m_control_host);
 
     if (!sock) {
         throw std::runtime_error("Cannot create socket");
     }
 
-    if (!ConnectSocketDirectly(m_control_host, sock->Get(), nConnectTimeout, true)) {
+    if (!ConnectSocketDirectly(m_control_host, *sock, nConnectTimeout, true)) {
         throw std::runtime_error(strprintf("Cannot connect to %s", m_control_host.ToString()));
     }
 
     SendRequestAndGetReply(*sock, "HELLO VERSION MIN=3.1 MAX=3.1");
 
-    return std::move(*sock);
+    return sock;
 }
 
 void Session::CheckControlSock()
 {
     LOCK(m_mutex);
 
     std::string errmsg;
-    if (!m_control_sock.IsConnected(errmsg)) {
+    if (!m_control_sock->IsConnected(errmsg)) {
         Log("Control socket error: %s", errmsg);
         Disconnect();
     }
@@ -341,26 +343,26 @@ Binary Session::MyDestination() const
 void Session::CreateIfNotCreatedAlready()
 {
     std::string errmsg;
-    if (m_control_sock.IsConnected(errmsg)) {
+    if (m_control_sock->IsConnected(errmsg)) {
         return;
     }
 
     Log("Creating SAM session with %s", m_control_host.ToString());
 
-    Sock sock = Hello();
+    auto sock = Hello();
 
     const auto& [read_ok, data] = ReadBinaryFile(m_private_key_file);
     if (read_ok) {
         m_private_key.assign(data.begin(), data.end());
     } else {
-        GenerateAndSavePrivateKey(sock);
+        GenerateAndSavePrivateKey(*sock);
     }
 
     const std::string& session_id = GetRandHash().GetHex().substr(0, 10); // full is an overkill, too verbose in the logs
     const std::string& private_key_b64 = SwapBase64(EncodeBase64(m_private_key));
 
-    SendRequestAndGetReply(sock, strprintf("SESSION CREATE STYLE=STREAM ID=%s DESTINATION=%s",
-                                           session_id, private_key_b64));
+    SendRequestAndGetReply(*sock, strprintf("SESSION CREATE STYLE=STREAM ID=%s DESTINATION=%s",
+                                            session_id, private_key_b64));
 
     m_my_addr = CService(DestBinToAddr(MyDestination()), Params().GetDefaultPort());
     m_session_id = session_id;
@@ -370,12 +372,12 @@ void Session::CreateIfNotCreatedAlready()
               m_my_addr.ToString());
 }
 
-Sock Session::StreamAccept()
+std::unique_ptr<Sock> Session::StreamAccept()
 {
-    Sock sock = Hello();
+    auto sock = Hello();
 
     const Reply& reply = SendRequestAndGetReply(
-        sock, strprintf("STREAM ACCEPT ID=%s SILENT=false", m_session_id), false);
+        *sock, strprintf("STREAM ACCEPT ID=%s SILENT=false", m_session_id), false);
 
     const std::string& result = reply.Get("RESULT");
 
@@ -393,14 +395,14 @@ Sock Session::StreamAccept()
 
 void Session::Disconnect()
 {
-    if (m_control_sock.Get() != INVALID_SOCKET) {
+    if (m_control_sock->Get() != INVALID_SOCKET) {
         if (m_session_id.empty()) {
             Log("Destroying incomplete session");
         } else {
             Log("Destroying session %s", m_session_id);
         }
     }
-    m_control_sock.Reset();
+    m_control_sock->Reset();
     m_session_id.clear();
 }
 } // namespace sam

src/i2p.h

@@ -12,6 +12,7 @@
 #include <threadinterrupt.h>
 #include <util/sock.h>
 
+#include <memory>
 #include <optional>
 #include <string>
 #include <unordered_map>
@@ -29,7 +30,7 @@ using Binary = std::vector<uint8_t>;
  */
 struct Connection {
     /** Connected socket. */
-    Sock sock;
+    std::unique_ptr<Sock> sock;
 
     /** Our I2P address. */
     CService me;
@@ -166,7 +167,7 @@ class Session
      * @return a connected socket
      * @throws std::runtime_error if an error occurs
      */
-    Sock Hello() const EXCLUSIVE_LOCKS_REQUIRED(m_mutex);
+    std::unique_ptr<Sock> Hello() const EXCLUSIVE_LOCKS_REQUIRED(m_mutex);
 
     /**
      * Check the control socket for errors and possibly disconnect.
@@ -204,10 +205,11 @@ class Session
 
     /**
      * Open a new connection to the SAM proxy and issue "STREAM ACCEPT" request using the existing
-     * session id. Return the idle socket that is waiting for a peer to connect to us.
+     * session id.
+     * @return the idle socket that is waiting for a peer to connect to us
      * @throws std::runtime_error if an error occurs
      */
-    Sock StreamAccept() EXCLUSIVE_LOCKS_REQUIRED(m_mutex);
+    std::unique_ptr<Sock> StreamAccept() EXCLUSIVE_LOCKS_REQUIRED(m_mutex);
 
     /**
      * Destroy the session, closing the internally used sockets.
@@ -248,7 +250,7 @@ class Session
      * connections and make outgoing ones.
      * See https://geti2p.net/en/docs/api/samv3
      */
-    Sock m_control_sock GUARDED_BY(m_mutex);
+    std::unique_ptr<Sock> m_control_sock GUARDED_BY(m_mutex);
 
     /**
      * Our .b32.i2p address.

src/net.cpp

@@ -432,7 +432,7 @@ CNode* CConnman::ConnectNode(CAddress addrConnect, const char *pszDest, bool fCo
             i2p::Connection conn;
             if (m_i2p_sam_session->Connect(addrConnect, conn, proxyConnectionFailed)) {
                 connected = true;
-                sock = std::make_unique<Sock>(std::move(conn.sock));
+                sock = std::move(conn.sock);
                 addr_bind = CAddress{conn.me, NODE_NONE};
             }
         } else if (GetProxy(addrConnect.GetNetwork(), proxy)) {
@@ -448,7 +448,7 @@ CNode* CConnman::ConnectNode(CAddress addrConnect, const char *pszDest, bool fCo
             if (!sock) {
                 return nullptr;
             }
-            connected = ConnectSocketDirectly(addrConnect, sock->Get(), nConnectTimeout,
+            connected = ConnectSocketDirectly(addrConnect, *sock, nConnectTimeout,
                                               conn_type == ConnectionType::MANUAL);
         }
         if (!proxyConnectionFailed) {
@@ -2253,7 +2253,7 @@ void CConnman::ThreadI2PAcceptIncoming()
             continue;
         }
 
-        CreateNodeFromAcceptedSocket(conn.sock.Release(), NetPermissionFlags::PF_NONE,
+        CreateNodeFromAcceptedSocket(conn.sock->Release(), NetPermissionFlags::PF_NONE,
                                      CAddress{conn.me, NODE_NONE}, CAddress{conn.peer, NODE_NONE});
     }
 }

src/netbase.cpp

@@ -537,12 +537,12 @@ static void LogConnectFailure(bool manual_connection, const char* fmt, const Arg
     }
 }
 
-bool ConnectSocketDirectly(const CService &addrConnect, const SOCKET& hSocket, int nTimeout, bool manual_connection)
+bool ConnectSocketDirectly(const CService &addrConnect, const Sock& sock, int nTimeout, bool manual_connection)
 {
     // Create a sockaddr from the specified service.
     struct sockaddr_storage sockaddr;
     socklen_t len = sizeof(sockaddr);
-    if (hSocket == INVALID_SOCKET) {
+    if (sock.Get() == INVALID_SOCKET) {
         LogPrintf("Cannot connect to %s: invalid socket\n", addrConnect.ToString());
         return false;
     }
@@ -552,55 +552,42 @@ bool ConnectSocketDirectly(const CService &addrConnect, const SOCKET& hSocket, i
     }
 
     // Connect to the addrConnect service on the hSocket socket.
-    if (connect(hSocket, (struct sockaddr*)&sockaddr, len) == SOCKET_ERROR)
-    {
+    if (sock.Connect(reinterpret_cast<struct sockaddr*>(&sockaddr), len) == SOCKET_ERROR) {
         int nErr = WSAGetLastError();
         // WSAEINVAL is here because some legacy version of winsock uses it
         if (nErr == WSAEINPROGRESS || nErr == WSAEWOULDBLOCK || nErr == WSAEINVAL)
         {
             // Connection didn't actually fail, but is being established
             // asynchronously. Thus, use async I/O api (select/poll)
             // synchronously to check for successful connection with a timeout.
-#ifdef USE_POLL
-            struct pollfd pollfd = {};
-            pollfd.fd = hSocket;
-            pollfd.events = POLLIN | POLLOUT;
-            int nRet = poll(&pollfd, 1, nTimeout);
-#else
-            struct timeval timeout = MillisToTimeval(nTimeout);
-            fd_set fdset;
-            FD_ZERO(&fdset);
-            FD_SET(hSocket, &fdset);
-            int nRet = select(hSocket + 1, nullptr, &fdset, nullptr, &timeout);
-#endif
-            // Upon successful completion, both select and poll return the total
-            // number of file descriptors that have been selected. A value of 0
-            // indicates that the call timed out and no file descriptors have
-            // been selected.
-            if (nRet == 0)
-            {
-                LogPrint(BCLog::NET, "connection to %s timeout\n", addrConnect.ToString());
+            const Sock::Event requested = Sock::RECV | Sock::SEND;
+            Sock::Event occurred;
+            if (!sock.Wait(std::chrono::milliseconds{nTimeout}, requested, &occurred)) {
+                LogPrintf("wait for connect to %s failed: %s\n",
+                          addrConnect.ToString(),
+                          NetworkErrorString(WSAGetLastError()));
                 return false;
-            }
-            if (nRet == SOCKET_ERROR)
-            {
-                LogPrintf("select() for %s failed: %s\n", addrConnect.ToString(), NetworkErrorString(WSAGetLastError()));
+            } else if (occurred == 0) {
+                LogPrint(BCLog::NET, "connection attempt to %s timed out\n", addrConnect.ToString());
                 return false;
             }
 
-            // Even if the select/poll was successful, the connect might not
+            // Even if the wait was successful, the connect might not
             // have been successful. The reason for this failure is hidden away
             // in the SO_ERROR for the socket in modern systems. We read it into
-            // nRet here.
-            socklen_t nRetSize = sizeof(nRet);
-            if (getsockopt(hSocket, SOL_SOCKET, SO_ERROR, (sockopt_arg_type)&nRet, &nRetSize) == SOCKET_ERROR)
-            {
+            // sockerr here.
+            int sockerr;
+            socklen_t sockerr_len = sizeof(sockerr);
+            if (sock.GetSockOpt(SOL_SOCKET, SO_ERROR, (sockopt_arg_type)&sockerr, &sockerr_len) ==
+                SOCKET_ERROR) {
                 LogPrintf("getsockopt() for %s failed: %s\n", addrConnect.ToString(), NetworkErrorString(WSAGetLastError()));
                 return false;
             }
-            if (nRet != 0)
-            {
-                LogConnectFailure(manual_connection, "connect() to %s failed after select(): %s", addrConnect.ToString(), NetworkErrorString(nRet));
+            if (sockerr != 0) {
+                LogConnectFailure(manual_connection,
+                                  "connect() to %s failed after wait: %s",
+                                  addrConnect.ToString(),
+                                  NetworkErrorString(sockerr));
                 return false;
             }
         }
@@ -668,7 +655,7 @@ bool IsProxy(const CNetAddr &addr) {
 bool ConnectThroughProxy(const proxyType& proxy, const std::string& strDest, uint16_t port, const Sock& sock, int nTimeout, bool& outProxyConnectionFailed)
 {
     // first connect to proxy server
-    if (!ConnectSocketDirectly(proxy.proxy, sock.Get(), nTimeout, true)) {
+    if (!ConnectSocketDirectly(proxy.proxy, sock, nTimeout, true)) {
         outProxyConnectionFailed = true;
         return false;
     }

src/netbase.h

@@ -194,15 +194,15 @@ extern std::function<std::unique_ptr<Sock>(const CService&)> CreateSock;
  * Try to connect to the specified service on the specified socket.
  *
  * @param addrConnect The service to which to connect.
- * @param hSocket The socket on which to connect.
+ * @param sock The socket on which to connect.
  * @param nTimeout Wait this many milliseconds for the connection to be
  *                 established.
  * @param manual_connection Whether or not the connection was manually requested
  *                          (e.g. through the addnode RPC)
  *
  * @returns Whether or not a connection was successfully made.
  */
-bool ConnectSocketDirectly(const CService &addrConnect, const SOCKET& hSocket, int nTimeout, bool manual_connection);
+bool ConnectSocketDirectly(const CService &addrConnect, const Sock& sock, int nTimeout, bool manual_connection);
 
 /**
  * Connect to a specified destination service through a SOCKS5 proxy by first