Merge bitcoin/bitcoin#27679: ZMQ: Support UNIX domain sockets

21d0e6c doc: release notes for PR 27679 (Matthew Zipkin)
791dea2 test: cover unix sockets in zmq interface (Matthew Zipkin)
c87b0a0 zmq: accept unix domain socket address for notifier (Matthew Zipkin)

Pull request description:

  This is a follow-up to bitcoin/bitcoin#27375, allowing ZMQ notifications to be published to a UNIX domain socket.

  Fortunately, libzmq handles unix sockets already, all we really have to do to support it is allow the format in the actual option.

  [libzmq](https://libzmq.readthedocs.io/en/latest/zmq_ipc.html) uses the prefix `ipc://` as opposed to `unix:` which is [used by Tor](https://gitlab.torproject.org/tpo/core/tor/-/blob/main/doc/man/tor.1.txt?ref_type=heads#L1475) and now also by [bitcoind](https://github.com/bitcoin/bitcoin/blob/a85e5a7c9ab75209bc88e49be6991ba0a467034e/doc/release-notes-27375.md?plain=1#L5) so we need to switch that internally.

  As far as I can tell, [LND](https://github.com/lightninglabs/gozmq/blob/d20a764486bf506bc045642e455bc7f0d21b232a/zmq.go#L38) supports `ipc://` and `unix://` (notice the double slashes).

  With this patch, LND can connect to bitcoind using unix sockets:

  Example:

  *bitcoin.conf*:
  ```
  zmqpubrawblock=unix:/tmp/zmqsb
  zmqpubrawtx=unix:/tmp/zmqst
  ```

  *lnd.conf*:
  ```
  bitcoind.zmqpubrawblock=ipc:///tmp/zmqsb
  bitcoind.zmqpubrawtx=ipc:///tmp/zmqst
  ```

ACKs for top commit:
  laanwj:
    Code review ACK 21d0e6c
  tdb3:
    crACK for 21d0e6c.  Changes lgtm. Will follow up with some testing within the next few days as time allows.
  achow101:
    ACK 21d0e6c
  guggero:
    Tested and code review ACK 21d0e6c

Tree-SHA512: ffd50222e80dd029d903e5ddde37b83f72dfec1856a3f7ce49da3b54a45de8daaf80eea1629a30f58559f4b8ded0b29809548c0638cd1c2811b2736ad8b73030

Author: achow101

doc/release-notes-27679.md

@@ -0,0 +1,2 @@
+- unix socket paths are now accepted for `-zmqpubrawblock` and `-zmqpubrawtx` with
+the format `-zmqpubrawtx=unix:/path/to/file`

src/init.cpp

@@ -1301,30 +1301,33 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
         }
     }
 
