Merge bitcoin#30509: multiprocess: Add -ipcbind option to bitcoin-node

30073e6 multiprocess: Add -ipcbind option to bitcoin-node (Russell Yanofsky)
73fe7d7 multiprocess: Add unit tests for connect, serve, and listen functions (Ryan Ofsky)
955d407 multiprocess: Add IPC connectAddress and listenAddress methods (Russell Yanofsky)
4da2043 depends: Update libmultiprocess library for CustomMessage function and ThreadContext bugfix (Ryan Ofsky)

Pull request description:

  Add `-ipcbind` option to `bitcoin-node` to make it listen on a unix socket and accept connections from other processes. The default socket path is `<datadir>/node.sock`, but this can be customized.

  This option lets potential wallet, gui, index, and mining processes connect to the node and control it. See examples in bitcoin#19460, bitcoin#19461, and bitcoin#30437.

  Motivation for this PR, in combination with bitcoin#30510, is be able to release a bitcoin core node binary that can generate block templates for a separate Stratum v2 mining service, like the one being implemented in Sjors#48, that connects over IPC.

  Other things to know about this PR:

  - While the `-ipcbind` option lets other processes to connect to the `bitcoin-node` process, the only thing they can actually do after connecting is call methods on the [`Init`](https://github.com/bitcoin/bitcoin/blob/master/src/ipc/capnp/init.capnp#L17-L20) interface which is currently very limited and doesn't do much. But PRs [bitcoin#30510](bitcoin#30510), [bitcoin#29409](bitcoin#29409), and [bitcoin#10102](bitcoin#10102) expand the `Init` interface to expose mining, wallet, and gui functionality respectively.

  - This PR is not needed for [bitcoin#10102](bitcoin#10102), which runs GUI, node, and wallet code in different processes, because [bitcoin#10102](bitcoin#10102) does not use unix sockets or allow outside processes to connect to existing processes. [bitcoin#10102](bitcoin#10102) lets parent and child processes communicate over internal socketpairs, not externally accessible sockets.

  ---

  This PR is part of the [process separation project](bitcoin#28722).

ACKs for top commit:
  achow101:
    ACK 30073e6
  TheCharlatan:
    Re-ACK 30073e6
  itornaza:
    Code review ACK 30073e6

Tree-SHA512: 2b766e60535f57352e8afda9c3748a32acb5a57b2827371b48ba865fa9aa1df00f340732654f2e300c6823dbc6f3e14377fca87e4e959e613fe85a6d2312d9c8

Author: achow101

depends/packages/native_libmultiprocess.mk

@@ -1,8 +1,8 @@
 package=native_libmultiprocess
-$(package)_version=6aca5f389bacf2942394b8738bbe15d6c9edfb9b
+$(package)_version=c1b4ab4eb897d3af09bc9b3cc30e2e6fff87f3e2
 $(package)_download_path=https://github.com/chaincodelabs/libmultiprocess/archive
 $(package)_file_name=$($(package)_version).tar.gz
-$(package)_sha256_hash=2efeed53542bc1d8af3291f2b6f0e5d430d86a5e04e415ce33c136f2c226a51d
+$(package)_sha256_hash=6edf5ad239ca9963c78f7878486fb41411efc9927c6073928a7d6edf947cac4a
 $(package)_dependencies=native_capnp
 
 define $(package)_config_cmds

src/bitcoind.cpp

@@ -109,10 +109,11 @@ int fork_daemon(bool nochdir, bool noclose, TokenPipeEnd& endpoint)
 
 #endif
 
-static bool ParseArgs(ArgsManager& args, int argc, char* argv[])
+static bool ParseArgs(NodeContext& node, int argc, char* argv[])
 {
+    ArgsManager& args{*Assert(node.args)};
     // If Qt is used, parameters/bitcoin.conf are parsed in qt/bitcoin.cpp's main()
-    SetupServerArgs(args);
+    SetupServerArgs(args, node.init->canListenIpc());
     std::string error;
     if (!args.ParseParameters(argc, argv, error)) {
         return InitError(Untranslated(strprintf("Error parsing command line arguments: %s", error)));
@@ -268,7 +269,7 @@ MAIN_FUNCTION
 
     // Interpret command line arguments
     ArgsManager& args = *Assert(node.args);
-    if (!ParseArgs(args, argc, argv)) return EXIT_FAILURE;
+    if (!ParseArgs(node, argc, argv)) return EXIT_FAILURE;
     // Process early info return commands such as -help or -version
     if (ProcessInitCommands(args)) return EXIT_SUCCESS;
 

src/common/args.cpp

@@ -635,6 +635,9 @@ std::string ArgsManager::GetHelpMessage() const
             case OptionsCategory::RPC:
                 usage += HelpMessageGroup("RPC server options:");
                 break;
+            case OptionsCategory::IPC:
+                usage += HelpMessageGroup("IPC interprocess connection options:");
+                break;
             case OptionsCategory::WALLET:
                 usage += HelpMessageGroup("Wallet options:");
                 break;

src/common/args.h

@@ -64,6 +64,7 @@ enum class OptionsCategory {
     COMMANDS,
     REGISTER_COMMANDS,
     CLI_COMMANDS,
+    IPC,
 
     HIDDEN // Always the last option to avoid printing these in the help
 };

src/init.cpp

@@ -29,6 +29,7 @@
 #include <init/common.h>
 #include <interfaces/chain.h>
 #include <interfaces/init.h>
+#include <interfaces/ipc.h>
 #include <interfaces/mining.h>
 #include <interfaces/node.h>
 #include <kernel/context.h>
@@ -441,7 +442,7 @@ static void OnRPCStopped()
     LogDebug(BCLog::RPC, "RPC stopped.\n");
 }
 
-void SetupServerArgs(ArgsManager& argsman)
+void SetupServerArgs(ArgsManager& argsman, bool can_listen_ipc)
 {
     SetupHelpOptions(argsman);
     argsman.AddArg("-help-debug", "Print help message with debugging options and exit", ArgsManager::ALLOW_ANY, OptionsCategory::DEBUG_TEST); // server-only for now
@@ -676,6 +677,9 @@ void SetupServerArgs(ArgsManager& argsman)
     argsman.AddArg("-rpcwhitelistdefault", "Sets default behavior for rpc whitelisting. Unless rpcwhitelistdefault is set to 0, if any -rpcwhitelist is set, the rpc server acts as if all rpc users are subject to empty-unless-otherwise-specified whitelists. If rpcwhitelistdefault is set to 1 and no -rpcwhitelist is set, rpc server acts as if all rpc users are subject to empty whitelists.", ArgsManager::ALLOW_ANY, OptionsCategory::RPC);
     argsman.AddArg("-rpcworkqueue=<n>", strprintf("Set the depth of the work queue to service RPC calls (default: %d)", DEFAULT_HTTP_WORKQUEUE), ArgsManager::ALLOW_ANY | ArgsManager::DEBUG_ONLY, OptionsCategory::RPC);
     argsman.AddArg("-server", "Accept command line and JSON-RPC commands", ArgsManager::ALLOW_ANY, OptionsCategory::RPC);
+    if (can_listen_ipc) {
+        argsman.AddArg("-ipcbind=<address>", "Bind to Unix socket address and listen for incoming connections. Valid address values are \"unix\" to listen on the default path, <datadir>/node.sock, or \"unix:/custom/path\" to specify a custom path. Can be specified multiple times to listen on multiple paths. Default behavior is not to listen on any path. If relative paths are specified, they are interpreted relative to the network data directory. If paths include any parent directory components and the parent directories do not exist, they will be created.", ArgsManager::ALLOW_ANY, OptionsCategory::IPC);
+    }
 
 #if HAVE_DECL_FORK
     argsman.AddArg("-daemon", strprintf("Run in the background as a daemon and accept commands (default: %d)", DEFAULT_DAEMON), ArgsManager::ALLOW_ANY, OptionsCategory::OPTIONS);
@@ -1200,6 +1204,17 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
     g_wallet_init_interface.Construct(node);
     uiInterface.InitWallet();
 
+    if (interfaces::Ipc* ipc = node.init->ipc()) {
+        for (std::string address : gArgs.GetArgs("-ipcbind")) {
+            try {
+                ipc->listenAddress(address);
+            } catch (const std::exception& e) {
+                return InitError(strprintf(Untranslated("Unable to bind to IPC address '%s'. %s"), address, e.what()));
+            }
+            LogPrintf("Listening for IPC requests on address %s\n", address);
+        }
+    }
+
     /* Register RPC commands regardless of -server setting so they will be
      * available in the GUI RPC console even if external calls are disabled.
      */

src/init.h

@@ -74,7 +74,7 @@ bool AppInitMain(node::NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip
 /**
  * Register all arguments with the ArgsManager
  */
-void SetupServerArgs(ArgsManager& argsman);
+void SetupServerArgs(ArgsManager& argsman, bool can_listen_ipc=false);
 
 /** Validates requirements to run the indexes and spawns each index initial sync thread */
 bool StartIndexBackgroundSync(node::NodeContext& node);

src/init/bitcoin-gui.cpp

@@ -34,6 +34,11 @@ class BitcoinGuiInit : public interfaces::Init
     }
     std::unique_ptr<interfaces::Echo> makeEcho() override { return interfaces::MakeEcho(); }
     interfaces::Ipc* ipc() override { return m_ipc.get(); }
+    // bitcoin-gui accepts -ipcbind option even though it does not use it
+    // directly. It just returns true here to accept the option because
+    // bitcoin-node accepts the option, and bitcoin-gui accepts all bitcoin-node
+    // options and will start the node with those options.
+    bool canListenIpc() override { return true; }
     node::NodeContext m_node;
     std::unique_ptr<interfaces::Ipc> m_ipc;
 };

src/init/bitcoin-node.cpp

@@ -37,6 +37,7 @@ class BitcoinNodeInit : public interfaces::Init
     }
     std::unique_ptr<interfaces::Echo> makeEcho() override { return interfaces::MakeEcho(); }
     interfaces::Ipc* ipc() override { return m_ipc.get(); }
+    bool canListenIpc() override { return true; }
     node::NodeContext& m_node;
     std::unique_ptr<interfaces::Ipc> m_ipc;
 };

src/interfaces/init.h

@@ -37,6 +37,7 @@ class Init
     virtual std::unique_ptr<WalletLoader> makeWalletLoader(Chain& chain) { return nullptr; }
     virtual std::unique_ptr<Echo> makeEcho() { return nullptr; }
     virtual Ipc* ipc() { return nullptr; }
+    virtual bool canListenIpc() { return false; }
 };
 
 //! Return implementation of Init interface for the node process. If the argv

src/interfaces/ipc.h

@@ -41,6 +41,11 @@ class Init;
 //!    to make other proxy objects calling other remote interfaces. It can also
 //!    destroy the initial interfaces::Init object to close the connection and
 //!    shut down the spawned process.
+//!
+//! When connecting to an existing process, the steps are similar to spawning a
+//! new process, except a socket is created instead of a socketpair, and
+//! destroying an Init interface doesn't end the process, since there can be
+//! multiple connections.
 class Ipc
 {
 public:
@@ -54,6 +59,17 @@ class Ipc
     //! true. If this is not a spawned child process, return false.
     virtual bool startSpawnedProcess(int argc, char* argv[], int& exit_status) = 0;
 
+    //! Connect to a socket address and make a client interface proxy object
+    //! using provided callback. connectAddress returns an interface pointer if
+    //! the connection was established, returns null if address is empty ("") or
+    //! disabled ("0") or if a connection was refused but not required ("auto"),
+    //! and throws an exception if there was an unexpected error.
+    virtual std::unique_ptr<Init> connectAddress(std::string& address) = 0;
+
+    //! Connect to a socket address and make a client interface proxy object
+    //! using provided callback. Throws an exception if there was an error.
+    virtual void listenAddress(std::string& address) = 0;
+
     //! Add cleanup callback to remote interface that will run when the
     //! interface is deleted.
     template<typename Interface>

src/ipc/capnp/protocol.cpp

@@ -23,6 +23,8 @@
 #include <mutex>
 #include <optional>
 #include <string>
+#include <sys/socket.h>
+#include <system_error>
 #include <thread>
 
 namespace ipc {
@@ -51,11 +53,20 @@ class CapnpProtocol : public Protocol
         startLoop(exe_name);
         return mp::ConnectStream<messages::Init>(*m_loop, fd);
     }
-    void serve(int fd, const char* exe_name, interfaces::Init& init) override
+    void listen(int listen_fd, const char* exe_name, interfaces::Init& init) override
+    {
+        startLoop(exe_name);
+        if (::listen(listen_fd, /*backlog=*/5) != 0) {
+            throw std::system_error(errno, std::system_category());
+        }
+        mp::ListenConnections<messages::Init>(*m_loop, listen_fd, init);
+    }
+    void serve(int fd, const char* exe_name, interfaces::Init& init, const std::function<void()>& ready_fn = {}) override
     {
         assert(!m_loop);
         mp::g_thread_context.thread_name = mp::ThreadName(exe_name);
         m_loop.emplace(exe_name, &IpcLogFn, &m_context);
+        if (ready_fn) ready_fn();
         mp::ServeStream<messages::Init>(*m_loop, fd, init);
         m_loop->loop();
         m_loop.reset();

src/ipc/interfaces.cpp

@@ -2,6 +2,7 @@
 // Distributed under the MIT software license, see the accompanying
 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
+#include <common/args.h>
 #include <common/system.h>
 #include <interfaces/init.h>
 #include <interfaces/ipc.h>
@@ -56,6 +57,35 @@ class IpcImpl : public interfaces::Ipc
         exit_status = EXIT_SUCCESS;
         return true;
     }
+    std::unique_ptr<interfaces::Init> connectAddress(std::string& address) override
+    {
+        if (address.empty() || address == "0") return nullptr;
+        int fd;
+        if (address == "auto") {
+            // Treat "auto" the same as "unix" except don't treat it an as error
+            // if the connection is not accepted. Just return null so the caller
+            // can work offline without a connection, or spawn a new
+            // bitcoin-node process and connect to it.
+            address = "unix";
+            try {
+                fd = m_process->connect(gArgs.GetDataDirNet(), "bitcoin-node", address);
+            } catch (const std::system_error& e) {
+                // If connection type is auto and socket path isn't accepting connections, or doesn't exist, catch the error and return null;
+                if (e.code() == std::errc::connection_refused || e.code() == std::errc::no_such_file_or_directory) {
+                    return nullptr;
+                }
+                throw;
+            }
+        } else {
+            fd = m_process->connect(gArgs.GetDataDirNet(), "bitcoin-node", address);
+        }
+        return m_protocol->connect(fd, m_exe_name);
+    }
+    void listenAddress(std::string& address) override
+    {
+        int fd = m_process->bind(gArgs.GetDataDirNet(), m_exe_name, address);
+        m_protocol->listen(fd, m_exe_name, m_init);
+    }
     void addCleanup(std::type_index type, void* iface, std::function<void()> cleanup) override
     {
         m_protocol->addCleanup(type, iface, std::move(cleanup));

src/ipc/process.cpp

@@ -4,22 +4,28 @@
 
 #include <ipc/process.h>
 #include <ipc/protocol.h>
+#include <logging.h>
 #include <mp/util.h>
 #include <tinyformat.h>
 #include <util/fs.h>
 #include <util/strencodings.h>
+#include <util/syserror.h>
 
 #include <cstdint>
 #include <cstdlib>
+#include <errno.h>
 #include <exception>
 #include <iostream>
 #include <stdexcept>
 #include <string.h>
-#include <system_error>
+#include <sys/socket.h>
+#include <sys/un.h>
 #include <unistd.h>
 #include <utility>
 #include <vector>
 
+using util::RemovePrefixView;
+
 namespace ipc {
 namespace {
 class ProcessImpl : public Process
@@ -54,7 +60,95 @@ class ProcessImpl : public Process
         }
         return true;
     }
+    int connect(const fs::path& data_dir,
+                const std::string& dest_exe_name,
+                std::string& address) override;
+    int bind(const fs::path& data_dir, const std::string& exe_name, std::string& address) override;
 };
+
+static bool ParseAddress(std::string& address,
+                  const fs::path& data_dir,
+                  const std::string& dest_exe_name,
+                  struct sockaddr_un& addr,
+                  std::string& error)
+{
+    if (address.compare(0, 4, "unix") == 0 && (address.size() == 4 || address[4] == ':')) {
+        fs::path path;
+        if (address.size() <= 5) {
+            path = data_dir / fs::PathFromString(strprintf("%s.sock", RemovePrefixView(dest_exe_name, "bitcoin-")));
+        } else {
+            path = data_dir / fs::PathFromString(address.substr(5));
+        }
+        std::string path_str = fs::PathToString(path);
+        address = strprintf("unix:%s", path_str);
+        if (path_str.size() >= sizeof(addr.sun_path)) {
+            error = strprintf("Unix address path %s exceeded maximum socket path length", fs::quoted(fs::PathToString(path)));
+            return false;
+        }
+        memset(&addr, 0, sizeof(addr));
+        addr.sun_family = AF_UNIX;
+        strncpy(addr.sun_path, path_str.c_str(), sizeof(addr.sun_path)-1);
+        return true;
+    }
+
+    error = strprintf("Unrecognized address '%s'", address);
+    return false;
+}
+
+int ProcessImpl::connect(const fs::path& data_dir,
+                         const std::string& dest_exe_name,
+                         std::string& address)
+{
+    struct sockaddr_un addr;
+    std::string error;
+    if (!ParseAddress(address, data_dir, dest_exe_name, addr, error)) {
+        throw std::invalid_argument(error);
+    }
+
+    int fd;
+    if ((fd = ::socket(addr.sun_family, SOCK_STREAM, 0)) == -1) {
+        throw std::system_error(errno, std::system_category());
+    }
+    if (::connect(fd, (struct sockaddr*)&addr, sizeof(addr)) == 0) {
+        return fd;
+    }
+    int connect_error = errno;
+    if (::close(fd) != 0) {
+        LogPrintf("Error closing file descriptor %i '%s': %s\n", fd, address, SysErrorString(errno));
+    }
+    throw std::system_error(connect_error, std::system_category());
+}
+
+int ProcessImpl::bind(const fs::path& data_dir, const std::string& exe_name, std::string& address)
+{
+    struct sockaddr_un addr;
+    std::string error;
+    if (!ParseAddress(address, data_dir, exe_name, addr, error)) {
+        throw std::invalid_argument(error);
+    }
+
+    if (addr.sun_family == AF_UNIX) {
+        fs::path path = addr.sun_path;
+        if (path.has_parent_path()) fs::create_directories(path.parent_path());
+        if (fs::symlink_status(path).type() == fs::file_type::socket) {
+            fs::remove(path);
+        }
+    }
+
+    int fd;
+    if ((fd = ::socket(addr.sun_family, SOCK_STREAM, 0)) == -1) {
+        throw std::system_error(errno, std::system_category());
+    }
+
+    if (::bind(fd, (struct sockaddr*)&addr, sizeof(addr)) == 0) {
+        return fd;
+    }
+    int bind_error = errno;
+    if (::close(fd) != 0) {
+        LogPrintf("Error closing file descriptor %i: %s\n", fd, SysErrorString(errno));
+    }
+    throw std::system_error(bind_error, std::system_category());
+}
 } // namespace
 
 std::unique_ptr<Process> MakeProcess() { return std::make_unique<ProcessImpl>(); }

src/ipc/process.h

@@ -34,6 +34,16 @@ class Process
     //! process. If so, return true and a file descriptor for communicating
     //! with the parent process.
     virtual bool checkSpawned(int argc, char* argv[], int& fd) = 0;
+
+    //! Canonicalize and connect to address, returning socket descriptor.
+    virtual int connect(const fs::path& data_dir,
+                        const std::string& dest_exe_name,
+                        std::string& address) = 0;
+
+    //! Create listening socket, bind and canonicalize address, and return socket descriptor.
+    virtual int bind(const fs::path& data_dir,
+                     const std::string& exe_name,
+                     std::string& address) = 0;
 };
 
 //! Constructor for Process interface. Implementation will vary depending on