-    for (const std::string port_option : {
-        "-i2psam",
-        "-onion",
-        "-proxy",
-        "-rpcbind",
-        "-torcontrol",
-        "-whitebind",
-        "-zmqpubhashblock",
-        "-zmqpubhashtx",
-        "-zmqpubrawblock",
-        "-zmqpubrawtx",
-        "-zmqpubsequence",
+    for (const auto &port_option : std::vector<std::pair<std::string, bool>>{
+        // arg name            UNIX socket support
+        {"-i2psam",                 false},
+        {"-onion",                  true},
+        {"-proxy",                  true},
+        {"-rpcbind",                false},
+        {"-torcontrol",             false},
+        {"-whitebind",              false},
+        {"-zmqpubhashblock",        true},
+        {"-zmqpubhashtx",           true},
+        {"-zmqpubrawblock",         true},
+        {"-zmqpubrawtx",            true},
+        {"-zmqpubsequence",         true}
     }) {
-        for (const std::string& socket_addr : args.GetArgs(port_option)) {
+        const std::string arg{port_option.first};
+        const bool unix{port_option.second};
+        for (const std::string& socket_addr : args.GetArgs(arg)) {
             std::string host_out;
             uint16_t port_out{0};
             if (!SplitHostPort(socket_addr, port_out, host_out)) {
 #if HAVE_SOCKADDR_UN
-                // Allow unix domain sockets for -proxy and -onion e.g. unix:/some/file/path
-                if ((port_option != "-proxy" && port_option != "-onion") || socket_addr.find(ADDR_PREFIX_UNIX) != 0) {
-                    return InitError(InvalidPortErrMsg(port_option, socket_addr));
+                // Allow unix domain sockets for some options e.g. unix:/some/file/path
+                if (!unix || socket_addr.find(ADDR_PREFIX_UNIX) != 0) {
+                    return InitError(InvalidPortErrMsg(arg, socket_addr));
                 }
 #else
-                return InitError(InvalidPortErrMsg(port_option, socket_addr));
+                return InitError(InvalidPortErrMsg(arg, socket_addr));
 #endif
             }
         }

src/zmq/zmqnotificationinterface.cpp

@@ -8,6 +8,7 @@
 #include <kernel/chain.h>
 #include <kernel/mempool_entry.h>
 #include <logging.h>
+#include <netbase.h>
 #include <primitives/block.h>
 #include <primitives/transaction.h>
 #include <validationinterface.h>
@@ -57,7 +58,12 @@ std::unique_ptr<CZMQNotificationInterface> CZMQNotificationInterface::Create(std
     {
         std::string arg("-zmq" + entry.first);
         const auto& factory = entry.second;
-        for (const std::string& address : gArgs.GetArgs(arg)) {
+        for (std::string& address : gArgs.GetArgs(arg)) {
+            // libzmq uses prefix "ipc://" for UNIX domain sockets
+            if (address.substr(0, ADDR_PREFIX_UNIX.length()) == ADDR_PREFIX_UNIX) {
+                address.replace(0, ADDR_PREFIX_UNIX.length(), ADDR_PREFIX_IPC);
+            }
+
             std::unique_ptr<CZMQAbstractNotifier> notifier = factory();
             notifier->SetType(entry.first);
             notifier->SetAddress(address);

src/zmq/zmqutil.h

@@ -9,4 +9,7 @@
 
 void zmqError(const std::string& str);
 
+/** Prefix for unix domain socket addresses (which are local filesystem paths) */
+const std::string ADDR_PREFIX_IPC = "ipc://"; // used by libzmq, example "ipc:///root/path/to/file"
+
 #endif // BITCOIN_ZMQ_ZMQUTIL_H

test/functional/interface_zmq.py

@@ -3,7 +3,9 @@
 # Distributed under the MIT software license, see the accompanying
 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
 """Test the ZMQ notification interface."""
+import os
 import struct
+import tempfile
 from time import sleep
 from io import BytesIO
 
@@ -30,7 +32,7 @@
 from test_framework.wallet import (
     MiniWallet,
 )
-from test_framework.netutil import test_ipv6_local
+from test_framework.netutil import test_ipv6_local, test_unix_socket
 
 
 # Test may be skipped and not have zmq installed
@@ -119,6 +121,10 @@ def run_test(self):
         self.ctx = zmq.Context()
         try:
             self.test_basic()
+            if test_unix_socket():
+                self.test_basic(unix=True)
+            else:
+                self.log.info("Skipping ipc test, because UNIX sockets are not supported.")
             self.test_sequence()
             self.test_mempool_sync()
             self.test_reorg()
@@ -139,7 +145,7 @@ def setup_zmq_test(self, services, *, recv_timeout=60, sync_blocks=True, ipv6=Fa
                 socket.setsockopt(zmq.IPV6, 1)
             subscribers.append(ZMQSubscriber(socket, topic.encode()))
 
-        self.restart_node(0, [f"-zmqpub{topic}={address}" for topic, address in services])
+        self.restart_node(0, [f"-zmqpub{topic}={address.replace('ipc://', 'unix:')}" for topic, address in services])
 
         for i, sub in enumerate(subscribers):
             sub.socket.connect(services[i][1])
@@ -176,12 +182,19 @@ def setup_zmq_test(self, services, *, recv_timeout=60, sync_blocks=True, ipv6=Fa
 
         return subscribers
 
-    def test_basic(self):
+    def test_basic(self, unix = False):
+        self.log.info(f"Running basic test with {'ipc' if unix else 'tcp'} protocol")
 
         # Invalid zmq arguments don't take down the node, see #17185.
         self.restart_node(0, ["-zmqpubrawtx=foo", "-zmqpubhashtx=bar"])
 
         address = f"tcp://127.0.0.1:{self.zmq_port_base}"
+
+        if unix:
+            # Use the shortest temp path possible since paths may have as little as 92-char limit
+            socket_path = tempfile.NamedTemporaryFile().name
+            address = f"ipc://{socket_path}"
+
         subs = self.setup_zmq_test([(topic, address) for topic in ["hashblock", "hashtx", "rawblock", "rawtx"]])
 
         hashblock = subs[0]
@@ -247,6 +260,8 @@ def test_basic(self):
         ])
 
         assert_equal(self.nodes[1].getzmqnotifications(), [])
+        if unix:
+            os.unlink(socket_path)
 
     def test_reorg(self):